Fission benchmark

.gitignore

@@ -170,3 +170,6 @@ dmypy.json
 sebs-*
 # cache
 cache
+
+# ide
+.vscode

benchmarks/wrappers/fission/nodejs/handler.js

@@ -0,0 +1,31 @@
+const path = require('path'), fs = require('fs');
+
+exports.handler = async function(context) {
+    var body = JSON.stringify(context.request.body);
+    var unbody = JSON.parse(body);
+    var func = require('./function/function');
+    var begin = Date.now()/1000;
+    var start = process.hrtime();
+    var ret = await func.handler(unbody);
+    var elapsed = process.hrtime(start);
+    var end = Date.now()/1000;
+    var micro = elapsed[1] / 1e3 + elapsed[0] * 1e6;
+    var is_cold = false;
+    var fname = path.join('/tmp','cold_run');
+    if(!fs.existsSync(fname)) {
+        is_cold = true;
+        fs.closeSync(fs.openSync(fname, 'w'));
+    }
+
+    return {
+        status: 200,
+        body: JSON.stringify({
+            begin,
+            end,
+            compute_time: micro,
+            results_time: 0,
+            result: ret,
+            is_cold
+        })
+    };
+};

benchmarks/wrappers/fission/python/handler.py

@@ -0,0 +1,42 @@
+from flask import request, jsonify, current_app
+
+import json
+import datetime
+import os
+
+
+def handler():
+    body = request.get_data().decode("utf-8")
+    current_app.logger.info("Body: " + body)
+    event = json.loads(body)
+    current_app.logger.info("Event: " + str(event))
+    begin = datetime.datetime.now()
+    from function import function
+
+    ret = function.handler(event)
+    end = datetime.datetime.now()
+    current_app.logger.info("Function result: " + str(ret))
+    log_data = {"result": ret["result"]}
+    if "measurement" in ret:
+        log_data["measurement"] = ret["measurement"]
+
+    results_time = (end - begin) / datetime.timedelta(microseconds=1)
+
+    # cold test
+    is_cold = False
+    fname = "cold_run"
+    if not os.path.exists(fname):
+        is_cold = True
+        open(fname, "a").close()
+
+    return jsonify(
+        json.dumps(
+            {
+                "begin": begin.strftime("%s.%f"),
+                "end": end.strftime("%s.%f"),
+                "results_time": results_time,
+                "is_cold": is_cold,
+                "result": log_data,
+            }
+        )
+    )

benchmarks/wrappers/fission/python/storage.py

@@ -0,0 +1,60 @@
+import os
+import uuid
+import json
+import minio
+from flask import current_app
+
+
+class storage:
+    instance = None
+    client = None
+
+    def __init__(self):
+        file = open(os.path.join(os.path.dirname(__file__), "minioConfig.json"), "r")
+        minioConfig = json.load(file)
+        try:
+            self.client = minio.Minio(
+                minioConfig["url"],
+                access_key=minioConfig["access_key"],
+                secret_key=minioConfig["secret_key"],
+                secure=False,
+            )
+        except Exception as e:
+            current_app.logger.info(e)
+
+    @staticmethod
+    def unique_name(name):
+        name, extension = name.split(".")
+        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_v2(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()
+
+    def get_instance():
+        if storage.instance is None:
+            storage.instance = storage()
+        return storage.instance

config/fission.json

@@ -0,0 +1,16 @@
+{
+  "experiments": {
+    "update_code": false,
+    "update_storage": false,
+    "download_results": false,
+    "deployment": "fission",
+    "runtime": {
+      "language": "python",
+      "version": "3.6"
+    }
+  },
+  "deployment": {
+    "name": "fission",
+    "shouldShutdown": false
+  }
+}

config/systems.json

@@ -100,5 +100,35 @@
         "username": "docker_user"
       }
     }
+  },
+  "fission": { 
+    "languages": {
+      "python": {
+        "base_images": {
+          "env":"fission/python-env:latest",
+          "builder":"fission/python-builder:latest"
+        },
+        "versions": ["3.6"],
+        "images": [],
+        "username": "docker_user",
+        "deployment": {
+          "files": ["handler.py", "storage.py"],
+          "packages": {"minio" : "^5.0.10"}
+        }
+      },
+      "nodejs": {
+        "base_images": {
+          "env":"fission/node-env:latest",
+          "builder":"fission/node-builder:latest"
+        },
+        "versions": ["10.x", "12.x"],
+        "images": [],
+        "username": "docker_user",
+        "deployment": {
+          "files": ["handler.js"],
+          "packages": {"minio" : "^7.0.16"}
+        }
+      }
+    }
   }
 }

install.py

@@ -4,10 +4,15 @@
 import os
 import subprocess
 
+from tools.fission_preparation import check_if_minikube_installed, check_if_k8s_installed, check_if_helm_installed, \
+    check_if_fission_cli_installed, install_fission_cli
+
 parser = argparse.ArgumentParser(description="Install SeBS and dependencies.")
 parser.add_argument("--with-pypapi", action="store_true")
+parser.add_argument("--with-fission", action="store_true")
 args = parser.parse_args()
 
+
 def execute(cmd):
     ret = subprocess.run(
         cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, shell=True
@@ -18,19 +23,20 @@ def execute(cmd):
         )
     return ret.stdout.decode("utf-8")
 
-env_dir="sebs-virtualenv"
-
-print("Creating Python virtualenv at {}".format(env_dir))
-execute("python3 -mvenv {}".format(env_dir))
+# env_dir="sebs-virtualenv"
+#
+# print("Creating Python virtualenv at {}".format(env_dir))
+# execute("python3 -mvenv {}".format(env_dir))
+#
+# print("Install Python dependencies with pip")
+# execute(". {}/bin/activate && pip3 install -r requirements.txt".format(env_dir))
+#
+# print("Configure mypy extensions")
+# execute(". {}/bin/activate && mypy_boto3".format(env_dir))
+#
+# print("Initialize git submodules")
+# execute("git submodule update --init --recursive")
 
-print("Install Python dependencies with pip")
-execute(". {}/bin/activate && pip3 install -r requirements.txt".format(env_dir))
-
-print("Configure mypy extensions")
-execute(". {}/bin/activate && mypy_boto3".format(env_dir))
-
-print("Initialize git submodules")
-execute("git submodule update --init --recursive")
 
 if args.with_pypapi:
     print("Build and install pypapi")
@@ -41,3 +47,13 @@ def execute(cmd):
     execute("python3 setup.py build")
     execute("python3 pypapi/papi_build.py")
     os.chdir(cur_dir)
+
+if args.with_fission:
+    check_if_minikube_installed()
+    check_if_k8s_installed()
+    check_if_helm_installed()
+    try:
+        check_if_fission_cli_installed(throws_error=True)
+    except subprocess.CalledProcessError:
+        install_fission_cli()
+        check_if_fission_cli_installed()

mypy.ini

@@ -2,3 +2,6 @@
 
 [mypy-docker]
 ignore_missing_imports = True
+
+[mypy-minio]
+ignore_missing_imports = True

sebs/aws/s3.py

@@ -95,7 +95,7 @@ def add_output_bucket(
         self, name: str, suffix: str = "output", cache: bool = True
     ) -> Tuple[str, int]:
 
-        idx = self.request_input_buckets
+        idx = self.request_input_buckets  # are you sure?
         name = "{}-{}-{}".format(name, idx + 1, suffix)
         if cache:
             self.request_input_buckets += 1

sebs/config.py

@@ -35,3 +35,10 @@ def supported_language_versions(
         return self._system_config[deployment_name]["languages"][language_name][
             "base_images"
         ].keys()
+
+    def benchmark_base_images(
+        self, deployment_name: str, language_name: str
+    ) -> Dict[str, str]:
+        return self._system_config[deployment_name]["languages"][language_name][
+            "base_images"
+        ]

sebs/fission/__init__.py

@@ -0,0 +1,3 @@
+from .fission import Fission  # noqa
+from .config import FissionConfig  # noqa
+from .minio import Minio  # noqa

sebs/fission/config.py

@@ -0,0 +1,50 @@
+from sebs.faas.config import Config, Credentials, Resources
+from sebs.cache import Cache
+
+
+class FissionCredentials(Credentials):
+    def __init__(self):
+        pass
+
+    @staticmethod
+    def initialize(config: dict, cache: Cache) -> Credentials:
+        return FissionCredentials()
+
+    def serialize(self) -> dict:
+        pass
+
+
+class FissionResources(Resources):
+    def __init__(self):
+        pass
+
+    def serialize(self) -> dict:
+        return {}
+
+    @staticmethod
+    def initialize(config: dict, cache: Cache) -> Resources:
+        pass
+
+
+class FissionConfig(Config):
+    name: str
+    cache: Cache
+    shouldShutdown: bool
+
+    def __init__(self, config: dict, cache: Cache):
+        self.name = config["name"]
+        self.shouldShutdown = config["shouldShutdown"]
+        self.cache = cache
+
+    @staticmethod
+    def initialize(config: dict, cache: Cache) -> Config:
+        return FissionConfig(config, cache)
+
+    def credentials(self) -> Credentials:
+        pass
+
+    def resources(self) -> Resources:
+        pass
+
+    def serialize(self) -> dict:
+        pass