pushing payload to runner/ docker run&disptach

Author: piotrek6641

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

@@ -22,6 +22,7 @@ import { STH_DOCKER_NETWORK, isHostSpawnedInDockerContainer, getHostname } from
 import { ObjLogger } from "@scramjet/obj-logger";
 import { getRunnerEnvEntries } from "./get-runner-env";
 import { Readable } from "stream";
+import { RunnerConnectInfo } from "@scramjet/types/src/runner-connect";
 
 /**
  * Adapter for running Instance by Runner executed in Docker container.
@@ -168,13 +169,13 @@ IComponent {
         };
     }
 
-    async run(config: InstanceConfig, instancesServerPort: number, instanceId: string, sequenceInfo: SequenceInfo): Promise<ExitCode> {
-        await this.dispatch(config, instancesServerPort, instanceId, sequenceInfo);
+    async run(config: InstanceConfig, instancesServerPort: number, instanceId: string, sequenceInfo: SequenceInfo, payload: RunnerConnectInfo): Promise<ExitCode> {
+        await this.dispatch(config, instancesServerPort, instanceId, sequenceInfo, payload);
         return this.waitUntilExit(config, instanceId, sequenceInfo);
     }
 
     // eslint-disable-next-line complexity
-    async dispatch(config: InstanceConfig, instancesServerPort: number, instanceId: string, sequenceInfo: SequenceInfo): Promise<void> {
+    async dispatch(config: InstanceConfig, instancesServerPort: number, instanceId: string, sequenceInfo: SequenceInfo, payload: RunnerConnectInfo): Promise<void> {
         if (config.type !== "docker") {
             throw new Error("Docker instance adapter run with invalid runner config");
         }
@@ -198,7 +199,8 @@ IComponent {
             instancesServerHost: networkSetup.host,
             instanceId,
             pipesPath: "",
-            sequenceInfo
+            sequenceInfo,
+            payload
         }).map(([k, v]) => `${k}=${v}`);
 
         this.logger.debug("Runner will start with envs", envs);
@@ -228,7 +230,7 @@ IComponent {
         this.logger.trace("Container is running", containerId);
     }
 
-    async waitUntilExit(instanceId: string): Promise<number> {
+    async waitUntilExit(config: InstanceConfig, instanceId:string, _sequenceInfo: SequenceInfo): Promise<number> {
         try {
             const containerId = await this.dockerHelper.getContainerIdByLabel("scramjet.instance.id", instanceId);
             const { statusCode } = await this.dockerHelper.wait(containerId);

packages/adapters/src/types.ts

@@ -1,4 +1,5 @@
 import { ExitCode, InstanceId, IObjectLogger, SequenceInfo } from "@scramjet/types";
+import { StartSequencePayload } from "@scramjet/types/src/rest-api-sth";
 import { ContainerStats, NetworkInspectInfo } from "dockerode";
 import { PathLike } from "fs";
 import { Stream, Writable } from "stream";
@@ -323,6 +324,7 @@ export type RunnerEnvConfig = {
     instancesServerHost: string;
     instanceId: InstanceId;
     sequenceInfo: SequenceInfo
+    payload?: StartSequencePayload
 }
 
 export type RunnerEnvironmentVariables = Partial<{

packages/host/src/lib/csi-controller.ts

@@ -4,7 +4,6 @@ import {
     DownstreamStreamsConfig,
     EncodedMessage,
     HandshakeAcknowledgeMessage,
-    ICommunicationHandler,
     ParsedMessage,
     PassThroughStreamsConfig,
     ReadableStream,
@@ -22,13 +21,14 @@ import {
     OpResponse,
     StopSequenceMessageData,
     HostProxy,
+    ICommunicationHandler,
 } from "@scramjet/types";
 import {
     AppError,
     CSIControllerError,
     HostError,
     MessageUtilities,
-    InstanceAdapterError,
+    InstanceAdapterError
 } from "@scramjet/model";
 import { CommunicationChannel as CC, RunnerExitCode, RunnerMessageCode } from "@scramjet/symbols";
 import { Duplex, PassThrough, Readable } from "stream";
@@ -165,18 +165,18 @@ export class CSIController extends TypedEmitter<Events> {
         public communicationHandler: ICommunicationHandler,
         private sthConfig: STHConfiguration,
         private hostProxy: HostProxy,
-        private adapter: STHConfiguration["runtimeAdapter"] = sthConfig.runtimeAdapter
+        private adapter: STHConfiguration["runtimeAdapter"] = sthConfig.runtimeAdapter,
     ) {
         super();
 
         this.id = this.handshakeMessage.id;
-        this.sequence = this.handshakeMessage.sequenceInfo;
-        this.appConfig = payload.appConfig;
-        this.args = payload.args;
-        this.outputTopic = payload.outputTopic;
-        this.inputTopic = payload.inputTopic;
+        this.sequence = this.handshakeMessage.sequence;
+        this.appConfig = this.handshakeMessage.payload.appConfig;
+        this.args = this.handshakeMessage.payload.args;
+        this.outputTopic = this.handshakeMessage.payload.outputTopic;
+        this.inputTopic = this.handshakeMessage.payload.inputTopic;
         this.limits = {
-            memory: payload.limits?.memory || sthConfig.docker.runner.maxMem
+            memory: handshakeMessage.payload.limits?.memory || sthConfig.docker.runner.maxMem
         };
 
         this.instanceLifetimeExtensionDelay = +sthConfig.timings.instanceLifetimeExtensionDelay;
@@ -507,7 +507,7 @@ export class CSIController extends TypedEmitter<Events> {
         }
 
         this.info.ports = message[1].ports;
-        this.sequence = message[1].sequenceInfo;
+        this.sequence = message[1].sequence;
 
         // TODO: add message to initiate the instance adapter
 

packages/host/src/lib/csi-dispatcher.ts

@@ -1,9 +1,9 @@
 import { ObjLogger } from "@scramjet/obj-logger";
-import { HostProxy, IObjectLogger, InstanceConfig, MessageDataType, STHConfiguration, STHRestAPI, SequenceInfo } from "@scramjet/types";
+import { HostProxy, ICommunicationHandler, IObjectLogger, InstanceConfig, MessageDataType, STHConfiguration, STHRestAPI, SequenceInfo } from "@scramjet/types";
 import { SocketServer } from "./socket-server";
 import { InstanceStore } from "./instance-store";
 import { CSIController } from "./csi-controller";
-import { CommunicationHandler, IDProvider } from "@scramjet/model";
+import { IDProvider } from "@scramjet/model";
 import { StartSequencePayload } from "@scramjet/types/src/rest-api-sth";
 import { getInstanceAdapter } from "@scramjet/adapters";
 import { TypedEmitter } from "@scramjet/utility";
@@ -36,9 +36,15 @@ export class CSIDispatcher extends TypedEmitter<Events> {
         this.STHConfig = STHConfig;
     }
 
-    createCSIController(id: string, sequence: SequenceInfo, payload: StartSequencePayload, communicationHandler: CommunicationHandler, config: STHConfiguration, instanceProxy: HostProxy) {
+    async createCSIController(
+        id: string,
+        sequence: SequenceInfo,
+        payload: StartSequencePayload,
+        communicationHandler: ICommunicationHandler,
+        config: STHConfiguration,
+        instanceProxy: HostProxy) {
         sequence.instances = sequence.instances || new Set();
-        const csiController = new CSIController(id, sequence, payload, communicationHandler, config, instanceProxy);
+        const csiController = new CSIController({ id, sequence, payload }, communicationHandler, config, instanceProxy, this.STHConfig.runtimeAdapter);
 
         csiController.logger.pipe(this.logger);
         this.logger.trace("CSIController created", id);
@@ -132,7 +138,7 @@ export class CSIDispatcher extends TypedEmitter<Events> {
             this.emit("terminated", { id, code });
         });
 
-        csiController.start();
+        await csiController.start();
 
         this.logger.trace("csiController started", id);
 
@@ -161,7 +167,7 @@ export class CSIDispatcher extends TypedEmitter<Events> {
             this.STHConfig.host.instancesServerPort,
             id,
             sequence,
-            //payload
+            payload
         );
         // @todo more instance info
         return {

packages/host/src/lib/host.ts

@@ -901,7 +901,7 @@ export class Host implements IComponent {
         this.logger.info("Start sequence", sequence.id, sequence.config.name);
 
         try {
-            const runner = this.csiDispatcher.startRunner(sequence, payload);
+            const runner = await this.csiDispatcher.startRunner(sequence, payload);
 
             // @todo more info
             await this.cpmConnector?.sendInstanceInfo({
@@ -948,7 +948,13 @@ export class Host implements IComponent {
             // @todo need more instance info
             if (!this.instancesStore[id]) {
                 this.logger.info("creating new CSIController for runner connecting");
-                this.csiDispatcher.createCSIController(id, {} as SequenceInfo, {} as STHRestAPI.StartSequencePayload, new CommunicationHandler(), this.config, this.instanceProxy);
+                await this.csiDispatcher.createCSIController(
+                    id,
+                    {} as SequenceInfo,
+                    {} as STHRestAPI.StartSequencePayload,
+                    new CommunicationHandler(),
+                    this.config,
+                    this.instanceProxy);
             }
             await this.instancesStore[id].handleInstanceConnect(
                 streams

packages/types/src/lifecycle-adapters.ts

@@ -4,6 +4,7 @@ import { InstanceConfig } from "./runner-config";
 import { IObjectLogger } from "./object-logger";
 import { InstanceLimits } from "./instance-limits";
 import { SequenceInfo } from "./sequence-adapter";
+import { RunnerConnectInfo } from "./runner-connect";
 
 export type ExitCode = number;
 
@@ -30,7 +31,7 @@ export interface ILifeCycleAdapterMain {
 
     getCrashLog(): Promise<string[]>;
 
-    waitUntilExit(instanceId: string): Promise<ExitCode>;
+    waitUntilExit(config: InstanceConfig, instanceId: string, sequenceInfo: SequenceInfo): Promise<ExitCode>;
 }
 // @TODO create ISequenceAdapter interface
 
@@ -43,15 +44,15 @@ export interface ILifeCycleAdapterRun extends ILifeCycleAdapterMain {
       * @param {InstanceConfig} Runner configuration.
       * @returns {ExitCode} Runner exit code.
       */
-    dispatch(config: InstanceConfig, instancesServerPort: number, instanceId: string, sequenceInfo: SequenceInfo): Promise<void>;
+    dispatch(config: InstanceConfig, instancesServerPort: number, instanceId: string, sequenceInfo: SequenceInfo, payload: RunnerConnectInfo): Promise<void>;
 
     /**
       * Starts Runner - in essence does `dispatch` and then `waitUntilExit`.
       *
       * @param {InstanceConfig} Runner configuration.
       * @returns {ExitCode} Runner exit code.
       */
-    run(config: InstanceConfig, instancesServerPort: number, instanceId: string, sequenceInfo: SequenceInfo): Promise<ExitCode>;
+    run(config: InstanceConfig, instancesServerPort: number, instanceId: string, sequenceInfo: SequenceInfo, payload: RunnerConnectInfo): Promise<ExitCode>;
 }
 
 export type LifeCycleError = any | (Error & { exitCode?: number, errorMessage?: string });

packages/types/src/messages/handshake.ts

@@ -13,7 +13,7 @@ export type HandshakeMessage = {
     payload: StartSequencePayload
 };
 
-export type PingMessageData = { id: string, ports?: Record<string, string>; sequenceInfo: SequenceInfo, payload: StartSequencePayload }
+export type PingMessageData = { id: string, ports?: Record<string, string>; sequence: SequenceInfo, payload: StartSequencePayload }
 
 export type PangMessageData = {
     requires?: string,