/
sagbescheid.py
162 lines (138 loc) · 5.65 KB
/
sagbescheid.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
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
#!/usr/bin/env python2
# coding: utf-8
# Copyright © 2015, 2017, Wieland Hoffmann
# License: MIT, see LICENSE for details
import argparse
import logging
from .argparse_ext import TestAction
from .notifier import get_all_notifiers, get_enabled_notifiers, NotifierRegistry
from .unit import get_all_unit_paths, Unit, UNIT_IFACE
from functools import partial
from operator import attrgetter
from sys import exit
from twisted.internet import defer, reactor
from twisted.python import log
from txdbus import client, error
try:
from systemd.daemon import booted, notify
def systemd_ready():
"""Signal to systemd that the service has successfully started.
"""
notify("READY=1")
def systemd_status(message):
"""Send a status `message` to systemd.
:type message: str
"""
notify("STATUS={message}".format(message=message))
except ImportError:
# The systemd-python package can't be installed successfully on
# ReadTheDocs. Just ignore its functions during doc building.
def systemd_ready():
"""Signal to systemd that the service has successfully started.
"""
pass
def systemd_status(message):
"""Send a status `message` to systemd.
:type message: str
"""
pass
@defer.inlineCallbacks
def setup(args):
"""
:type args: :class:`argparse.Namespace`
"""
con = yield client.connect(reactor, "system")
registry = NotifierRegistry(get_enabled_notifiers(args.notifier))
try:
if args.all_units:
# Get the names of all units
units = yield get_all_unit_paths(con)
for unit in units:
yield Unit.from_child_object_path(unit, registry).connect(con)
systemd_ready()
systemd_status("Monitoring {} units.".format(len(units)))
else:
for unit in args.unit:
Unit.from_unit_filename(unit, registry).connect(con)
systemd_ready()
systemd_status("Monitoring {}.".format(args.unit))
except error.DBusException:
logging.exception(
"The following exception occured during the initial setup:")
reactor.stop()
def test(args):
registry = NotifierRegistry(get_enabled_notifiers(args.notifier))
units = []
for unit in args.unit:
units.append(Unit.from_unit_filename(unit, registry))
def emit_signals():
for unit in units:
# Set the initial state. We can enter every state because the
# default state of a unit is unknown.
initial_state_meth = getattr(
unit, "become_{}".format(args.test_state_from))
initial_state_meth()
change = {"ActiveState": args.test_state_to}
unit.onSignal(UNIT_IFACE, change, None)
reactor.callLater(2, reactor.stop)
reactor.callLater(2, emit_signals)
def build_arg_parser():
parser = argparse.ArgumentParser(prog='sagbescheid',
fromfile_prefix_chars='@',
description='Monitor systemd unit states',
formatter_class=argparse.ArgumentDefaultsHelpFormatter) # noqa
available_notifiers = list(get_all_notifiers())
available_notifier_names = map(attrgetter("name"),
available_notifiers)
parser.add_argument("--notifier", action="append", default=[],
choices=available_notifier_names,
help="A notifier to enable.")
parser.add_argument("-v", "--verbose", action="store_true", default=False,
help="Be more verbose.")
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument("--unit", action="append",
help="A unit to monitor.")
group.add_argument("--all-units", action="store_true", default=False,
help="Monitor all units.")
for notifier in available_notifiers:
arg_group = parser.add_argument_group(notifier.name,
"Arguments for the %s notifier" %
notifier.name)
notifier.add_arguments(arg_group)
test_group = parser.add_argument_group("Test notifications", """Send a test
notification for enabled units""")
test_group.add_argument("--test",
action=TestAction,
default=False,
nargs=0,
help="Enable testing mode")
test_group.add_argument("--test-state-from", action="store",
# choices=State.__members__.keys(),
default="failed",
help="The start state for all units")
test_group.add_argument("--test-state-to", action="store",
# choices=State.__members__.keys(),
default="active",
help="The state units will transition to")
return parser
def main():
observer = log.PythonLoggingObserver(loggerName="")
observer.start()
parser = build_arg_parser()
args = parser.parse_args()
if args.verbose:
logging.basicConfig(level=logging.DEBUG)
else:
logging.basicConfig(level=logging.INFO)
for notifier in get_enabled_notifiers(args.notifier):
notifier.handle_arguments(args)
if not args.test:
reactor.callWhenRunning(partial(setup, args))
else:
test(args)
systemd_status("Discovering units")
reactor.run()
if __name__ == "__main__":
if not booted():
exit("This system doesn't run systemd!")
main()