reanahub / reana-server / 937

pending completion

Build # 937

Build Type

Pull #255

travis-ci

Committed by

web-flow

Commit Message

<a href="https://github.com/reanahub/reana-server/commit/<a class=hub.com/reanahub/reana-server/commit/<a class="double-link" href="https://git"><a class=hub.com/reanahub/reana-server/commit/<a class="double-link" href="https://git"><a class=hub.com/reanahub/reana-server/commit/c6381bd912c0ed55b2ddec749c1d74fea4857ef1">c6381bd91<a href="https://github.com/reanahub/reana-server/commit/c6381bd912c0ed55b2ddec749c1d74fea4857ef1">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/reanahub/reana-server/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/reanahub/reana-server/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/reanahub/reana-server/commit/c6381bd912c0ed55b2ddec749c1d74fea4857ef1&quot;&gt;c6381bd91&lt;/a&gt;&lt;a href=&quot;https://github.com/reanahub/reana-server/commit/c6381bd912c0ed55b2ddec749c1d74fea4857ef1&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/reanahub/reana-server/commit/c6381bd912c0ed55b2ddec749c1d74fe&lt;/a&gt;a4857ef1&quot;>&quot;&gt;Merge &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/reanahub/reana-server/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/reanahub/reana-server/commit/<a class="double-link" href="https://github.com/reanahub/reana-server/commit/1ed94f13aa526a884dbc54e1fc3ee2935e89ada1">1ed94f13a</a><a href="https://github.com/reanahub/reana-server/commit/c6381bd912c0ed55b2ddec749c1d74fea4857ef1">&amp;quot;&amp;gt;1ed94f13a&amp;lt;/a&amp;gt;&amp;quot;&amp;gt;1ed94f13a&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/reanahub/reana-server/commit/c6381bd912c0ed55b2ddec749c1d74fea4857&lt;/a&gt;ef1&quot;&gt; into &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/reanahub/reana-server/commit/<a class="double-link" href="https://github.com/reanahub/reana-server/commit/b912ee7bf">b912ee7bf">b912ee7bf</a>

Pull Request Pull Request #255: global: Black support

Run Details

342 of 342 new or added lines in 15 files covered. (100.0%)

707 of 1290 relevant lines covered (54.81%)

0.55 hits per line

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

83.33

/reana_server/scheduler.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 MIT License; see LICENSE file for more details.

"""REANA Server Workflow Execution Scheduler."""

import json
import logging

from bravado.exception import HTTPBadGateway, HTTPNotFound
from reana_commons.consumer import BaseConsumer
from reana_commons.tasks import reana_ready
from reana_db.database import Session
from reana_db.models import Workflow

from reana_server.api_client import (
    current_rwc_api_client,
    current_workflow_submission_publisher,
)


class WorkflowExecutionScheduler(BaseConsumer):
    """Scheduler of workflow execution.

    Class responsible for consuming from the workflow-submission queue
    and scheduling workflows for execution based on policies and system
    availability.
    """

    def __init__(self, **kwargs):
        """Initialise the WorkflowExecutionScheduler class."""
        super(WorkflowExecutionScheduler, self).__init__(
            queue="workflow-submission", **kwargs
        )

    def get_consumers(self, Consumer, channel):
        """Implement providing kombu.Consumers with queues/callbacks."""
        return [
            Consumer(
                queues=self.queue,
                callbacks=[self.on_message],
                accept=[self.message_default_format],
            )
        ]

    def requeue_workflow(self, **kwargs):
        """Send a workflow back to the queue.

        We do not use ``message.requeue()`` because it cannot be used after
        ``message.ack()``, and we cannot wait to validate all the checks
        (``reana_ready()`` and calling RWC) without calling ``message.ack()``
        and getting a new call to on_message with the same workflow.
        """
        try:
            current_workflow_submission_publisher.publish_workflow_submission(
                kwargs["user"], kwargs["workflow_id_or_name"], kwargs["parameters"]
            )
            logging.error(
                f"Requeueing workflow " f'{kwargs["workflow_id_or_name"]} ...'
            )
        except KeyError:
            logging.error(
                f"Wrong parameters to requeue workflow:\n"
                f"{kwargs}\n"
                f"Did reana_commons.publisher.WorkflowSubmissionPublisher's "
                f"method publish_workflow_submission change its signature?",
                exc_info=True,
            )
        except Exception:
            logging.error(
                "An error has occurred while requeueing worfklow", exc_info=True
            )

    def on_message(self, workflow_submission, message):
        """On new workflow_submission event handler."""
        message.ack()
        workflow_submission = json.loads(workflow_submission)
        if reana_ready():
            logging.info("Starting queued workflow: {}".format(workflow_submission))
            workflow_submission["status"] = "start"
            try:
                requeue = True
                started = False
                (
                    response,
                    http_response,
                ) = current_rwc_api_client.api.set_workflow_status(
                    **workflow_submission
                ).result()
                http_response_json = http_response.json()
                if http_response.status_code == 200:
                    started = True
                    logging.info(
                        f"Workflow "
                        f'{http_response_json["workflow_id"]} '
                        f"successfully started."
                    )
                else:
                    logging.error(
                        f"RWC returned an unexpected status code:\n"
                        f"{http_response_json}"
                    )

            except HTTPBadGateway as api_e:
                logging.error(
                    f"Workflow failed to start because "
                    f"RWC got an error while calling an external"
                    f"service (i.e. DB):\n"
                    f"{api_e}",
                    exc_info=True,
                )
            except HTTPNotFound as not_found_e:
                logging.error(
                    f"Workflow failed to start because "
                    f"workflow does not exist or was deleted \n"
                    f"{not_found_e}",
                    exc_info=True,
                )
                requeue = False
            except Exception as e:
                logging.error(
                    f"Something went wrong while calling RWC :\n" f"{e}", exc_info=True
                )
            finally:
                if not started and requeue:
                    self.requeue_workflow(**workflow_submission)
        else:
            logging.info(
                f"REANA not ready to run workflow "
                f'{workflow_submission["workflow_id_or_name"]}, '
                f"requeueing ..."
            )
            self.requeue_workflow(**workflow_submission)

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
7	# under the terms of the MIT License; see LICENSE file for more details.
8
9	"""REANA Server Workflow Execution Scheduler."""	1✔
10
11	import json	1✔
12	import logging	1✔
13
14	from bravado.exception import HTTPBadGateway, HTTPNotFound	1✔
15	from reana_commons.consumer import BaseConsumer	1✔
16	from reana_commons.tasks import reana_ready	1✔
17	from reana_db.database import Session	1✔
18	from reana_db.models import Workflow	1✔
19
20	from reana_server.api_client import (	1✔
21	current_rwc_api_client,
22	current_workflow_submission_publisher,
23	)
24
25
26	class WorkflowExecutionScheduler(BaseConsumer):	1✔
27	"""Scheduler of workflow execution.
28
29	Class responsible for consuming from the workflow-submission queue
30	and scheduling workflows for execution based on policies and system
31	availability.
32	"""
33
34	def __init__(self, **kwargs):	1✔
35	"""Initialise the WorkflowExecutionScheduler class."""
36	super(WorkflowExecutionScheduler, self).__init__(	1✔
37	queue="workflow-submission", **kwargs
38	)
39
40	def get_consumers(self, Consumer, channel):	1✔
41	"""Implement providing kombu.Consumers with queues/callbacks."""
42	return [	1✔
43	Consumer(
44	queues=self.queue,
45	callbacks=[self.on_message],
46	accept=[self.message_default_format],
47	)
48	]
49
50	def requeue_workflow(self, **kwargs):	1✔
51	"""Send a workflow back to the queue.
52
53	We do not use ``message.requeue()`` because it cannot be used after
54	``message.ack()``, and we cannot wait to validate all the checks
55	(``reana_ready()`` and calling RWC) without calling ``message.ack()``
56	and getting a new call to on_message with the same workflow.
57	"""
58	try:	1✔
59	current_workflow_submission_publisher.publish_workflow_submission(	1✔
60	kwargs["user"], kwargs["workflow_id_or_name"], kwargs["parameters"]
61	)
62	logging.error(	1✔
63	f"Requeueing workflow " f'{kwargs["workflow_id_or_name"]} ...'
64	)
65	except KeyError:	×
66	logging.error(	×
67	f"Wrong parameters to requeue workflow:\n"
68	f"{kwargs}\n"
69	f"Did reana_commons.publisher.WorkflowSubmissionPublisher's "
70	f"method publish_workflow_submission change its signature?",
71	exc_info=True,
72	)
73	except Exception:	×
74	logging.error(	×
75	"An error has occurred while requeueing worfklow", exc_info=True
76	)
77
78	def on_message(self, workflow_submission, message):	1✔
79	"""On new workflow_submission event handler."""
80	message.ack()	1✔
81	workflow_submission = json.loads(workflow_submission)	1✔
82	if reana_ready():	1✔
83	logging.info("Starting queued workflow: {}".format(workflow_submission))	1✔
84	workflow_submission["status"] = "start"	1✔
85	try:	1✔
86	requeue = True	1✔
87	started = False	1✔
88	(	1✔
89	response,
90	http_response,
91	) = current_rwc_api_client.api.set_workflow_status(
92	**workflow_submission
93	).result()
94	http_response_json = http_response.json()	1✔
95	if http_response.status_code == 200:	1✔
96	started = True	1✔
97	logging.info(	1✔
98	f"Workflow "
99	f'{http_response_json["workflow_id"]} '
100	f"successfully started."
101	)
102	else:
103	logging.error(	×
104	f"RWC returned an unexpected status code:\n"
105	f"{http_response_json}"
106	)
107
108	except HTTPBadGateway as api_e:	1✔
109	logging.error(	×
110	f"Workflow failed to start because "
111	f"RWC got an error while calling an external"
112	f"service (i.e. DB):\n"
113	f"{api_e}",
114	exc_info=True,
115	)
116	except HTTPNotFound as not_found_e:	1✔
117	logging.error(	×
118	f"Workflow failed to start because "
119	f"workflow does not exist or was deleted \n"
120	f"{not_found_e}",
121	exc_info=True,
122	)
123	requeue = False	×
124	except Exception as e:	1✔
125	logging.error(	1✔
126	f"Something went wrong while calling RWC :\n" f"{e}", exc_info=True
127	)
128	finally:
129	if not started and requeue:	1✔
130	self.requeue_workflow(**workflow_submission)	1✔
131	else:
132	logging.info(	1✔
133	f"REANA not ready to run workflow "
134	f'{workflow_submission["workflow_id_or_name"]}, '
135	f"requeueing ..."
136	)
137	self.requeue_workflow(**workflow_submission)	1✔

reanahub / reana-server / 937

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