Release/0.8.3

FunctionWorker/python/FunctionWorker.py

@@ -330,10 +330,22 @@ def _fork_and_handle_message(self, key, encapsulated_value):
 
                 # 3. Apply InputPath, if available
                 timestamp_map["t_start_inputpath"] = time.time() * 1000.0
+                #self._logger.debug("[FunctionWorker] Before Path/Parameters processing, input: " + str(type(raw_state_input)) + " : " + str(raw_state_input) + ", metadata: " + str(metadata) + " has_error: " + str(has_error))
                 if not has_error:
                     try:
-                        function_input = self._state_utils.applyInputPath(raw_state_input)
-                        #self._logger.debug("[FunctionWorker] User code input(After InputPath processing):" + str(type(function_input)) + ":" + str(function_input))
+                        if "__state_action" not in metadata or (metadata["__state_action"] != "post_map_processing" and metadata["__state_action"] != "post_parallel_processing"):
+                             #self._logger.debug("[FunctionWorker] User code input(Before InputPath processing):" + str(type(raw_state_input)) + ":" + str(raw_state_input))
+                             function_input = self._state_utils.applyInputPath(raw_state_input)
+                             #self._logger.debug("[FunctionWorker] User code input(Before applyParameter processing):" + str(type(function_input)) + ":" + str(function_input))
+                             function_input = self._state_utils.applyParameters(function_input)
+                             #self._logger.debug("[FunctionWorker] User code input(Before ItemsPath processing):" + str(type(function_input)) + ":" + str(function_input))
+                             function_input = self._state_utils.applyItemsPath(function_input) # process map items path
+
+                        #elif "Action" not in metadata or metadata["Action"] != "post_parallel_processing":
+                        #     function_input = self._state_utils.applyInputPath(raw_state_input)
+
+                        else:
+                             function_input = raw_state_input
                     except Exception as exc:
                         self._logger.exception("InputPath processing exception: %s\n%s", str(instance_pid), str(exc))
                         error_type = "InputPath processing exception"
@@ -592,6 +604,7 @@ def _get_and_handle_message(self):
 
     def run(self):
         self._is_running = True
+
         self._logger.info("[FunctionWorker] Started:" \
             + self._function_state_name \
             + ", user: " + self._userid \

FunctionWorker/python/PublicationUtils.py

@@ -356,6 +356,13 @@ def _generate_trigger_metadata(self, topic_next):
 
         output_instance_id = self._output_counter_map[topic_next]
         next_function_execution_id = self._metadata["__function_execution_id"] + "_" + str(output_instance_id)
+
+        # get current state type. if map state add marker to execution Id
+        state_type = self._state_utils.functionstatetype
+        self._logger.debug("self._state_utils.functionstatetype: " + str(state_type))
+
+        if state_type == 'Map':
+            next_function_execution_id = self._metadata["__function_execution_id"] + "_" + str(output_instance_id)+"-M"
         self._output_counter_map[topic_next] += 1
 
         trigger_metadata = copy.deepcopy(self._metadata)

GUI/.gitignore

@@ -0,0 +1,2 @@
+app/pages/docs/sdk/sdk.md
+app/pages/docs/sdk/cli.md

GUI/Makefile

@@ -23,7 +23,9 @@ app/pages/docs/sdk/cli.md: ../mfn_cli/README.md
 -include ../proxy.mk
 include ../docker.mk
 
-image: $(shell find ./ -type f|grep -v Makefile)
+image: $(shell find ./ -type f|grep -v -e Makefile -e app/pages/docs/sdk/sdk.md -e app/pages/docs/sdk/cli.md) \
+	app/pages/docs/sdk/cli.md \
+	app/pages/docs/sdk/sdk.md
 	$(call build_image,Dockerfile,microfn/nginx)
 
 push: image

GUI/app/pages/workflows/WorkflowEditorCtrl.js

@@ -31,6 +31,9 @@
      var fail = $.getJSON("lib/asl-validator/schemas/fail.json", function(json) {
      });
 
+     var mapState = $.getJSON("lib/asl-validator/schemas/map.json", function(json) {
+     });
+
      var parallel = $.getJSON("lib/asl-validator/schemas/parallel.json", function(json) {
      });
 
@@ -1209,6 +1212,7 @@
                 choice.responseJSON,
                 fail.responseJSON,
                 parallel.responseJSON,
+		mapState.responseJSON,
                 pass.responseJSON,
                 stateMachine.responseJSON,
                 state.responseJSON,

GUI/lib/asl-validator/schemas/map.json

@@ -0,0 +1,95 @@
+{
+  "$id": "http://asl-validator.cloud/map#",
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "type": "object",
+  "properties": {
+    "Type": {
+      "type": "string",
+      "pattern": "^Map$"
+    },
+    "Next": {
+      "type": "string"
+    },
+    "End": {
+      "enum": [true]
+    },
+    "Comment": {
+      "type": "string"
+    },
+    "OutputPath": {
+      "type": ["string", "null"]
+    },
+    "InputPath": {
+      "type": ["string", "null"]
+    },
+    "ResultPath": {
+      "type": ["string", "null"]
+    },
+    "ItemsPath": {
+      "type": ["string", "null"]
+    },
+    "MaxConcurrency": {
+      "type": "number",
+      "minimum": 0
+    },
+    "Iterator": {
+      "$ref": "http://asl-validator.cloud/state-machine#"
+    },
+    "Parameters": {
+      "type": "object"
+    },
+    "Retry": {
+      "type": "array",
+      "items": {
+        "types": "object",
+        "properties": {
+          "ErrorEquals": {
+            "type": "array",
+            "items": {
+              "types": "string"
+            }
+          },
+          "IntervalSeconds": {
+            "type": "number",
+            "minimum": 0
+          },
+          "MaxAttempts": {
+            "type": "number",
+            "minimum": 0
+          },
+          "BackoffRate": {
+            "type": "number",
+            "minimum": 0
+          }
+        },
+        "required": ["ErrorEquals"]
+      }
+    },
+    "Catch": {
+      "type": "array",
+      "items": {
+        "types": "object",
+        "properties": {
+          "ErrorEquals": {
+            "type": "array",
+            "items": {
+              "types": "string"
+            }
+          },
+          "Next": {
+            "type": "string"
+          }
+        },
+        "required": ["ErrorEquals", "Next"]
+      }
+    }
+  },
+  "oneOf": [{
+    "required": ["Next"]
+  }, {
+    "required": ["End"]
+  }],
+  "required": ["Type", "Iterator"],
+  "additionalProperties": false
+}
+

GUI/lib/asl-validator/schemas/pass.json

@@ -25,6 +25,9 @@
     "ResultPath": {
       "type": "string"
     },
+    "Parameters": {
+      "type": "object"
+    },
     "Result": {}
   },
   "oneOf": [{

GUI/lib/asl-validator/schemas/state.json

@@ -11,6 +11,9 @@
     {
       "$ref": "http://asl-validator.cloud/parallel#"
     },
+    {
+      "$ref": "http://asl-validator.cloud/map#"
+    },
     {
       "$ref": "http://asl-validator.cloud/pass#"
     },

ManagementService/Makefile

@@ -28,7 +28,7 @@ thrift: $(THRIFT)
 
 $(THRIFT): ../DataLayerService/thrift/DataLayerMessage.thrift ../DataLayerService/thrift/DataLayerService.thrift
 	mkdir -p data_layer
-	docker run --rm -v $(CURDIR)/..:/root -w /root thrift:0.12.0 bash -c '\
+	docker run --user $$(id -u):$$(id -g) --rm -v $(CURDIR)/..:/root -w /root thrift:0.12.0 bash -c '\
 		thrift --gen py -out ManagementService/ DataLayerService/thrift/DataLayerMessage.thrift; \
 		thrift --gen py -out ManagementService/ DataLayerService/thrift/DataLayerService.thrift'
 

ManagementService/management_init.py

@@ -74,6 +74,7 @@ def parse_states(state_map):
 
         if stype == "Task":
             functions.append(wfs["Resource"])
+
         elif stype == 'Parallel':
             # add the parallel state function worker
             states.append({'name':sresource,'type':stype})
@@ -83,7 +84,18 @@ def parse_states(state_map):
                 # add found functions to list of function workers
                 functions.extend(sub_functions)
                 states.extend(sub_states)
-        elif stype in {'Choice', 'Pass', 'Wait', 'Fail', 'Succeed'}: #, 'Parallel'}:
+
+        elif stype == 'Map':
+            # add the Map state iterator function worker
+            states.append({'name':sresource,'type':stype})
+            # find recursively everything that is in the branch
+            branch = state_map[sresource]['Iterator']
+            sub_functions, sub_states = parse_states(branch['States'])
+            # add found functions to list of function workers
+            functions.extend(sub_functions)
+            states.extend(sub_states)
+
+        elif stype in {'Choice', 'Pass', 'Wait', 'Fail', 'Succeed'}: 
             states.append({'name':sresource,'type':stype})
         else:
             raise Exception("Unknown state type: " + stype)

ManagementService/python/deleteWorkflow.py

@@ -39,7 +39,7 @@ def handle(value, sapi):
                     wf = sapi.get(email + "_workflow_" + workflow["id"], True)
                     if wf is not None and wf != "":
                         wf = json.loads(wf)
-                        if wf["status"] == "undeployed":
+                        if wf["status"] == "undeployed" or wf["status"] == "failed":
                             for wn in workflows:
                                 if workflows[wn] == workflow["id"]:
                                     del workflows[wn]

ManagementService/python/deployWorkflow.py

@@ -48,6 +48,22 @@ def check_workflow_functions(wf_type, wfobj, email, sapi):
                     for state in list(branches['States'].keys()): # state is the key
                         wf_state_map[state] = branches['States'][state] # add the state to the state map root
 
+            if wf_state_map[state_names]["Type"] == "Map":
+                mapStateName = state_names
+                iterator = wf_state_map[mapStateName]['Iterator'] # this is a dict 
+
+                states_dict = iterator['States'] # this is a also dict
+                print (json.dumps(states_dict))
+                for state in states_dict.keys():
+                    print ("FOUND MAP STATE: "+str(state))
+                    wf_state_map[state] = states_dict[state]
+
+                """
+                for iterators in wf_state_map[mapStateName]['Iterator']:
+                    for state in list(iterators['States'].keys()): # state is the key
+                        wf_state_map[state] = iterators['States'][state] # add the state to the state map root
+                """
+
         for wfsname in wf_state_map:
             wfs = wf_state_map[wfsname]
             if wfs["Type"] == "Task":

ManagementService/python/getWorkflows.py

@@ -138,9 +138,9 @@ def handle(value, sapi):
     try:
         workflows = sapi.get(email + "_list_workflows", True)
         if workflows is None or workflows == "":
-            raise Exception("Couldn't retrieve workflow status; no such workflow.")
-
-        workflows = json.loads(workflows)
+            workflows = {}
+        else:
+            workflows = json.loads(workflows)
 
         # get single workflow status
         if "workflow" in data and "id" in data["workflow"]: