Merge pull request #100 from knix-microfunctions/release/0.8.9

Release/0.8.9

Author: Istemi Ekin Akkus

DataLayerService/Dockerfile

@@ -25,4 +25,4 @@ RUN mkdir -p /opt/mfn/datalayer/logs
 COPY target/datalayerservice.jar /opt/mfn/datalayer/
 WORKDIR /opt/mfn/datalayer
 
-CMD ["java","-jar","datalayerservice.jar"]
+CMD java ${JAVA_OPTS} -jar datalayerservice.jar

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

@@ -40,10 +40,11 @@
 import org.apache.thrift.TException;
 import org.apache.thrift.protocol.TCompactProtocol;
 import org.apache.thrift.server.TServer;
-import org.apache.thrift.server.TThreadedSelectorServer;
+import org.apache.thrift.server.TThreadPoolServer;
+//import org.apache.thrift.server.TThreadedSelectorServer;
 import org.apache.thrift.transport.TFramedTransport;
-import org.apache.thrift.transport.TNonblockingServerSocket;
-import org.apache.thrift.transport.TNonblockingServerTransport;
+import org.apache.thrift.transport.TServerSocket;
+import org.apache.thrift.transport.TServerTransport;
 import org.apache.thrift.transport.TTransportException;
 import org.microfunctions.data_layer.DataLayerService.Iface;
 
@@ -65,8 +66,9 @@ public class DataLayerServer implements Iface, Callable<Object> {
 	private static final List<AbstractMap.SimpleEntry<String, Integer>> NO_KEYSPACES = new ArrayList<AbstractMap.SimpleEntry<String, Integer>>(0);
 	private static final List<String> NO_KEYS = new ArrayList<String>(0);
 
-	public static final int DEFAULT_SELECTOR_THREADS = 50; //Math.max(2, 2 * Runtime.getRuntime().availableProcessors());
-    public static final int DEFAULT_WORKER_THREADS = 100;  //Math.max(4, 4 * Runtime.getRuntime().availableProcessors());
+//	public static final int DEFAULT_SELECTOR_THREADS = 50; //Math.max(2, 2 * Runtime.getRuntime().availableProcessors());
+//    public static final int DEFAULT_WORKER_THREADS = 100;  //Math.max(4, 4 * Runtime.getRuntime().availableProcessors());
+	public static final int DEFAULT_MAX_WORKER_THREADS = Integer.MAX_VALUE;
 	public static final int DEFAULT_CLIENT_TIMEOUT = 0;
 	public static final int DEFAULT_MAX_FRAME_LENGTH = Integer.MAX_VALUE;
 
@@ -1625,14 +1627,22 @@ public boolean updateTableTypeCache(String action, String table, String tableTyp
 
 
 	public void start(InetSocketAddress bindAddr) throws TTransportException {
-	    TNonblockingServerTransport transport = new TNonblockingServerSocket(bindAddr, DEFAULT_CLIENT_TIMEOUT);
+/*	    TServerTransport transport = new TServerSocket(bindAddr, DEFAULT_CLIENT_TIMEOUT);
         TThreadedSelectorServer.Args args = new TThreadedSelectorServer.Args(transport)
                 .transportFactory(new TFramedTransport.Factory(DEFAULT_MAX_FRAME_LENGTH))
                 .protocolFactory(new TCompactProtocol.Factory())
                 .processor(new DataLayerService.Processor<Iface>(this))
                 .selectorThreads(DEFAULT_SELECTOR_THREADS)
                 .workerThreads(DEFAULT_WORKER_THREADS);
         server = new TThreadedSelectorServer(args);
+*/        
+        TServerTransport transport = new TServerSocket(bindAddr, DEFAULT_CLIENT_TIMEOUT);
+        TThreadPoolServer.Args args = new TThreadPoolServer.Args(transport)
+                .transportFactory(new TFramedTransport.Factory(DEFAULT_MAX_FRAME_LENGTH))
+                .protocolFactory(new TCompactProtocol.Factory())
+                .processor(new DataLayerService.Processor<Iface>(this))
+                .maxWorkerThreads(DEFAULT_MAX_WORKER_THREADS);
+        server = new TThreadPoolServer(args);
 
 		LOGGER.info("Listening on "+bindAddr);
 		server.serve();

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

@@ -393,7 +393,7 @@ private boolean createTableWithType (String keyspace, String table, String table
 			}
 
 			Namespace bucket = new Namespace(tableType, keyspace + ";" + table);
-			StoreBucketProperties props = new StoreBucketProperties.Builder(bucket).withNVal(replicationFactor).build();
+			StoreBucketProperties props = new StoreBucketProperties.Builder(bucket).withNVal(replicationFactor).withW(replicationFactor).withR(replicationFactor).withNotFoundOk(false).build();
 			client.execute(props);
 			LOGGER.info("createTableWithType() Keyspace: " + keyspace + "  Table: " + table + "  TableType: " + tableType);
 			boolean success = this.insertRow(keyspace, null, table, ByteBuffer.wrap(tableType.getBytes(StandardCharsets.UTF_8)));

FunctionWorker/MicroFunctionsAPI.thrift

@@ -60,6 +60,7 @@ service MicroFunctionsAPIService {
     void put(1: string key, 2: string value, 3: bool is_private, 4: bool is_queued),
     string get(1: string key, 2: bool is_private),
     void remove(1: string key, 2: bool is_private, 3: bool is_queued),
+    list<string> getKeys(1: i32 start_index, 2: i32 end_index, 3: bool is_private),
 
     void createMap(1: string mapname, 2: bool is_private, 3: bool is_queued),
     void putMapEntry(1: string mapname, 2: string key, 3: string value, 4: bool is_private, 5: bool is_queued),

FunctionWorker/python/DataLayerClient.py

@@ -271,7 +271,7 @@ def getMapKeys(self, mapname):
             try:
                 keyset = self.datalayer.retrieveKeysetFromMap(self.keyspace, self.maptablename, mapname, self.locality)
                 if keyset.key != "" and keyset.key == mapname:
-                    keys = keyset.items
+                    keys = list(keyset.items)
                 break
             except TTransport.TTransportException as exc:
                 print("[DataLayerClient] Reconnecting because of failed getMapKeys: " + str(exc))
@@ -330,7 +330,7 @@ def deleteMap(self, mapname):
         return status
 
     def getMapNames(self, start_index=0, end_index=2147483647):
-        maps = None
+        maps = []
         for retry in range(MAX_RETRIES):
             try:
                 maps = self.datalayer.selectMaps(self.keyspace, self.maptablename, start_index, end_index, self.locality)
@@ -406,7 +406,7 @@ def retrieveSet(self, setname):
             try:
                 itemsset = self.datalayer.retrieveSet(self.keyspace, self.settablename, setname, self.locality)
                 if itemsset.key != "" and itemsset.key == setname:
-                    items = itemsset.items
+                    items = list(itemsset.items)
                 break
             except TTransport.TTransportException as exc:
                 print("[DataLayerClient] Reconnecting because of failed retrieveSet: " + str(exc))
@@ -446,7 +446,7 @@ def deleteSet(self, setname):
         return status
 
     def getSetNames(self, start_index=0, end_index=2147483647):
-        sets = None
+        sets = []
         for retry in range(MAX_RETRIES):
             try:
                 sets = self.datalayer.selectSets(self.keyspace, self.settablename, start_index, end_index, self.locality)
@@ -541,7 +541,7 @@ def deleteCounter(self, countername, tableName=None):
         return status
 
     def getCounterNames(self, start_index=0, end_index=2147483647, tableName=None):
-        counters = None
+        counters = []
         table = self.countertablename if tableName is None else tableName
         for retry in range(MAX_RETRIES):
             try:
@@ -570,14 +570,21 @@ def createTriggerableTable(self, tableName):
         return status
 
     def listKeys(self, start, count, tableName=None):
-        listkeys_response = []
+        keys_response = []
         table = self.tablename if tableName is None else tableName
 
         for retry in range(MAX_RETRIES):
             try:
                 listkeys_response = self.datalayer.selectKeys(self.keyspace, table, start, count, self.locality)
-                if listkeys_response == None or type(listkeys_response) != type([]):
-                    listkeys_response = []
+                if listkeys_response is not None or isinstance(listkeys_response, list):
+                    for key in listkeys_response:
+                        if key.startswith("grain_requirements_") or\
+                            key.startswith("grain_source_") or\
+                            key.startswith("workflow_json_") or\
+                            key.endswith("_metadata"):
+                            continue
+                        else:
+                            keys_response.append(key)
                 break
             except TTransport.TTransportException as exc:
                 print("[DataLayerClient] Reconnecting because of failed selectKeys: " + str(exc))
@@ -586,7 +593,7 @@ def listKeys(self, start, count, tableName=None):
                 print("[DataLayerClient] failed selectKeys: " + str(exc))
                 raise
 
-        return listkeys_response
+        return keys_response
 
     def shutdown(self):
         try:

FunctionWorker/python/DataLayerOperator.py

@@ -95,6 +95,25 @@ def delete(self, key, is_private=False, is_queued=False, table=None):
             data_layer_client = self._get_data_layer_client(is_private)
             data_layer_client.delete(key, tableName=table)
 
+    def getKeys(self, start_index, end_index, is_private=False):
+        keys = set()
+
+        # XXX: should follow "read your writes"
+        # the final result should include:
+        # 1. all created locally
+        # 2. all existing globally minus the ones deleted locally
+
+        # TODO: 1. check local data layer first: get locally created and deleted
+
+        # 2. retrieve all existing globally
+        dlc = self._get_data_layer_client(is_private)
+        m2 = dlc.listKeys(start_index, end_index)
+        if m2 is not None:
+            # TODO: 3. remove the ones deleted locally
+            keys = keys.union(m2)
+
+        return list(keys)
+
     # map operations
     def createMap(self, mapname, is_private=False, is_queued=False):
         if is_queued:

FunctionWorker/python/LocalQueueClient.py

@@ -42,9 +42,9 @@ def connect(self):
             try:
                 self.socket = TSocket.TSocket(host, int(port))
                 self.transport = TTransport.TFramedTransport(self.socket)
-                self.transport.open()
                 self.protocol = TCompactProtocol.TCompactProtocol(self.transport)
                 self.queue = LocalQueueService.Client(self.protocol)
+                self.transport.open()
                 break
             except Thrift.TException as exc:
                 if retry < 60: