init gke target scanner queue

.github/workflows/apps-job-dispatcher.yaml → .github/workflows/apps-event-worker.yaml

@@ -4,16 +4,16 @@ on:
   pull_request:
     branches: ["*"]
     paths:
-      - apps/job-dispatcher/**
+      - apps/event-worker/**
       - packages/db/**
-      - .github/workflows/apps-job-dispatcher.yaml
+      - .github/workflows/apps-event-worker.yaml
       - pnpm-lock.yaml
   push:
     branches: ["main"]
     paths:
-      - apps/job-dispatcher/**
+      - apps/event-worker/**
       - packages/db/**
-      - .github/workflows/apps-job-dispatcher.yaml
+      - .github/workflows/apps-event-worker.yaml
       - pnpm-lock.yaml
 
 jobs:
@@ -41,7 +41,7 @@ jobs:
         id: meta
         uses: docker/metadata-action@v4
         with:
-          images: ctrlplane/job-dispatchers
+          images: ctrlplane/event-worker
           tags: |
             type=sha,format=short,prefix=
 
@@ -50,13 +50,13 @@ jobs:
         if: github.ref != 'refs/heads/main'
         with:
           push: false
-          file: apps/job-dispatchers/Dockerfile
+          file: apps/event-worker/Dockerfile
           tags: ${{ steps.meta.outputs.tags }}
 
       - name: Build and Push
         uses: docker/build-push-action@v6
         if: github.ref == 'refs/heads/main'
         with:
           push: true
-          file: apps/job-dispatchers/Dockerfile
+          file: apps/event-worker/Dockerfile
           tags: ${{ steps.meta.outputs.tags }}

apps/job-dispatcher/Dockerfile → apps/event-worker/Dockerfile

@@ -25,16 +25,16 @@ COPY tooling/typescript/package.json ./tooling/typescript/package.json
 COPY packages/db/package.json ./packages/db/package.json
 COPY packages/job-dispatch/package.json ./packages/job-dispatch/package.json
 
-COPY apps/job-dispatcher/package.json ./apps/job-dispatcher/package.json
+COPY apps/event-worker/package.json ./apps/event-worker/package.json
 
 RUN pnpm install --frozen-lockfile
 
 COPY . .
 
-RUN turbo build --filter=...@ctrlplane/job-dispatcher
+RUN turbo build --filter=...@ctrlplane/event-worker
 
 RUN addgroup --system --gid 1001 nodejs
 RUN adduser --system --uid 1001 nodejs
 USER nodejs
 
-CMD node apps/job-dispatcher/dist/index.js
+CMD node apps/event-worker/dist/index.js

apps/event-worker/README.md

@@ -0,0 +1 @@
+# Event Worker

apps/job-dispatcher/eslint.config.js → apps/event-worker/eslint.config.js

@@ -1,9 +1,10 @@
-import baseConfig from "@ctrlplane/eslint-config/base";
+import baseConfig, { requireJsSuffix } from "@ctrlplane/eslint-config/base";
 
 /** @type {import('typescript-eslint').Config} */
 export default [
   {
     ignores: [".nitro/**", ".output/**"],
   },
+  requireJsSuffix,
   ...baseConfig,
 ];

apps/job-dispatcher/package.json → apps/event-worker/package.json

@@ -1,5 +1,5 @@
 {
-  "name": "@ctrlplane/job-dispatcher",
+  "name": "@ctrlplane/event-worker",
   "private": true,
   "type": "module",
   "scripts": {
@@ -12,12 +12,17 @@
   },
   "dependencies": {
     "@ctrlplane/db": "workspace:*",
+    "@ctrlplane/logger": "workspace:*",
+    "@ctrlplane/validators": "workspace:*",
     "@google-cloud/container": "^5.16.0",
     "@kubernetes/client-node": "^0.21.0",
     "@t3-oss/env-core": "^0.10.1",
+    "bullmq": "^5.12.10",
     "cron": "^3.1.7",
     "dotenv": "^16.4.5",
     "google-auth-library": "^9.13.0",
+    "ioredis": "^5.4.1",
+    "ms": "^2.1.3",
     "semver": "^7.6.2",
     "zod": "catalog:"
   },

apps/job-dispatcher/src/config.ts → apps/event-worker/src/config.ts

@@ -5,6 +5,6 @@ import { z } from "zod";
 dotenv.config();
 
 export const env = createEnv({
-  server: { POSTGRES_URL: z.string().url() },
+  server: { POSTGRES_URL: z.string().url(), REDIS_URL: z.string().url() },
   runtimeEnv: process.env,
 });

apps/event-worker/src/index.ts

@@ -0,0 +1,18 @@
+import { logger } from "@ctrlplane/logger";
+
+import { redis } from "./redis";
+import { createTargetScanWorker } from "./target-scan";
+
+const targetScanWorker = createTargetScanWorker();
+
+const shutdown = () => {
+  logger.warn("Exiting...");
+
+  targetScanWorker.close();
+  redis.quit();
+
+  process.exit(0);
+};
+
+process.on("SIGTERM", shutdown);
+process.on("SIGINT", shutdown);

apps/event-worker/src/redis.ts

@@ -0,0 +1,7 @@
+import IORedis from "ioredis";
+
+import { env } from "./config";
+
+export const redis = new IORedis(env.REDIS_URL, {
+  maxRetriesPerRequest: null,
+});

apps/event-worker/src/target-scan/gke.ts

@@ -0,0 +1,107 @@
+import { CoreV1Api } from "@kubernetes/client-node";
+import { Job } from "bullmq";
+import _ from "lodash";
+
+import { TargetProviderGoogle, Workspace } from "@ctrlplane/db/schema";
+import { logger } from "@ctrlplane/logger";
+
+import {
+  clusterToTarget,
+  connectToCluster,
+  getClusters,
+  getGoogleClusterClient,
+} from "./google.js";
+import { UpsertTarget } from "./upsert.js";
+
+const log = logger.child({ label: "target-scan/gke" });
+
+export const getGkeTargets = async (
+  workspace: Workspace,
+  config: TargetProviderGoogle,
+) => {
+  const { googleServiceAccountEmail } = workspace;
+  log.info(
+    `Scaning ${config.projectIds.join(", ")} using ${googleServiceAccountEmail}`,
+    { workspaceId: workspace.id, config, googleServiceAccountEmail },
+  );
+  const googleClusterClient = await getGoogleClusterClient(
+    googleServiceAccountEmail,
+  );
+
+  const clusters = (
+    await Promise.allSettled(
+      config.projectIds.map(async (project) => {
+        const clusters = await getClusters(googleClusterClient, project);
+        return { project, clusters };
+      }),
+    )
+  )
+    .filter((result) => result.status === "fulfilled")
+    .map((v) => v.value);
+
+  const kubernetesApiTargets: UpsertTarget[] = clusters.flatMap(
+    ({ project, clusters }) =>
+      clusters.map((cluster) =>
+        clusterToTarget(
+          workspace.id,
+          config.targetProviderId,
+          project,
+          cluster,
+        ),
+      ),
+  );
+  const kubernetesNamespaceTargets = (
+    await Promise.all(
+      clusters.flatMap(({ project, clusters }, idx) => {
+        return clusters.flatMap(async (cluster) => {
+          if (cluster.name == null || cluster.location == null) return [];
+
+          const kubeConfig = await connectToCluster(
+            googleClusterClient,
+            project,
+            cluster.name,
+            cluster.location,
+          );
+
+          const k8sApi = kubeConfig.makeApiClient(CoreV1Api);
+
+          try {
+            const response = await k8sApi.listNamespace();
+            const namespaces = response.body.items;
+            return namespaces
+              .filter((n) => n.metadata != null)
+              .map((n) =>
+                _.merge(
+                  clusterToTarget(
+                    workspace.id,
+                    config.targetProviderId,
+                    project,
+                    cluster,
+                  ),
+                  {
+                    name: `${cluster.name ?? cluster.id ?? ""}/${n.metadata!.name}`,
+                    kind: "KubernetesNamespace",
+                    identifier: `${project}/${cluster.name}/${n.metadata!.name}`,
+                    config: {
+                      namespace: n.metadata!.name,
+                    },
+                    labels: {
+                      ...n.metadata?.labels,
+                      "kubernetes/namespace": n.metadata!.name,
+                    },
+                  },
+                ),
+              );
+          } catch {
+            console.log(
+              `Unable to connect to cluster: ${cluster.name}/${cluster.id}`,
+            );
+            return [];
+          }
+        });
+      }),
+    )
+  ).flat();
+
+  return [...kubernetesApiTargets, ...kubernetesNamespaceTargets];
+};

apps/event-worker/src/target-scan/google.ts

@@ -0,0 +1,134 @@
+import type { ClusterManagerClient } from "@google-cloud/container";
+import Container from "@google-cloud/container";
+import { google } from "@google-cloud/container/build/protos/protos.js";
+import { KubeConfig } from "@kubernetes/client-node";
+import { GoogleAuth } from "google-auth-library";
+import { SemVer } from "semver";
+
+import { omitNullUndefined } from "../utils.js";
+
+const sourceCredentials = new GoogleAuth({
+  scopes: ["https://www.googleapis.com/auth/cloud-platform"],
+});
+
+export const getGoogleClusterClient = async (
+  targetPrincipal?: string | null,
+) => {
+  return new Container.v1.ClusterManagerClient({
+    clientOptions:
+      targetPrincipal != null
+        ? {
+            sourceClient: await sourceCredentials.getClient(),
+            targetPrincipal,
+            lifetime: 3600, // Token lifetime in seconds
+            delegates: [],
+            targetScopes: ["https://www.googleapis.com/auth/cloud-platform"],
+          }
+        : {},
+  });
+};
+
+export const getClusters = async (
+  clusterClient: ClusterManagerClient,
+  projectId: string,
+) => {
+  const request = { parent: `projects/${projectId}/locations/-` };
+  const [response] = await clusterClient.listClusters(request);
+  const { clusters } = response;
+  return clusters ?? [];
+};
+
+export const connectToCluster = async (
+  clusterClient: ClusterManagerClient,
+  project: string,
+  clusterName: string,
+  clusterLocation: string,
+) => {
+  const [credentials] = await clusterClient.getCluster({
+    name: `projects/${project}/locations/${clusterLocation}/clusters/${clusterName}`,
+  });
+  const kubeConfig = new KubeConfig();
+  kubeConfig.loadFromOptions({
+    clusters: [
+      {
+        name: clusterName,
+        server: `https://${credentials.endpoint}`,
+        caData: credentials.masterAuth!.clusterCaCertificate!,
+      },
+    ],
+    users: [
+      {
+        name: clusterName,
+        token: (await sourceCredentials.getAccessToken())!,
+      },
+    ],
+    contexts: [
+      {
+        name: clusterName,
+        user: clusterName,
+        cluster: clusterName,
+      },
+    ],
+    currentContext: clusterName,
+  });
+  return kubeConfig;
+};
+
+export const clusterToTarget = (
+  workspaceId: string,
+  providerId: string,
+  project: string,
+  cluster: google.container.v1.ICluster,
+) => {
+  const masterVersion = new SemVer(cluster.currentMasterVersion ?? "0");
+  const nodeVersion = new SemVer(cluster.currentNodeVersion ?? "0");
+  const autoscaling = String(
+    cluster.autoscaling?.enableNodeAutoprovisioning ?? false,
+  );
+
+  const appUrl = `https://console.cloud.google.com/kubernetes/clusters/details/${cluster.location}/${cluster.name}/details?project=${project}`;
+  return {
+    workspaceId,
+    name: cluster.name ?? cluster.id ?? "",
+    providerId,
+    identifier: `${project}/${cluster.name}`,
+    version: "kubernetes/v1",
+    kind: "KubernetesAPI",
+    config: {
+      name: cluster.name,
+      status: cluster.status,
+      cluster: {
+        certificateAuthorityData: cluster.masterAuth?.clusterCaCertificate,
+        endpoint: `https://${cluster.endpoint}`,
+      },
+    },
+    labels: omitNullUndefined({
+      "ctrlplane/url": appUrl,
+
+      "google/self-link": cluster.selfLink,
+      "google/project": project,
+      "google/location": cluster.location,
+      "google/autopilot": cluster.autopilot?.enabled,
+
+      "kubernetes/cluster-name": cluster.name,
+      "kubernetes/cluster-id": cluster.id,
+      "kubernetes/distribution": "gke",
+      "kubernetes/status": cluster.status,
+      "kubernetes/node-count": String(cluster.currentNodeCount ?? "unknown"),
+
+      "kubernetes/master-version": masterVersion.version,
+      "kubernetes/master-version-major": String(masterVersion.major),
+      "kubernetes/master-version-minor": String(masterVersion.minor),
+      "kubernetes/master-version-patch": String(masterVersion.patch),
+
+      "kubernetes/node-version": nodeVersion.version,
+      "kubernetes/node-version-major": String(nodeVersion.major),
+      "kubernetes/node-version-minor": String(nodeVersion.minor),
+      "kubernetes/node-version-patch": String(nodeVersion.patch),
+
+      "kubernetes/autoscaling-enabled": autoscaling,
+
+      ...(cluster.resourceLabels ?? {}),
+    }),
+  };
+};