JobHandler._final() - Code Metrics - Inspection of "Code and comment enhancements." - SetBased/py-enarksh - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( ad34bf...a9a32b )

by P.R.

created 2016-10-20 07:40 UTC

JobHandler._final() A

↳ Parent: JobHandler

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	20

Importance

Changes

Metric	Value
cc	4
c	0
b	0
f	0
dl	0
loc	17
ccs	0
cts	8
cp	0
crap	20
rs	9.2

"""
Enarksh

Copyright 2013-2016 Set Based IT Consultancy

Licence MIT
"""
import os
import pwd
import sys
import traceback
from configparser import ConfigParser

import enarksh
from enarksh.event.Event import Event
from enarksh.event.EventActor import EventActor
from enarksh.logger.message.LogFileMessage import LogFileMessage
from enarksh.spawner.ChunkLogger import ChunkLogger


class JobHandler(EventActor):
    """
    Class for reading the stdout and stderr and monitoring the processes of a job.
    """
    __allowed_users = []
    """
    The list of user names under which a process can be started.

    :type: list[str]
    """

    # ------------------------------------------------------------------------------------------------------------------
    def __init__(self, sch_id, rnd_id, user_name, args):
        """
        Creates a job handler for starting a job.

        :param int sch_id: The ID of the schedule of the job.
        :param int rnd_id: The ID of the job.
        :param str user_name: The user under which the job must run.
        :param args: The arguments for the job.
        """
        EventActor.__init__(self)

        self.__sch_id = sch_id
        """
        The ID of the schedule of the job.

        :type: int
        """

        self.__rnd_id = rnd_id
        """
        The ID of the job.

        :type: int
        """

        self.__user_name = user_name
        """
        The user under which the job must run.

        :type: str
        """

        self.__args = args
        """
        The arguments for the job.

        :type: list[str]
        """

        self.stdout_logger = ChunkLogger()
        """
        The chunk logger for STDOUT of the job.

        :type: enarksh.spawner.ChunkLogger.ChunkLogger
        """

        self.stderr_logger = ChunkLogger()
        """
        The chunk logger for STDERR of the job.

        :type: enarksh.spawner.ChunkLogger.ChunkLogger
        """

        self.__child_pid = -1
        """
        The PID of the child process.

        :type: int
        """

        self.__stdout = -1
        """
        The fd for reading the STDOUT of the child process.

        :type: int
        """

        self.__stderr = -1
        """
        The fd for reading the STDERR of the child process.

        :type: int
        """

        self.final_event = Event(self)
        """
        The event that will be fired when the job has been finally done.

        :type: enarksh.event.Event.Event
        """

    # ------------------------------------------------------------------------------------------------------------------
    @property
    def pid(self):
        """
        Returns the PID of the job. If the job is finished the pid is -1.

        :rtype int: The PID of the job.
        """
        return self.__child_pid

    # ------------------------------------------------------------------------------------------------------------------
    @property
    def stderr(self):
        """
        Returns the file descriptor for reading the stderr of the child process.

        :rtype int: file descriptor
        """
        return self.__stderr

    # ------------------------------------------------------------------------------------------------------------------
    @property
    def stdout(self):
        """
        Returns the file descriptor for reading the stdout of the child process.

        :rtype int: file descriptor
        """
        return self.__stdout

    # ------------------------------------------------------------------------------------------------------------------
    @property
    def rnd_id(self):
        """
        Returns the ID of the job.

        :rtype: int
        """
        return self.__rnd_id

    # ------------------------------------------------------------------------------------------------------------------
    @property
    def sch_id(self):
        """
        Returns the ID of the schedule of the job.

        :rtype: int
        """
        return self.__sch_id

    # ------------------------------------------------------------------------------------------------------------------
    def __log_job_start(self):
        """
        Logs the starting of a job.
        """
        print("Start rnd_id: %10d, %8s, %s" % (self.__rnd_id, self.__user_name, str(self.__args)))

    # ------------------------------------------------------------------------------------------------------------------
    def __log_job_stop(self):
        """
        Logs the end of job.
        """
        print("End   rnd_id: %10d, %8s, %s" % (self.__rnd_id, self.__user_name, str(self.__args)))

    # ------------------------------------------------------------------------------------------------------------------
    def __final(self):
        """
        When the job is finally done fires a done event.
        """
        if self.__child_pid == -1 and self.__stdout == -1 and self.__stderr == -1:
            self.__log_job_stop()

            # Close the files of the chunk loggers.
            self.stdout_logger.close()
            self.stderr_logger.close()

            # Send messages to logger daemon that the stdout and stderr of the job can be loaded into the database.
            self.get_logger_message('out').send_message('logger')
            self.get_logger_message('err').send_message('logger')

            # Fire the event that this job has been done completely.
            self.final_event.fire()

    # ------------------------------------------------------------------------------------------------------------------
    def set_job_has_finished(self):
        """
        Marks that the job has finished.
        """
        self.__child_pid = -1
        self.__final()

    # ------------------------------------------------------------------------------------------------------------------
    def get_logger_message(self, std):
        """
        Returns a message for the logger.

        :param str std: log for stdout, err for stderr

        :rtype: enarksh.message.logger.LogFileMessage.LogFileMessage
        """
        if std == 'out':
            chunk_logger = self.stdout_logger
        elif std == 'err':
            chunk_logger = self.stderr_logger
        else:
            raise Exception("Unknown output '%s'." % std)

        return LogFileMessage(self.__rnd_id,
                              std,
                              chunk_logger.get_total_log_size(),
                              chunk_logger.filename1,
                              chunk_logger.filename2)

    # ------------------------------------------------------------------------------------------------------------------
    def read(self, fd):
        """
        Reads data from the file descriptor and stores the data in a chunk logger.

        :param int fd: The file descriptor.
        """
        if fd == self.__stdout:
            data = os.read(fd, 1000)
            if data == b'':
                # The pipe has been closed by the child process.
                os.close(self.__stdout)
                self.__stdout = -1
                self.__final()
            else:
                self.stdout_logger.write(data)

        elif fd == self.__stderr:
            data = os.read(fd, 1000)
            if data == b'':
                # The pipe has been closed by the child process.
                os.close(self.__stderr)
                self.__stderr = -1
                self.__final()
            else:
                self.stdout_logger.write(data)
                self.stderr_logger.write(data)

        else:
            raise ValueError('Unknown file descriptor %d.' % fd)

    # ------------------------------------------------------------------------------------------------------------------
    @staticmethod
    def read_allowed_users():
        """
        Reads the user names under which enarksh is allowed to start processes.
        """
        config = ConfigParser()
        config.read(os.path.join(enarksh.HOME, 'etc/enarksh.cfg'))

        JobHandler.__allowed_users = config.get('spawner', 'users').split()

    # ------------------------------------------------------------------------------------------------------------------
    def start_job(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        self.__log_job_start()

        # Create pipes for stdout and stderr.
        pipe_stdout = os.pipe()
        pipe_stderr = os.pipe()

        self.__child_pid = os.fork()
        if self.__child_pid == 0:
            # Child process.
            try:
                # Close the read ends from the pipes.
                os.close(pipe_stdout[0])
                os.close(pipe_stderr[0])

                # Duplicate stdout and stderr on the pipes.
                sys.stdout.flush()
                sys.stderr.flush()
                os.dup2(pipe_stdout[1], sys.stdout.fileno())
                os.dup2(pipe_stderr[1], sys.stderr.fileno())

                # Set the effective user and group.
                if self.__user_name in self.__allowed_users:
                    _, _, uid, gid, _, _, _ = pwd.getpwnam(self.__user_name)
                    os.setuid(0)

                    os.initgroups(self.__user_name, gid)
                    os.setuid(uid)
                else:
                    raise RuntimeError("Spanner is not allowed to start processes under user '%s'." % self.__user_name)

                # Set variable for subprocess.
                os.putenv('ENK_RND_ID', str(self.__rnd_id))
                os.putenv('ENK_SCH_ID', str(self.__sch_id))

                # Replace this child process with the actual job.
                os.execv(self.__args[0], self.__args)

            except Exception as e:

                print('Unable to start job.', file=sys.stderr)
                print('Reason: %s' % e, file=sys.stderr)
                traceback.print_exc(file=sys.stderr)
                # Exit immediately without running the exit handlers (e.g. from daemon) from the parent process.
                os._exit(-1)
        else:
            # Parent process.
            # Close the write ends from the pipes.
            os.close(pipe_stdout[1])
            os.close(pipe_stderr[1])

            # Remember the fds for reading the stdout and stderr from the child process.
            self.__stdout = pipe_stdout[0]
            self.__stderr = pipe_stderr[0]

            # Make reading from the pipes non-blocking.
            # fcntl.fcntl(self._stdout, 0)
            # fcntl.fcntl(self._stderr, 0)

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


1			"""
2			Enarksh
3
4			Copyright 2013-2016 Set Based IT Consultancy
5
6			Licence MIT
7			"""
8			import os
9			import pwd
10			import sys
11			import traceback
12			from configparser import ConfigParser
13
14			import enarksh
15			from enarksh.event.Event import Event
16			from enarksh.event.EventActor import EventActor
17			from enarksh.logger.message.LogFileMessage import LogFileMessage
18			from enarksh.spawner.ChunkLogger import ChunkLogger
19
20
21			class JobHandler(EventActor):
22			"""
23			Class for reading the stdout and stderr and monitoring the processes of a job.
24			"""
25			__allowed_users = []
26			"""
27			The list of user names under which a process can be started.
28
29			:type: list[str]
30			"""
31
32			# ------------------------------------------------------------------------------------------------------------------
33			def __init__(self, sch_id, rnd_id, user_name, args):
34			"""
35			Creates a job handler for starting a job.
36
37			:param int sch_id: The ID of the schedule of the job.
38			:param int rnd_id: The ID of the job.
39			:param str user_name: The user under which the job must run.
40			:param args: The arguments for the job.
41			"""
42			EventActor.__init__(self)
43
44			self.__sch_id = sch_id
45			"""
46			The ID of the schedule of the job.
47
48			:type: int
49			"""
50
51			self.__rnd_id = rnd_id
52			"""
53			The ID of the job.
54
55			:type: int
56			"""
57
58			self.__user_name = user_name
59			"""
60			The user under which the job must run.
61
62			:type: str
63			"""
64
65			self.__args = args
66			"""
67			The arguments for the job.
68
69			:type: list[str]
70			"""
71
72			self.stdout_logger = ChunkLogger()
73			"""
74			The chunk logger for STDOUT of the job.
75
76			:type: enarksh.spawner.ChunkLogger.ChunkLogger
77			"""
78
79			self.stderr_logger = ChunkLogger()
80			"""
81			The chunk logger for STDERR of the job.
82
83			:type: enarksh.spawner.ChunkLogger.ChunkLogger
84			"""
85
86			self.__child_pid = -1
87			"""
88			The PID of the child process.
89
90			:type: int
91			"""
92
93			self.__stdout = -1
94			"""
95			The fd for reading the STDOUT of the child process.
96
97			:type: int
98			"""
99
100			self.__stderr = -1
101			"""
102			The fd for reading the STDERR of the child process.
103
104			:type: int
105			"""
106
107			self.final_event = Event(self)
108			"""
109			The event that will be fired when the job has been finally done.
110
111			:type: enarksh.event.Event.Event
112			"""
113
114			# ------------------------------------------------------------------------------------------------------------------
115			@property
116			def pid(self):
117			"""
118			Returns the PID of the job. If the job is finished the pid is -1.
119
120			:rtype int: The PID of the job.
121			"""
122			return self.__child_pid
123
124			# ------------------------------------------------------------------------------------------------------------------
125			@property
126			def stderr(self):
127			"""
128			Returns the file descriptor for reading the stderr of the child process.
129
130			:rtype int: file descriptor
131			"""
132			return self.__stderr
133
134			# ------------------------------------------------------------------------------------------------------------------
135			@property
136			def stdout(self):
137			"""
138			Returns the file descriptor for reading the stdout of the child process.
139
140			:rtype int: file descriptor
141			"""
142			return self.__stdout
143
144			# ------------------------------------------------------------------------------------------------------------------
145			@property
146			def rnd_id(self):
147			"""
148			Returns the ID of the job.
149
150			:rtype: int
151			"""
152			return self.__rnd_id
153
154			# ------------------------------------------------------------------------------------------------------------------
155			@property
156			def sch_id(self):
157			"""
158			Returns the ID of the schedule of the job.
159
160			:rtype: int
161			"""
162			return self.__sch_id
163
164			# ------------------------------------------------------------------------------------------------------------------
165			def __log_job_start(self):
166			"""
167			Logs the starting of a job.
168			"""
169			print("Start rnd_id: %10d, %8s, %s" % (self.__rnd_id, self.__user_name, str(self.__args)))
170
171			# ------------------------------------------------------------------------------------------------------------------
172			def __log_job_stop(self):
173			"""
174			Logs the end of job.
175			"""
176			print("End rnd_id: %10d, %8s, %s" % (self.__rnd_id, self.__user_name, str(self.__args)))
177
178			# ------------------------------------------------------------------------------------------------------------------
179			def __final(self):
180			"""
181			When the job is finally done fires a done event.
182			"""
183			if self.__child_pid == -1 and self.__stdout == -1 and self.__stderr == -1:
184			self.__log_job_stop()
185
186			# Close the files of the chunk loggers.
187			self.stdout_logger.close()
188			self.stderr_logger.close()
189
190			# Send messages to logger daemon that the stdout and stderr of the job can be loaded into the database.
191			self.get_logger_message('out').send_message('logger')
192			self.get_logger_message('err').send_message('logger')
193
194			# Fire the event that this job has been done completely.
195			self.final_event.fire()
196
197			# ------------------------------------------------------------------------------------------------------------------
198			def set_job_has_finished(self):
199			"""
200			Marks that the job has finished.
201			"""
202			self.__child_pid = -1
203			self.__final()
204
205			# ------------------------------------------------------------------------------------------------------------------
206			def get_logger_message(self, std):
207			"""
208			Returns a message for the logger.
209
210			:param str std: log for stdout, err for stderr
211
212			:rtype: enarksh.message.logger.LogFileMessage.LogFileMessage
213			"""
214			if std == 'out':
215			chunk_logger = self.stdout_logger
216			elif std == 'err':
217			chunk_logger = self.stderr_logger
218			else:
219			raise Exception("Unknown output '%s'." % std)
220
221			return LogFileMessage(self.__rnd_id,
222			std,
223			chunk_logger.get_total_log_size(),
224			chunk_logger.filename1,
225			chunk_logger.filename2)
226
227			# ------------------------------------------------------------------------------------------------------------------
228			def read(self, fd):
229			"""
230			Reads data from the file descriptor and stores the data in a chunk logger.
231
232			:param int fd: The file descriptor.
233			"""
234			if fd == self.__stdout:
235			data = os.read(fd, 1000)
236			if data == b'':
237			# The pipe has been closed by the child process.
238			os.close(self.__stdout)
239			self.__stdout = -1
240			self.__final()
241			else:
242			self.stdout_logger.write(data)
243
244			elif fd == self.__stderr:
245			data = os.read(fd, 1000)
246			if data == b'':
247			# The pipe has been closed by the child process.
248			os.close(self.__stderr)
249			self.__stderr = -1
250			self.__final()
251			else:
252			self.stdout_logger.write(data)
253			self.stderr_logger.write(data)
254
255			else:
256			raise ValueError('Unknown file descriptor %d.' % fd)
257
258			# ------------------------------------------------------------------------------------------------------------------
259			@staticmethod
260			def read_allowed_users():
261			"""
262			Reads the user names under which enarksh is allowed to start processes.
263			"""
264			config = ConfigParser()
265			config.read(os.path.join(enarksh.HOME, 'etc/enarksh.cfg'))
266
267			JobHandler.__allowed_users = config.get('spawner', 'users').split()
268
269			# ------------------------------------------------------------------------------------------------------------------
270			def start_job(self):
			0 ignored issues – show Coding Style introduced 2016-07-27 12:34 UTC by Report Bug Copy Issue Report This method should have a docstring. The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods: class SomeClass: def some_method(self): """Do x and return foo.""" If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions. Loading history...
271			self.__log_job_start()
272
273			# Create pipes for stdout and stderr.
274			pipe_stdout = os.pipe()
275			pipe_stderr = os.pipe()
276
277			self.__child_pid = os.fork()
278			if self.__child_pid == 0:
279			# Child process.
280			try:
281			# Close the read ends from the pipes.
282			os.close(pipe_stdout[0])
283			os.close(pipe_stderr[0])
284
285			# Duplicate stdout and stderr on the pipes.
286			sys.stdout.flush()
287			sys.stderr.flush()
288			os.dup2(pipe_stdout[1], sys.stdout.fileno())
289			os.dup2(pipe_stderr[1], sys.stderr.fileno())
290
291			# Set the effective user and group.
292			if self.__user_name in self.__allowed_users:
293			_, _, uid, gid, _, _, _ = pwd.getpwnam(self.__user_name)
294			os.setuid(0)
295
296			os.initgroups(self.__user_name, gid)
297			os.setuid(uid)
298			else:
299			raise RuntimeError("Spanner is not allowed to start processes under user '%s'." % self.__user_name)
300
301			# Set variable for subprocess.
302			os.putenv('ENK_RND_ID', str(self.__rnd_id))
303			os.putenv('ENK_SCH_ID', str(self.__sch_id))
304
305			# Replace this child process with the actual job.
306			os.execv(self.__args[0], self.__args)
307
308			except Exception as e:
			1 ignored issue – show Best Practice introduced 2016-07-27 12:34 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... Coding Style Naming introduced 2016-07-27 12:34 UTC by Report Bug Copy Issue Report The name `e` does not conform to the variable naming conventions (`[a-z_][a-z0-9_]{1,60}$`). This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
309			print('Unable to start job.', file=sys.stderr)
310			print('Reason: %s' % e, file=sys.stderr)
311			traceback.print_exc(file=sys.stderr)
312			# Exit immediately without running the exit handlers (e.g. from daemon) from the parent process.
313			os._exit(-1)
314			else:
315			# Parent process.
316			# Close the write ends from the pipes.
317			os.close(pipe_stdout[1])
318			os.close(pipe_stderr[1])
319
320			# Remember the fds for reading the stdout and stderr from the child process.
321			self.__stdout = pipe_stdout[0]
322			self.__stderr = pipe_stderr[0]
323
324			# Make reading from the pipes non-blocking.
325			# fcntl.fcntl(self._stdout, 0)
326			# fcntl.fcntl(self._stderr, 0)
327
328			# ----------------------------------------------------------------------------------------------------------------------
329

SetBased / py-enarksh

GitHub Access Token became invalid

Push — master ( ad34bf...a9a32b )

JobHandler._final() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like