JobHandler._final() - Code Metrics - Inspection of "Use the PEP 3143 standard daemon process library,..." - SetBased/py-enarksh - Measure and Improve Code Quality continuously with Scrutinizer

Test Setup Failed

Push — master ( 19fcc2...697e45 )

by P.R.

created 2016-08-04 09:20 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
rs	9.2
ccs	0
cts	7
cp	0
crap	20

"""
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 SomeClass:
    def some_method(self):
        """Do x and return foo."""
    _allowed_users = []

    # ------------------------------------------------------------------------------------------------------------------
    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 SystemExit("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(-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):
			0 ignored issues – show Coding Style introduced 2016-08-04 09:24 UTC by Report Bug Copy Issue Report This class 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...
22			_allowed_users = []
23
24			# ------------------------------------------------------------------------------------------------------------------
25			def __init__(self, sch_id, rnd_id, user_name, args):
26			"""
27			Creates a job handler for starting a job.
28
29			:param int sch_id: The ID of the schedule of the job.
30			:param int rnd_id: The ID of the job.
31			:param str user_name: The user under which the job must run.
32			:param args: The arguments for the job.
33			"""
34			EventActor.__init__(self)
35
36			self._sch_id = sch_id
37			"""
38			The ID of the schedule of the job.
39
40			:type: int
41			"""
42
43			self._rnd_id = rnd_id
44			"""
45			The ID of the job.
46
47			:type: int
48			"""
49
50			self._user_name = user_name
51			"""
52			The user under which the job must run.
53
54			:type: str
55			"""
56
57			self._args = args
58			"""
59			The arguments for the job.
60
61			:type: list[str]
62			"""
63
64			self.stdout_logger = ChunkLogger()
65			"""
66			The chunk logger for STDOUT of the job.
67
68			:type: enarksh.spawner.ChunkLogger.ChunkLogger
69			"""
70
71			self.stderr_logger = ChunkLogger()
72			"""
73			The chunk logger for STDERR of the job.
74
75			:type: enarksh.spawner.ChunkLogger.ChunkLogger
76			"""
77
78			self._child_pid = -1
79			"""
80			The PID of the child process.
81
82			:type: int
83			"""
84
85			self._stdout = -1
86			"""
87			The fd for reading the STDOUT of the child process.
88
89			:type: int
90			"""
91
92			self._stderr = -1
93			"""
94			The fd for reading the STDERR of the child process.
95
96			:type: int
97			"""
98
99			self.final_event = Event(self)
100			"""
101			The event that will be fired when the job has been finally done.
102
103			:type: enarksh.event.Event.Event
104			"""
105
106			# ------------------------------------------------------------------------------------------------------------------
107			@property
108			def pid(self):
109			"""
110			Returns the PID of the job. If the job is finished the pid is -1.
111
112			:rtype int: The PID of the job.
113			"""
114			return self._child_pid
115
116			# ------------------------------------------------------------------------------------------------------------------
117			@property
118			def stderr(self):
119			"""
120			Returns the file descriptor for reading the stderr of the child process.
121
122			:rtype int: file descriptor
123			"""
124			return self._stderr
125
126			# ------------------------------------------------------------------------------------------------------------------
127			@property
128			def stdout(self):
129			"""
130			Returns the file descriptor for reading the stdout of the child process.
131
132			:rtype int: file descriptor
133			"""
134			return self._stdout
135
136			# ------------------------------------------------------------------------------------------------------------------
137			@property
138			def rnd_id(self):
139			"""
140			Returns the ID of the job.
141
142			:rtype: int
143			"""
144			return self._rnd_id
145
146			# ------------------------------------------------------------------------------------------------------------------
147			@property
148			def sch_id(self):
149			"""
150			Returns the ID of the schedule of the job.
151
152			:rtype: int
153			"""
154			return self._sch_id
155
156			# ------------------------------------------------------------------------------------------------------------------
157			def _log_job_start(self):
158			"""
159			Logs the starting of a job.
160			"""
161			print("Start rnd_id: %10d, %8s, %s" % (self._rnd_id, self._user_name, str(self._args)))
162
163			# ------------------------------------------------------------------------------------------------------------------
164			def _log_job_stop(self):
165			"""
166			Logs the end of job.
167			"""
168			print("End rnd_id: %10d, %8s, %s" % (self._rnd_id, self._user_name, str(self._args)))
169
170			# ------------------------------------------------------------------------------------------------------------------
171			def _final(self):
172			"""
173			When the job is finally done fires a done event.
174			"""
175			if self._child_pid == -1 and self._stdout == -1 and self._stderr == -1:
176			self._log_job_stop()
177
178			# Close the files of the chunk loggers.
179			self.stdout_logger.close()
180			self.stderr_logger.close()
181
182			# Send messages to logger daemon that the stdout and stderr of the job can be loaded into the database.
183			self.get_logger_message('out').send_message('logger')
184			self.get_logger_message('err').send_message('logger')
185
186			# Fire the event that this job has been done completely.
187			self.final_event.fire()
188
189			# ------------------------------------------------------------------------------------------------------------------
190			def set_job_has_finished(self):
191			"""
192			Marks that the job has finished.
193			"""
194			self._child_pid = -1
195			self._final()
196
197			# ------------------------------------------------------------------------------------------------------------------
198			def get_logger_message(self, std):
199			"""
200			Returns a message for the logger.
201
202			:param str std: log for stdout, err for stderr
203
204			:rtype: enarksh.message.logger.LogFileMessage.LogFileMessage
205			"""
206			if std == 'out':
207			chunk_logger = self.stdout_logger
208			elif std == 'err':
209			chunk_logger = self.stderr_logger
210			else:
211			raise Exception("Unknown output '%s'." % std)
212
213			return LogFileMessage(self._rnd_id,
214			std,
215			chunk_logger.get_total_log_size(),
216			chunk_logger.filename1,
217			chunk_logger.filename2)
218
219			# ------------------------------------------------------------------------------------------------------------------
220			def read(self, fd):
221			"""
222			Reads data from the file descriptor and stores the data in a chunk logger.
223
224			:param int fd: The file descriptor.
225			"""
226			if fd == self._stdout:
227			data = os.read(fd, 1000)
228			if data == b'':
229			# The pipe has been closed by the child process.
230			os.close(self._stdout)
231			self._stdout = -1
232			self._final()
233			else:
234			self.stdout_logger.write(data)
235
236			elif fd == self._stderr:
237			data = os.read(fd, 1000)
238			if data == b'':
239			# The pipe has been closed by the child process.
240			os.close(self._stderr)
241			self._stderr = -1
242			self._final()
243			else:
244			self.stdout_logger.write(data)
245			self.stderr_logger.write(data)
246
247			else:
248			raise ValueError('Unknown file descriptor %d.' % fd)
249
250			# ------------------------------------------------------------------------------------------------------------------
251			@staticmethod
252			def read_allowed_users():
253			"""
254			Reads the user names under which enarksh is allowed to start processes.
255			"""
256			config = ConfigParser()
257			config.read(os.path.join(enarksh.HOME, 'etc/enarksh.cfg'))
258
259			JobHandler._allowed_users = config.get('spawner', 'users').split()
260
261			# ------------------------------------------------------------------------------------------------------------------
262			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...
263			self._log_job_start()
264
265			# Create pipes for stdout and stderr.
266			pipe_stdout = os.pipe()
267			pipe_stderr = os.pipe()
268
269			self._child_pid = os.fork()
270			if self._child_pid == 0:
271			# Child process.
272			try:
273			# Close the read ends from the pipes.
274			os.close(pipe_stdout[0])
275			os.close(pipe_stderr[0])
276
277			# Duplicate stdout and stderr on the pipes.
278			sys.stdout.flush()
279			sys.stderr.flush()
280			os.dup2(pipe_stdout[1], sys.stdout.fileno())
281			os.dup2(pipe_stderr[1], sys.stderr.fileno())
282
283			# Set the effective user and group.
284			if self._user_name in self._allowed_users:
285			_, _, uid, gid, _, _, _ = pwd.getpwnam(self._user_name)
286			os.setuid(0)
287
288			os.initgroups(self._user_name, gid)
289			os.setuid(uid)
290			else:
291			raise SystemExit("Spanner is not allowed to start processes under user '%s'." % self._user_name)
292
293			# Set variable for subprocess.
294			os.putenv('ENK_RND_ID', str(self._rnd_id))
295			os.putenv('ENK_SCH_ID', str(self._sch_id))
296
297			# Replace this child process with the actual job.
298			os.execv(self._args[0], self._args)
299
300			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...
301			print('Unable to start job.', file=sys.stderr)
302			print('Reason: %s' % e, file=sys.stderr)
303			traceback.print_exc(file=sys.stderr)
304			exit(-1)
305			else:
306			# Parent process.
307			# Close the write ends from the pipes.
308			os.close(pipe_stdout[1])
309			os.close(pipe_stderr[1])
310
311			# Remember the fds for reading the stdout and stderr from the child process.
312			self._stdout = pipe_stdout[0]
313			self._stderr = pipe_stderr[0]
314
315			# Make reading from the pipes non-blocking.
316			# fcntl.fcntl(self._stdout, 0)
317			# fcntl.fcntl(self._stderr, 0)
318
319			# ----------------------------------------------------------------------------------------------------------------------
320

SetBased / py-enarksh

GitHub Access Token became invalid

Push — master ( 19fcc2...697e45 )

JobHandler._final() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like