Allow to get binary output from /output

bdd/data/sequences/bin-out-seq/dist/index.js

@@ -0,0 +1,73 @@
+"use strict";
+var __awaiter = (this && this.__awaiter) || function (thisArg, _arguments, P, generator) {
+    function adopt(value) { return value instanceof P ? value : new P(function (resolve) { resolve(value); }); }
+    return new (P || (P = Promise))(function (resolve, reject) {
+        function fulfilled(value) { try { step(generator.next(value)); } catch (e) { reject(e); } }
+        function rejected(value) { try { step(generator["throw"](value)); } catch (e) { reject(e); } }
+        function step(result) { result.done ? resolve(result.value) : adopt(result.value).then(fulfilled, rejected); }
+        step((generator = generator.apply(thisArg, _arguments || [])).next());
+    });
+};
+var __generator = (this && this.__generator) || function (thisArg, body) {
+    var _ = { label: 0, sent: function() { if (t[0] & 1) throw t[1]; return t[1]; }, trys: [], ops: [] }, f, y, t, g;
+    return g = { next: verb(0), "throw": verb(1), "return": verb(2) }, typeof Symbol === "function" && (g[Symbol.iterator] = function() { return this; }), g;
+    function verb(n) { return function (v) { return step([n, v]); }; }
+    function step(op) {
+        if (f) throw new TypeError("Generator is already executing.");
+        while (_) try {
+            if (f = 1, y && (t = op[0] & 2 ? y["return"] : op[0] ? y["throw"] || ((t = y["return"]) && t.call(y), 0) : y.next) && !(t = t.call(y, op[1])).done) return t;
+            if (y = 0, t) op = [op[0] & 2, t.value];
+            switch (op[0]) {
+                case 0: case 1: t = op; break;
+                case 4: _.label++; return { value: op[1], done: false };
+                case 5: _.label++; y = op[1]; op = [0]; continue;
+                case 7: op = _.ops.pop(); _.trys.pop(); continue;
+                default:
+                    if (!(t = _.trys, t = t.length > 0 && t[t.length - 1]) && (op[0] === 6 || op[0] === 2)) { _ = 0; continue; }
+                    if (op[0] === 3 && (!t || (op[1] > t[0] && op[1] < t[3]))) { _.label = op[1]; break; }
+                    if (op[0] === 6 && _.label < t[1]) { _.label = t[1]; t = op; break; }
+                    if (t && _.label < t[2]) { _.label = t[2]; _.ops.push(op); break; }
+                    if (t[2]) _.ops.pop();
+                    _.trys.pop(); continue;
+            }
+            op = body.call(thisArg, _);
+        } catch (e) { op = [6, e]; y = 0; } finally { f = t = 0; }
+        if (op[0] & 5) throw op[1]; return { value: op[0] ? op[1] : void 0, done: true };
+    }
+};
+exports.__esModule = true;
+var fs_1 = require("fs");
+var stream_1 = require("stream");
+var crypto = require("crypto");
+var mod = [
+    function (_stream, filePath) {
+        if (filePath === void 0) { filePath = "".concat(__dirname, "/random.bin"); }
+        return __awaiter(this, void 0, void 0, function () {
+            var ps, stream, hash_1;
+            var _this = this;
+            return __generator(this, function (_a) {
+                this.logger.info("Sequence started");
+                try {
+                    ps = new stream_1.PassThrough({ encoding: "binary" });
+                    stream = (0, fs_1.createReadStream)(filePath);
+                    hash_1 = crypto.createHash("sha256");
+                    stream.pipe(ps);
+                    stream.on("data", function (data) {
+                        hash_1.update(data);
+                    });
+                    stream.on("end", function () {
+                        var checksum = hash_1.digest("hex");
+                        console.log(checksum);
+                        _this.logger.info("".concat(filePath, " checksum written to stdout: ").concat(checksum));
+                    });
+                    return [2 /*return*/, stream];
+                }
+                catch (e) {
+                    this.logger.error(e);
+                }
+                return [2 /*return*/];
+            });
+        });
+    }
+];
+exports["default"] = mod;

bdd/data/sequences/bin-out-seq/dist/package.json

@@ -0,0 +1,23 @@
+{
+  "name": "@scramjet/bin-out-seq",
+  "version": "0.23.2",
+  "description": "A Sequence that reads binary file and writes data to output.",
+  "main": "index.js",
+  "scripts": {
+    "build": "tsc -p tsconfig.json",
+    "postbuild": "cp -r package.json random.bin dist/ && (cd dist && npm i --only=production)"
+  },
+  "author": "Scramjet <[email protected]>",
+  "license": "ISC",
+  "devDependencies": {
+    "@scramjet/types": "^0.38.0",
+    "@types/node": "15.12.5"
+  },
+  "repository": {
+    "type": "git",
+    "url": "https://github.com/scramjetorg/transform-hub.git"
+  },
+  "assets": [
+    "random.bin"
+  ]
+}

bdd/data/sequences/bin-out-seq/index.ts

@@ -0,0 +1,37 @@
+/* eslint-disable consistent-return */
+/* eslint-disable no-console */
+import { AppConfig, AppContext } from "@scramjet/types";
+import { createReadStream } from "fs";
+import { PassThrough } from "stream";
+import * as crypto from "crypto";
+
+const mod = [
+    async function(this: AppContext<AppConfig, any>, _stream, filePath: string = `${__dirname}/random.bin`) {
+        this.logger.info("Sequence started");
+
+        try {
+            const ps = new PassThrough({ encoding: "binary" });
+            const stream = createReadStream(filePath);
+            const hash = crypto.createHash("sha256");
+
+            stream.pipe(ps);
+
+            stream.on("data", (data) => {
+                hash.update(data);
+            });
+
+            stream.on("end", () => {
+                const checksum = hash.digest("hex");
+
+                console.log(checksum);
+                this.logger.info(`${filePath} checksum written to stdout: ${checksum}`);
+            });
+
+            return stream;
+        } catch (e) {
+            this.logger.error(e);
+        }
+    }
+];
+
+export default mod;

bdd/data/sequences/bin-out-seq/package.json

@@ -0,0 +1,23 @@
+{
+  "name": "@scramjet/bin-out-seq",
+  "version": "0.23.2",
+  "description": "A Sequence that reads binary file and writes data to output.",
+  "main": "index.js",
+  "scripts": {
+    "build": "tsc -p tsconfig.json",
+    "postbuild": "cp -r package.json random.bin dist/ && (cd dist && npm i --only=production)"
+  },
+  "author": "Scramjet <[email protected]>",
+  "license": "ISC",
+  "devDependencies": {
+    "@scramjet/types": "^0.38.0",
+    "@types/node": "15.12.5"
+  },
+  "repository": {
+    "type": "git",
+    "url": "https://github.com/scramjetorg/transform-hub.git"
+  },
+  "assets": [
+    "random.bin"
+  ]
+}

bdd/data/sequences/bin-out-seq/tsconfig.json

@@ -0,0 +1,8 @@
+{
+  "compilerOptions": {
+    "outDir": "./dist"
+  },
+  "include": [
+    "./index.ts"
+  ]
+}

bdd/features/performance-tests/PT-004-checksum.feature

@@ -12,3 +12,16 @@ Feature: Verify the checksums of payloads are correct
         When runner has ended execution
         Then host is still running
 
+    @ci-performance
+    Scenario: PT-004 TC-002 Checksum of binary payload
+        Given file in the location "data/sequences/bin-out-seq/random.bin" exists on hard drive
+        And host is running
+        When I execute CLI with "seq pack data/sequences/bin-out-seq/dist -o data/sequences/bin-out-seq.tar.gz"
+        When sequence "data/sequences/bin-out-seq.tar.gz" loaded
+        And instance started
+        When wait for instance healthy is "true"
+        And get runner PID
+        And confirm file checksum match output checksum
+        When runner has ended execution
+        Then host is still running
+

bdd/step-definitions/e2e/host-steps.ts

@@ -20,7 +20,7 @@ import fs, { createReadStream, existsSync, ReadStream } from "fs";
 import { HostClient, InstanceOutputStream } from "@scramjet/api-client";
 import { HostUtils } from "../../lib/host-utils";
 import { PassThrough, Readable, Stream, Writable } from "stream";
-import crypto from "crypto";
+import crypto, { BinaryLike } from "crypto";
 import { promisify } from "util";
 import Dockerode from "dockerode";
 import { CustomWorld } from "../world";
@@ -67,6 +67,25 @@ const streamToString = async (stream: Stream): Promise<string> => {
 
     return chunks.join("");
 };
+const streamToBinary = async (stream: Readable): Promise<BinaryLike> => {
+    const chunks: Uint8Array[] = [];
+
+    return new Promise((resolve, reject) => {
+        stream.on("data", (chunk: Buffer | Uint8Array) => {
+            chunks.push(chunk instanceof Buffer ? chunk : Uint8Array.from(chunk));
+        });
+
+        stream.on("end", () => {
+            const binaryData = Buffer.concat(chunks);
+
+            resolve(binaryData);
+        });
+
+        stream.on("error", (error: Error) => {
+            reject(error);
+        });
+    });
+};
 const waitForContainerToClose = async () => {
     if (!containerId) assert.fail();
 
@@ -505,6 +524,23 @@ When("compare checksums of content sent from file {string}", async function(this
     await this.resources.instance?.sendInput("null");
 });
 
+When("confirm file checksum match output checksum", async function(this: CustomWorld) {
+    // the random.bin hex is written to instance stdout
+    const stdout = await this.resources.instance!.getStream("stdout");
+    const fileHexFromStdout = await streamToString(stdout);
+    const output = await this.resources.instance?.getStream("output");
+
+    if (!output || !stdout) assert.fail("No output or stdout, or both.");
+
+    const dataFromOutput = await streamToBinary(output);
+    const outputHex: string = crypto
+        .createHash("sha256")
+        .update(dataFromOutput)
+        .digest("hex");
+
+    assert.strictEqual(outputHex, fileHexFromStdout.trim());
+});
+
 When(
     "send stop message to instance with arguments timeout {int} and canCallKeepAlive {string}",
     async function(this: CustomWorld, timeout: number, canCallKeepalive: string) {

packages/api-server/src/handlers/stream.ts

@@ -67,7 +67,7 @@ export function createStreamHandlers(router: SequentialCeroRouter) {
                 ? `${type}; charset=${encoding}`
                 : type;
 
-            out.setEncoding(encoding);
+            logger.debug("encoding, cType, readableEncoding", encoding, cType, data.readableEncoding);
 
             res.setHeader("content-type", cType);
             res.setHeader("transfer-encoding", "chunked");

packages/client-utils/src/types/index.ts

@@ -15,6 +15,7 @@ export type SendStreamOptions = Partial<{
  */
 export type GetStreamOptions = Partial<{
     type: string;
+    encoding?: BufferEncoding;
 }>;
 
 /**

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

@@ -103,8 +103,10 @@ export class CSIController extends TypedEmitter<Events> {
     info: CSIControllerInfo = {};
     status: InstanceStatus;
     terminated?: { exitcode: number; reason: string; };
+
     provides?: string;
     requires?: string;
+    outputEncoding: BufferEncoding = "utf-8";
 
     initResolver?: { res: Function; rej: Function };
     heartBeatResolver?: { res: Function; rej: Function };
@@ -440,6 +442,9 @@ export class CSIController extends TypedEmitter<Events> {
                 this.apiInputEnabled = false;
             }
 
+            this.outputEncoding = pangData.outputEncoding || "utf-8";
+            //this.upStreams[CC.OUT].setDefaultEncoding(pangData.outputEncoding || "utf-8");
+
             this.emit("pang", {
                 provides: this.provides,
                 requires: this.requires,
@@ -595,7 +600,8 @@ export class CSIController extends TypedEmitter<Events> {
         if (development()) {
             this.router.upstream("/monitoring", this.upStreams[CC.MONITORING]);
         }
-        this.router.upstream("/output", this.upStreams[CC.OUT]);
+
+        this.router.upstream("/output", this.upStreams[CC.OUT], { encoding: this.outputEncoding });
 
         this.router.downstream("/input", (req) => {
             if (this.apiInputEnabled) {