Implements pipeline process opening in the execute_pipeline API endpoint

src/service/clmcservice/graphapi/tests.py

@@ -24,10 +24,10 @@
 
 from json import dumps
 import pytest
-from unittest.mock import patch, Mock
+from unittest.mock import patch, Mock, PropertyMock
 from pyramid import testing
 from clmcservice.graphapi.views import GraphAPI
-from pyramid.httpexceptions import HTTPBadRequest, HTTPNotFound
+from pyramid.httpexceptions import HTTPBadRequest, HTTPNotFound, HTTPInternalServerError
 
 
 class TestGraphAPI(object):
@@ -497,7 +497,7 @@ class TestGraphAPI(object):
 
     @patch('clmcservice.graphapi.views.Popen')
     @patch('clmcservice.graphapi.views.uuid4')
-    def test_execute_pipeline_graph(self, uuid_mock, popen_mock):
+    def test_execute_graph_pipeline(self, uuid_mock, popen_mock):
         """
         Tests the functionality to start a pipeline script executing the graph API workflow - build, query, delete
         """
@@ -506,10 +506,14 @@ class TestGraphAPI(object):
         uuid_mock.return_value = "monitor_test_uuid1"
 
         # mock the behaviour of the Popen class
+        pid_property_mock = PropertyMock(return_value=111)
+        returncode_property_mock = PropertyMock(return_value=None)
         popen_intance_mock = Mock()
-        popen_intance_mock.pid = Mock(return_value=111)
+        type(popen_intance_mock).pid = pid_property_mock  # a property mock cannot be attached directly to the mock object, hence use its type object
+        type(popen_intance_mock).returncode = returncode_property_mock  # a property mock cannot be attached directly to the mock object, hence use its type object
         popen_mock.return_value = popen_intance_mock
 
+        # check proper behaviour
         service_functions = dict(nginx={"measurement_name": "nginx", "response_time_field": "mean(avg_processing_time)",
                                         "request_size_field": "mean(avg_request_size)", "response_size_field": "mean(avg_response_size)"},
                                  minio={"measurement_name": "minio_http", "response_time_field": "mean(total_processing_time)/mean(total_requests_count)",
@@ -523,8 +527,25 @@ class TestGraphAPI(object):
         response = GraphAPI(request).execute_graph_pipeline()
         assert response == {"uuid": uuid_mock.return_value, "database": "test_sfc"}
 
-        popen_mock.assert_called_with("graph_pipeline.sh", body)  # assert that the graph pipeline script is ran with the JSON config that was received in the request
-        popen_intance_mock.pid.assert_called()  # assert that the process ID attribute was called and saved
+        popen_mock.assert_called_once_with(["./graph_pipeline.sh", body])  # assert that the graph pipeline script is ran with the JSON config that was received in the request
+        pid_property_mock.assert_called_once_with()  # assert that the process ID attribute was called and saved
+        returncode_property_mock.assert_called_once_with()  # assert that the process return code attribute was called to check if the process has started successfully
+
+        # check erroneous behaviour
+        returncode_property_mock.return_value = -1
+        request = testing.DummyRequest()
+        request.body = body.encode(request.charset)
+        error_raised = False
+        try:
+            GraphAPI(request).execute_graph_pipeline()
+        except HTTPInternalServerError:
+            error_raised = True
+        assert error_raised, "Expecting a 500 HTTP error if the process terminated immediately after it was started"
+
+        popen_mock.assert_called_with(["./graph_pipeline.sh", body])  # assert that the graph pipeline script is ran with the JSON config that was received in the request
+        pid_property_mock.assert_called_with()  # assert that the process ID attribute was called and saved
+        returncode_property_mock.assert_called_with()  # assert that the process return code attribute was called to check if the process has started successfully
+
 
     @staticmethod
     def check_exist_relationship(relationships_tuple, graph, uuid):

src/service/clmcservice/graphapi/views.py

@@ -27,11 +27,12 @@ from clmcservice.graphapi.utilities import validate_build_request_body, validate
     build_network_graph, delete_network_graph, build_temporal_subgraph, delete_temporal_subgraph, validate_graph_rtt_params, find_node_with_possible_types
 from influxdb import InfluxDBClient
 from py2neo import Graph
-from pyramid.httpexceptions import HTTPBadRequest, HTTPNotFound, HTTPServiceUnavailable, HTTPNotImplemented
+from pyramid.httpexceptions import HTTPBadRequest, HTTPNotFound, HTTPServiceUnavailable, HTTPNotImplemented, HTTPInternalServerError
 from pyramid.view import view_defaults, view_config
 from requests import exceptions, get
 from uuid import uuid4
 from json import load
+from subprocess import Popen
 import logging
 
 
@@ -359,7 +360,15 @@ class GraphAPI(object):
             raise HTTPBadRequest("Database for service function chain {0} not found.".format(database_name))
 
         request_uuid = str(uuid4())
-
-        # TODO start a background process running the pipeline script
-
-        return {"database": database_name, "uuid": request_uuid}
+        sfc = json_queries["service_function_chain"]
+
+        process = Popen(["./graph_pipeline.sh", body])
+        process_pid = process.pid
+        process_return_code = process.returncode
+
+        if process_return_code is None:  # process has started running
+            log.info("Started a graph pipeline process for SFC {0} with PID {1}".format(sfc, process_pid))
+            return {"database": database_name, "uuid": request_uuid}
+        else:  # a valid returned code was returned, hence the process has terminated one way or another - we do not expect this since the pipeline script must be continuously running
+            log.warning("Graph pipeline process for SFC {0} with PID {1} has finished executing unexpectedly with return code {2}".format(sfc, process_pid, process_return_code))
+            raise HTTPInternalServerError("An unexpected error occurred while trying to start monitoring graph measurements for service function chain {0}".format(sfc))