notify.py 5.41 KB
Newer Older
Sven Panne's avatar
Sven Panne committed
1
#!/usr/bin/env python3
2
3
4
5
# -*- coding: utf-8 -*-
# Copyright (C) 2019 tribe29 GmbH - License: GNU General Public License v2
# This file is part of Checkmk (https://checkmk.com). It is subject to the terms and
# conditions defined in the file COPYING, which is part of this source code package.
6

7
8
9
import os
import subprocess
from logging import Logger
10
from typing import List, NewType, Optional, TypedDict
11

12
import cmk.utils.defines
13
14
from cmk.utils.exceptions import MKGeneralException
from cmk.utils.i18n import _
15

16
17
18
19
# NOTE: Keep in sync with values in MonitoringLog.cc.
MAX_COMMENT_LENGTH = 2000
MAX_PLUGIN_OUTPUT_LENGTH = 1000

20
21
22
23
# 0 -> OK
# 1 -> temporary issue
# 2 -> permanent issue
NotificationResultCode = NewType("NotificationResultCode", int)
24
NotificationPluginName = NewType("NotificationPluginName", str)
25
NotificationContext = NewType("NotificationContext", dict[str, str])
26

27

28
29
30
31
32
33
34
35
class NotificationResult(TypedDict, total=False):
    plugin: NotificationPluginName
    status: NotificationResultCode
    output: List[str]
    forward: bool
    context: NotificationContext


Sven Panne's avatar
Sven Panne committed
36
def _state_for(exit_code: NotificationResultCode) -> str:
37
    return cmk.utils.defines.service_state_name(exit_code, "UNKNOWN")
38
39


Sven Panne's avatar
Sven Panne committed
40
def find_wato_folder(context: NotificationContext) -> str:
41
42
43
    for tag in context.get("HOSTTAGS", "").split():
        if tag.startswith("/wato/"):
            return tag[6:].rstrip("/")
44
    return ""
45
46


Sven Panne's avatar
Sven Panne committed
47
def notification_message(plugin: NotificationPluginName, context: NotificationContext) -> str:
48
49
50
51
52
53
54
55
56
57
58
59
60
    contact = context["CONTACTNAME"]
    hostname = context["HOSTNAME"]
    service = context.get("SERVICEDESC")
    if service:
        what = "SERVICE NOTIFICATION"
        spec = "%s;%s" % (hostname, service)
        state = context["SERVICESTATE"]
        output = context["SERVICEOUTPUT"]
    else:
        what = "HOST NOTIFICATION"
        spec = hostname
        state = context["HOSTSTATE"]
        output = context["HOSTOUTPUT"]
61
62
63
64
65
66
67
68
69
    # NOTE: There are actually 3 more additional fields, which we don't use: author, comment and long plugin output.
    return "%s: %s;%s;%s;%s;%s" % (
        what,
        contact,
        spec,
        state,
        plugin,
        output[:MAX_PLUGIN_OUTPUT_LENGTH],
    )
70
71


72
73
74
75
76
77
def notification_progress_message(
    plugin: NotificationPluginName,
    context: NotificationContext,
    exit_code: NotificationResultCode,
    output: str,
) -> str:
78
79
80
81
82
83
84
85
86
87
    contact = context["CONTACTNAME"]
    hostname = context["HOSTNAME"]
    service = context.get("SERVICEDESC")
    if service:
        what = "SERVICE NOTIFICATION PROGRESS"
        spec = "%s;%s" % (hostname, service)
    else:
        what = "HOST NOTIFICATION PROGRESS"
        spec = hostname
    state = _state_for(exit_code)
88
89
90
91
92
93
94
95
    return "%s: %s;%s;%s;%s;%s" % (
        what,
        contact,
        spec,
        state,
        plugin,
        output[:MAX_PLUGIN_OUTPUT_LENGTH],
    )
96
97


98
99
100
101
def notification_result_message(
    plugin: NotificationPluginName,
    context: NotificationContext,
    exit_code: NotificationResultCode,
102
    output: list[str],
103
) -> str:
104
105
106
107
108
109
110
111
112
113
114
    contact = context["CONTACTNAME"]
    hostname = context["HOSTNAME"]
    service = context.get("SERVICEDESC")
    if service:
        what = "SERVICE NOTIFICATION RESULT"
        spec = "%s;%s" % (hostname, service)
    else:
        what = "HOST NOTIFICATION RESULT"
        spec = hostname
    state = _state_for(exit_code)
    comment = " -- ".join(output)
115
    short_output = output[-1] if output else ""
116
117
118
119
120
121
122
123
124
    return "%s: %s;%s;%s;%s;%s;%s" % (
        what,
        contact,
        spec,
        state,
        plugin,
        short_output[:MAX_PLUGIN_OUTPUT_LENGTH],
        comment[:MAX_COMMENT_LENGTH],
    )
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169


def ensure_utf8(logger: Optional[Logger] = None) -> None:
    # Make sure that mail(x) is using UTF-8. Otherwise we cannot send notifications
    # with non-ASCII characters. Unfortunately we do not know whether C.UTF-8 is
    # available. If e.g. mail detects a non-Ascii character in the mail body and
    # the specified encoding is not available, it will silently not send the mail!
    # Our resultion in future: use /usr/sbin/sendmail directly.
    # Our resultion in the present: look with locale -a for an existing UTF encoding
    # and use that.
    proc: subprocess.Popen = subprocess.Popen(  # pylint:disable=consider-using-with
        ["locale", "-a"],
        close_fds=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.DEVNULL,
    )
    locales_list: List[str] = []
    std_out: bytes = proc.communicate()[0]
    exit_code: int = proc.returncode
    error_msg: str = _("Command 'locale -a' could not be executed. Exit code of command was")
    not_found_msg: str = _(
        "No UTF-8 encoding found in your locale -a! " "Please install appropriate locales."
    )
    if exit_code != 0:
        if not logger:
            raise MKGeneralException("%s: %r. %s" % (error_msg, exit_code, not_found_msg))
        logger.info("%s: %r" % (error_msg, exit_code))
        logger.info(not_found_msg)
        return

    locales_list = std_out.decode("utf-8", "ignore").split("\n")
    for encoding in locales_list:
        el: str = encoding.lower()
        if "utf8" in el or "utf-8" in el or "utf.8" in el:
            encoding = encoding.strip()
            os.putenv("LANG", encoding)
            if logger:
                logger.debug("Setting locale for mail to %s.", encoding)
            break
    else:
        if not logger:
            raise MKGeneralException(not_found_msg)
        logger.info(not_found_msg)

    return