feat: Knative deployment support

Signed-off-by: Abhishek Kumar <[email protected]>

config/systems.json

@@ -234,5 +234,47 @@
         }
       }
     }
+  },
+  "knative": {
+    "languages": {
+      "python": {
+        "base_images": {
+          "3.9": "python:3.9-slim",
+          "3.10": "python:3.10-slim"
+        },
+        "images": [
+          "build",
+          "run"
+        ],
+        "username": "docker_user",
+        "deployment": {
+          "files": [
+            "handler.py",
+            "storage.py"
+          ],
+          "packages": {
+            "parliament-functions": "0.1.0"
+          }
+        }
+      },
+      "nodejs": {
+        "base_images": {
+          "20": "node:20",
+          "18": "node:18"
+        },
+        "images": [
+          "build",
+          "run"
+        ],
+        "username": "docker_user",
+        "deployment": {
+          "files": [
+            "handler.js",
+            "storage.js"
+          ],
+          "packages": []
+        }
+      }
+    }
   }
 }

sebs.py

@@ -88,7 +88,7 @@ def common_params(func):
     @click.option(
         "--deployment",
         default=None,
-        type=click.Choice(["azure", "aws", "gcp", "local", "openwhisk"]),
+        type=click.Choice(["azure", "aws", "gcp", "local", "openwhisk", "knative"]),
         help="Cloud deployment to use.",
     )
     @click.option(

sebs/faas/config.py

@@ -204,6 +204,10 @@ def deserialize(config: dict, cache: Cache, handlers: LoggingHandlers) -> Config
             from sebs.openwhisk.config import OpenWhiskConfig
 
             implementations["openwhisk"] = OpenWhiskConfig.deserialize
+        if has_platform("knative"):
+            from sebs.knative.config import KnativeConfig
+
+            implementations["knative"] = KnativeConfig.deserialize
         func = implementations.get(name)
         assert func, "Unknown config type!"
         return func(config[name] if name in config else config, cache, handlers)

sebs/knative/config.py

@@ -0,0 +1,196 @@
+from sebs.cache import Cache
+from sebs.faas.config import Credentials, Resources, Config
+from sebs.utils import LoggingHandlers
+from sebs.storage.config import MinioConfig
+
+from typing import cast, Optional
+
+
+class KnativeResources(Resources):
+    def __init__(
+        self,
+        registry: Optional[str] = None,
+        username: Optional[str] = None,
+        password: Optional[str] = None,
+        registry_updated: bool = False,
+    ):
+        super().__init__(name="knative")
+        self._docker_registry = registry if registry != "" else None
+        self._docker_username = username if username != "" else None
+        self._docker_password = password if password != "" else None
+        self._registry_updated = registry_updated
+        self._storage: Optional[MinioConfig] = None
+        self._storage_updated = False
+
+    @staticmethod
+    def typename() -> str:
+        return "Knative.Resources"
+
+    @property
+    def docker_registry(self) -> Optional[str]:
+        return self._docker_registry
+
+    @property
+    def docker_username(self) -> Optional[str]:
+        return self._docker_username
+
+    @property
+    def docker_password(self) -> Optional[str]:
+        return self._docker_password
+
+    @property
+    def storage_config(self) -> Optional[MinioConfig]:
+        return self._storage
+
+    @property
+    def storage_updated(self) -> bool:
+        return self._storage_updated
+
+    @property
+    def registry_updated(self) -> bool:
+        return self._registry_updated
+
+    @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"]
+
+    @staticmethod
+    def deserialize(config: dict, cache: Cache, handlers: LoggingHandlers) -> Resources:
+
+        cached_config = cache.get_config("knative")
+        ret = KnativeResources()
+        if cached_config:
+            super(KnativeResources, KnativeResources).initialize(
+                ret, cached_config["resources"]
+            )
+
+        # Check for new config - overrides but check if it's different
+        if "docker_registry" in config:
+
+            KnativeResources.initialize(ret, config["docker_registry"])
+            ret.logging.info("Using user-provided Docker registry for Knative.")
+            ret.logging_handlers = handlers
+
+            # check if there has been an update
+            if not (
+                cached_config
+                and "resources" in cached_config
+                and "docker" in cached_config["resources"]
+                and cached_config["resources"]["docker"] == config["docker_registry"]
+            ):
+                ret._registry_updated = True
+
+        # Load cached values
+        elif (
+            cached_config
+            and "resources" in cached_config
+            and "docker" in cached_config["resources"]
+        ):
+            KnativeResources.initialize(ret, cached_config["resources"]["docker"])
+            ret.logging_handlers = handlers
+            ret.logging.info("Using cached Docker registry for Knative")
+        else:
+            ret = KnativeResources()
+            ret.logging.info("Using default Docker registry for Knative.")
+            ret.logging_handlers = handlers
+            ret._registry_updated = True
+
+        # 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.")
+
+            # check if there has been an update
+            if not (
+                cached_config
+                and "resources" in cached_config
+                and "storage" in cached_config["resources"]
+                and cached_config["resources"]["storage"] == config["storage"]
+            ):
+                ret.logging.info(
+                    "User-provided configuration is different from cached storage, "
+                    "we will update existing Knative actions."
+                )
+                ret._storage_updated = True
+
+        # Load cached values
+        elif (
+            cached_config
+            and "resources" in cached_config
+            and "storage" in cached_config["resources"]
+        ):
+            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"]
+        )
+        cache.update_config(
+            val=self.docker_username, keys=["knative", "resources", "docker", "username"]
+        )
+        cache.update_config(
+            val=self.docker_password, keys=["knative", "resources", "docker", "password"]
+        )
+        if self._storage:
+            self._storage.update_cache(["knative", "resources", "storage"], cache)
+
+    def serialize(self) -> dict:
+        out: dict = {
+            **super().serialize(),
+            "docker_registry": self.docker_registry,
+            "docker_username": self.docker_username,
+            "docker_password": self.docker_password,
+        }
+        if self._storage:
+            out = {**out, "storage": self._storage.serialize()}
+        return out
+
+
+class KnativeConfig(Config):
+    name: str
+    cache: Cache
+
+    def __init__(self, config: dict, cache: Cache):
+        super().__init__(name="knative")
+        self._resources = KnativeResources()
+        self.knative_exec = config["knativeExec"]
+        self.cache = cache
+
+    @property
+    def resources(self) -> KnativeResources:
+        return self._resources
+
+    @staticmethod
+    def initialize(cfg: Config, dct: dict):
+        cfg._region = dct["region"]
+
+    def serialize(self) -> dict:
+        return {
+            "name": self._name,
+            "region": self._region,
+            "knativeExec": self.knative_exec,
+            "resources": self._resources.serialize(),
+        }
+
+    @staticmethod
+    def deserialize(config: dict, cache: Cache, handlers: LoggingHandlers) -> Config:
+        cached_config = cache.get_config("knative")
+        resources = cast(
+            KnativeResources, KnativeResources.deserialize(config, cache, handlers)
+        )
+
+        res = KnativeConfig(config, cached_config)
+        res.logging_handlers = handlers
+        res._resources = resources
+        return res
+
+    def update_cache(self, cache: Cache):
+        cache.update_config(val=self.knative_exec, keys=["knative", "knativeExec"])
+        self.resources.update_cache(cache)

sebs/knative/function.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+from typing import cast, Optional
+from dataclasses import dataclass
+
+from sebs.benchmark import Benchmark
+from sebs.faas.function import Function, FunctionConfig, Runtime
+from sebs.storage.config import MinioConfig
+
+@dataclass
+class KnativeFunctionConfig(FunctionConfig):
+    docker_image: str = ""
+    namespace: str = "default"
+    storage: Optional[MinioConfig] = None
+    url: str = ""
+
+    @staticmethod
+    def deserialize(data: dict) -> KnativeFunctionConfig:
+        keys = list(KnativeFunctionConfig.__dataclass_fields__.keys())
+        data = {k: v for k, v in data.items() if k in keys}
+        data["runtime"] = Runtime.deserialize(data["runtime"])
+        data["storage"] = MinioConfig.deserialize(data["storage"])
+        return KnativeFunctionConfig(**data)
+
+    def serialize(self) -> dict:
+        return self.__dict__
+
+    @staticmethod
+    def from_benchmark(benchmark: Benchmark) -> KnativeFunctionConfig:
+        return super(KnativeFunctionConfig, KnativeFunctionConfig)._from_benchmark(
+            benchmark, KnativeFunctionConfig
+        )
+
+class KnativeFunction(Function):
+    def __init__(
+        self, name: str, benchmark: str, code_package_hash: str, cfg: KnativeFunctionConfig
+    ):
+        super().__init__(benchmark, name, code_package_hash, cfg)
+
+    @property
+    def config(self) -> KnativeFunctionConfig:
+        return cast(KnativeFunctionConfig, self._cfg)
+
+    @staticmethod
+    def typename() -> str:
+        return "Knative.Function"
+
+    def serialize(self) -> dict:
+        return {**super().serialize(), "config": self._cfg.serialize()}
+
+    @staticmethod
+    def deserialize(cached_config: dict) -> KnativeFunction:
+        from sebs.faas.function import Trigger
+        from sebs.knative.triggers import KnativeLibraryTrigger, KnativeHTTPTrigger
+
+        cfg = KnativeFunctionConfig.deserialize(cached_config["config"])
+        ret = KnativeFunction(
+            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"]),
+            )
+            assert trigger_type, "Unknown trigger type {}".format(trigger["type"])
+            ret.add_trigger(trigger_type.deserialize(trigger))
+        return ret