Reconnect. Fix starting instance

Author: patuwwy

packages/adapters/src/get-runner-env.ts

@@ -9,7 +9,7 @@ import { RunnerEnvConfig, RunnerEnvironmentVariables } from "./types";
  * @returns env vars
  */
 export function getRunnerEnvVariables({
-    sequencePath, instancesServerPort, instancesServerHost, instanceId, pipesPath, paths = "posix", sequenceInfo
+    sequencePath, instancesServerPort, instancesServerHost, instanceId, pipesPath, paths = "posix", sequenceInfo, payload
 }: RunnerEnvConfig, extra: Record<string, string> = {}): RunnerEnvironmentVariables {
     const join = path[paths].join;
 
@@ -24,6 +24,7 @@ export function getRunnerEnvVariables({
         PIPES_LOCATION: pipesPath,
         CRASH_LOG: join(pipesPath, "crash_log"),
         SEQUENCE_INFO: JSON.stringify(sequenceInfo),
+        RUNNER_CONNECT_INFO: JSON.stringify(payload),
         ...extra
     };
 }

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

@@ -21,6 +21,7 @@ import { adapterConfigDecoder } from "./kubernetes-config-decoder";
 import { getRunnerEnvEntries } from "./get-runner-env";
 import { PassThrough } from "stream";
 import { RunnerExitCode } from "@scramjet/symbols";
+import { RunnerConnectInfo } from "@scramjet/types/src/runner-connect";
 
 /**
  * Adapter for running Instance by Runner executed in separate process.
@@ -88,6 +89,9 @@ IComponent {
             }
         };
     }
+    async dispatch(config: InstanceConfig, instancesServerPort: number, instanceId: string, sequenceInfo: SequenceInfo, payload: RunnerConnectInfo): Promise<void> {
+        throw Error("not implemented");
+    }
 
     async run(config: InstanceConfig, instancesServerPort: number, instanceId: string, sequenceInfo: SequenceInfo): Promise<ExitCode> {
         if (config.type !== "kubernetes") {
@@ -181,6 +185,11 @@ IComponent {
         return 0;
     }
 
+    async waitUntilExit(config: InstanceConfig, instanceId:string, _sequenceInfo: SequenceInfo): Promise<number> {
+        throw Error("Not implemented");
+    }
+
+
     async cleanup(): Promise<void> {
         await this.remove(this.adapterConfig.timeout);
     }

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

@@ -16,6 +16,7 @@ import { ChildProcess, spawn } from "child_process";
 
 import path from "path";
 import { getRunnerEnvVariables } from "./get-runner-env";
+import { RunnerConnectInfo } from "@scramjet/types/src/runner-connect";
 
 const isTSNode = !!(process as any)[Symbol.for("ts-node.register.instance")];
 const gotPython = "\n                              _ \n __      _____  _ __  ___ ___| |\n \\ \\ /\\ / / _ \\| '_ \\/ __|_  / |\n  \\ V  V / (_) | | | \\__ \\/ /|_|\n   \\_/\\_/ \\___/|_| |_|___/___(_)  🐍\n";
@@ -116,8 +117,11 @@ class ProcessInstanceAdapter implements
         return pythonpath;
     }
 
+    async dispatch(config: InstanceConfig, instancesServerPort: number, instanceId: string, sequenceInfo: SequenceInfo, payload: RunnerConnectInfo): Promise<void> {
+        throw Error("not implemented");
+    }
+
     async run(config: InstanceConfig, instancesServerPort: number, instanceId: string, sequenceInfo: SequenceInfo): Promise<ExitCode> {
-        console.log("config type", config.type)
         if (config.type !== "process") {
             throw new Error("Process instance adapter run with invalid runner config");
         }
@@ -177,6 +181,10 @@ class ProcessInstanceAdapter implements
         return statusCode;
     }
 
+    async waitUntilExit(config: InstanceConfig, instanceId:string, _sequenceInfo: SequenceInfo): Promise<number> {
+        throw Error("Not implemented");
+    }
+
     /**
      * Performs cleanup after Runner end.
      * Removes fifos used to communication with runner.

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

@@ -21,7 +21,7 @@ import {
     OpResponse,
     StopSequenceMessageData,
     HostProxy,
-    ICommunicationHandler,
+    ICommunicationHandler
 } from "@scramjet/types";
 import {
     AppError,
@@ -165,12 +165,12 @@ 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.sequence;
+        this.sequence = this.handshakeMessage.sequenceInfo;
         this.appConfig = this.handshakeMessage.payload.appConfig;
         this.args = this.handshakeMessage.payload.args;
         this.outputTopic = this.handshakeMessage.payload.outputTopic;
@@ -186,12 +186,14 @@ export class CSIController extends TypedEmitter<Events> {
         this.logger.debug("Constructor executed");
 
         this.status = InstanceStatus.INITIALIZING;
+
+
     }
 
     async start() {
         const i = new Promise((res, rej) => {
             this.initResolver = { res, rej };
-            //this.startInstance();
+            this.startInstance();
         });
 
         i.then(() => this.main()).catch(async (e) => {
@@ -256,6 +258,8 @@ export class CSIController extends TypedEmitter<Events> {
 
         this._instanceAdapter.logger.pipe(this.logger, { end: false });
 
+        this.endOfSequence = this._instanceAdapter.waitUntilExit(undefined, this.id, this.sequence)
+
         // @todo this also is moved to CSIDispatcher in entirety
         const instanceMain = async () => {
             try {
@@ -507,7 +511,7 @@ export class CSIController extends TypedEmitter<Events> {
         }
 
         this.info.ports = message[1].ports;
-        this.sequence = message[1].sequence;
+        this.sequence = message[1].sequenceInfo;
 
         // TODO: add message to initiate the instance adapter
 
@@ -525,7 +529,7 @@ export class CSIController extends TypedEmitter<Events> {
         }
 
         this.info.started = new Date();
-        this.logger.info("Instance started", this.info);
+        this.logger.info("Instance started", JSON.stringify(message, undefined, 4));
     }
 
     async handleInstanceConnect(streams: DownstreamStreamsConfig) {

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

@@ -38,16 +38,16 @@ export class CSIDispatcher extends TypedEmitter<Events> {
 
     async createCSIController(
         id: string,
-        sequence: SequenceInfo,
+        sequenceInfo: 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, this.STHConfig.runtimeAdapter);
+        sequenceInfo.instances = sequenceInfo.instances || new Set();
+        const csiController = new CSIController({ id, sequenceInfo, payload }, communicationHandler, config, instanceProxy, this.STHConfig.runtimeAdapter);
 
         csiController.logger.pipe(this.logger);
-        this.logger.trace("CSIController created", id);
+        this.logger.trace("CSIController created", id, sequenceInfo);
 
         csiController.logger.pipe(this.logger, { end: false });
         communicationHandler.logger.pipe(this.logger, { end: false });
@@ -108,7 +108,7 @@ export class CSIDispatcher extends TypedEmitter<Events> {
 
             delete InstanceStore[csiController.id];
 
-            sequence.instances.delete(id);
+            sequenceInfo.instances.delete(id);
 
             // await this.cpmConnector?.sendInstanceInfo({
             //     id: csiController.id,
@@ -138,13 +138,14 @@ export class CSIDispatcher extends TypedEmitter<Events> {
             this.emit("terminated", { id, code });
         });
 
-        await csiController.start();
+        csiController.start();
 
         this.logger.trace("csiController started", id);
 
-        sequence.instances.add(id);
+        sequenceInfo.instances.add(id);
 
         this.instancesStore[id] = csiController;
+
         return csiController;
     }
 
@@ -169,6 +170,7 @@ export class CSIDispatcher extends TypedEmitter<Events> {
             sequence,
             payload
         );
+
         // @todo more instance info
         return {
             id,

packages/host/src/lib/host.ts

@@ -956,6 +956,7 @@ export class Host implements IComponent {
                     this.config,
                     this.instanceProxy);
             }
+
             await this.instancesStore[id].handleInstanceConnect(
                 streams
             );

packages/model/src/stream-handler.ts

@@ -7,6 +7,7 @@ import {
     EncodedMonitoringMessage,
     ICommunicationHandler,
     IObjectLogger,
+    InstanceConnectionInfo,
     LoggerOutput,
     MaybePromise,
     MessageDataType,
@@ -147,6 +148,14 @@ export class CommunicationHandler implements ICommunicationHandler {
         return this;
     }
 
+    waitForHandshake(): Promise<InstanceConnectionInfo> {
+        return new Promise((res) => {
+            this.addMonitoringHandler(RunnerMessageCode.PING, (msg) => {
+                res(msg);
+            });
+        })
+    }
+
     pipeMessageStreams() {
         if (this._piped) {
             this.logger.error("pipeMessageStreams called twice");

packages/runner/src/bin/start-runner.ts

@@ -5,14 +5,26 @@ import fs from "fs";
 import { AppConfig, SequenceInfo } from "@scramjet/types";
 import { HostClient } from "../host-client";
 import { RunnerExitCode } from "@scramjet/symbols";
+import { RunnerConnectInfo } from "@scramjet/types/src/runner-connect";
 
 const sequencePath: string = process.env.SEQUENCE_PATH?.replace(/.js$/, "") + ".js";
 const instancesServerPort = process.env.INSTANCES_SERVER_PORT;
 const instancesServerHost = process.env.INSTANCES_SERVER_HOST;
 const instanceId = process.env.INSTANCE_ID;
 const sequenceInfo = process.env.SEQUENCE_INFO;
+const runnerConnectInfo = process.env.RUNNER_CONNECT_INFO;
 
 let connectInfo: SequenceInfo;
+let parsedRunnerConnectInfo: RunnerConnectInfo;
+
+try {
+    if (!runnerConnectInfo) throw new Error("Connection JSON is required.");
+    parsedRunnerConnectInfo = JSON.parse(runnerConnectInfo);
+} catch {
+    console.error("Error while parsing connection information.");
+    process.exit(RunnerExitCode.INVALID_ENV_VARS);
+}
+
 
 try {
     if (!sequenceInfo) throw new Error("Connection JSON is required.");
@@ -54,7 +66,7 @@ const hostClient = new HostClient(+instancesServerPort, instancesServerHost);
  * @param fifosPath - fifo files path
  */
 
-const runner: Runner<AppConfig> = new Runner(sequencePath, hostClient, instanceId, connectInfo);
+const runner: Runner<AppConfig> = new Runner(sequencePath, hostClient, instanceId, connectInfo, parsedRunnerConnectInfo);
 
 runner.main()
     .catch(e => {

packages/runner/src/runner.ts

@@ -33,6 +33,7 @@ import { mapToInputDataStream, readInputStreamHeaders } from "./input-stream";
 import { MessageUtils } from "./message-utils";
 import { HostClient as HostApiClient } from "@scramjet/api-client";
 import { ClientUtilsCustomAgent } from "@scramjet/client-utils";
+import { RunnerConnectInfo } from "@scramjet/types/src/runner-connect";
 
 // async function flushStream(source: Readable | undefined, target: Writable) {
 //     if (!source) return;
@@ -122,15 +123,22 @@ export class Runner<X extends AppConfig> implements IComponent {
     private outputDataStream: DataStream;
     private sequenceInfo: SequenceInfo;
 
+    private runnerConnectInfo: RunnerConnectInfo = {
+        appConfig: {}
+    };
+
     constructor(
         private sequencePath: string,
         private hostClient: IHostClient,
         private instanceId: string,
-        sequenceInfo: SequenceInfo
+        sequenceInfo: SequenceInfo,
+        runnerConnectInfo: RunnerConnectInfo
     ) {
         this.sequenceInfo = sequenceInfo;
         this.emitter = new EventEmitter();
 
+        this.runnerConnectInfo = runnerConnectInfo;
+
         this.logger = new ObjLogger(this, { id: instanceId });
         hostClient.logger.pipe(this.logger);
 
@@ -455,7 +463,12 @@ export class Runner<X extends AppConfig> implements IComponent {
 
     sendHandshakeMessage() {
         // TODO: send connection info
-        MessageUtils.writeMessageOnStream([RunnerMessageCode.PING, { id: this.instanceId, sequenceInfo: this.sequenceInfo }], this.hostClient.monitorStream);
+        MessageUtils.writeMessageOnStream([
+            RunnerMessageCode.PING, {
+                id: this.instanceId,
+                sequenceInfo: this.sequenceInfo,
+                payload: this.runnerConnectInfo
+            }], this.hostClient.monitorStream);
 
         this.logger.trace("Handshake sent");
     }

packages/types/src/lifecycle-adapters.ts

@@ -31,7 +31,7 @@ export interface ILifeCycleAdapterMain {
 
     getCrashLog(): Promise<string[]>;
 
-    waitUntilExit(config: InstanceConfig, instanceId: string, sequenceInfo: SequenceInfo): Promise<ExitCode>;
+    waitUntilExit(config: InstanceConfig | undefined, instanceId: string, sequenceInfo: SequenceInfo): Promise<ExitCode>;
 }
 // @TODO create ISequenceAdapter interface