NagiosClient.__test_daemon_is_running() - Code Metrics - Inspection of "Used new features of PyStratum." - SetBased/py-enarksh - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 3b532d...9b942b )

by P.R.

created 2016-11-21 09:45 UTC

NagiosClient.__test_daemon_is_running() A

↳ Parent: NagiosClient

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	12

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	3
c	1
b	0
f	0
dl	0
loc	21
ccs	0
cts	10
cp	0
crap	12
rs	9.3142

"""
Enarksh

Copyright 2013-2016 Set Based IT Consultancy

Licence MIT
"""
import os

import zmq

import enarksh

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

SetBased / py-enarksh

GitHub Access Token became invalid

Push — master ( 3b532d...9b942b )

NagiosClient.__test_daemon_is_running() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like