Mount src directory from path on host running worker container

3 files changed, 59 insertions, 34 deletions

diff --git a/worker/executor.ts b/worker/executor.ts
index 47b337c..11c24f6 100644
--- a/worker/executor.ts
+++ b/worker/executor.ts
@@ -28,7 +28,7 @@ export const executeJob = (tJob: ITraceable<Job, LogMetricTraceSupplier>) => {
                 .flatMapAsync((args) => getStdout(tJob.move(tJob.get().type), { env: args })),
         )
         .flatMapAsync(TraceUtil.promiseify(TraceUtil.traceResultingEither(metric)))
-        .peek(TraceUtil.promiseify(t => t.traceScope(() => LogLevel.DEBUG).trace.trace(JSON.stringify(t.get()))))
+        .peek(TraceUtil.promiseify((t) => t.traceScope(() => LogLevel.DEBUG).trace.trace(JSON.stringify(t.get()))))
         .get();
 };
 // -- </job.exectuor> --
diff --git a/worker/jobs/ci_pipeline.run b/worker/jobs/ci_pipeline.run
index 87bcc84..6bee929 100755
--- a/worker/jobs/ci_pipeline.run
+++ b/worker/jobs/ci_pipeline.run
@@ -1,5 +1,16 @@
 #!/bin/sh
 
+# add scripts executed by the pipeline
 export PATH=$PATH:$PIPELINE_PATH
 
+containerid=$(cat /etc/hostname)
+isindocker=$(docker ps -q -f "id=$containerid")
+if [ -n "$isindocker" ]; then
+    executorLaminarPath=$(docker inspect "$containerid" | jq -r '.[0].Mounts[] | select(.Destination == "/var/lib/laminar") | .Source')
+else
+    executorLaminarPath=$(pwd)
+fi
+
+export executorLaminarPath
+
 checkout_ci.js
diff --git a/worker/scripts/checkout_ci.ts b/worker/scripts/checkout_ci.ts
index d71df6e..cba6a2f 100755
--- a/worker/scripts/checkout_ci.ts
+++ b/worker/scripts/checkout_ci.ts
@@ -5,23 +5,22 @@ import {
     Either,
     LogTraceable,
     getRequiredEnvVars,
-    getStdout,
     isObject,
     LogMetricTraceable,
     Metric,
     prependWith,
     TraceUtil,
-    IEither,
+    getStdoutMany,
 } from '@emprespresso/pengueno';
 import { mkdir, readFile, rm } from 'fs/promises';
-import { basename, join } from 'path';
+import { join } from 'path';
 import { type CheckoutCiJob, type FetchCodeJob, PipelineImpl } from '@emprespresso/ci_model';
 import { executeJob, executePipeline } from '@emprespresso/ci_worker';
 
-export interface CiWorkflow {
+interface CiWorkflow {
     workflow: string;
 }
-export function isCiWorkflow(t: unknown): t is CiWorkflow {
+function isCiWorkflow(t: unknown): t is CiWorkflow {
     return isObject(t) && 'workflow' in t && typeof t.workflow === 'string' && !t.workflow.includes('..');
 }
 
@@ -30,8 +29,19 @@ const OCI_REGISTRY = 'oci.liz.coffee';
 const PIPELINE_IMAGE = OCI_REGISTRY + '/emprespresso/ci_worker:release';
 const READONLY_CREDENTIALS = { username: 'readonly', password: 'readonly' };
 
-const run = Date.now().toString();
-const eitherJob = getRequiredEnvVars(['remote', 'refname', 'rev']).mapRight(
+// use a different directory per job run
+// even though the Laminar run is unique per job, there's potential for "children" we spawn
+// to be a checkout_ci job as well. in which case, we don't want any conflicts with whatever
+// the "parent" was doing, so we create a unique directory for each.
+// i.e.
+// Laminar Run: 57, CWD=/var/lib/laminar/run/57
+//   ci_pipeline (1000.uuidA)
+//     -> checkout_ci
+//       -> [...children]
+//          -> checkout_ci (1000.uuidB)
+const run = `${Date.now()}.${crypto.randomUUID().replaceAll('-', '')}`;
+
+const eitherJob = getRequiredEnvVars(['remote', 'refname', 'rev', 'executorLaminarPath']).mapRight(
     (baseArgs) =>
         <CheckoutCiJob>{
             type: 'checkout_ci.js',
@@ -46,9 +56,9 @@ const afterJob = eitherJob.flatMapAsync((job) =>
     Either.fromFailableAsync(() => rm(getWorkingDirectoryForCiJob(job), { recursive: true })),
 );
 
+const logTraceableJob = LogTraceable.of(eitherJob).flatMap(TraceUtil.withTrace(`checkout_ci.run.${run}`));
 const ciRunMetric = Metric.fromName('checkout_ci.run');
-const _logJob = LogTraceable.of(eitherJob).flatMap(TraceUtil.withTrace(`checkout_ci.${run}`));
-await LogMetricTraceable.ofLogTraceable(_logJob)
+await LogMetricTraceable.ofLogTraceable(logTraceableJob)
     .flatMap(TraceUtil.withMetricTrace(ciRunMetric))
     .map((tEitherJob) =>
         tEitherJob.get().flatMapAsync((ciJob) => {
@@ -84,32 +94,25 @@ await LogMetricTraceable.ofLogTraceable(_logJob)
             ),
         );
         return repoCiFileContents
-            .flatMap((fileText) => Either.fromFailable<Error, unknown>(() => JSON.parse(fileText)))
-            .flatMap((json) =>
-                eitherCiJob.flatMap((ciJob): IEither<Error, { commands: Array<Command>; job: CheckoutCiJob }> => {
-                    if (!isCiWorkflow(json)) {
-                        const e = new Error("couldn't find any valid ci configuration (｡•́︿•̀｡), that's okay~");
-                        return Either.left(e);
-                    }
-                    return Either.right({
-                        commands: getPipelineGenerationCommand(ciJob, json.workflow),
-                        job: ciJob,
-                    });
-                }),
-            );
+            .flatMap((fileText) =>
+                Either.fromFailable<Error, CiWorkflow>(() => JSON.parse(fileText)).filter((json) => isCiWorkflow(json)),
+            )
+            .joinRight(eitherCiJob, (job: CheckoutCiJob, { workflow }) => ({
+                job,
+                commands: getPipelineGenerationCommand(job, workflow),
+            }));
     })
     .map(async (tEitherPipelineGenerationCommand) => {
         const eitherJobCommand = await tEitherPipelineGenerationCommand.get();
-        const pipelineSerialized = await eitherJobCommand.flatMapAsync(({ commands }) => {
-            return Either.joinRightAsync(
-                commands,
-                (command) => tEitherPipelineGenerationCommand.move(command).map(getStdout).get(),
-                Either.right<Error, string>(''),
-            );
-        });
+        const pipelineSerialized = await eitherJobCommand.flatMapAsync(({ commands }) =>
+            getStdoutMany(tEitherPipelineGenerationCommand.move(commands)),
+        );
         return pipelineSerialized
-            .flatMap((s) => PipelineImpl.from(s))
-            .flatMap((pipeline) => eitherJobCommand.mapRight(({ job }) => ({ job, pipeline })));
+            .flatMap((results) => {
+                const pipeline = results.at(-1)!;
+                return PipelineImpl.from(pipeline);
+            })
+            .joinRight(eitherJobCommand, (job, pipeline) => ({ job, pipeline }));
     })
     .peek(
         TraceUtil.promiseify((tEitherPipeline) =>
@@ -147,7 +150,18 @@ function getWorkingDirectoryForCiJob(job: CheckoutCiJob) {
 }
 
 function getSrcDirectoryForCiJob(job: CheckoutCiJob) {
-    return `${job.arguments.returnPath}/${job.arguments.run}/src`;
+    return `${getWorkingDirectoryForCiJob(job)}/src`;
+}
+
+function getSrcDirectoryForChildContainer(job: CheckoutCiJob) {
+    // the container which runs the pipeline synthesizer (A) might be spawned by another container
+    // (B) ((which should be the one running this job)) by talking to the host's docker daemon
+    // (mounting /var/run/docker.sock) and executing the {@link getPipelineGenerationCommand}.
+    //
+    // so mounting {@link getSrcDirectoryForCiJob} has no meaning as it doesn't exist on the host;
+    // here we replace the path in (B) with the actual volume source on the host, where the src
+    // exists.
+    return getSrcDirectoryForCiJob(job).replace('/var/lib/laminar', job.arguments.executorLaminarPath);
 }
 
 function getPipelineGenerationCommand(
@@ -170,7 +184,7 @@ function getPipelineGenerationCommand(
                 '-e',
             ),
             '-v',
-            `${getSrcDirectoryForCiJob(job)}:/src`,
+            `${getSrcDirectoryForChildContainer(job)}:/src`,
             image,
             `/src/${pipelineGeneratorPath}`,
         ],