[#64] Queue and Storage triggers for AWS, GCP and Azure

benchmarks/wrappers/aws/python/handler.py

@@ -1,18 +1,31 @@
-
 import datetime, io, json, os, sys, uuid
 
 # Add current directory to allow location of packages
 sys.path.append(os.path.join(os.path.dirname(__file__), '.python_packages/lib/site-packages'))
 
-# TODO: usual trigger
-# implement support for S3 and others
 def handler(event, context):
 
     income_timestamp = datetime.datetime.now().timestamp()
 
+    # Queue trigger
+    if ("Records" in event and event["Records"][0]["eventSource"] == 'aws:sqs'):
+        event = json.loads(event["Records"][0]["body"])
+
+    # Storage trigger
+    if ("Records" in event and "s3" in event["Records"][0]):
+        bucket_name = event["Records"][0]["s3"]["bucket"]["name"]
+        file_name = event["Records"][0]["s3"]["object"]["key"]
+
+        from function import storage
+        storage_inst = storage.storage.get_instance()
+
+        obj = storage_inst.get_object(bucket_name, file_name)
+        event = json.loads(obj['Body'].read())
+
     # HTTP trigger with API Gateaway
     if 'body' in event:
         event = json.loads(event['body'])
+
     req_id = context.aws_request_id
     event['request-id'] = req_id
     event['income-timestamp'] = income_timestamp

benchmarks/wrappers/aws/python/storage.py

@@ -46,6 +46,9 @@ def download_stream(self, bucket, file):
         data = io.BytesIO()
         self.client.download_fileobj(bucket, file, data)
         return data.getbuffer()
+
+    def get_object(self, bucket, file):
+        return self.client.get_object(Bucket=bucket, Key=file)
 
     def get_instance():
         if storage.instance is None:

benchmarks/wrappers/azure/python/handler.py

@@ -1,12 +1,11 @@
 
-import datetime, io, json, os, uuid
+import base64
+import datetime, io, json, logging, os, uuid
 
 import azure.functions as func
 
 
-# TODO: usual trigger
-# implement support for blob and others
-def main(req: func.HttpRequest, context: func.Context) -> func.HttpResponse:
+def handler_http(req: func.HttpRequest, context: func.Context) -> func.HttpResponse:
     income_timestamp = datetime.datetime.now().timestamp()
     req_json = req.get_json()
     if 'connection_string' in req_json:
@@ -73,3 +72,20 @@ def main(req: func.HttpRequest, context: func.Context) -> func.HttpResponse:
         mimetype="application/json"
     )
 
+def handler_queue(msg: func.QueueMessage):
+    logging.info('Python queue trigger function processed a queue item.')
+    payload = msg.get_body().decode('utf-8')
+
+    from . import function
+    ret = function.handler(payload)
+
+    # TODO(oana)
+
+def handler_storage(blob: func.InputStream):
+    logging.info('Python Blob trigger function processed %s', blob.name)
+    payload = blob.readline().decode('utf-8')  # TODO(oana)
+
+    from . import function
+    ret = function.handler(payload)
+
+    # TODO(oana)

benchmarks/wrappers/gcp/python/handler.py

@@ -1,9 +1,10 @@
-import datetime, io, json, os, uuid, sys
+import base64, datetime, io, json, os, uuid, sys
 
-sys.path.append(os.path.join(os.path.dirname(__file__), '.python_packages/lib/site-packages'))
+from google.cloud import storage as gcp_storage
 
+sys.path.append(os.path.join(os.path.dirname(__file__), '.python_packages/lib/site-packages'))
 
-def handler(req):
+def handler_http(req):
     income_timestamp = datetime.datetime.now().timestamp()
     req_id = req.headers.get('Function-Execution-Id')
 
@@ -62,3 +63,31 @@ def handler(req):
             'cold_start_var': cold_start_var,
             'container_id': container_id,
         }), 200, {'ContentType': 'application/json'}
+
+def handler_queue(data, context):
+    serialized_payload = data.get('data')
+    payload = json.loads(base64.b64decode(serialized_payload).decode("utf-8"))
+
+    from function import function
+    ret = function.handler(payload)
+
+    # TODO(oana)
+
+def handler_storage(data, context):
+    bucket_name = data.get('bucket')
+    name = data.get('name')
+    filepath = '/tmp/bucket_contents'
+
+    from function import storage
+    storage_inst = storage.storage.get_instance()
+    storage_inst.download(bucket_name, name, filepath)
+
+    payload = {}
+
+    with open(filepath, 'r') as fp:
+        payload = json.load(fp)
+
+    from function import function
+    ret = function.handler(payload)
+
+    # TODO(oana)

config/example.json

@@ -6,7 +6,7 @@
     "download_results": false,
     "runtime": {
       "language": "python",
-      "version": "3.7"
+      "version": "3.9"
     },
     "type": "invocation-overhead",
     "perf-cost": {

config/systems.json

@@ -18,7 +18,8 @@
       "python": {
         "base_images": {
           "3.7": "python:3.7-slim",
-          "3.8": "python:3.8-slim"
+          "3.8": "python:3.8-slim",
+          "3.9": "python:3.9-slim"
         },
         "images": [
           "run",

docs/modularity.md

@@ -303,6 +303,7 @@ Implement this step in the following function:
         language_version: str,
         benchmark: str,
         is_cached: bool,
+        trigger: Optional[Trigger.TriggerType],
     ) -> Tuple[str, int]
 ```
 

docs/platforms.md

@@ -85,9 +85,9 @@ AZURE_SECRET_PASSWORD = XXXXXXXXXXXXX
 You can pass the credentials either using the environment variables:
 
 ```
-export AZURE_SECRET_APPLICATION_ID = XXXXXXXXXXXXXXXX
-export AZURE_SECRET_TENANT = XXXXXXXXXXXX
-export AZURE_SECRET_PASSWORD = XXXXXXXXXXXXX
+export AZURE_SECRET_APPLICATION_ID=XXXXXXXXXXXXXXXX
+export AZURE_SECRET_TENANT=XXXXXXXXXXXX
+export AZURE_SECRET_PASSWORD=XXXXXXXXXXXXX
 ```
 
 or in the JSON input configuration:

requirements.azure.txt

@@ -1 +1,3 @@
 azure-storage-blob==12.10.0
+azure-storage-queue==12.9.0
+azure-identity==1.16.0

sebs.py

@@ -173,7 +173,7 @@ def benchmark():
 @click.option("--repetitions", default=5, type=int, help="Number of experimental repetitions.")
 @click.option(
     "--trigger",
-    type=click.Choice(["library", "http"]),
+    type=click.Choice(["library", "http", "queue", "storage"]),
     default="http",
     help="Function trigger to be used.",
 )
@@ -224,6 +224,10 @@ def invoke(
     if image_tag_prefix is not None:
         sebs_client.config.image_tag_prefix = image_tag_prefix
 
+    # Insert trigger into (experiment) config. Required by Azure when packaging.
+    trigger = trigger if trigger is not None else "http"
+    update_nested_dict(config, ["experiments", "trigger"], trigger)
+
     experiment_config = sebs_client.get_experiment_config(config["experiments"])
     update_nested_dict(config, ["experiments", "benchmark"], benchmark)
     benchmark_obj = sebs_client.get_benchmark(
@@ -237,9 +241,18 @@ def invoke(
     if timeout is not None:
         benchmark_obj.benchmark_config.timeout = timeout
 
+    function_name = function_name if function_name else deployment_client.default_function_name(benchmark_obj)
+
+    # GCP and Azure only allow one trigger per function, so augment function name with
+    # trigger type: _http, _queue etc.
+    #
+    # Additionally, Azure requires for the trigger to be defined at deployment time.
+    if deployment_client.name() == "gcp" or deployment_client.name() == "azure":
+        function_name = "{}-{}".format(function_name, trigger)
+
     func = deployment_client.get_function(
         benchmark_obj,
-        function_name if function_name else deployment_client.default_function_name(benchmark_obj),
+        function_name,
     )
     storage = deployment_client.get_storage(replace_existing=experiment_config.update_storage)
     input_config = benchmark_obj.prepare_input(storage=storage, size=benchmark_input_size)

sebs/aws/aws.py

@@ -132,6 +132,7 @@ def package_code(
         language_version: str,
         benchmark: str,
         is_cached: bool,
+        trigger: Optional[Trigger.TriggerType],
     ) -> Tuple[str, int]:
 
         CONFIG_FILES = {
@@ -258,13 +259,19 @@ def create_function(self, code_package: Benchmark, func_name: str) -> "LambdaFun
 
     def cached_function(self, function: Function):
 
-        from sebs.aws.triggers import LibraryTrigger
+        from sebs.aws.triggers import LibraryTrigger, QueueTrigger, StorageTrigger
 
         for trigger in function.triggers(Trigger.TriggerType.LIBRARY):
             trigger.logging_handlers = self.logging_handlers
             cast(LibraryTrigger, trigger).deployment_client = self
         for trigger in function.triggers(Trigger.TriggerType.HTTP):
             trigger.logging_handlers = self.logging_handlers
+        for trigger in function.triggers(Trigger.TriggerType.QUEUE):
+            trigger.logging_handlers = self.logging_handlers
+            cast(QueueTrigger, trigger).deployment_client = self
+        for trigger in function.triggers(Trigger.TriggerType.STORAGE):
+            trigger.logging_handlers = self.logging_handlers
+            cast(StorageTrigger, trigger).deployment_client = self
 
     """
         Update function code and configuration on AWS.
@@ -484,10 +491,11 @@ def download_metrics(
         )
 
     def create_trigger(self, func: Function, trigger_type: Trigger.TriggerType) -> Trigger:
-        from sebs.aws.triggers import HTTPTrigger
+        from sebs.aws.triggers import HTTPTrigger, QueueTrigger, StorageTrigger
 
         function = cast(LambdaFunction, func)
 
+        trigger: Trigger
         if trigger_type == Trigger.TriggerType.HTTP:
 
             api_name = "{}-http-api".format(function.name)
@@ -511,6 +519,14 @@ def create_trigger(self, func: Function, trigger_type: Trigger.TriggerType) -> T
         elif trigger_type == Trigger.TriggerType.LIBRARY:
             # should already exist
             return func.triggers(Trigger.TriggerType.LIBRARY)[0]
+        elif trigger_type == Trigger.TriggerType.QUEUE:
+            trigger = QueueTrigger(func.name, self)
+            trigger.logging_handlers = self.logging_handlers
+            self.logging.info(f"Created Queue trigger for {func.name} function.")
+        elif trigger_type == Trigger.TriggerType.STORAGE:
+            trigger = StorageTrigger(func.name, self)
+            trigger.logging_handlers = self.logging_handlers
+            self.logging.info(f"Created Storage trigger for {func.name} function.")
         else:
             raise RuntimeError("Not supported!")
 

sebs/aws/function.py

@@ -39,7 +39,7 @@ def serialize(self) -> dict:
     @staticmethod
     def deserialize(cached_config: dict) -> "LambdaFunction":
         from sebs.faas.function import Trigger
-        from sebs.aws.triggers import LibraryTrigger, HTTPTrigger
+        from sebs.aws.triggers import LibraryTrigger, HTTPTrigger, QueueTrigger, StorageTrigger
 
         cfg = FunctionConfig.deserialize(cached_config["config"])
         ret = LambdaFunction(
@@ -55,7 +55,12 @@ def deserialize(cached_config: dict) -> "LambdaFunction":
         for trigger in cached_config["triggers"]:
             trigger_type = cast(
                 Trigger,
-                {"Library": LibraryTrigger, "HTTP": HTTPTrigger}.get(trigger["type"]),
+                {
+                    "Library": LibraryTrigger,
+                    "HTTP": HTTPTrigger,
+                    "Queue": QueueTrigger,
+                    "Storage": StorageTrigger,
+                }.get(trigger["type"]),
             )
             assert trigger_type, "Unknown trigger type {}".format(trigger["type"])
             ret.add_trigger(trigger_type.deserialize(trigger))