Revamp Job flow

lib/server/core/JobManager.js

@@ -0,0 +1,71 @@
+import { v4 } from "uuid";
+import { getTest } from "../database/queries/catalog.js";
+import applyJobInternally from "../k8s/k8s-internal.js";
+import applyJob from "../k8s/k8s.js";
+
+const maxJobs = process.env.MAX_JOBS ? parseInt(process.env.MAX_JOBS) : 3;
+const internalDeploy = process.env.INTERNAL_DEPLOY === "true";
+const launchJob = internalDeploy ? applyJobInternally : applyJob;
+
+async function getTests(job) {
+  if (job.pipeline) return [await getTest(job.pipeline.__test)];
+  if (!job.testNames) return [];
+  const tests = await Promise.all(job.testNames.map((name) => getTest(name)));
+  return tests;
+}
+
+class JobManager {
+  constructor() {
+    this.clientMaxJobs = maxJobs;
+    this.clients = {};
+  }
+
+  getJob(clientId, jobId) {
+    return this.clients[clientId].jobs.find((j) => j.id === jobId);
+  }
+
+  getJobById(jobId) {
+    for (var client of Object.values(this.clients)) {
+      const job = client.jobs.find((j) => j.id === jobId);
+      if (!job) continue;
+      return job;
+    }
+  }
+
+  pushLog(jobId, log) {
+    const job = this.getJobById(jobId);
+    if (!job) return;
+
+    if (log instanceof Array) job.log.push(...log);
+    else job.log.push(log);
+  }
+
+  closeJob(jobId, exitcode) {
+    const job = this.getJobById(jobId);
+    if (!job) return;
+    job.exitcode = exitcode;
+  }
+
+  async newJob(jobRequest, id) {
+    if (!jobRequest) throw Error("Request Must Be Object!");
+    if (!this.clients[id]) this.clients[id] = { jobs: [] };
+    const job = { ...jobRequest };
+    job.image = "registry.dunemask.net/garden/dev/reed:latest";
+    job.id = v4();
+    job.log = [];
+    this.clients[id].jobs.push(job);
+    job.dashboardSocketId = id;
+    job.tests = await getTests(job);
+    for (var t of job.tests) if (!t) throw Error("1 or more tests not found!");
+    launchJob(job);
+    return { ...job };
+  }
+
+  removeJob(clientId, id) {
+    this.clients[clientId].jobs = this.clients[clientId].jobs.filter(
+      (j) => j.id !== id
+    );
+  }
+}
+
+export default new JobManager();

lib/server/core/Qualiteer.js

@@ -0,0 +1,56 @@
+// Imports
+import fig from "figlet";
+import http from "http";
+import express from "express";
+import { INFO, OK, logInfo } from "../util/logging.js";
+
+// Import Core Modules
+import buildRoutes from "../routes/router.js";
+import pg from "../database/postgres.js";
+import injectSockets from "./socket-server.js";
+import JobManager from "./JobManager.js";
+import buildRabbiteer from "../rabbit/rabbit-workers.js";
+
+// Constants
+const title = "QLTR";
+const rabbiteerEnabled = process.env.QUALITEER_RABBITEER_ENABLED !== "false";
+const port = process.env.QUALITEER_DEV_PORT ?? 52000;
+
+// Class
+export default class Qualiteer {
+  constructor(options = {}) {
+    for (var k in options) this[k] = options[k];
+    this.jobs = JobManager;
+    this.port = options.port ?? port;
+  }
+
+  async _preinitialize() {
+    logInfo(fig.textSync(title, "Cyberlarge"));
+    INFO("INIT", "Initializing...");
+    this.app = express();
+    this.pg = pg;
+    this.server = http.createServer(this.app);
+    this.sockets = injectSockets(this.server, this.jobs);
+    this.routes = buildRoutes(this.pg, this.sockets);
+    this.rabbiteer = buildRabbiteer(this.pg, this.sockets);
+    this.app.use(this.routes);
+  }
+
+  async _connect() {
+    await this.pg.connect();
+    if (!rabbiteerEnabled) return;
+    await this.rabbiteer.connect();
+  }
+
+  start() {
+    const qt = this;
+    return new Promise(async function init(res) {
+      qt._preinitialize();
+      await qt._connect();
+      qt.server.listen(qt.port, function onStart() {
+        OK("SERVER", `Running on ${qt.port}`);
+        res();
+      });
+    });
+  }
+}

lib/server/core/client-listeners.js

@@ -0,0 +1,34 @@
+import evt from "../../common/sockets/events.js";
+
+export const initiator = async (socket, jobs) => {
+  const jobStr = socket.handshake.query.job;
+  const jobReq = JSON.parse(jobStr);
+
+  if (!jobReq || !(jobReq instanceof Object))
+    throw Error("No 'job' was included in the handshake query");
+
+  const job = await jobs.newJob(jobReq, socket.id);
+  socket.join(job.id);
+  socket.emit(evt.JOB_CRT, job.id);
+};
+
+export const executor = (io, socket, jobs) => {
+  const jobId = socket.handshake.query.jobId;
+  if (!jobId) throw Error("No 'jobId' was included in the handshake query");
+
+  socket.join(jobId);
+  socket.on(evt.JOB_REP, function onReport(log) {
+    jobs.pushLog(jobId, log);
+    io.to(jobId).emit(evt.JOB_LOG, log);
+  });
+  socket.on(evt.JOB_CLS, function onClose(code) {
+    jobs.closeJob(jobId, code);
+    io.to(jobId).emit(evt.JOB_CLS, code);
+  });
+};
+
+export const viewer = (socket) => {
+  const jobId = socket.handshake.query.jobId;
+  if (!jobId) throw Error("No 'jobId' was included in the handshake query");
+  socket.join(jobId);
+};

lib/server/core/crons.js

@@ -0,0 +1,13 @@
+import cron from "cron";
+const { CronJob } = cron;
+
+// Remove Expired Silenced Tests
+const expiredSilenced = () => {
+  console.log("Would Update Silenced Tests");
+};
+
+const silencedCron = new CronJob("* * * * * *", expiredSilenced);
+
+export default async function startCrons() {
+  silencedCron.start();
+}

lib/server/core/socket-server.js

@@ -0,0 +1,55 @@
+import { Server as Skio } from "socket.io";
+import evt from "../../common/sockets/events.js";
+import modes from "../../common/sockets/modes.js";
+
+import { initiator, executor, viewer } from "./client-listeners.js";
+
+const socketDrop = (io, room, id) => {
+  const { rooms } = io.of("/").adapter;
+  const clients = rooms.get(room);
+  if (clients.size > 1 || clients.size === 0) return;
+  const socketId = Array.from(clients)[0];
+  const s = io.sockets.sockets.get(socketId);
+  s.disconnect();
+};
+
+const socketConnect = async (io, socket, jobs) => {
+  const { mode } = socket.handshake.query;
+  try {
+    switch (mode) {
+      case modes.INIT:
+        await initiator(socket, jobs);
+        break;
+      case modes.EXEC:
+        executor(io, socket, jobs);
+        break;
+      case modes.VIEW:
+        viewer(socket);
+        break;
+      default:
+        socket.send(evt.ERR, "Invalid Mode!");
+        socket.disconnect();
+        break;
+    }
+  } catch (err) {
+    console.log(err);
+    socket.send(evt.ERR, err);
+    socket.disconnect();
+  }
+};
+
+const socketAuth = (socket, next) => {
+  const { token } = socket.handshake.auth;
+  // next(new Error("Bad Token"));
+  next();
+};
+
+const applySockets = (server, jobs, options) => {
+  const io = new Skio(server);
+  io.on("connection", (socket) => socketConnect(io, socket, jobs));
+  io.of("/").adapter.on("leave-room", (room, id) => socketDrop(io, room, id));
+  return io;
+  cle;
+};
+
+export default applySockets;