Adds API endpoint to continuously execute the graph workflow - implements...

Adds API endpoint to continuously execute the graph workflow - implements error handling functionality only

src/service/clmcservice/__init__.py

@@ -70,6 +70,7 @@ def main(global_config, **settings):
     config.add_route('graph_manage', '/graph/temporal/{graph_id}')
     config.add_route('graph_algorithms_rtt', '/graph/temporal/{graph_id}/round-trip-time')
     config.add_route('graph_network_topology', '/graph/network')
+    config.add_route('graph_execute_pipeline', '/graph/monitor')
 
     # add routes of the Alerts Configuration API
     config.add_route('alerts_configuration', '/alerts')

src/service/clmcservice/graphapi/utilities.py

@@ -29,6 +29,8 @@ import logging
 
 GRAPH_ROUND_TRIP_TIME_URL_PARAMS = ("startpoint", "endpoint")
 
+GRAPH_MONITOR_QUERY_PARAMS = {"query_period", "results_measurement_name", "service_function_chain", "service_function_chain_instance", "service_functions"}
+
 GRAPH_BUILD_QUERY_PARAMS = {"from", "to", "service_function_chain", "service_function_chain_instance", "service_functions"}
 GRAPH_BUILD_SF_QUERY_PARAMS = {"response_time_field", "request_size_field", "response_size_field", "measurement_name"}
 
@@ -61,9 +63,9 @@ RETURN latencies  as forward_latencies, reverse(latencies) as reverse_latencies,
 log = logging.getLogger('service_logger')
 
 
-def validate_json_queries_body(body):
+def validate_build_request_body(body):
     """
-    Validates the request body containing mappings from service functions to queries to execute.
+    Validates the request body, with mappings from service functions to queries to execute, given in a request to build temporal graph.
 
     :param body: the request body to validate
 
@@ -72,7 +74,7 @@ def validate_json_queries_body(body):
     :raise AssertionError: if the body is invalid
     """
 
-    global GRAPH_BUILD_QUERY_PARAMS
+    global GRAPH_BUILD_QUERY_PARAMS, GRAPH_BUILD_SF_QUERY_PARAMS
 
     try:
         body = loads(body)
@@ -92,7 +94,7 @@ def validate_json_queries_body(body):
     # except ValueError:
     #     assert False, "Incorrect format of service function chain instance ID - use format <sfcID>_<instanceNum>"
 
-    assert type(body["service_functions"]) == dict, "The service function description should be represented with a dictionary."
+    assert type(body["service_functions"]) == dict, "The service functions description should be represented with a JSON object."
 
     for sf in body["service_functions"]:
         query_data = body["service_functions"][sf]
@@ -107,6 +109,39 @@ def validate_json_queries_body(body):
     return body
 
 
+def validate_monitor_request_body(body):
+    """
+    Validates the request body, with mappings from service functions to queries to execute, given in a request to monitor graph metrics.
+
+    :param body: the request body to validate
+
+    :return the validated json queries dictionary object
+
+    :raise AssertionError: if the body is invalid
+    """
+
+    global GRAPH_MONITOR_QUERY_PARAMS, GRAPH_BUILD_SF_QUERY_PARAMS
+
+    try:
+        body = loads(body)
+    except Exception:
+        raise AssertionError("Configuration must be a JSON object.")
+
+    assert GRAPH_MONITOR_QUERY_PARAMS == set(body.keys()), "Invalid JSON query document."
+
+    assert type(body["service_functions"]) == dict, "The service functions description should be represented with a JSON object."
+
+    for sf in body["service_functions"]:
+        query_data = body["service_functions"][sf]
+        assert type(query_data) == dict, "Each service function must be associated with a respective JSON object."
+        assert GRAPH_BUILD_SF_QUERY_PARAMS == set(query_data.keys()), "Invalid query data for service function {0} in the JSON query document".format(sf)
+
+    assert type(body["query_period"]) == int, "'query_period' parameter must be an integer"
+    assert body["query_period"] > 0, "'query_period' parameter must be a positive integer."
+
+    return body
+
+
 def validate_graph_rtt_params(params):
     """
     Validates the request url parameters used in running a round trip time cypher query.

src/service/clmcservice/graphapi/views.py

@@ -23,7 +23,7 @@
 """
 
 
-from clmcservice.graphapi.utilities import validate_json_queries_body, RTT_CYPHER_QUERY_TEMPLATE, \
+from clmcservice.graphapi.utilities import validate_build_request_body, validate_monitor_request_body, RTT_CYPHER_QUERY_TEMPLATE, \
     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
@@ -64,7 +64,7 @@ class GraphAPI(object):
 
         try:
             body = self.request.body.decode(self.request.charset)
-            json_queries = validate_json_queries_body(body)  # validate the content and receive a json dictionary object
+            json_queries = validate_build_request_body(body)  # validate the content and receive a json dictionary object
         except AssertionError as e:
             raise HTTPBadRequest("Bad request content: {0}".format(e.args))
 
@@ -334,3 +334,21 @@ class GraphAPI(object):
         deleted_switches, deleted_clusters, deleted_ues = delete_network_graph(graph)
 
         return {"deleted_switches_count": deleted_switches, "deleted_clusters_count": deleted_clusters, "deleted_ues_count": deleted_ues}
+
+    @view_config(route_name='graph_execute_pipeline', request_method='POST')
+    def execute_graph_pipeline(self):
+
+        try:
+            body = self.request.body.decode(self.request.charset)
+            json_queries = validate_monitor_request_body(body)  # validate the content and receive a json dictionary object
+        except AssertionError as e:
+            raise HTTPBadRequest("Bad request content: {0}".format(e.args))
+
+        graph = Graph(host=self.request.registry.settings['neo4j_host'], password=self.request.registry.settings['neo4j_password'])
+        influx_client = InfluxDBClient(host=self.request.registry.settings['influx_host'], port=self.request.registry.settings['influx_port'], timeout=10)
+
+        database_name = json_queries["service_function_chain"]
+        if database_name not in [db["name"] for db in influx_client.get_list_database()]:
+            raise HTTPBadRequest("Database for service function chain {0} not found.".format(database_name))
+
+        return json_queries