reanahub / reana-workflow-controller / 433

pending completion

Build # 433

Build Type

Pull #118

travis-ci

Committed by

web-flow

Commit Message

global: remove reana-workflow-commons

* Removes reana-workflow-commons as it is merged with reana-commons.

Signed-off-by: Dinos Kousidis <dinos.kousidis@cern.ch>
(cherry picked from commit 4ac303de7)

Pull Request Pull Request #118: global: remove reana-workflow-commons

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

265 of 502 relevant lines covered (52.79%)

0.53 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/reana_workflow_controller/consumer.py

# -*- coding: utf-8 -*-
#
# This file is part of REANA.
# Copyright (C) 2018 CERN.
#
# REANA is free software; you can redistribute it and/or modify it under the
# terms of the GNU General Public License as published by the Free Software
# Foundation; either version 2 of the License, or (at your option) any later
# version.
#
# REANA is distributed in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
# A PARTICULAR PURPOSE. See the GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along with
# REANA; if not, write to the Free Software Foundation, Inc., 59 Temple Place,
# Suite 330, Boston, MA 02111-1307, USA.
#
# In applying this license, CERN does not waive the privileges and immunities
# granted to it by virtue of its status as an Intergovernmental Organization or
# submit itself to any jurisdiction.

"""REANA Workflow Controller MQ Consumer."""

import json

import pika
from reana_commons.consumer import Consumer
from reana_db.database import Session
from reana_db.models import WorkflowStatus

from .config import STATUS_QUEUE
from .tasks import (_update_job_cache, _update_job_progress,
                    _update_run_progress, _update_workflow_status)


class JobStatusConsumer(Consumer):
    """Consumer of jobs-status queue."""

    def __init__(self):
        """Constructor."""
        super(JobStatusConsumer, self).__init__(STATUS_QUEUE)

    def on_message(self, channel, method_frame, header_frame, body):
        """On new message event handler."""
        self._channel.basic_ack(delivery_tag=method_frame.delivery_tag)
        body_dict = json.loads(body)
        workflow_uuid = body_dict.get('workflow_uuid')
        if workflow_uuid:
            status = body_dict.get('status')
            if status:
                status = WorkflowStatus(status)
                print(" [x] Received workflow_uuid: {0} status: {1}".
                      format(workflow_uuid, status))
            logs = body_dict.get('logs') or ''
            _update_workflow_status(workflow_uuid, status, logs)
            if 'message' in body_dict and body_dict.get('message'):
                msg = body_dict['message']
                if 'progress' in msg:
                    _update_run_progress(workflow_uuid, msg)
                    _update_job_progress(workflow_uuid, msg)
                # Caching: calculate input hash and store in JobCache
                if 'caching_info' in msg:
                    _update_job_cache(msg)
                Session.commit()

    def consume(self):
        """Start consuming incoming messages."""
        while True:
            self.connect()
            self._channel.basic_consume(self.on_message, STATUS_QUEUE)
            try:
                self._channel.start_consuming()
            except KeyboardInterrupt:
                self._channel.stop_consuming()
                self._conn.close()
            except pika.exceptions.ConnectionClosed:
                # Uncomment this to make the example not attempt recovery
                # from server-initiated connection closure, including
                # when the node is stopped cleanly
                # except pika.exceptions.ConnectionClosedByBroker:
                #     pass
                continue

1	# -- coding: utf-8 --
2	#
3	# This file is part of REANA.
4	# Copyright (C) 2018 CERN.
5	#
6	# REANA is free software; you can redistribute it and/or modify it under the
7	# terms of the GNU General Public License as published by the Free Software
8	# Foundation; either version 2 of the License, or (at your option) any later
9	# version.
10	#
11	# REANA is distributed in the hope that it will be useful, but WITHOUT ANY
12	# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
13	# A PARTICULAR PURPOSE. See the GNU General Public License for more details.
14	#
15	# You should have received a copy of the GNU General Public License along with
16	# REANA; if not, write to the Free Software Foundation, Inc., 59 Temple Place,
17	# Suite 330, Boston, MA 02111-1307, USA.
18	#
19	# In applying this license, CERN does not waive the privileges and immunities
20	# granted to it by virtue of its status as an Intergovernmental Organization or
21	# submit itself to any jurisdiction.
22
23	"""REANA Workflow Controller MQ Consumer."""	×
24
25	import json	×
26
27	import pika	×
28	from reana_commons.consumer import Consumer	×
29	from reana_db.database import Session	×
30	from reana_db.models import WorkflowStatus	×
31
32	from .config import STATUS_QUEUE	×
33	from .tasks import (_update_job_cache, _update_job_progress,	×
34	_update_run_progress, _update_workflow_status)
35
36
37	class JobStatusConsumer(Consumer):	×
38	"""Consumer of jobs-status queue."""
39
40	def __init__(self):	×
41	"""Constructor."""
42	super(JobStatusConsumer, self).__init__(STATUS_QUEUE)	×
43
44	def on_message(self, channel, method_frame, header_frame, body):	×
45	"""On new message event handler."""
46	self._channel.basic_ack(delivery_tag=method_frame.delivery_tag)	×
47	body_dict = json.loads(body)	×
48	workflow_uuid = body_dict.get('workflow_uuid')	×
49	if workflow_uuid:	×
50	status = body_dict.get('status')	×
51	if status:	×
52	status = WorkflowStatus(status)	×
53	print(" [x] Received workflow_uuid: {0} status: {1}".	×
54	format(workflow_uuid, status))
55	logs = body_dict.get('logs') or ''	×
56	_update_workflow_status(workflow_uuid, status, logs)	×
57	if 'message' in body_dict and body_dict.get('message'):	×
58	msg = body_dict['message']	×
59	if 'progress' in msg:	×
60	_update_run_progress(workflow_uuid, msg)	×
61	_update_job_progress(workflow_uuid, msg)	×
62	# Caching: calculate input hash and store in JobCache
63	if 'caching_info' in msg:	×
64	_update_job_cache(msg)	×
65	Session.commit()	×
66
67	def consume(self):	×
68	"""Start consuming incoming messages."""
69	while True:	×
70	self.connect()	×
71	self._channel.basic_consume(self.on_message, STATUS_QUEUE)	×
72	try:	×
73	self._channel.start_consuming()	×
74	except KeyboardInterrupt:	×
75	self._channel.stop_consuming()	×
76	self._conn.close()	×
77	except pika.exceptions.ConnectionClosed:	×
78	# Uncomment this to make the example not attempt recovery
79	# from server-initiated connection closure, including
80	# when the node is stopped cleanly
81	# except pika.exceptions.ConnectionClosedByBroker:
82	# pass
83	continue	×

reanahub / reana-workflow-controller / 433

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous