Fission benchmark

.gitignore

@@ -179,6 +179,8 @@ sebs-*
 # cache
 cache
 
+# ide
+.vscode
 # IntelliJ IDEA files
 .idea
 *.iml

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/nodejs/storage.js

@@ -0,0 +1,63 @@
+
+const minio = require('minio'),
+      uuid = require('uuid'),
+      util = require('util'),
+      stream = require('stream'),
+      fs = require('fs');
+
+class minio_storage {
+
+  constructor() {
+    let minioConfig = JSON.parse(fs.readFileSync('minioConfig.json'));
+    let address = minioConfig["url"];
+    let access_key = minioConfig["access_key"];
+    let secret_key = minioConfig["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, extension] = file.split('.');
+    let uuid_name = uuid.v4().split('-')[0];
+    return util.format('%s.%s.%s', name, uuid_name, extension);
+  }
+
+  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;

benchmarks/wrappers/fission/python/__main__.py

@@ -0,0 +1,51 @@
+import logging
+import datetime
+import os
+
+import minio
+
+def main(args):
+    logging.getLogger().setLevel(logging.INFO)
+    begin = datetime.datetime.now()
+    args['request-id'] = os.getenv('__OW_ACTIVATION_ID')
+    args['income-timestamp'] = begin.timestamp()
+
+    for arg in ["MINIO_STORAGE_CONNECTION_URL", "MINIO_STORAGE_ACCESS_KEY", "MINIO_STORAGE_SECRET_KEY"]:
+        os.environ[arg] = args[arg]
+        del args[arg]
+
+    try:
+        from function import function
+        ret = function.handler(args)
+        end = datetime.datetime.now()
+        logging.info("Function result: {}".format(ret))
+        log_data = {"result": ret["result"]}
+        if "measurement" in ret:
+            log_data["measurement"] = ret["measurement"]
+
+        results_time = (end - begin) / datetime.timedelta(microseconds=1)
+
+        is_cold = False
+        fname = "cold_run"
+        if not os.path.exists(fname):
+            is_cold = True
+            open(fname, "a").close()
+
+        return {
+            "begin": begin.strftime("%s.%f"),
+            "end": end.strftime("%s.%f"),
+            "request_id": os.getenv('__OW_ACTIVATION_ID'),
+            "results_time": results_time,
+            "is_cold": is_cold,
+            "result": log_data,
+        }
+    except Exception as e:
+        end = datetime.datetime.now()
+        results_time = (end - begin) / datetime.timedelta(microseconds=1)
+        return {
+            "begin": begin.strftime("%s.%f"),
+            "end": end.strftime("%s.%f"),
+            "request_id": os.getenv('__OW_ACTIVATION_ID'),
+            "results_time": results_time,
+            "result": f"Error - invocation failed! Reason: {e}"
+        }

benchmarks/wrappers/fission/python/handler.py

@@ -0,0 +1,67 @@
+import datetime, io, json, os, sys, uuid
+
+from flask import request
+from flask import current_app
+
+# Add current directory to allow location of packages
+sys.path.append(os.path.join(os.path.dirname(__file__), '.python_packages/lib/site-packages'))
+
+
+def handler():
+    current_app.logger.info("Received request")
+    event = request.json
+    income_timestamp = datetime.datetime.now().timestamp()
+
+    req_id = str(uuid.uuid4())
+    event['income-timestamp'] = income_timestamp
+    begin = datetime.datetime.now()
+    from function import function
+    ret = function.handler(event)
+    end = datetime.datetime.now()
+
+    log_data = {
+        'output': ret['result']
+    }
+    if 'measurement' in ret:
+        log_data['measurement'] = ret['measurement']
+    if 'logs' in event:
+        log_data['time'] = (end - begin) / datetime.timedelta(microseconds=1)
+        results_begin = datetime.datetime.now()
+        from function import storage
+        storage_inst = storage.storage.get_instance()
+        b = event.get('logs').get('bucket')
+        storage_inst.upload_stream(b, '{}.json'.format(req_id),
+                io.BytesIO(json.dumps(log_data).encode('utf-8')))
+        results_end = datetime.datetime.now()
+        results_time = (results_end - results_begin) / datetime.timedelta(microseconds=1)
+    else:
+        results_time = 0
+
+    # cold test
+    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()
+
+    cold_start_var = ""
+    if "cold_start" in os.environ:
+        cold_start_var = os.environ["cold_start"]
+
+    return {
+        'statusCode': 200,
+        'begin': begin.strftime('%s.%f'),
+        'end': end.strftime('%s.%f'),
+        'results_time': results_time,
+        'is_cold': is_cold,
+        'result': log_data,
+        'request_id': req_id,
+        'cold_start_var': cold_start_var,
+        'container_id': container_id,
+    }
+

benchmarks/wrappers/fission/python/setup.py

@@ -0,0 +1,14 @@
+from distutils.core import setup
+from glob import glob
+from pkg_resources import parse_requirements
+
+with open('requirements.txt') as f:
+    requirements = [str(r) for r in parse_requirements(f)]
+
+setup(
+    name='function',
+    install_requires=requirements,
+    packages=['function'],
+    package_dir={'function': '.'},
+    package_data={'function': glob('**', recursive=True)},
+)

benchmarks/wrappers/fission/python/storage.py

@@ -0,0 +1,150 @@
+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]
+                )
+            )
+            print("WE get from os")
+            print(os.getenv("MINIO_STORAGE_CONNECTION_URL"))
+            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
+
+
+
+
+
+
+
+
+# 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