Adapters with IRuntimeAdapter interface

Author: patuwwy

packages/adapter-docker/src/docker-instance-adapter.ts

@@ -11,7 +11,7 @@ import {
     InstanceConfig,
     RunnerContainerConfiguration,
     InstanceLimits,
-    STHConfiguration,
+    DockerAdapterConfiguration,
 } from "@scramjet/types";
 import path from "path";
 import { DockerodeDockerHelper } from "./dockerode-docker-helper";
@@ -32,6 +32,7 @@ IComponent {
     private dockerHelper: IDockerHelper;
     private _limits?: InstanceLimits = {};
     private resources: DockerAdapterResources = {};
+
     id: string = "";
 
     logger: IObjectLogger;
@@ -41,7 +42,7 @@ IComponent {
     get limits() { return this._limits || {} as InstanceLimits; }
     private set limits(value: InstanceLimits) { this._limits = value; }
 
-    constructor(_sthConfig: STHConfiguration, id: string = "") {
+    constructor(_config: DockerAdapterConfiguration, id: string = "") {
         this.dockerHelper = new DockerodeDockerHelper();
 
         this.logger = new ObjLogger(this, { id });

packages/adapter-docker/src/index.ts

@@ -1,13 +1,39 @@
+import { DockerAdapterConfiguration, IInstanceAdapter, IRuntimeAdapter, ISequenceAdapter } from "@scramjet/types";
 import { DockerodeDockerHelper } from "./dockerode-docker-helper";
+import { DockerSequenceAdapter } from "./docker-sequence-adapter";
+import { DockerInstanceAdapter } from "./docker-instance-adapter";
 
 /**
  * Adapter module must provide SequenceAdapter, InstanceAdapter classes, init method and name field.
  */
-export { DockerSequenceAdapter as SequenceAdapter } from "./docker-sequence-adapter";
-export { DockerInstanceAdapter as InstanceAdapter } from "./docker-instance-adapter";
+export default class DockerAdapter implements IRuntimeAdapter {
+    name = "docker";
 
-export const init = async (..._args: any[]) => {
-    return await DockerodeDockerHelper.isDockerConfigured() ? {} : { error: "Docker initialization failed" };
-};
+    #_instanceAdapter: IInstanceAdapter;
+    #_sequenceAdapter: ISequenceAdapter;
 
-export const name = "docker";
+    get instanceAdapter() {
+        return this.#_instanceAdapter;
+    }
+
+    get sequenceAdapter() {
+        return this.#_sequenceAdapter;
+    }
+
+    config: DockerAdapterConfiguration;
+
+    constructor(config: DockerAdapterConfiguration) {
+        this.config = config;
+
+        this.#_instanceAdapter = new DockerInstanceAdapter(this.config);
+        this.#_sequenceAdapter = new DockerSequenceAdapter(this.config);
+    }
+
+    getSequenceAdapter(): ISequenceAdapter {
+        return this.#_sequenceAdapter;
+    }
+
+    async init(config: any): Promise<{ error?: string }> {
+        return await DockerodeDockerHelper.isDockerConfigured() ? {} : { error: "Docker initialization failed" };
+    }
+}

packages/adapter-k8s/src/index.ts

@@ -1,11 +1,41 @@
+import { IInstanceAdapter, IRuntimeAdapter, ISequenceAdapter, K8SAdapterConfiguration } from "@scramjet/types";
+import { KubernetesInstanceAdapter } from "./kubernetes-instance-adapter";
+import { KubernetesSequenceAdapter } from "./kubernetes-sequence-adapter";
+
 /**
  * Adapter module must provide SequenceAdapter, InstanceAdapter classes, init method and name field.
  */
 export { KubernetesSequenceAdapter as SequenceAdapter } from "./kubernetes-sequence-adapter";
 export { KubernetesInstanceAdapter as InstanceAdapter } from "./kubernetes-instance-adapter";
 
-export const init = async (..._args: any[]) => {
-    return Promise.resolve({ });
-};
+export default class KubernetesAdapter implements IRuntimeAdapter {
+    name = "kubernetes";
+
+    #_instanceAdapter: IInstanceAdapter;
+    #_sequenceAdapter: ISequenceAdapter;
+
+    get instanceAdapter() {
+        return this.#_instanceAdapter;
+    }
+
+    get sequenceAdapter() {
+        return this.#_sequenceAdapter;
+    }
+
+    config: K8SAdapterConfiguration;
+
+    constructor(config: K8SAdapterConfiguration) {
+        this.config = config;
+
+        this.#_instanceAdapter = new KubernetesInstanceAdapter(this.config);
+        this.#_sequenceAdapter = new KubernetesSequenceAdapter(this.config);
+    }
+
+    getSequenceAdapter(): ISequenceAdapter {
+        return this.#_sequenceAdapter;
+    }
 
-export const name = "kubernetes";
+    async init(config: any): Promise<{ error?: string }> {
+        return Promise.resolve({});
+    }
+}

packages/adapter-k8s/src/kubernetes-instance-adapter.ts

@@ -15,7 +15,7 @@ import path from "path";
 import { ObjLogger } from "@scramjet/obj-logger";
 import { createReadStream } from "fs";
 import { KubernetesClientAdapter } from "./kubernetes-client-adapter";
-import { adapterConfigDecoder } from "./kubernetes-config-decoder";
+//import { adapterConfigDecoder } from "./kubernetes-config-decoder";
 import { getRunnerEnvEntries } from "@scramjet/adapters-utils";
 import { PassThrough } from "stream";
 import { RunnerExitCode } from "@scramjet/symbols";
@@ -42,13 +42,13 @@ IComponent {
     constructor(config: K8SAdapterConfiguration) {
         // @TODO this is a redundant check (it was already checked in sequence adapter)
         // We should move this to config service decoding: https://github.com/scramjetorg/transform-hub/issues/279
-        const decodedAdapterConfig = adapterConfigDecoder.decode(config);
+        // const decodedAdapterConfig = adapterConfigDecoder.decode(config);
 
-        if (!decodedAdapterConfig.isOk()) {
-            throw new Error("Invalid Kubernetes Adapter configuration");
-        }
+        // if (!decodedAdapterConfig.isOk()) {
+        //     throw new Error("Invalid Kubernetes Adapter configuration");
+        // }
 
-        this.adapterConfig = decodedAdapterConfig.value;
+        this.adapterConfig = config;//decodedAdapterConfig.value;
         this.logger = new ObjLogger(this);
     }
 

packages/adapter-k8s/src/kubernetes-sequence-adapter.ts

@@ -57,7 +57,7 @@ class KubernetesSequenceAdapter implements ISequenceAdapter {
 
     private adapterConfig: K8SAdapterConfiguration;
 
-    constructor(config: KubernetesSequenceAdapter) {
+    constructor(config: K8SAdapterConfiguration) {
         const decodedAdapterConfig = adapterConfigDecoder.decode(config);
 
         if (!decodedAdapterConfig.isOk()) {

packages/adapter-process/src/index.ts

@@ -1,15 +1,43 @@
+import { IInstanceAdapter, IRuntimeAdapter, ISequenceAdapter, ProcessAdapterConfiguration } from "@scramjet/types";
+import { ProcessInstanceAdapter } from "./process-instance-adapter";
+import { ProcessSequenceAdapter } from "./process-sequence-adapter";
+
 /**
- * Adapter module must provide SequenceAdapter, InstanceAdapter classes, init method and name field.
+ * Adapter class must implement IRuntimeAdapter Interface.
  */
-export { ProcessSequenceAdapter as SequenceAdapter } from "./process-sequence-adapter";
-export { ProcessInstanceAdapter as InstanceAdapter } from "./process-instance-adapter";
 
-export const init = async (config: any) => {
-    if (!config.sequencesRoot) {
-        return Promise.reject({ error: "No 'sequencesRoot' in config" });
+export default class ProcessAdapter implements IRuntimeAdapter {
+    name = "process";
+
+    #_instanceAdapter: IInstanceAdapter;
+    #_sequenceAdapter: ISequenceAdapter;
+
+    get instanceAdapter() {
+        return this.#_instanceAdapter;
+    }
+
+    get sequenceAdapter() {
+        return this.#_sequenceAdapter;
+    }
+
+    config: ProcessAdapterConfiguration;
+
+    constructor(config: ProcessAdapterConfiguration) {
+        this.config = config;
+
+        this.#_instanceAdapter = new ProcessInstanceAdapter(this.config);
+        this.#_sequenceAdapter = new ProcessSequenceAdapter(this.config);
     }
 
-    return Promise.resolve({});
-};
+    getSequenceAdapter(): ISequenceAdapter {
+        return this.#_sequenceAdapter;
+    }
+
+    init(config: any): Promise<{ error?: string }> {
+        if (!config.sequencesRoot) {
+            return Promise.reject({ error: "No 'sequencesRoot' in config" });
+        }
 
-export const name = "process";
+        return Promise.resolve({});
+    }
+}

packages/adapter-process/src/process-instance-adapter.ts

@@ -1,11 +1,12 @@
-import { STHConfiguration,
+import {
     ExitCode,
     InstanceConfig,
     InstanceLimits,
     IObjectLogger,
     MonitoringMessageData,
     SequenceConfig,
-    IInstanceAdapter
+    IInstanceAdapter,
+    ProcessAdapterConfiguration
 } from "@scramjet/types";
 import { ObjLogger } from "@scramjet/obj-logger";
 import { streamToString } from "@scramjet/utility";
@@ -22,7 +23,7 @@ const gotPython = "\n                              _ \n __      _____  _ __  ___
  */
 class ProcessInstanceAdapter implements IInstanceAdapter {
     logger: IObjectLogger;
-    sthConfig: STHConfiguration;
+    config: ProcessAdapterConfiguration;
 
     private runnerProcess?: ChildProcess;
     private crashLogStreams?: Promise<string[]>;
@@ -34,9 +35,9 @@ class ProcessInstanceAdapter implements IInstanceAdapter {
         this.logger.warn("Limits are not yet supported in process runner");
     }
 
-    constructor(config: STHConfiguration) {
+    constructor(config: ProcessAdapterConfiguration) {
         this.logger = new ObjLogger(this);
-        this.sthConfig = config;
+        this.config = config;
     }
 
     async init(): Promise<void> {
@@ -71,7 +72,7 @@ class ProcessInstanceAdapter implements IInstanceAdapter {
             this.logger.trace(gotPython);
             const runnerPath = path.resolve(__dirname, require.resolve("@scramjet/python-runner"));
 
-            if (this.sthConfig.debug)
+            if (this.config.debug)
                 debugFlags = ["-m", "pdb", "-c", "continue"];
 
             return [
@@ -82,7 +83,7 @@ class ProcessInstanceAdapter implements IInstanceAdapter {
                 "./python-runner-startup.log",
             ];
         }
-        if (this.sthConfig.debug)
+        if (this.config.debug)
             debugFlags = ["--inspect-brk=9229"];
 
         return [

packages/host/src/lib/adapter-manager.ts

@@ -1,8 +1,13 @@
 import { ObjLogger } from "@scramjet/obj-logger";
 import { IRuntimeAdapter, InstanceRequirements, STHConfiguration } from "@scramjet/types";
 
+export type InitializedRuntimeAdapter = IRuntimeAdapter & {
+    pkgName: string;
+    status: "ready" | { error?: string };
+};
+
 export class AdapterManager {
-    adapters: { [key: string]: IRuntimeAdapter } = {};
+    adapters: { [key: string]: InitializedRuntimeAdapter } = {};
     sthConfig: STHConfiguration;
 
     logger = new ObjLogger(this);
@@ -23,8 +28,8 @@ export class AdapterManager {
             Object.keys(this.sthConfig.adapters).map(
                 async (pkgName) => {
                     const typedPkgName = pkgName as unknown as keyof STHConfiguration["adapters"];
-                    const adapter = Object.assign(await import(pkgName), { pkgName }) as IRuntimeAdapter;
-
+                    const adapter = Object.assign(new ((await import(pkgName)).default), { pkgName }) as InitializedRuntimeAdapter;
+console.log(adapter)
                     if (!AdapterManager.validateAdapter(adapter)) {
                         throw new Error(`Invalid adapter provided ${adapter.pkgName}`);
                     }
@@ -63,7 +68,7 @@ export class AdapterManager {
      * @returns {boolean} True if required fields are available.
      */
     static validateAdapter(adapter: IRuntimeAdapter): boolean {
-        return !!(adapter.name.trim() && ["SequenceAdapter", "InstanceAdapter", "init"].every((className: string) => className in adapter));
+        return !!(adapter.name?.trim() && ["sequenceAdapter", "instanceAdapter", "init"].every((className: string) => className in adapter));
     }
 
     /**
@@ -84,7 +89,7 @@ export class AdapterManager {
      * @param {string} pkgName Adapter package name.
      * @returns {IRuntimeAdapter} Adapter
      */
-    getAdapterByName(pkgName: string): IRuntimeAdapter | undefined {
+    getAdapterByName(pkgName: string): InitializedRuntimeAdapter | undefined {
         return Object.values(this.adapters).find(a => a.pkgName === pkgName);
     }
 

packages/host/src/lib/host.ts

@@ -104,6 +104,10 @@ export class Host implements IComponent {
      * Instance of CPMConnector used to communicate with Manager.
      */
     cpmConnector?: CPMConnector;
+
+    /**
+     *
+     */
     adapterManager!: AdapterManager;
 
     /**
@@ -338,7 +342,7 @@ export class Host implements IComponent {
 
             this.logger.info(`Default adapter for running Sequences: "${this.adapterName}" (${defaultAdapter?.name})`);
         } else {
-            this.logger.warn("No RuntimeAdapters.");
+            this.logger.warn("No Runtime Adapters.");
         }
 
         this.pushTelemetry("Host started");
@@ -499,7 +503,6 @@ export class Host implements IComponent {
         });
 
         this.api.op("delete", `${this.apiBase}/sequence/:id`, (req: ParsedMessage) => this.handleDeleteSequence(req));
-
         this.api.op("post", `${this.apiBase}/sequence/:id/start`, async (req: ParsedMessage) => this.handleStartSequence(req));
 
         this.api.get(`${this.apiBase}/sequence/:id`, (req) => this.getSequence(req.params?.id));
@@ -668,7 +671,7 @@ export class Host implements IComponent {
                 };
             }
 
-            const sequenceAdapter = new adapter.SequenceAdapter(this.config);
+            const sequenceAdapter = adapter.sequenceAdapter;
 
             await sequenceAdapter.remove(sequenceInfo.config);
             this.sequencesStore.delete(id);
@@ -739,7 +742,7 @@ export class Host implements IComponent {
             throw new Error("Error identifying existing sequences. Adapter unavailable");
         }
 
-        const sequenceAdapter = new adapter!.SequenceAdapter(this.config);
+        const sequenceAdapter = adapter!.sequenceAdapter;
 
         try {
             await sequenceAdapter.init();
@@ -779,7 +782,7 @@ export class Host implements IComponent {
                 };
             }
 
-            const sequenceAdapter = new adapter.SequenceAdapter(this.config);
+            const sequenceAdapter = adapter.sequenceAdapter;
 
             sequenceAdapter.logger.updateBaseLog({ id });
             sequenceAdapter.logger.pipe(this.logger);
@@ -1030,7 +1033,7 @@ export class Host implements IComponent {
             throw new Error("Failed to use adapter");
         }
 
-        const instanceAdapter = new adapter.InstanceAdapter(this.config);
+        const instanceAdapter = adapter.instanceAdapter;
 
         const communicationHandler = new CommunicationHandler();
         const id = payload.instanceId || IDProvider.generate();

packages/sth-config/src/config-service.ts

@@ -1,8 +1,10 @@
 import { DeepPartial, PublicSTHConfiguration, STHConfiguration } from "@scramjet/types";
 
 import { merge } from "@scramjet/utility";
-import path from "path";
 import { homedir } from "os";
+import path from "path";
+// import path from "path";
+// import { homedir } from "os";
 
 const imageConfig = require("./image-config.json");
 
@@ -54,7 +56,7 @@ const _defaultConfig: STHConfiguration = {
     },
     adapters: {
         "@scramjet/adapter-process": {
-            sequencesRoot: path.join(homedir(), ".scramjet_sequences"),
+            sequencesRoot: path.join(homedir(), ".scramjet_sequences")
         },
         "@scramjet/adapter-docker": {
             prerunner: {
@@ -122,7 +124,7 @@ export class ConfigService {
     }
 
     static getConfigInfo(config: STHConfiguration): PublicSTHConfiguration {
-        const kubeFull = config.adapters["@scramjet/adapter-k8s"]!;
+        const kubeFull = config.adapters["@scramjet/adapter-k8s"] || {};
         const {
             cpmSslCaPath: optionsCpmSslCaPath,
             ...safe