Merge pull request #91 from knix-microfunctions/release/0.8.7

Release/0.8.7

Author: Istemi Ekin Akkus

DataLayerService/.gitignore

@@ -4,3 +4,4 @@
 /target/
 /.settings
 /.factorypath
+/.m2/

DataLayerService/Makefile

@@ -24,29 +24,37 @@ all: build_thrift \
 build_thrift:
 	cd ../build_tools/thrift/$(THRIFT_VERSION)/; ./build_thrift.sh; cd -
 
-src/main/java/org/microfunctions/data_layer/DataLayerService.java: ../build_env.mk \
-	thrift/DataLayerMessage.thrift \
-	thrift/DataLayerService.thrift
-	docker run --rm -it --name datalayer-thrift \
+src/main/java/org/microfunctions/data_layer/DataLayerService.java: \
+		../build_env.mk \
+		../build_tools/thrift/$(THRIFT_VERSION)/build_thrift.sh \
+		thrift/DataLayerMessage.thrift \
+		thrift/DataLayerService.thrift
+	cd ../build_tools/thrift/$(THRIFT_VERSION)/; ./build_thrift.sh; cd -
+	docker run --rm -i --name datalayer-thrift \
 		--user $(UID):$(GID) \
 		-v $(CURDIR):/usr/src/datalayer -w /usr/src/datalayer \
 		$(THRIFT_IMAGE_NAME):$(THRIFT_VERSION) \
 		bash -c 'thrift --gen java -out ./src/main/java/ ./thrift/DataLayerMessage.thrift; thrift --gen java -out ./src/main/java/ ./thrift/DataLayerService.thrift'
 	touch src/main/java/org/microfunctions/data_layer/DataLayerService.java
 
-PythonClient/data_layer: ../build_env.mk \
-	thrift/DataLayerMessage.thrift \
-	thrift/DataLayerService.thrift
-	docker run --rm -it --name datalayer-thrift \
+PythonClient/data_layer: \
+		../build_env.mk \
+		../build_tools/thrift/$(THRIFT_VERSION)/build_thrift.sh \
+		thrift/DataLayerMessage.thrift \
+		thrift/DataLayerService.thrift
+	cd ../build_tools/thrift/$(THRIFT_VERSION)/; ./build_thrift.sh; cd -
+	docker run --rm -i --name datalayer-thrift \
 		--user $(UID):$(GID) \
 		-v $(CURDIR):/usr/src/datalayer -w /usr/src/datalayer \
 		$(THRIFT_IMAGE_NAME):$(THRIFT_VERSION) \
 		bash -c 'thrift --gen py -out ./PythonClient ./thrift/DataLayerMessage.thrift; thrift --gen py -out ./PythonClient ./thrift/DataLayerService.thrift'
 
-# In case the user doesn't have maven configurations in the home directory, we need to create it
-~/.m2/settings-docker.xml: settings-docker.xml.sample
-	if [ ! -e ~/.m2 ]; then mkdir ~/.m2; fi
+# Always create maven configuration and repo
+.PHONY: .m2/settings-docker.xml
+.m2/settings-docker.xml:
+	if [ ! -e .m2 ]; then mkdir .m2; fi
 	cp settings-docker.xml.sample $@.tmp; \
+	sed "s#<localRepository>[^<]\+</localRepository>#<localRepository>/usr/src/datalayer/.m2/repository</localRepository>#" -i $@.tmp; \
 	noproxyhosts=$$(echo $$no_proxy|sed 's/,/|/g'); \
 	http_proxy=$${http_proxy:-$${HTTP_PROXY}}; \
 	https_proxy=$${https_proxy:-$${HTTPS_PROXY}}; \
@@ -62,15 +70,14 @@ PythonClient/data_layer: ../build_env.mk \
 JAVA_SOURCES=$(shell find src/ -type f|grep -v src/main/java/org/microfunctions/data_layer/DataLayerService.java)
 GIT_REVISION:=`git describe --tags --always --dirty=-dirty`
 target/datalayerservice.jar: $(JAVA_SOURCES) \
-	src/main/java/org/microfunctions/data_layer/DataLayerService.java \
-	~/.m2/settings-docker.xml
-	if [ ! -e ~/.m2/repository ]; then mkdir -p ~/.m2/repository; fi
-	docker run --rm -it --name datalayer-maven \
+		src/main/java/org/microfunctions/data_layer/DataLayerService.java \
+		.m2/settings-docker.xml
+	if [ ! -e .m2/repository ]; then mkdir -p .m2/repository; fi
+	docker run --rm -i --name datalayer-maven \
 		--user $(UID):$(GID) \
-		-v ~/.m2:/usr/share/maven/ref \
-		-v $(CURDIR):/usr/src/datalayer -w /usr/src/datalayer \
+		-e HOME=/usr/src/datalayer -v $(CURDIR):/usr/src/datalayer -w /usr/src/datalayer \
 		maven:3.6.1-jdk-8 \
-		mvn -Drevision=$(GIT_REVISION) -Dlibthrift.version=$(MVN_LIBTHRIFT_VERSION) -DargLine="-Duser.home=/tmp" -s /usr/share/maven/ref/settings-docker.xml package
+		mvn -Drevision=$(GIT_REVISION) -Dlibthrift.version=$(MVN_LIBTHRIFT_VERSION) -DargLine="-Duser.home=/tmp" -s /usr/src/datalayer/.m2/settings-docker.xml package
 
 clean:
 	rm -rf ./target

DataLayerService/src/main/java/org/microfunctions/data_layer/RiakAccess.java

@@ -74,10 +74,10 @@ public class RiakAccess {
     private static final Logger LOGGER = LogManager.getLogger(RiakAccess.class);
 
 	public static String BUCKET_TYPE_DEFAULT = "default";
-    public static final String BUCKET_TYPE_STRONG_CONSISTENCY = "strong"; // make sure there is a Riak bucket type called "strong" for strong consistency.
-	public static final String BUCKET_TYPE_TRIGGERS = "triggers"; // make sure there is a Riak bucket type called "triggers".
+    //public static final String BUCKET_TYPE_STRONG_CONSISTENCY = "strong"; // make sure there is a Riak bucket type called "strong" for strong consistency.
+	// public static final String BUCKET_TYPE_TRIGGERS = "triggers"; // make sure there is a Riak bucket type called "triggers".
 	public static final String BUCKET_TYPE_COUNTERS = "counters";	// make sure there is a Riak bucket type called "counters" with its data type being "counter".
-    public static final String BUCKET_TYPE_TRIGGERABLE_COUNTERS = "mfn_counter_trigger"; // make sure there is a Riak bucket type called "mfn_counter_trigger".
+    // public static final String BUCKET_TYPE_TRIGGERABLE_COUNTERS = "mfn_counter_trigger"; // make sure there is a Riak bucket type called "mfn_counter_trigger".
 	public static final String BUCKET_TYPE_SETS = "sets";	// make sure there is a Riak bucket type called "sets" with its data type being "set".
 	public static final String BUCKET_TYPE_MAPS = "maps";	// make sure there is a Riak bucket type called "maps" with its data type being "map".
 	public static final String BUCKET_TYPE_ALL = "all";
@@ -119,28 +119,28 @@ public void connect (Map<String,Integer> riakNodes) {
 		cluster.start();
 
         ALL_BUCKET_TYPES.add(BUCKET_TYPE_DEFAULT);
-        ALL_BUCKET_TYPES.add(BUCKET_TYPE_STRONG_CONSISTENCY);
-		ALL_BUCKET_TYPES.add(BUCKET_TYPE_TRIGGERS);
+        //ALL_BUCKET_TYPES.add(BUCKET_TYPE_STRONG_CONSISTENCY);
+		// ALL_BUCKET_TYPES.add(BUCKET_TYPE_TRIGGERS);
         ALL_BUCKET_TYPES.add(BUCKET_TYPE_COUNTERS);
-        ALL_BUCKET_TYPES.add(BUCKET_TYPE_TRIGGERABLE_COUNTERS);
+        // ALL_BUCKET_TYPES.add(BUCKET_TYPE_TRIGGERABLE_COUNTERS);
         ALL_BUCKET_TYPES.add(BUCKET_TYPE_SETS);
         ALL_BUCKET_TYPES.add(BUCKET_TYPE_MAPS);
         ALL_BUCKET_TYPES.add(BUCKET_TYPE_ALL);
 
         KV_BUCKET_TYPES.add(BUCKET_TYPE_DEFAULT);
-        KV_BUCKET_TYPES.add(BUCKET_TYPE_STRONG_CONSISTENCY);
-		KV_BUCKET_TYPES.add(BUCKET_TYPE_TRIGGERS);
+        //KV_BUCKET_TYPES.add(BUCKET_TYPE_STRONG_CONSISTENCY);
+		// KV_BUCKET_TYPES.add(BUCKET_TYPE_TRIGGERS);
 
         COUNTER_BUCKET_TYPES.add(BUCKET_TYPE_COUNTERS);
-        COUNTER_BUCKET_TYPES.add(BUCKET_TYPE_TRIGGERABLE_COUNTERS);
+        // COUNTER_BUCKET_TYPES.add(BUCKET_TYPE_TRIGGERABLE_COUNTERS);
 
         CONSISTENCY_BUCKET_TYPES.add(BUCKET_TYPE_DEFAULT);
-        CONSISTENCY_BUCKET_TYPES.add(BUCKET_TYPE_STRONG_CONSISTENCY);
+        //CONSISTENCY_BUCKET_TYPES.add(BUCKET_TYPE_STRONG_CONSISTENCY);
 
 		NUM_NODES = cluster.getNodes().size();
-		if (NUM_NODES >= MIN_NODES_FOR_STRONG_CONSISTENCY) {
-		    BUCKET_TYPE_DEFAULT = BUCKET_TYPE_STRONG_CONSISTENCY;
-		}
+		//if (NUM_NODES >= MIN_NODES_FOR_STRONG_CONSISTENCY) {
+		//    BUCKET_TYPE_DEFAULT = BUCKET_TYPE_STRONG_CONSISTENCY;
+		//}
 
         client = new RiakClient(cluster);
 

FunctionWorker/.gitignore

@@ -2,4 +2,5 @@
 /python/local_queue/
 /doc/
 __init__.py
-__pycache__
+__pycache__
+/.m2

FunctionWorker/Makefile

@@ -23,14 +23,16 @@ default: doc/MicroFunctionsAPI.html
 
 PYTHON_DATALAYER=$(shell find python/data_layer -type f -size +1c) python/data_layer
 $(PYTHON_DATALAYER): ../build_env.mk \
-	../DataLayerService/thrift/DataLayerMessage.thrift \
-	../DataLayerService/thrift/DataLayerService.thrift
+		../DataLayerService/thrift/DataLayerMessage.thrift \
+		../DataLayerService/thrift/DataLayerService.thrift \
+		../build_tools/thrift/$(THRIFT_VERSION)/build_thrift.sh
+	cd ../build_tools/thrift/$(THRIFT_VERSION)/; ./build_thrift.sh; cd -
 	docker run --user $(UID):$(GID) --rm -v $(CURDIR)/..:/root -w /root $(THRIFT_IMAGE_NAME):$(THRIFT_VERSION) bash -c '\
 		thrift --gen py -out FunctionWorker/python/ DataLayerService/thrift/DataLayerMessage.thrift; \
 		thrift --gen py -out FunctionWorker/python/ DataLayerService/thrift/DataLayerService.thrift'
 
 doc/MicroFunctionsAPI.html: $(PYTHON_DATALAYER)
-	docker run --user $(UID):$(GID) --rm -it \
+	docker run --user $(UID):$(GID) --rm -i \
 		--name mfnapi_doc_gen \
 		-v $(CURDIR):/opt/mfnapi \
 		-w /opt/mfnapi \

FunctionWorker/MicroFunctionsAPI.thrift

@@ -16,6 +16,9 @@
 namespace java org.microfunctions.mfnapi
 
 service MicroFunctionsAPIService {
+
+    string get_context_object_properties(),
+
     string ping(1:i32 n),
 
     void update_metadata(1: string metadata_name, 2: string metadata_value, 3: bool is_privileged_metadata),
@@ -25,7 +28,7 @@ service MicroFunctionsAPIService {
     void send_to_all_running_functions_in_session(1: string message, 2: bool send_now),  //message
     void send_to_running_function_in_session_with_alias(1: string als, 2: string message, 3: bool send_now),  // message
 
-    list<string> get_session_update_messages(1: i32 count),
+    list<string> get_session_update_messages(1: i32 count, 2: bool blck),
 
     void set_session_alias(1: string als),
     void unset_session_alias(),

FunctionWorker/python/DataLayerClient.py

@@ -27,7 +27,7 @@
 
 class DataLayerClient:
 
-    def __init__(self, locality=1, sid=None, wid=None, suid=None, is_wf_private=False, for_mfn=False, connect="127.0.0.1:4998", init_tables=False, drop_keyspace=False):
+    def __init__(self, locality=1, sid=None, wid=None, suid=None, is_wf_private=False, for_mfn=False, connect="127.0.0.1:4998", init_tables=False, drop_keyspace=False, tableName=None):
         self.dladdress = connect
 
         if for_mfn:
@@ -52,7 +52,10 @@ def __init__(self, locality=1, sid=None, wid=None, suid=None, is_wf_private=Fals
 
             else:
                 self.keyspace = "storage_" + suid
-                self.tablename = "defaultTable"
+                if tableName is not None:
+                    self.tablename = tableName
+                else:
+                    self.tablename = "defaultTable"
                 self.maptablename = "defaultMapTable"
                 self.settablename = "defaultSetTable"
                 self.countertablename = "defaultCounterTable"
@@ -84,7 +87,7 @@ def _initialize_tables(self):
             self.datalayer.createSetTable(self.keyspace, self.settablename, self.locality)
             self.datalayer.createCounterTable(self.keyspace, self.countertablename, Metadata(tableType="counters"), self.locality)
             self.datalayer.createTable(self.keyspace, self.triggersinfotablename, Metadata(tableType="default"), self.locality)
-            self.datalayer.createCounterTable(self.keyspace, self.countertriggerstable, Metadata(tableType="mfn_counter_trigger"), self.locality)
+            self.datalayer.createCounterTable(self.keyspace, self.countertriggerstable, Metadata(tableType="counters"), self.locality)
             self.datalayer.createTable(self.keyspace, self.countertriggersinfotable, Metadata(tableType="default"), self.locality)
         except Thrift.TException as exc:
             print("Could not initialize tables: " + str(exc))
@@ -263,7 +266,7 @@ def containsMapKey(self, mapname, key):
         return ret
 
     def getMapKeys(self, mapname):
-        keys = None
+        keys = []
         for retry in range(MAX_RETRIES):
             try:
                 keyset = self.datalayer.retrieveKeysetFromMap(self.keyspace, self.maptablename, mapname, self.locality)
@@ -280,7 +283,7 @@ def getMapKeys(self, mapname):
         return keys
 
     def retrieveMap(self, mapname):
-        mapentries = None
+        mapentries = {}
         for retry in range(MAX_RETRIES):
             try:
                 mapentries = self.datalayer.retrieveAllEntriesFromMap(self.keyspace, self.maptablename, mapname, self.locality)
@@ -398,7 +401,7 @@ def containsSetItem(self, setname, item):
         return ret
 
     def retrieveSet(self, setname):
-        items = None
+        items = []
         for retry in range(MAX_RETRIES):
             try:
                 itemsset = self.datalayer.retrieveSet(self.keyspace, self.settablename, setname, self.locality)
@@ -557,7 +560,7 @@ def createTriggerableTable(self, tableName):
         status = False
         for retry in range(MAX_RETRIES):
             try:
-                status = self.datalayer.createTable(self.keyspace, tableName, Metadata(tableType="triggers"), self.locality)
+                status = self.datalayer.createTable(self.keyspace, tableName, Metadata(tableType="default"), self.locality)
                 break
             except TTransport.TTransportException as exc:
                 self.connect()

FunctionWorker/python/FunctionWorker.py

@@ -20,10 +20,11 @@
 import imp
 import json
 import logging
+import random
 import socket
 import subprocess
 import shlex
-#import hashlib
+import hashlib
 from threading import Timer
 
 import thriftpy2
@@ -116,6 +117,10 @@ def __init__(self, args_dict):
 
         signal(SIGCHLD, SIG_IGN)
 
+        # do this once rather than at every forked process
+        if self._state_utils.isTaskState():
+            os.chdir(self._function_folder)
+
         self._is_running = False
         #self._print_self()
 
@@ -137,6 +142,7 @@ def _set_args(self, args):
         self._datalayer = args["datalayer"]
         self._external_endpoint = args["externalendpoint"]
         self._internal_endpoint = args["internalendpoint"]
+        self._management_endpoints = args["managementendpoints"]
         self._wf_next = args["fnext"]
         self._wf_pot_next = args["fpotnext"]
         self._function_runtime = args["fruntime"]
@@ -208,6 +214,7 @@ def _print_self(self):
         self._logger.debug("\tself._datalayer: %s", self._datalayer)
         self._logger.debug("\tself._external_endpoint: %s", str(self._external_endpoint))
         self._logger.debug("\tself._internal_endpoint: %s", str(self._internal_endpoint))
+        self._logger.debug("\tself._management_endpoints: %s", str(self._management_endpoints))
         self._logger.debug("\tself._wf_next: %s", ",".join(self._wf_next))
         self._logger.debug("\tself._wf_pot_next: %s", ",".join(self._wf_pot_next))
         self._logger.debug("\tself._wf_function_list: %s", ",".join(self._wf_function_list))
@@ -301,10 +308,7 @@ def _fork_and_handle_message(self, key, encapsulated_value):
                         error_type = "User Input Decapsulation Error"
                         has_error = True
 
-                timestamp_map["t_start_chdir"] = time.time() * 1000.0
                 signal(SIGCHLD, SIG_DFL)
-                if self._state_utils.isTaskState():
-                    os.chdir(self._function_folder)
 
                 # 2. Decode input. Input (value) must be a valid JSON Text.
                 # Note: JSON Text is not the same as JSON string. JSON string a one variable type that can be contained inside a JSON Text.
@@ -393,7 +397,7 @@ def _fork_and_handle_message(self, key, encapsulated_value):
                         # Maybe allow only if the destination is a session function? Requires a list of session functions and passing them to the MicroFunctionsAPI and SessionUtils
                         # Nonetheless, currently, MicroFunctionsAPI and SessionUtils write warning messages to the workflow log to indicate such problems
                         # (e.g., when this is not a workflow session or session function, when the destination running function instance does not exist)
-                        sapi = MicroFunctionsAPI(self._storage_userid, self._sandboxid, self._workflowid, self._function_state_name, key, publication_utils, self._is_session_workflow, self._is_session_function, session_utils, self._logger, self._datalayer, self._external_endpoint, self._internal_endpoint, self._userid, self._usertoken)
+                        sapi = MicroFunctionsAPI(self._storage_userid, self._sandboxid, self._workflowid, self._function_state_name, key, publication_utils, self._is_session_workflow, self._is_session_function, session_utils, self._logger, self._datalayer, self._external_endpoint, self._internal_endpoint, self._userid, self._usertoken, self._management_endpoints)
                         # need this to retrieve and publish the in-memory, transient data (i.e., stored/deleted via is_queued = True)
                         publication_utils.set_sapi(sapi)
                     except Exception as exc:
@@ -450,7 +454,10 @@ def _fork_and_handle_message(self, key, encapsulated_value):
                     elif self._function_runtime == "java":
                         exec_arguments = {}
 
-                        api_uds = "/tmp/" + self._function_state_name + "_" + key + ".uds"
+                        random.seed()
+                        name = self._function_state_name + "_" + key + "_" + str(time.time() * 1000.0) + "_" + str(random.uniform(0, 100000))
+                        sha = hashlib.sha256(name.encode()).hexdigest()
+                        api_uds = "/tmp/" + sha + ".uds"
 
                         exec_arguments["api_uds"] = api_uds
                         exec_arguments["thriftAPIService"] = self._api_thrift.MicroFunctionsAPIService