NagiosClient.__get_pid() - Code Metrics - Inspection of "Added a Nagios check command." - SetBased/py-enarksh - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( f1939b...e2cb7a )

by P.R.

created 2016-10-21 13:24 UTC

NagiosClient.__get_pid() 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	18
ccs	0
cts	8
cp	0
crap	12
rs	9.4285

"""
Enarksh

Copyright 2013-2016 Set Based IT Consultancy

Licence MIT
"""
import os

import zmq

import enarksh
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:
            pid_file = os.path.join(enarksh.HOME, 'var/lock', 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 ['controllerd', 'loggerd', 'spawnerd']:
            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'][enarksh.ENK_RST_ID_WAITING],
                   response['rst_count'][enarksh.ENK_RST_ID_QUEUED],
                   response['rst_count'][enarksh.ENK_RST_ID_RUNNING],
                   response['rst_count'][enarksh.ENK_RST_ID_COMPLETED],
                   response['rst_count'][enarksh.ENK_RST_ID_ERROR])

        self.__message += 'Schedules = {}, Waiting = {}, Queued = {}, Running = {}, Completed = {}, Error = {}'. \
            format(response['sch_count'],
                   response['rst_count'][enarksh.ENK_RST_ID_WAITING],
                   response['rst_count'][enarksh.ENK_RST_ID_QUEUED],
                   response['rst_count'][enarksh.ENK_RST_ID_RUNNING],
                   response['rst_count'][enarksh.ENK_RST_ID_COMPLETED],
                   response['rst_count'][enarksh.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.
        """
        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(enarksh.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
13			from enarksh.controller.message.NagiosMessage import NagiosMessage
14
15
16			class NagiosClient:
17			"""
18			A client for Nagios.
19			"""
20
21			# ------------------------------------------------------------------------------------------------------------------
22			def __init__(self, io):
23			"""
24			Object constructor.
25
26			:param enarksh.style.EnarkshStyle.EnarkshStyle io: The output decorator.
27			"""
28			self._zmq_context = None
29			"""
30			The ZMQ context.
31
32			:type: Context
33			"""
34
35			self._zmq_controller = None
36			"""
37			The socket for communicating with the controller.
38
39			:type: zmq.sugar.socket.Socket
40			"""
41
42			self._io = io
43			"""
44			The output decorator.
45
46			:type: enarksh.style.EnarkshStyle.EnarkshStyle
47			"""
48
49			self.__state = ''
50			"""
51			The state for Nagios.
52
53			:type: str
54			"""
55
56			self.__message = ''
57			"""
58			The message for Nagios.
59
60			:type: str
61			"""
62
63			self.__performance_data = ''
64			"""
65			The performance data for Nagios.
66
67			:type: str
68			"""
69
70			# ------------------------------------------------------------------------------------------------------------------
71			@staticmethod
72			def __get_pid(daemon):
73			"""
74			Returns the PID of a daemon of Enarksh.
75
76			:param str daemon: The name of the daemon.
77
78			:rtype: str\|None
79			"""
80			try:
81			pid_file = os.path.join(enarksh.HOME, 'var/lock', daemon + '.pid')
82			with open(pid_file) as handle:
83			pid = handle.read(1000)
84
85			return int(pid)
86
87			except Exception:
			1 ignored issue – show Best Practice introduced 2016-10-21 13:27 UTC by Report Bug Copy Issue Report Catching very general exceptions such as `Exception` is usually not recommended. Generally, you would want to handle very specific errors in the exception handler. This ensure that you do not hide other types of errors which should be fixed. So, unless you specifically plan to handle any error, consider adding a more specific exception. Loading history...
88			return None
89
90			# ------------------------------------------------------------------------------------------------------------------
91			def __test_daemon_is_running(self, daemon):
92			"""
93			Test whether a daemon is running.
94
95			:param str daemon: The name of the daemon.
96
97			:rtype: bool
98			"""
99			pid = self.__get_pid(daemon)
100
101			if pid is None:
102			return False
103
104			try:
105			proc_file = os.path.join('/proc', str(pid), 'comm')
106			name = open(proc_file).read(1000).strip()
107
108			return name == daemon
109
110			except Exception:
			1 ignored issue – show Best Practice introduced 2016-10-21 13:27 UTC by Report Bug Copy Issue Report Catching very general exceptions such as `Exception` is usually not recommended. Generally, you would want to handle very specific errors in the exception handler. This ensure that you do not hide other types of errors which should be fixed. So, unless you specifically plan to handle any error, consider adding a more specific exception. Loading history...
111			return False
112
113			# ------------------------------------------------------------------------------------------------------------------
114			def __test_daemons_are_running(self):
115			"""
116			Tests all 3 daemons are running.
117			"""
118			# Test controller is running
119			for daemon in ['controllerd', 'loggerd', 'spawnerd']:
120			if not self.__test_daemon_is_running(daemon):
121			if self.__message:
122			self.__message += ' '
123			self.__message = '{} is not running'.format(daemon)
124
125			if self.__message:
126			self.__message = 'Enarksh CRITICAL - ' + self.__message
127			self.__state = 2
128			else:
129			self.__message = 'Enarksh OK - '
130			self.__state = 0
131
132			# ------------------------------------------------------------------------------------------------------------------
133			def __get_performance_data(self):
134			"""
135			Retrieves performance data from the controller.
136			"""
137			# Compose the message for the controller.
138			message = NagiosMessage()
139
140			# Send the message to the controller.
141			self._zmq_controller.send_pyobj(message)
142
143			# Await the response from the controller.
144			response = self._zmq_controller.recv_pyobj()
145
146			self.__performance_data += 'schedules={}, waiting={}, queued={}, running={}, completed={}, error={}'.\
147			format(response['sch_count'],
148			response['rst_count'][enarksh.ENK_RST_ID_WAITING],
149			response['rst_count'][enarksh.ENK_RST_ID_QUEUED],
150			response['rst_count'][enarksh.ENK_RST_ID_RUNNING],
151			response['rst_count'][enarksh.ENK_RST_ID_COMPLETED],
152			response['rst_count'][enarksh.ENK_RST_ID_ERROR])
153
154			self.__message += 'Schedules = {}, Waiting = {}, Queued = {}, Running = {}, Completed = {}, Error = {}'. \
155			format(response['sch_count'],
156			response['rst_count'][enarksh.ENK_RST_ID_WAITING],
157			response['rst_count'][enarksh.ENK_RST_ID_QUEUED],
158			response['rst_count'][enarksh.ENK_RST_ID_RUNNING],
159			response['rst_count'][enarksh.ENK_RST_ID_COMPLETED],
160			response['rst_count'][enarksh.ENK_RST_ID_ERROR])
161
162			# ------------------------------------------------------------------------------------------------------------------
163			def main(self):
164			"""
165			The main function of Nagios.
166			"""
167			# Initialize ZMQ.
168			self._zmq_init()
169
170			self.__test_daemons_are_running()
171
172			if self.__state == 0:
173			self.__get_performance_data()
174
175			print(self.__message + '\|' + self.__performance_data)
176
177			return self.__state
178
179			# ------------------------------------------------------------------------------------------------------------------
180			def _zmq_init(self):
181			"""
182			Initializes ZMQ.
183			"""
184			self._zmq_context = zmq.Context()
185
186			# Create socket for communicating with the controller.
187			self._zmq_controller = self._zmq_context.socket(zmq.REQ)
			1 ignored issue – show Bug introduced 2016-10-21 13:27 UTC by Report Bug Copy Issue Report The Module `zmq` does not seem to have a member named `REQ`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
188			self._zmq_controller.connect(enarksh.CONTROLLER_LOCKSTEP_END_POINT)
189
190			# ----------------------------------------------------------------------------------------------------------------------
191

SetBased / py-enarksh

GitHub Access Token became invalid

Push — master ( f1939b...e2cb7a )

NagiosClient.__get_pid() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like