NagiosClient.__test_daemons_are_running() - Code Metrics - SetBased/py-enarksh - Measure and Improve Code Quality continuously with Scrutinizer

NagiosClient.__test_daemons_are_running() B
last analyzed 2017-08-02 20:49 UTC

↳ Parent: NagiosClient

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	30

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	5
c	1
b	0
f	0
dl	0
loc	17
ccs	0
cts	11
cp	0
crap	30
rs	8.5454

"""
Enarksh

Copyright 2013-2016 Set Based IT Consultancy

Licence MIT
"""
import os

import zmq

from enarksh.C import C
from enarksh.Config import Config
from enarksh.controller.message.NagiosMessage import NagiosMessage


class NagiosClient:
    """
    A client for Nagios.
    """

    # ------------------------------------------------------------------------------------------------------------------
    def __init__(self, io):
        """
        Object constructor.

        :param enarksh.style.EnarkshStyle.EnarkshStyle io: The output decorator.
        """
        self.__zmq_context = None
        """
        The ZMQ context.

        :type: Context
        """

        self.__zmq_controller = None
        """
        The socket for communicating with the controller.

        :type: zmq.sugar.socket.Socket
        """

        self._io = io
        """
        The output decorator.

        :type: enarksh.style.EnarkshStyle.EnarkshStyle
        """

        self.__state = ''
        """
        The state for Nagios.

        :type: str
        """

        self.__message = ''
        """
        The message for Nagios.

        :type: str
        """

        self.__performance_data = ''
        """
        The performance data for Nagios.

        :type: str
        """

    # ------------------------------------------------------------------------------------------------------------------
    @staticmethod
    def __get_pid(daemon):
        """
        Returns the PID of a daemon of Enarksh.

        :param str daemon: The name of the daemon.

        :rtype: str|None
        """
        try:
            config = Config.get()

            pid_file = os.path.join(C.HOME, config.get_enarksh_lock_dir(), daemon + '.pid')
            with open(pid_file) as handle:
                pid = handle.read(1000)

            return int(pid)

        except Exception:
            return None

    # ------------------------------------------------------------------------------------------------------------------
    def __test_daemon_is_running(self, daemon):
        """
        Test whether a daemon is running.

        :param str daemon: The name of the daemon.

        :rtype: bool
        """
        pid = self.__get_pid(daemon)

        if pid is None:
            return False

        try:
            proc_file = os.path.join('/proc', str(pid), 'comm')
            name = open(proc_file).read(1000).strip()

            return name == daemon

        except Exception:
            return False

    # ------------------------------------------------------------------------------------------------------------------
    def __test_daemons_are_running(self):
        """
        Tests all 3 daemons are running.
        """
        # Test controller is running
        for daemon in ['controller', 'logger', 'spawner']:
            if not self.__test_daemon_is_running(daemon):
                if self.__message:
                    self.__message += ' '
                self.__message = '{} is not running'.format(daemon)

        if self.__message:
            self.__message = 'Enarksh CRITICAL - ' + self.__message
            self.__state = 2
        else:
            self.__message = 'Enarksh OK - '
            self.__state = 0

    # ------------------------------------------------------------------------------------------------------------------
    def __get_performance_data(self):
        """
        Retrieves performance data from the controller.
        """
        # Compose the message for the controller.
        message = NagiosMessage()

        # Send the message to the controller.
        self.__zmq_controller.send_pyobj(message)

        # Await the response from the controller.
        response = self.__zmq_controller.recv_pyobj()

        self.__performance_data += 'schedules={}, waiting={}, queued={}, running={}, completed={}, error={}'. \
            format(response['sch_count'],
                   response['rst_count'][C.ENK_RST_ID_WAITING],
                   response['rst_count'][C.ENK_RST_ID_QUEUED],
                   response['rst_count'][C.ENK_RST_ID_RUNNING],
                   response['rst_count'][C.ENK_RST_ID_COMPLETED],
                   response['rst_count'][C.ENK_RST_ID_ERROR])

        self.__message += 'Schedules = {}, Waiting = {}, Queued = {}, Running = {}, Completed = {}, Error = {}'. \
            format(response['sch_count'],
                   response['rst_count'][C.ENK_RST_ID_WAITING],
                   response['rst_count'][C.ENK_RST_ID_QUEUED],
                   response['rst_count'][C.ENK_RST_ID_RUNNING],
                   response['rst_count'][C.ENK_RST_ID_COMPLETED],
                   response['rst_count'][C.ENK_RST_ID_ERROR])

    # ------------------------------------------------------------------------------------------------------------------
    def main(self):
        """
        The main function of Nagios.
        """
        # Initialize ZMQ.
        self.__zmq_init()

        self.__test_daemons_are_running()

        if self.__state == 0:
            self.__get_performance_data()

        print(self.__message + '|' + self.__performance_data)

        return self.__state

    # ------------------------------------------------------------------------------------------------------------------
    def __zmq_init(self):
        """
        Initializes ZMQ.
        """
        config = Config.get()

        self.__zmq_context = zmq.Context()

        # Create socket for communicating with the controller.
        self.__zmq_controller = self.__zmq_context.socket(zmq.REQ)
        self.__zmq_controller.connect(config.get_controller_lockstep_end_point())

# ----------------------------------------------------------------------------------------------------------------------


1			"""
2			Enarksh
3
4			Copyright 2013-2016 Set Based IT Consultancy
5
6			Licence MIT
7			"""
8			import os
9
10			import zmq
11
12			from enarksh.C import C
13			from enarksh.Config import Config
14			from enarksh.controller.message.NagiosMessage import NagiosMessage
15
16
17			class NagiosClient:
18			"""
19			A client for Nagios.
20			"""
21
22			# ------------------------------------------------------------------------------------------------------------------
23			def __init__(self, io):
24			"""
25			Object constructor.
26
27			:param enarksh.style.EnarkshStyle.EnarkshStyle io: The output decorator.
28			"""
29			self.__zmq_context = None
30			"""
31			The ZMQ context.
32
33			:type: Context
34			"""
35
36			self.__zmq_controller = None
37			"""
38			The socket for communicating with the controller.
39
40			:type: zmq.sugar.socket.Socket
41			"""
42
43			self._io = io
44			"""
45			The output decorator.
46
47			:type: enarksh.style.EnarkshStyle.EnarkshStyle
48			"""
49
50			self.__state = ''
51			"""
52			The state for Nagios.
53
54			:type: str
55			"""
56
57			self.__message = ''
58			"""
59			The message for Nagios.
60
61			:type: str
62			"""
63
64			self.__performance_data = ''
65			"""
66			The performance data for Nagios.
67
68			:type: str
69			"""
70
71			# ------------------------------------------------------------------------------------------------------------------
72			@staticmethod
73			def __get_pid(daemon):
74			"""
75			Returns the PID of a daemon of Enarksh.
76
77			:param str daemon: The name of the daemon.
78
79			:rtype: str\|None
80			"""
81			try:
82			config = Config.get()
83
84			pid_file = os.path.join(C.HOME, config.get_enarksh_lock_dir(), daemon + '.pid')
85			with open(pid_file) as handle:
86			pid = handle.read(1000)
87
88			return int(pid)
89
90			except Exception:
91			return None
92
93			# ------------------------------------------------------------------------------------------------------------------
94			def __test_daemon_is_running(self, daemon):
95			"""
96			Test whether a daemon is running.
97
98			:param str daemon: The name of the daemon.
99
100			:rtype: bool
101			"""
102			pid = self.__get_pid(daemon)
103
104			if pid is None:
105			return False
106
107			try:
108			proc_file = os.path.join('/proc', str(pid), 'comm')
109			name = open(proc_file).read(1000).strip()
110
111			return name == daemon
112
113			except Exception:
114			return False
115
116			# ------------------------------------------------------------------------------------------------------------------
117			def __test_daemons_are_running(self):
118			"""
119			Tests all 3 daemons are running.
120			"""
121			# Test controller is running
122			for daemon in ['controller', 'logger', 'spawner']:
123			if not self.__test_daemon_is_running(daemon):
124			if self.__message:
125			self.__message += ' '
126			self.__message = '{} is not running'.format(daemon)
127
128			if self.__message:
129			self.__message = 'Enarksh CRITICAL - ' + self.__message
130			self.__state = 2
131			else:
132			self.__message = 'Enarksh OK - '
133			self.__state = 0
134
135			# ------------------------------------------------------------------------------------------------------------------
136			def __get_performance_data(self):
137			"""
138			Retrieves performance data from the controller.
139			"""
140			# Compose the message for the controller.
141			message = NagiosMessage()
142
143			# Send the message to the controller.
144			self.__zmq_controller.send_pyobj(message)
145
146			# Await the response from the controller.
147			response = self.__zmq_controller.recv_pyobj()
148
149			self.__performance_data += 'schedules={}, waiting={}, queued={}, running={}, completed={}, error={}'. \
150			format(response['sch_count'],
151			response['rst_count'][C.ENK_RST_ID_WAITING],
152			response['rst_count'][C.ENK_RST_ID_QUEUED],
153			response['rst_count'][C.ENK_RST_ID_RUNNING],
154			response['rst_count'][C.ENK_RST_ID_COMPLETED],
155			response['rst_count'][C.ENK_RST_ID_ERROR])
156
157			self.__message += 'Schedules = {}, Waiting = {}, Queued = {}, Running = {}, Completed = {}, Error = {}'. \
158			format(response['sch_count'],
159			response['rst_count'][C.ENK_RST_ID_WAITING],
160			response['rst_count'][C.ENK_RST_ID_QUEUED],
161			response['rst_count'][C.ENK_RST_ID_RUNNING],
162			response['rst_count'][C.ENK_RST_ID_COMPLETED],
163			response['rst_count'][C.ENK_RST_ID_ERROR])
164
165			# ------------------------------------------------------------------------------------------------------------------
166			def main(self):
167			"""
168			The main function of Nagios.
169			"""
170			# Initialize ZMQ.
171			self.__zmq_init()
172
173			self.__test_daemons_are_running()
174
175			if self.__state == 0:
176			self.__get_performance_data()
177
178			print(self.__message + '\|' + self.__performance_data)
179
180			return self.__state
181
182			# ------------------------------------------------------------------------------------------------------------------
183			def __zmq_init(self):
184			"""
185			Initializes ZMQ.
186			"""
187			config = Config.get()
188
189			self.__zmq_context = zmq.Context()
190
191			# Create socket for communicating with the controller.
192			self.__zmq_controller = self.__zmq_context.socket(zmq.REQ)
193			self.__zmq_controller.connect(config.get_controller_lockstep_end_point())
194
195			# ----------------------------------------------------------------------------------------------------------------------
196

SetBased / py-enarksh

GitHub Access Token became invalid

NagiosClient.__test_daemons_are_running() B last analyzed 2017-08-02 20:49 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

NagiosClient.__test_daemons_are_running() B
last analyzed 2017-08-02 20:49 UTC