added wrappers and updated knative functions

octonawish-akcodes · octonawish-akcodes · commit 8779f507d7b1 · 2024-07-08T09:53:34.000+05:30
Signed-off-by: Abhishek Kumar &lt;abhishek22512@gmail.com&gt;
diff --git a/benchmarks/wrappers/knative/nodejs/index.js b/benchmarks/wrappers/knative/nodejs/index.js
@@ -0,0 +1,54 @@
+const { CloudEvent, HTTP } = require('cloudevents');
+const handler = require('./function').handler;
+
+async function handle(context, event) {
+  const startTime = new Date();
+
+  try {
+    // Ensure event data is parsed correctly
+    const eventData = event ? event : context.body;
+    context.log.info(`Received event: ${JSON.stringify(eventData)}`);
+
+    // Call the handler function with the event data
+    const result = await handler(eventData);
+    const endTime = new Date();
+
+    context.log.info(`Function result: ${JSON.stringify(result)}`);
+    const resultTime = (endTime - startTime) / 1000; // Time in seconds
+
+    // Create a response
+    const response = {
+      begin: startTime.toISOString(),
+      end: endTime.toISOString(),
+      results_time: resultTime,
+      result: result
+    };
+
+    // Return the response
+    return {
+      data: response,
+      headers: { 'Content-Type': 'application/json' },
+      statusCode: 200
+    };
+  } catch (error) {
+    const endTime = new Date();
+    const resultTime = (endTime - startTime) / 1000; // Time in seconds
+
+    context.log.error(`Error - invocation failed! Reason: ${error.message}`);
+    const response = {
+      begin: startTime.toISOString(),
+      end: endTime.toISOString(),
+      results_time: resultTime,
+      result: `Error - invocation failed! Reason: ${error.message}`
+    };
+
+    // Return the error response
+    return {
+      data: response,
+      headers: { 'Content-Type': 'application/json' },
+      statusCode: 500
+    };
+  }
+}
+
+module.exports = handle;
diff --git a/benchmarks/wrappers/knative/nodejs/storage.js b/benchmarks/wrappers/knative/nodejs/storage.js
@@ -0,0 +1,63 @@
+
+const minio = require('minio'),
+  path = require('path'),
+  uuid = require('uuid'),
+  util = require('util'),
+  stream = require('stream'),
+  fs = require('fs');
+
+class minio_storage {
+
+  constructor() {
+    let address = process.env.MINIO_STORAGE_CONNECTION_URL;
+    let access_key = process.env.MINIO_STORAGE_ACCESS_KEY;
+    let secret_key = process.env.MINIO_STORAGE_SECRET_KEY;
+
+    this.client = new minio.Client(
+      {
+        endPoint: address.split(':')[0],
+        port: parseInt(address.split(':')[1], 10),
+        accessKey: access_key,
+        secretKey: secret_key,
+        useSSL: false
+      }
+    );
+  }
+
+  unique_name(file) {
+    let name = path.parse(file);
+    let uuid_name = uuid.v4().split('-')[0];
+    return path.join(name.dir, util.format('%s.%s%s', name.name, uuid_name, name.ext));
+  }
+
+  upload(bucket, file, filepath) {
+    let uniqueName = this.unique_name(file);
+    return [uniqueName, this.client.fPutObject(bucket, uniqueName, filepath)];
+  };
+
+  download(bucket, file, filepath) {
+    return this.client.fGetObject(bucket, file, filepath);
+  };
+
+  uploadStream(bucket, file) {
+    var write_stream = new stream.PassThrough();
+    let uniqueName = this.unique_name(file);
+    let promise = this.client.putObject(bucket, uniqueName, write_stream, write_stream.size);
+    return [write_stream, promise, uniqueName];
+  };
+
+  downloadStream(bucket, file) {
+    var read_stream = new stream.PassThrough();
+    return this.client.getObject(bucket, file);
+  };
+
+  static get_instance() {
+    if(!this.instance) {
+      this.instance = new storage();
+    }
+    return this.instance;
+  }
+
+
+};
+exports.storage = minio_storage;
diff --git a/benchmarks/wrappers/knative/python/func.py b/benchmarks/wrappers/knative/python/func.py
@@ -0,0 +1,41 @@
+import logging
+import datetime
+from flask import jsonify
+from parliament import Context
+from function import handler
+
+
+def main(context: Context):
+    logging.getLogger().setLevel(logging.INFO)
+
+    try:
+        # Extract JSON data from the request
+        event = context.request.json
+
+        begin = datetime.datetime.now()
+        # Pass the extracted JSON data to the handler function
+        ret = handler(event)
+        end = datetime.datetime.now()
+        logging.info(f"Function result: {ret}")
+        results_time = (end - begin) / datetime.timedelta(microseconds=1)
+
+        response = {
+            "begin": begin.strftime("%s.%f"),
+            "end": end.strftime("%s.%f"),
+            "results_time": results_time,
+            "result": ret,
+        }
+
+        return jsonify(response), 200
+
+    except Exception as e:
+        end = datetime.datetime.now()
+        results_time = (end - begin) / datetime.timedelta(microseconds=1)
+        logging.error(f"Error - invocation failed! Reason: {e}")
+        response = {
+            "begin": begin.strftime("%s.%f"),
+            "end": end.strftime("%s.%f"),
+            "results_time": results_time,
+            "result": f"Error - invocation failed! Reason: {e}",
+        }
+        return jsonify(response), 500
diff --git a/benchmarks/wrappers/knative/python/storage.py b/benchmarks/wrappers/knative/python/storage.py
@@ -0,0 +1,80 @@
+import os
+import uuid
+import json
+import minio
+import logging
+
+
+class storage:
+    instance = None
+    client = None
+
+    def __init__(self):
+        try:
+            """
+            Minio does not allow another way of configuring timeout for connection.
+            The rest of configuration is copied from source code of Minio.
+            """
+            import urllib3
+            from datetime import timedelta
+
+            timeout = timedelta(seconds=1).seconds
+
+            mgr = urllib3.PoolManager(
+                timeout=urllib3.util.Timeout(connect=timeout, read=timeout),
+                maxsize=10,
+                retries=urllib3.Retry(
+                    total=5, backoff_factor=0.2, status_forcelist=[500, 502, 503, 504]
+                )
+            )
+            self.client = minio.Minio(
+                os.getenv("MINIO_STORAGE_CONNECTION_URL"),
+                access_key=os.getenv("MINIO_STORAGE_ACCESS_KEY"),
+                secret_key=os.getenv("MINIO_STORAGE_SECRET_KEY"),
+                secure=False,
+                http_client=mgr
+            )
+        except Exception as e:
+            logging.info(e)
+            raise e
+
+    @staticmethod
+    def unique_name(name):
+        name, extension = os.path.splitext(name)
+        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.fput_object(bucket, key_name, filepath)
+        return key_name
+
+    def download(self, bucket, file, filepath):
+        self.client.fget_object(bucket, file, filepath)
+
+    def download_directory(self, bucket, prefix, path):
+        objects = self.client.list_objects(bucket, prefix, recursive=True)
+        for obj in objects:
+            file_name = obj.object_name
+            self.download(bucket, file_name, os.path.join(path, file_name))
+
+    def upload_stream(self, bucket, file, bytes_data):
+        key_name = storage.unique_name(file)
+        self.client.put_object(
+            bucket, key_name, bytes_data, bytes_data.getbuffer().nbytes
+        )
+        return key_name
+
+    def download_stream(self, bucket, file):
+        data = self.client.get_object(bucket, file)
+        return data.read()
+
+    @staticmethod
+    def get_instance():
+        if storage.instance is None:
+            storage.instance = storage()
+        return storage.instance
diff --git a/config/systems.json b/config/systems.json
@@ -249,7 +249,7 @@
         "username": "docker_user",
         "deployment": {
           "files": [
-            "handler.py",
+            "func.py",
             "storage.py"
           ],
           "packages": {
@@ -269,7 +269,7 @@
         "username": "docker_user",
         "deployment": {
           "files": [
-            "handler.js",
+            "index.js",
             "storage.js"
           ],
           "packages": []
diff --git a/sebs/knative/__init__.py b/sebs/knative/__init__.py
@@ -0,0 +1,2 @@
+from .knative import KnativeSystem  # noqa
+from .config import KnativeConfig  # noqa
diff --git a/sebs/knative/config.py b/sebs/knative/config.py
@@ -1,5 +1,5 @@
 from sebs.cache import Cache
-from sebs.faas.config import Credentials, Resources, Config
+from sebs.faas.config import Resources, Config
 from sebs.utils import LoggingHandlers
 from sebs.storage.config import MinioConfig
 
@@ -53,9 +53,9 @@ def registry_updated(self) -> bool:
     @staticmethod
     def initialize(res: Resources, dct: dict):
         ret = cast(KnativeResources, res)
-        ret._docker_registry = dct["registry"]
-        ret._docker_username = dct["username"]
-        ret._docker_password = dct["password"]
+        ret._docker_registry = dct.get("registry")
+        ret._docker_username = dct.get("username")
+        ret._docker_password = dct.get("password")
 
     @staticmethod
     def deserialize(config: dict, cache: Cache, handlers: LoggingHandlers) -> Resources:
@@ -101,7 +101,9 @@ def deserialize(config: dict, cache: Cache, handlers: LoggingHandlers) -> Resour
         # Check for new config
         if "storage" in config:
             ret._storage = MinioConfig.deserialize(config["storage"])
-            ret.logging.info("Using user-provided configuration of storage for Knative.")
+            ret.logging.info(
+                "Using user-provided configuration of storage for Knative."
+            )
 
             # check if there has been an update
             if not (
@@ -122,21 +124,26 @@ def deserialize(config: dict, cache: Cache, handlers: LoggingHandlers) -> Resour
             and "resources" in cached_config
             and "storage" in cached_config["resources"]
         ):
-            ret._storage = MinioConfig.deserialize(cached_config["resources"]["storage"])
+            ret._storage = MinioConfig.deserialize(
+                cached_config["resources"]["storage"]
+            )
             ret.logging.info("Using cached configuration of storage for Knative.")
 
         return ret
 
     def update_cache(self, cache: Cache):
         super().update_cache(cache)
         cache.update_config(
-            val=self.docker_registry, keys=["knative", "resources", "docker", "registry"]
+            val=self.docker_registry,
+            keys=["knative", "resources", "docker", "registry"],
         )
         cache.update_config(
-            val=self.docker_username, keys=["knative", "resources", "docker", "username"]
+            val=self.docker_username,
+            keys=["knative", "resources", "docker", "username"],
         )
         cache.update_config(
-            val=self.docker_password, keys=["knative", "resources", "docker", "password"]
+            val=self.docker_password,
+            keys=["knative", "resources", "docker", "password"],
         )
         if self._storage:
             self._storage.update_cache(["knative", "resources", "storage"], cache)
@@ -186,7 +193,7 @@ def deserialize(config: dict, cache: Cache, handlers: LoggingHandlers) -> Config
             KnativeResources, KnativeResources.deserialize(config, cache, handlers)
         )
 
-        res = KnativeConfig(config, cached_config)
+        res = KnativeConfig(config, cache)
         res.logging_handlers = handlers
         res._resources = resources
         return res
diff --git a/sebs/knative/function.py b/sebs/knative/function.py
@@ -7,6 +7,7 @@
 from sebs.faas.function import Function, FunctionConfig, Runtime
 from sebs.storage.config import MinioConfig
 
+
 @dataclass
 class KnativeFunctionConfig(FunctionConfig):
     docker_image: str = ""
@@ -31,9 +32,14 @@ def from_benchmark(benchmark: Benchmark) -> KnativeFunctionConfig:
             benchmark, KnativeFunctionConfig
         )
 
+
 class KnativeFunction(Function):
     def __init__(
-        self, name: str, benchmark: str, code_package_hash: str, cfg: KnativeFunctionConfig
+        self,
+        name: str,
+        benchmark: str,
+        code_package_hash: str,
+        cfg: KnativeFunctionConfig,
     ):
         super().__init__(benchmark, name, code_package_hash, cfg)
 
@@ -55,12 +61,17 @@ def deserialize(cached_config: dict) -> KnativeFunction:
 
         cfg = KnativeFunctionConfig.deserialize(cached_config["config"])
         ret = KnativeFunction(
-            cached_config["name"], cached_config["benchmark"], cached_config["hash"], cfg
+            cached_config["name"],
+            cached_config["benchmark"],
+            cached_config["hash"],
+            cfg,
         )
         for trigger in cached_config["triggers"]:
             trigger_type = cast(
                 Trigger,
-                {"Library": KnativeLibraryTrigger, "HTTP": KnativeHTTPTrigger}.get(trigger["type"]),
+                {"Library": KnativeLibraryTrigger, "HTTP": KnativeHTTPTrigger}.get(
+                    trigger["type"]
+                ),
             )
             assert trigger_type, "Unknown trigger type {}".format(trigger["type"])
             ret.add_trigger(trigger_type.deserialize(trigger))
diff --git a/sebs/knative/knative.py b/sebs/knative/knative.py
diff --git a/sebs/knative/storage.py b/sebs/knative/storage.py
diff --git a/sebs/knative/triggers.py b/sebs/knative/triggers.py
diff --git a/tools/knative_preparation.py b/tools/knative_preparation.py

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,2 @@`
	`1`	`+from .knative import KnativeSystem # noqa`
	`2`	`+from .config import KnativeConfig # noqa`