Workflows

.mypy.ini

@@ -24,11 +24,11 @@ ignore_missing_imports = True
 [mypy-minio]
 ignore_missing_imports = True
 
-[mypy-google.cloud]
-ignore_missing_imports = True
-
-[mypy-google.api_core]
+[mypy-google.*]
 ignore_missing_imports = True
 
 [mypy-testtools]
 ignore_missing_imports = True
+
+[mypy-redis]
+ignore_missing_imports = True

benchmarks/600.workflows/610.gen/config.json

@@ -0,0 +1,5 @@
+{
+  "timeout": 120,
+  "memory": 128,
+  "languages": ["python"]
+}

benchmarks/600.workflows/610.gen/definition.json

@@ -0,0 +1,48 @@
+{
+    "root": "get_astros",
+    "states": {
+        "get_astros": {
+            "type": "task",
+            "func_name": "get_astros",
+            "next": "select_astros_number"
+        },
+        "select_astros_number": {
+            "type": "switch",
+            "cases": [
+                {
+                    "var": "astros.number",
+                    "op": "<",
+                    "val": 10,
+                    "next": "few_people"
+                },
+                {
+                    "var": "astros.number",
+                    "op": ">=",
+                    "val": 10,
+                    "next": "many_people"
+                }
+            ],
+            "default": "few_people"
+        },
+        "few_people": {
+            "type": "task",
+            "func_name": "few_people",
+            "next": "map_astros"
+        },
+        "many_people": {
+            "type": "task",
+            "func_name": "many_people",
+            "next": "map_astros"
+        },
+        "map_astros": {
+            "type": "map",
+            "array": "astros.people",
+            "func_name": "map_astros",
+            "next": "process_astros"
+        },
+        "process_astros": {
+            "type": "task",
+            "func_name": "process_astros"
+        }
+    }
+}

benchmarks/600.workflows/610.gen/input.py

@@ -0,0 +1,5 @@
+def buckets_count():
+    return (0, 0)
+
+def generate_input(data_dir, size, input_buckets, output_buckets, upload_func):
+    return dict()

benchmarks/600.workflows/610.gen/python/few_people.py

@@ -0,0 +1,5 @@
+def handler(event):
+    return {
+        "many_astros": False,
+        **event
+    }

benchmarks/600.workflows/610.gen/python/get_astros.py

@@ -0,0 +1,8 @@
+import requests
+
+def handler(event):
+    res = requests.get("http://api.open-notify.org/astros.json")
+
+    return {
+        "astros": res.json()
+    }

benchmarks/600.workflows/610.gen/python/many_people.py

@@ -0,0 +1,5 @@
+def handler(event):
+    return {
+        "many_astros": True,
+        **event
+    }

benchmarks/600.workflows/610.gen/python/map_astros.py

@@ -0,0 +1,7 @@
+def handler(elem):
+    name = elem["name"]
+    fn, ln = name.split(" ")
+    name = " ".join([ln, fn])
+    elem["name_rev"] = name
+
+    return elem

benchmarks/600.workflows/610.gen/python/process_astros.py

@@ -0,0 +1,5 @@
+def handler(arr):
+    return {
+        "astros": arr,
+        "done": True
+    }

benchmarks/600.workflows/620.func_invo/config.json

@@ -0,0 +1,5 @@
+{
+  "timeout": 120,
+  "memory": 128,
+  "languages": ["python"]
+}

benchmarks/600.workflows/620.func_invo/definition.json

@@ -0,0 +1,10 @@
+{
+    "root": "process",
+    "states": {
+        "process": {
+            "type": "loop",
+            "func_name": "process",
+            "count": 10
+        }
+    }
+}

benchmarks/600.workflows/620.func_invo/input.py

@@ -0,0 +1,11 @@
+size_generators = {
+    'test' : 10,
+    'small' : 2**10,
+    'large': 2**15
+}
+
+def buckets_count():
+    return (0, 0)
+
+def generate_input(data_dir, size, input_buckets, output_buckets, upload_func):
+    return { 'size': size_generators[size] }

benchmarks/600.workflows/620.func_invo/python/process.py

@@ -0,0 +1,14 @@
+from random import shuffle
+
+def handler(event):
+    size = int(event["size"]) if isinstance(event, dict) else len(event)
+    elems = list(range(size))
+    shuffle(elems)
+
+    data = ""
+    for i in elems:
+        data += str(i % 255)
+        if len(data) > size:
+            break
+
+    return data[:size]

benchmarks/600.workflows/630.parallel/config.json

@@ -0,0 +1,5 @@
+{
+  "timeout": 120,
+  "memory": 128,
+  "languages": ["python"]
+}

benchmarks/600.workflows/630.parallel/definition.json

@@ -0,0 +1,15 @@
+{
+    "root": "generate",
+    "states": {
+        "generate": {
+            "type": "task",
+            "func_name": "generate",
+            "next": "process"
+        },
+        "process": {
+            "type": "map",
+            "func_name": "process",
+            "array": "buffer"
+        }
+    }
+}

benchmarks/600.workflows/630.parallel/input.py

@@ -0,0 +1,11 @@
+size_generators = {
+    'test' : 5,
+    'small' : 100,
+    'large': 1000
+}
+
+def buckets_count():
+    return (0, 0)
+
+def generate_input(data_dir, size, input_buckets, output_buckets, upload_func):
+    return { 'size': size_generators[size] }

benchmarks/600.workflows/630.parallel/python/generate.py

@@ -0,0 +1,7 @@
+def handler(event):
+    size = int(event["size"])
+    buffer = size * ["asdf"]
+
+    return {
+        "buffer": buffer
+    }

benchmarks/600.workflows/630.parallel/python/process.py

@@ -0,0 +1,2 @@
+def handler(elem):
+    return elem[::-1]

benchmarks/wrappers/aws/python/handler_workflow.py

@@ -0,0 +1,56 @@
+
+import datetime
+import io
+import json
+import os
+import sys
+import uuid
+import importlib
+
+# Add current directory to allow location of packages
+sys.path.append(os.path.join(os.path.dirname(__file__), '.python_packages/lib/site-packages'))
+
+from redis import Redis
+
+def probe_cold_start():
+    is_cold = False
+    fname = os.path.join("/tmp", "cold_run")
+    if not os.path.exists(fname):
+        is_cold = True
+        container_id = str(uuid.uuid4())[0:8]
+        with open(fname, "a") as f:
+            f.write(container_id)
+    else:
+        with open(fname, "r") as f:
+            container_id = f.read()
+
+    return is_cold, container_id
+
+
+def handler(event, context):
+    start = datetime.datetime.now().timestamp()
+
+    workflow_name, func_name = context.function_name.split("___")
+    function = importlib.import_module(f"function.{func_name}")
+    res = function.handler(event)
+
+    end = datetime.datetime.now().timestamp()
+
+    is_cold, container_id = probe_cold_start()
+    payload = json.dumps({
+        "func": func_name,
+        "start": start,
+        "end": end,
+        "is_cold": is_cold,
+        "container_id": container_id
+    })
+
+    redis = Redis(host={{REDIS_HOST}},
+                  port=6379,
+                  decode_responses=True,
+                  socket_connect_timeout=10)
+
+    key = os.path.join(workflow_name, func_name, str(uuid.uuid4())[0:8])
+    redis.set(key, payload)
+
+    return res

benchmarks/wrappers/aws/python/storage.py

@@ -15,17 +15,17 @@ def __init__(self):
     @staticmethod
     def unique_name(name):
         name, extension = os.path.splitext(name)
-        return '{name}.{random}.{extension}'.format(
+        return '{name}.{random}{extension}'.format(
                     name=name,
                     extension=extension,
                     random=str(uuid.uuid4()).split('-')[0]
                 )
-    
+
     def upload(self, bucket, file, filepath):
         key_name = storage.unique_name(file)
         self.client.upload_file(filepath, bucket, key_name)
         return key_name
-    
+
     def download(self, bucket, file, filepath):
         self.client.download_file(bucket, file, filepath)
 
@@ -46,7 +46,7 @@ def download_stream(self, bucket, file):
         data = io.BytesIO()
         self.client.download_fileobj(bucket, file, data)
         return data.getbuffer()
-    
+
     def get_instance():
         if storage.instance is None:
             storage.instance = storage()

benchmarks/wrappers/azure/python/fsm.py

@@ -0,0 +1 @@
+../../../../../serverless-benchmarks/sebs/faas/fsm.py

benchmarks/wrappers/azure/python/handler.py → benchmarks/wrappers/azure/python/handler_function.py

@@ -6,7 +6,7 @@
 
 # TODO: usual trigger
 # implement support for blob and others
-def main(req: func.HttpRequest, context: func.Context) -> func.HttpResponse:
+def main(req: func.HttpRequest, starter: str, context: func.Context) -> func.HttpResponse:
     income_timestamp = datetime.datetime.now().timestamp()
     req_json = req.get_json()
     if 'connection_string' in req_json:

benchmarks/wrappers/azure/python/handler_workflow.py

@@ -0,0 +1,59 @@
+import datetime
+import json
+import os
+import uuid
+import importlib
+
+from azure.storage.blob import BlobServiceClient
+import azure.functions as func
+from redis import Redis
+
+def probe_cold_start():
+    is_cold = False
+    fname = os.path.join("/tmp", "cold_run")
+    if not os.path.exists(fname):
+        is_cold = True
+        container_id = str(uuid.uuid4())[0:8]
+        with open(fname, "a") as f:
+            f.write(container_id)
+    else:
+        with open(fname, "r") as f:
+            container_id = f.read()
+
+    return is_cold, container_id
+
+
+def main(event):
+    start = datetime.datetime.now().timestamp()
+
+    workflow_name = os.getenv("APPSETTING_WEBSITE_SITE_NAME")
+    func_name = os.path.basename(os.path.dirname(__file__))
+
+    module_name = f"{func_name}.{func_name}"
+    module_path = f"{func_name}/{func_name}.py"
+    spec = importlib.util.spec_from_file_location(module_name, module_path)
+    function = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(function)
+
+    res = function.handler(event)
+
+    end = datetime.datetime.now().timestamp()
+
+    is_cold, container_id = probe_cold_start()
+    payload = json.dumps({
+        "func": func_name,
+        "start": start,
+        "end": end,
+        "is_cold": is_cold,
+        "container_id": container_id,
+    })
+
+    redis = Redis(host={{REDIS_HOST}},
+          port=6379,
+          decode_responses=True,
+          socket_connect_timeout=10)
+
+    key = os.path.join(workflow_name, func_name, str(uuid.uuid4())[0:8])
+    redis.set(key, payload)
+
+    return res