v1 of log sockets

package.json

@@ -53,6 +53,8 @@
     "react-dom": "^18.3.1",
     "react-hook-form": "^7.53.1",
     "reflect-metadata": "^0.2.2",
+    "socket.io": "^4.8.1",
+    "socket.io-client": "^4.8.1",
     "sonner": "^1.5.0",
     "tailwind-merge": "^2.5.4",
     "tailwindcss-animate": "^1.0.7",

src/app/api/route.ts

@@ -0,0 +1,23 @@
+import { redirect } from "next/navigation";
+import { NextRequest, NextResponse } from "next/server";
+//import { Server } from 'socket.io'
+
+// redirects to default route "general" for the app
+export async function GET(request: NextRequest, response: NextResponse) {
+/*
+  if (response.socket.server.io) {
+    console.log('Socket is already running')
+  } else {
+    console.log('Socket is initializing')
+    const io = new Server(res.socket.server)
+    res.socket.server.io = io
+
+    io.on('connection', socket => {
+      socket.on('input-change', msg => {
+        socket.broadcast.emit('update-input', msg)
+      })
+    })
+  }
+  res.end()*/
+    return redirect(`/project/app/overview?appId=${new URL(request.url).searchParams.get("appId")}`);
+}

src/app/project/app/[tabName]/action.ts

@@ -1,5 +1,6 @@
 'use server'
 
+import { SuccessActionResult } from "@/model/server-action-error-return.model";
 import appService from "@/server/services/app.service";
 import deploymentService from "@/server/services/deployment.service";
 import { getAuthUserSession, simpleAction } from "@/server/utils/action-wrapper.utils";
@@ -9,11 +10,21 @@ export const deploy = async (appId: string) =>
     simpleAction(async () => {
         await getAuthUserSession();
         await appService.buildAndDeploy(appId);
+        return new SuccessActionResult(undefined, 'Successfully started deployment.');
     });
 
-export const test = async (appId: string) =>
+export const stopApp = async (appId: string) =>
     simpleAction(async () => {
         await getAuthUserSession();
         const app = await appService.getExtendedById(appId);
-        await deploymentService.getDeploymentHistory(app.projectId, app.id);
+        await deploymentService.setReplicasForDeployment(app.projectId, app.id, 0);
+        return new SuccessActionResult(undefined, 'Successfully stopped app.');
+    });
+
+export const startApp = async (appId: string) =>
+    simpleAction(async () => {
+        await getAuthUserSession();
+        const app = await appService.getExtendedById(appId);
+        await deploymentService.setReplicasForDeployment(app.projectId, app.id, app.replicas);
+        return new SuccessActionResult(undefined, 'Successfully started app.');
     });

src/app/project/app/[tabName]/app-action-buttons.tsx

@@ -2,19 +2,20 @@
 
 import { Button } from "@/components/ui/button";
 import { Card, CardContent } from "@/components/ui/card";
-import { deploy, test } from "./action";
+import { deploy, startApp, stopApp } from "./action";
 import { AppExtendedModel } from "@/model/app-extended.model";
+import { Toast } from "@/lib/toast.utils";
 
 export default function AppActionButtons({
     app
 }: {
     app: AppExtendedModel;
 }) {
-
     return <Card>
         <CardContent className="p-4 flex gap-4">
-            <Button onClick={() => deploy(app.id)}>Deploy</Button>
-            <Button  onClick={() => test(app.id)} variant="secondary">Start</Button>
+            <Button onClick={() => Toast.fromAction(() => deploy(app.id))}>Deploy</Button>
+            <Button onClick={() => Toast.fromAction(() => startApp(app.id))} variant="secondary">Start</Button>
+            <Button onClick={() => Toast.fromAction(() => stopApp(app.id))} variant="secondary">Stop</Button>
             <Button variant="secondary">Rebuild</Button>
         </CardContent>
     </Card >;

src/app/project/app/[tabName]/app-tabs.tsx

@@ -11,6 +11,7 @@ import StorageList from "./storage/storages";
 import { AppExtendedModel } from "@/model/app-extended.model";
 import { BuildJobModel } from "@/model/build-job";
 import BuildsTab from "./overview/builds-tab";
+import Logs from "./overview/logs";
 
 export default function AppTabs({
     app,
@@ -35,7 +36,9 @@ export default function AppTabs({
                 <TabsTrigger value="storage">Storage</TabsTrigger>
             </TabsList>
             <TabsContent value="overview" className="space-y-4">
+                <Logs app={app} />
                 <BuildsTab app={app} />
+
             </TabsContent>
             <TabsContent value="general" className="space-y-4">
                 <GeneralAppSource app={app} />

src/app/project/app/[tabName]/overview/actions.ts

@@ -1,7 +1,8 @@
 'use server'
 
 import { BuildJobModel } from "@/model/build-job";
-import { DeploymentInfoModel } from "@/model/deployment";
+import { DeploymentInfoModel } from "@/model/deployment-info.model";
+import { PodsInfoModel } from "@/model/pods-info.model";
 import { ServerActionResult, SuccessActionResult } from "@/model/server-action-error-return.model";
 import appService from "@/server/services/app.service";
 import buildService from "@/server/services/build.service";
@@ -21,4 +22,11 @@ export const deleteBuild = async (buildName: string) =>
         await getAuthUserSession();
         await buildService.deleteBuild(buildName);
         return new SuccessActionResult(undefined, 'Successfully stopped and deleted build.');
-    }) as Promise<ServerActionResult<unknown, void>>;
+    }) as Promise<ServerActionResult<unknown, void>>;
+
+export const getPodsForApp = async (appId: string) =>
+    simpleAction(async () => {
+        await getAuthUserSession();
+        const app = await appService.getExtendedById(appId);
+        return await deploymentService.getPodsForApp(app.projectId, appId);
+    }) as Promise<ServerActionResult<unknown, PodsInfoModel[]>>;

src/app/project/app/[tabName]/overview/builds-tab.tsx

@@ -13,8 +13,10 @@ import BuildStatusBadge from "./build-status-badge";
 import { Button } from "@/components/ui/button";
 import { useConfirmDialog } from "@/lib/zustand.states";
 import { Toast } from "@/lib/toast.utils";
-import { DeploymentInfoModel } from "@/model/deployment";
+import { DeploymentInfoModel } from "@/model/deployment-info.model";
 import DeploymentStatusBadge from "./deployment-status-badge";
+import { io } from "socket.io-client";
+import { podLogsSocket } from "@/socket";
 
 export default function BuildsTab({
     app
@@ -54,7 +56,6 @@ export default function BuildsTab({
         }
     }
 
-
     useEffect(() => {
         if (app.sourceType === 'container') {
             return;

src/app/project/app/[tabName]/overview/deployment-status-badge.tsx

@@ -1,6 +1,6 @@
 'use client'
 
-import { DeplyomentStatus } from "@/model/deployment";
+import { DeplyomentStatus } from "@/model/deployment-info.model";
 
 
 export default function DeploymentStatusBadge(

src/app/project/app/[tabName]/overview/logs-streamed.tsx

@@ -0,0 +1,64 @@
+import { Card, CardContent, CardDescription, CardHeader, CardTitle } from "@/components/ui/card";
+import { AppExtendedModel } from "@/model/app-extended.model";
+import { useEffect, useState } from "react";
+import { podLogsSocket } from "@/socket";
+import { Textarea } from "@/components/ui/textarea";
+
+export default function LogsStreamed({
+    app,
+    podName,
+}: {
+    app: AppExtendedModel;
+    podName: string;
+}) {
+    const [isConnected, setIsConnected] = useState(false);
+    const [transport, setTransport] = useState("N/A");
+    const [logs, setLogs] = useState<string>('');
+
+    useEffect(() => {
+        function onConnect() {
+            setIsConnected(true);
+            setTransport(podLogsSocket.io.engine.transport.name);
+
+            podLogsSocket.io.engine.on("upgrade", (transport) => {
+                setTransport(transport.name);
+            });
+        }
+
+        function onDisconnect() {
+            setIsConnected(false);
+            setTransport("N/A");
+        }
+
+        if (podLogsSocket.connected) {
+            onConnect();
+        }
+
+        podLogsSocket.emit('joinPodLog', { appId: app.id, podName });
+
+        const myListener = (e: string) => {
+            setLogs(e);
+        }
+
+        podLogsSocket.on("connect", onConnect);
+        podLogsSocket.on("disconnect", onDisconnect);
+        podLogsSocket.on(`logs_${app.projectId}_${app.id}_${podName}`, myListener);
+
+        return () => {
+            podLogsSocket.off("connect", onConnect);
+            podLogsSocket.off("disconnect", onDisconnect);
+            podLogsSocket.off(`logs_${app.projectId}_${app.id}_${podName}`, myListener);
+        };
+    }, [app, podName]);
+
+
+    if (app.sourceType === 'container') {
+        return <></>;
+    }
+
+    return <>
+        <Textarea value={logs} readOnly className="h-[400px] bg-slate-900 text-white" />
+
+        <div className="text-sm pl-1">Status: {isConnected ? 'Connected' : 'Disconnected'}</div>
+    </>;
+}

src/app/project/app/[tabName]/overview/logs.tsx

@@ -0,0 +1,71 @@
+import { Card, CardContent, CardDescription, CardHeader, CardTitle } from "@/components/ui/card";
+import { AppExtendedModel } from "@/model/app-extended.model";
+import { useEffect, useState } from "react";
+import { podLogsSocket } from "@/socket";
+import LogsStreamed from "./logs-streamed";
+import { getPodsForApp } from "./actions";
+import { PodsInfoModel } from "@/model/pods-info.model";
+import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select";
+import FullLoadingSpinner from "@/components/ui/full-loading-spinnter";
+
+export default function Logs({
+    app
+}: {
+    app: AppExtendedModel;
+}) {
+    const [selectedPod, setSelectedPod] = useState<string | undefined>(undefined);
+    const [appPods, setAppPods] = useState<PodsInfoModel[] | undefined>(undefined);
+    const [error, setError] = useState<string | undefined>(undefined);
+
+    const updateBuilds = async () => {
+        setError(undefined);
+        try {
+            const response = await getPodsForApp(app.id);
+            if (response.status === 'success' && response.data) {
+                setAppPods(response.data);
+                if (!selectedPod && response.data.length > 0) {
+                    setSelectedPod(response.data[0].podName);
+                }
+            } else {
+                console.error(response);
+                setError(response.message ?? 'An unknown error occurred.');
+            }
+        } catch (ex) {
+            console.error(ex);
+            setError('An unknown error occurred.');
+        }
+    }
+
+    useEffect(() => {
+        if (app.sourceType === 'container') {
+            return;
+        }
+        updateBuilds();
+        const intervalId = setInterval(updateBuilds, 10000);
+        return () => clearInterval(intervalId);
+    }, [app]);
+
+
+
+    return <>
+        <Card>
+            <CardHeader>
+                <CardTitle>Logs</CardTitle>
+                <CardDescription>App Logs.</CardDescription>
+            </CardHeader>
+            <CardContent className="space-y-4">
+                {!appPods && <FullLoadingSpinner />}
+                {appPods && appPods.length === 0 && <div>No running pods found for this app.</div>}
+                {appPods && <Select defaultValue={appPods[0].podName} onValueChange={(val) => setSelectedPod(val)}>
+                    <SelectTrigger className="w-[180px]">
+                        <SelectValue placeholder="Pod wählen" />
+                    </SelectTrigger>
+                    <SelectContent>
+                        {appPods.map(pod => <SelectItem key={pod.podName} value={pod.podName}>{pod.podName}</SelectItem>)}
+                    </SelectContent>
+                </Select>}
+                {selectedPod && <LogsStreamed app={app} podName={selectedPod} />}
+            </CardContent>
+        </Card >
+    </>;
+}

src/model/pods-info.model.ts

@@ -0,0 +1,10 @@
+import { z } from "zod";
+
+export const podsInfoZodModel = z.object({
+    podName: z.string(),
+    containerName: z.string()
+});
+
+export type PodsInfoModel = z.infer<typeof podsInfoZodModel>;
+
+

src/server.ts

@@ -0,0 +1,32 @@
+import { createServer } from 'http'
+import { parse } from 'url'
+import next from 'next'
+import webSocketHandler from './socket-io'
+import { Server } from 'socket.io'
+
+// Source: https://nextjs.org/docs/app/building-your-application/configuring/custom-server
+
+const port = parseInt(process.env.PORT || '3000', 10)
+const dev = process.env.NODE_ENV !== 'production'
+const app = next({ dev })
+const handle = app.getRequestHandler()
+
+app.prepare().then(() => {
+
+    const server = createServer((req, res) => {
+        const parsedUrl = parse(req.url!, true)
+        handle(req, res, parsedUrl)
+    });
+
+    webSocketHandler.initializeSocketIo(server);
+    //const io = new Server(server);
+
+
+    server.listen(port)
+
+    console.log(
+        `> Server listening at http://localhost:${port} as ${dev ? 'development' : process.env.NODE_ENV
+        }`
+    )
+});
+

src/server/adapter/kubernetes-api.adapter.ts

@@ -6,6 +6,8 @@ const getK8sCoreApiClient = () => {
     const k8sCoreClient = kc.makeApiClient(k8s.CoreV1Api);
     return k8sCoreClient;
 }
+const k8sCoreClient = globalThis.k8sCoreGlobal ?? getK8sCoreApiClient()
+if (process.env.NODE_ENV !== 'production') globalThis.k8sCoreGlobal = k8sCoreClient
 
 const getK8sAppsApiClient = () => {
     const kc = new k8s.KubeConfig();
@@ -13,6 +15,8 @@ const getK8sAppsApiClient = () => {
     const k8sCoreClient = kc.makeApiClient(k8s.AppsV1Api);
     return k8sCoreClient;
 }
+const k8sAppsClient = globalThis.k8sAppsGlobal ?? getK8sAppsApiClient()
+if (process.env.NODE_ENV !== 'production') globalThis.k8sAppsGlobal = k8sAppsClient
 
 const getK8sBatchApiClient = () => {
     const kc = new k8s.KubeConfig();
@@ -20,27 +24,35 @@ const getK8sBatchApiClient = () => {
     const k8sJobClient = kc.makeApiClient(k8s.BatchV1Api);
     return k8sJobClient;
 }
+const k8sJobClient = globalThis.k8sJobGlobal ?? getK8sBatchApiClient()
+if (process.env.NODE_ENV !== 'production') globalThis.k8sJobGlobal = k8sJobClient
+
+
+const getK8sLogApiClient = () => {
+    const kc = new k8s.KubeConfig();
+    kc.loadFromFile('/workspace/kube-config.config'); // todo update --> use security role
+    const logClient = new k8s.Log(kc)
+    return logClient;
+}
+const k8sLogClient = globalThis.k8sLogGlobal ?? getK8sLogApiClient()
+if (process.env.NODE_ENV !== 'production') globalThis.k8sLogGlobal = k8sLogClient
 
 declare const globalThis: {
     k8sCoreGlobal: ReturnType<typeof getK8sCoreApiClient>;
     k8sAppsGlobal: ReturnType<typeof getK8sAppsApiClient>;
     k8sJobGlobal: ReturnType<typeof getK8sBatchApiClient>;
+    k8sLogGlobal: ReturnType<typeof getK8sLogApiClient>;
 } & typeof global;
 
-const k8sCoreClient = globalThis.k8sCoreGlobal ?? getK8sCoreApiClient()
-if (process.env.NODE_ENV !== 'production') globalThis.k8sCoreGlobal = k8sCoreClient
 
 
-const k8sAppsClient = globalThis.k8sAppsGlobal ?? getK8sAppsApiClient()
-if (process.env.NODE_ENV !== 'production') globalThis.k8sAppsGlobal = k8sAppsClient
 
-const k8sJobClient = globalThis.k8sJobGlobal ?? getK8sBatchApiClient()
-if (process.env.NODE_ENV !== 'production') globalThis.k8sJobGlobal = k8sJobClient
 
 class K3sApiAdapter {
     core = k8sCoreClient;
     apps = k8sAppsClient;
     batch = k8sJobClient;
+    log = k8sLogClient;
 }
 
 const k3s = new K3sApiAdapter();

src/server/services/build.service.ts

@@ -2,8 +2,6 @@ import { AppExtendedModel } from "@/model/app-extended.model";
 import k3s from "../adapter/kubernetes-api.adapter";
 import { V1Job, V1JobStatus } from "@kubernetes/client-node";
 import { StringUtils } from "../utils/string.utils";
-import { revalidateTag, unstable_cache } from "next/cache";
-import { Tags } from "../utils/cache-tag-generator.utils";
 import { BuildJobModel } from "@/model/build-job";
 import { ServiceException } from "@/model/service.exception.model";
 
@@ -61,7 +59,7 @@ class BuildService {
             ];
         }
         await k3s.batch.createNamespacedJob(buildNamespace, jobDefinition);
-        revalidateTag(Tags.appBuilds(app.id));
+        //revalidateTag(Tags.appBuilds(app.id));
 
         const buildJobPromise = this.waitForJobCompletion(jobDefinition.metadata!.name!)
 
@@ -80,29 +78,23 @@ class BuildService {
     }
 
     async getBuildsForApp(appId: string) {
-        return await unstable_cache(async (appId: string) => {
-            const jobNamePrefix = StringUtils.toJobName(appId);
-            const jobs = await k3s.batch.listNamespacedJob(buildNamespace);
-            const jobsOfBuild = jobs.body.items.filter((job) => job.metadata?.name?.startsWith(jobNamePrefix));
-            const builds = jobsOfBuild.map((job) => {
-                return {
-                    name: job.metadata?.name,
-                    startTime: job.status?.startTime,
-                    status: this.getJobStatusString(job.status),
-                } as BuildJobModel;
-            });
-            builds.sort((a, b) => {
-                if (a.startTime && b.startTime) {
-                    return new Date(b.startTime).getTime() - new Date(a.startTime).getTime();
-                }
-                return 0;
-            });
-            return builds;
-        },
-            [Tags.appBuilds(appId)], {
-            tags: [Tags.appBuilds(appId)],
-            revalidate: 10,
-        })(appId);
+        const jobNamePrefix = StringUtils.toJobName(appId);
+        const jobs = await k3s.batch.listNamespacedJob(buildNamespace);
+        const jobsOfBuild = jobs.body.items.filter((job) => job.metadata?.name?.startsWith(jobNamePrefix));
+        const builds = jobsOfBuild.map((job) => {
+            return {
+                name: job.metadata?.name,
+                startTime: job.status?.startTime,
+                status: this.getJobStatusString(job.status),
+            } as BuildJobModel;
+        });
+        builds.sort((a, b) => {
+            if (a.startTime && b.startTime) {
+                return new Date(b.startTime).getTime() - new Date(a.startTime).getTime();
+            }
+            return 0;
+        });
+        return builds;
     }
 
     async waitForJobCompletion(jobName: string) {

src/server/services/deployment.service.ts

@@ -3,8 +3,10 @@ import k3s from "../adapter/kubernetes-api.adapter";
 import { V1Deployment } from "@kubernetes/client-node";
 import buildService from "./build.service";
 import { ListUtils } from "../utils/list.utils";
-import { DeploymentInfoModel, DeplyomentStatus } from "@/model/deployment";
+import { DeploymentInfoModel, DeplyomentStatus } from "@/model/deployment-info.model";
 import { BuildJobStatus } from "@/model/build-job";
+import { ServiceException } from "@/model/service.exception.model";
+import { PodsInfoModel } from "@/model/pods-info.model";
 
 class DeploymentService {
 
@@ -140,6 +142,15 @@ class DeploymentService {
         await this.createOrUpdateService(app);
     }
 
+    async setReplicasForDeployment(projectId: string, appId: string, replicas: number) {
+        const existingDeployment = await this.getDeployment(projectId, appId);
+        if (!existingDeployment) {
+            throw new ServiceException("This app has not been deployed yet. Please deploy it first.");
+        }
+        existingDeployment.spec!.replicas = replicas;
+        return k3s.apps.replaceNamespacedDeployment(appId, projectId, existingDeployment);
+    }
+
     async createNamespaceIfNotExists(namespace: string) {
         const existingNamespaces = await this.getNamespaces();
         if (existingNamespaces.includes(namespace)) {
@@ -159,6 +170,22 @@ class DeploymentService {
         }
     }
 
+    async getPodsForApp(projectId: string, appId: string) {
+        const res = await k3s.core.listNamespacedPod(projectId, undefined, undefined, undefined, undefined, `app=${appId}`);
+        return res.body.items.map((item) => ({
+            podName: item.metadata?.name!,
+            containerName: item.spec?.containers?.[0].name!
+        })).filter((item) => !!item.podName && !!item.containerName) as PodsInfoModel[];
+    }
+
+    async getPodByName(projectId: string, podName: string) {
+        const res = await k3s.core.readNamespacedPod(podName, projectId);
+        return {
+            podName: res.body.metadata?.name!,
+            containerName: res.body.spec?.containers?.[0].name!
+        } as PodsInfoModel;
+    }
+
 
     /**
      * Searches for Build Jobs (only for Git Projects) and ReplicaSets (for all projects) and returns a list of DeploymentModel

src/server/services/log.service.ts

@@ -0,0 +1,24 @@
+import { revalidateTag, unstable_cache } from "next/cache";
+import dataAccess from "../adapter/db.client";
+import { Tags } from "../utils/cache-tag-generator.utils";
+import { Prisma, Project } from "@prisma/client";
+import { StringUtils } from "../utils/string.utils";
+import deploymentService from "./deployment.service";
+import k3s from "../adapter/kubernetes-api.adapter";
+
+class LogService {
+/*
+    async streamLogs(namespace: string, podName: string, containerName: string, logStream: NodeJS.WritableStream) {
+        const req = await k3s.log.log(namespace, podName, containerName, logStream, {
+            follow: true,
+            pretty: false,
+            tailLines: 100,
+        });
+
+        return req;
+    }*/
+
+}
+
+const logService = new LogService();
+export default logService;

src/socket-io.ts

@@ -0,0 +1,65 @@
+import type http from "node:http";
+import { Server } from "socket.io";
+import k3s from "./server/adapter/kubernetes-api.adapter";
+import stream from "stream";
+import appService from "./server/services/app.service";
+import deploymentService from "./server/services/deployment.service";
+import dataAccess from "./server/adapter/db.client";
+
+class WebSocketHandler {
+
+
+
+	initializeSocketIo(server: http.Server<typeof http.IncomingMessage, typeof http.ServerResponse>) {
+
+		const io = new Server(server);
+
+		const podLogsNamespace = io.of("/pod-logs");
+		podLogsNamespace.on("connection", (socket) => {
+
+			let req: any;
+
+			socket.on('joinPodLog', async (podInfo) => {
+				const { appId, podName } = podInfo;
+				if (!appId || !podName) {
+					return;
+				}
+				const app = await dataAccess.client.app.findFirstOrThrow({
+					where: {
+						id: appId
+					}
+				});
+
+				const pod = await deploymentService.getPodByName(app.projectId, podName);
+
+				const logStream = new stream.PassThrough();
+				logStream.on('data', (chunk) => {
+					socket.emit(`logs_${app.projectId}_${app.id}_${pod.podName}`, chunk.toString());
+				});
+
+				req = await k3s.log.log(app.projectId, pod.podName, pod.containerName, logStream, {
+					follow: true,
+					pretty: false,
+					tailLines: 100,
+				});
+
+
+
+				// on disconnect from client
+
+			});
+
+			socket.on('disconnect', () => {
+				if (req) {
+					req.abort();
+					console.log("Aborted request");
+				}
+			});
+		});
+
+	};
+
+}
+const webSocketHandler = new WebSocketHandler();
+export default webSocketHandler;
+

src/socket.ts

@@ -0,0 +1,8 @@
+"use client";
+
+import { Manager } from "socket.io-client";
+
+const manager = new Manager();
+
+export const podLogsSocket = manager.socket("/pod-logs");
+//export const deploymentStatusSocket = manager.socket("/deployment-status");