minor fix for logging backups; fixes to mfnmetrics.py to use the workflow log index

Author: Istemi Ekin Akkus

FunctionWorker/python/PublicationUtils.py

@@ -550,10 +550,6 @@ def publish_output_direct(self, key, value_output, has_error, error_type, timest
 
             encoded_result = self.encode_output(result)
 
-            encapsulated_result = self.encapsulate_output(encoded_result, self._metadata)
-
-            dlc.putMapEntry(self._execution_info_map_name, "result_" + current_function_instance_id, encapsulated_result)
-
             # publish a message to the 'exit' topic
             trigger = {}
             trigger["next"] = self._wf_exit
@@ -572,11 +568,8 @@ def publish_output_direct(self, key, value_output, has_error, error_type, timest
 
             if self._should_checkpoint:
                 timestamp_map["t_start_dlcbackup"] = time.time() * 1000.0
-                dlc = self.get_backup_data_layer_client()
-
                 timestamp_map["t_start_resultmap"] = time.time() * 1000.0
-                dlc.putMapEntry(self._execution_info_map_name, "result_" + current_function_instance_id, encapsulated_value_output)
-                #self._logger.info("[__mfn_backup] [%s] [%s] %s", self._execution_info_map_name, "result_" + current_function_instance_id, encapsulated_value_output)
+                self._logger.info("[__mfn_backup] [%s] [%s] %s", self._execution_info_map_name, "result_" + current_function_instance_id, encapsulated_value_output)
 
             timestamp_map["t_start_storeoutput"] = time.time() * 1000.0
             # store self._sapi.transient_output into the data layer

LoggingService/elasticsearch/mfnmetrics.py

@@ -26,9 +26,9 @@
 import re
 
 NGINX_INDEX= 'mfnnx'
-COMPONENT_INDEXES= 'mfnfe,mfnqs,mfnwf,mfnnx'
+COMPONENT_INDEXES= 'mfnfe,mfnqs,mfnnx'
 
-def get_metric_logs_from_all_components(eshost, esport=9200, uuids=None, proxies=None, debugReadFromFile=False):
+def get_metric_logs_from_all_components(workflow_id, eshost, esport=9200, uuids=None, proxies=None, debugReadFromFile=False):
     # Form elasticsearch query
     status, request = form_query(uuids)
     if status == False:
@@ -37,7 +37,7 @@ def get_metric_logs_from_all_components(eshost, esport=9200, uuids=None, proxies
 
     # Execute the query
     if debugReadFromFile == False:
-        status, output = query_elasticsearch(eshost, esport, request, proxies)
+        status, output = query_elasticsearch(workflow_id, eshost, esport, request, proxies)
         if status == False:
             logging.debug("query_elasticsearch error: " + output)
             return False, output
@@ -61,7 +61,7 @@ def get_metric_logs_from_all_components(eshost, esport=9200, uuids=None, proxies
             status, output = handleFrontendLogline(hit)
         elif index == 'mfnqs':
             status, output = handleQueueServiceLogline(hit)
-        elif index == 'mfnwf':
+        elif index == 'mfnwf-' + workflow_id:
             status, output = handleWorkflowTraceLogline(hit)
         elif index == 'mfnnx':
             status, output = handleNginxLogline(hit)
@@ -731,12 +731,12 @@ def handleNginxLogline(hit):
     },
     '''
 
-def query_elasticsearch(eshost, esport, request, proxies):
+def query_elasticsearch(wid, eshost, esport, request, proxies):
     url="http://"+eshost+":" + str(esport)
     logging.debug("Url: " + url)
 
     try:
-        r=requests.get(url+"/"+COMPONENT_INDEXES+'/_search', json=request, proxies=proxies)
+        r=requests.get(url+"/"+COMPONENT_INDEXES+ ",mfnwf-" + wid + '/_search', json=request, proxies=proxies)
 
         logging.debug('Http response code: ' + str(r.status_code))
         logging.debug('Http status reason: ' + r.reason)
@@ -778,6 +778,7 @@ def printlog(outlog):
 
 def main():
     parser = argparse.ArgumentParser(description='Generate overhead metrics and timestamps for microfunctions workflow executions', prog='mfnmetrics.py')
+    parser.add_argument('-wid', '--wid', metavar='WORKFLOW_ID', help='Workflow id to determine the index.')
     parser.add_argument('-eid', '--eid', nargs='+', metavar='EXECUTION_UUIDS', help='Generate metric for specific execution id(s).')
     parser.add_argument('-eidfile', '--eidfile', type=str, help="Read execution ids from the given file (with 1 uuid per line); overrides -eid option")
     parser.add_argument('-eshost', '--eshost', type=str, metavar='ELASTICSEARCH_HOST', default=socket.gethostname(), help='Elasticsearch host. Defaults to ' + socket.gethostname() + ':9200.')
@@ -789,6 +790,8 @@ def main():
 
     args = parser.parse_args()
 
+    workflow_id = args.wid
+
     eidfilename = args.eidfile
     if eidfilename is None:
         uuids = args.eid
@@ -820,6 +823,7 @@ def main():
         "https": https_proxy,
     }
 
+    logging.debug("Workflow id: " + str(workflow_id))
     logging.debug("Execution uuid(s): " + str(uuids))
     logging.debug("Elasticsearch host: " + str(eshost)+ ':9200')
     logging.debug("Proxies: " + str(proxies))
@@ -828,7 +832,7 @@ def main():
     if proxy == False:
         proxies = None
 
-    status, result = get_metric_logs_from_all_components(eshost, esport=9200, uuids=uuids, proxies=proxies, debugReadFromFile=debugReadFromFile)
+    status, result = get_metric_logs_from_all_components(workflow_id, eshost, esport=9200, uuids=uuids, proxies=proxies, debugReadFromFile=debugReadFromFile)
     if status == True:
         printlog(result)
     else:

tests/mfn_test_utils.py

@@ -498,7 +498,7 @@ def plot_latency_breakdown(self, num_last_executions=15):
         _, _ = run_command_return_output(cmd)
 
     def parse_metrics(self, eid_filename, timestamps_filename):
-        cmd = "python3 ../mfnmetrics.py -eidfile " + eid_filename
+        cmd = "python3 ../mfnmetrics.py -eidfile " + eid_filename + " -wid " + self._workflow.id
         output, error = run_command_return_output(cmd)
         log_lines = combine_output(output, error)
         with open(timestamps_filename, "w") as f:

tests/performance_interaction_latency/test.py

@@ -66,18 +66,24 @@ def test_chain_response_latency(self):
         print("----------------")
         print("Checkpoints: False")
 
-        test.exec_tests(self._test_tuple_list, check_duration=True, print_report=False)
+        test.exec_tests(self._test_tuple_list, check_duration=True, print_report=False, should_undeploy=False)
 
         #test.plot_latency_breakdown(COUNT_EXECUTIONS)
 
+        test.undeploy_workflow()
+        test.cleanup()
+
         test = MFNTest(test_name='chain_checkpoints', workflow_filename='wf_chain_checkpoints.json')
         print("----------------")
         print("Checkpoints: True")
 
-        test.exec_tests(self._test_tuple_list, check_duration=True, print_report=False)
+        test.exec_tests(self._test_tuple_list, check_duration=True, print_report=False, should_undeploy=False)
 
         #test.plot_latency_breakdown(COUNT_EXECUTIONS)
 
+        test.undeploy_workflow()
+        test.cleanup()
+
     def _get_and_print_statistics(self, test, logs, checkpoints_on):
         log = logs["log"]
         log_lines = log.split("\n")