diff --git a/cyclops-ctrl/internal/controller/modules.go b/cyclops-ctrl/internal/controller/modules.go index 1017b9ea..36c4ac00 100644 --- a/cyclops-ctrl/internal/controller/modules.go +++ b/cyclops-ctrl/internal/controller/modules.go @@ -2,12 +2,14 @@ package controller import ( "fmt" + "io" "net/http" "os" - "sigs.k8s.io/yaml" "strings" "time" + "sigs.k8s.io/yaml" + "github.com/gin-gonic/gin" "github.com/cyclops-ui/cyclops/cyclops-ctrl/api/v1alpha1" @@ -541,24 +543,43 @@ func (m *Modules) GetLogs(ctx *gin.Context) { ctx.Header("Access-Control-Allow-Origin", "*") logCount := int64(100) - rawLogs, err := m.kubernetesClient.GetPodLogs( - ctx.Param("namespace"), - ctx.Param("container"), - ctx.Param("name"), - &logCount, - ) - if err != nil { - fmt.Println(err) - ctx.JSON(http.StatusInternalServerError, dto.NewError("Error fetching logs", err.Error())) - return - } - logs := make([]string, 0, len(rawLogs)) - for _, log := range rawLogs { - logs = append(logs, trimLogLine(log)) - } + logChan := make(chan string) - ctx.JSON(http.StatusOK, logs) + go func() { + defer close(logChan) + + err := m.kubernetesClient.GetStreamedPodLogs( + ctx.Request.Context(), // we will have to pass the context for the k8s podClient - so it can stop the stream when the client disconnects + ctx.Param("namespace"), + ctx.Param("container"), + ctx.Param("name"), + &logCount, + logChan, + ) + if err != nil { + return + } + }() + + // stream logs to the client + ctx.Stream(func(w io.Writer) bool { + for { + select { + case log, ok := <-logChan: + if !ok { + return false + } + + ctx.SSEvent("pod-log", log) + return true + case <-ctx.Request.Context().Done(): + return false + case <-ctx.Done(): + return false + } + } + }) } func (m *Modules) GetDeploymentLogs(ctx *gin.Context) { diff --git a/cyclops-ctrl/internal/handler/handler.go b/cyclops-ctrl/internal/handler/handler.go index 47c6d8a9..cd2e5e2d 100644 --- a/cyclops-ctrl/internal/handler/handler.go +++ b/cyclops-ctrl/internal/handler/handler.go @@ -83,7 +83,7 @@ func (h *Handler) Start() error { h.router.GET("/modules/:name/helm-template", modulesController.HelmTemplate) //h.router.POST("/modules/resources", modulesController.ModuleToResources) - h.router.GET("/resources/pods/:namespace/:name/:container/logs", modulesController.GetLogs) + h.router.GET("/resources/pods/:namespace/:name/:container/logs", sse.HeadersMiddleware(), modulesController.GetLogs) h.router.GET("/resources/pods/:namespace/:name/:container/logs/download", modulesController.DownloadLogs) h.router.GET("/manifest", modulesController.GetManifest) diff --git a/cyclops-ctrl/pkg/cluster/k8sclient/client.go b/cyclops-ctrl/pkg/cluster/k8sclient/client.go index f1523b0a..43cb5d9d 100644 --- a/cyclops-ctrl/pkg/cluster/k8sclient/client.go +++ b/cyclops-ctrl/pkg/cluster/k8sclient/client.go @@ -7,11 +7,13 @@ import ( "errors" "fmt" "io" + networkingv1 "k8s.io/api/networking/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/util/intstr" "k8s.io/apimachinery/pkg/watch" "k8s.io/client-go/tools/cache" + "os" "os/exec" "sort" @@ -171,6 +173,34 @@ func (k *KubernetesClient) GetPods(namespace, name string) ([]apiv1.Pod, error) return podList.Items, err } +func (k *KubernetesClient) GetStreamedPodLogs(ctx context.Context, namespace, container, name string, logCount *int64, logChan chan<- string) error { + podLogOptions := apiv1.PodLogOptions{ + Container: container, + TailLines: logCount, + Timestamps: true, + Follow: true, + } + + podClient := k.clientset.CoreV1().Pods(namespace).GetLogs(name, &podLogOptions) + stream, err := podClient.Stream(ctx) + if err != nil { + return err + } + defer stream.Close() + + scanner := bufio.NewScanner(stream) + + for scanner.Scan() { + logChan <- scanner.Text() + } + + if err := scanner.Err(); err != nil { + return err + } + + return nil +} + func (k *KubernetesClient) GetPodLogs(namespace, container, name string, numLogs *int64) ([]string, error) { podLogOptions := apiv1.PodLogOptions{ Container: container, diff --git a/cyclops-ui/src/components/k8s-resources/common/PodTable/PodLogs.tsx b/cyclops-ui/src/components/k8s-resources/common/PodTable/PodLogs.tsx index cfd34609..69db32e7 100644 --- a/cyclops-ui/src/components/k8s-resources/common/PodTable/PodLogs.tsx +++ b/cyclops-ui/src/components/k8s-resources/common/PodTable/PodLogs.tsx @@ -1,16 +1,17 @@ import { ReadOutlined } from "@ant-design/icons"; import { Alert, Button, Col, Divider, Modal, Tabs, TabsProps } from "antd"; -import axios from "axios"; import { useState } from "react"; import ReactAce from "react-ace/lib/ace"; import { mapResponseError } from "../../../../utils/api/errors"; +import { isStreamingEnabled } from "../../../../utils/api/common"; +import { logStream } from "../../../../utils/api/sse/logs"; interface PodLogsProps { pod: any; } const PodLogs = ({ pod }: PodLogsProps) => { - const [logs, setLogs] = useState(""); + const [logs, setLogs] = useState([]); const [logsModal, setLogsModal] = useState({ on: false, namespace: "", @@ -18,6 +19,7 @@ const PodLogs = ({ pod }: PodLogsProps) => { containers: [], initContainers: [], }); + const [error, setError] = useState({ message: "", description: "", @@ -31,7 +33,7 @@ const PodLogs = ({ pod }: PodLogsProps) => { containers: [], initContainers: [], }); - setLogs(""); + setLogs([]); }; const getTabItems = () => { @@ -50,7 +52,7 @@ const PodLogs = ({ pod }: PodLogsProps) => { type="primary" // icon={} onClick={downloadLogs(container.name)} - disabled={logs === "No logs available"} + disabled={logs.length === 0} > Download @@ -58,7 +60,9 @@ const PodLogs = ({ pod }: PodLogsProps) => { @@ -78,7 +82,7 @@ const PodLogs = ({ pod }: PodLogsProps) => { type="primary" // icon={} onClick={downloadLogs(container.name)} - disabled={logs === "No logs available"} + disabled={logs.length === 0} > Download @@ -86,7 +90,9 @@ const PodLogs = ({ pod }: PodLogsProps) => { @@ -99,31 +105,36 @@ const PodLogs = ({ pod }: PodLogsProps) => { }; const onLogsTabsChange = (container: string) => { - axios - .get( - "/api/resources/pods/" + - logsModal.namespace + - "/" + - logsModal.pod + - "/" + - container + - "/logs", - ) - .then((res) => { - if (res.data) { - let log = ""; - res.data.forEach((s: string) => { - log += s; - log += "\n"; - }); - setLogs(log); - } else { - setLogs("No logs available"); - } - }) - .catch((error) => { - setError(mapResponseError(error)); - }); + const controller = new AbortController(); + setLogs(() => []); + + if (isStreamingEnabled()) { + logStream( + logsModal.pod, + logsModal.namespace, + container, + (log, isReset = false) => { + if (isReset) { + setLogs(() => []); + } else { + setLogs((prevLogs) => { + return [...prevLogs, log]; + }); + } + }, + (err, isReset = false) => { + if (isReset) { + setError({ + message: "", + description: "", + }); + } else { + setError(mapResponseError(err)); + } + }, + controller, + ); + } }; const downloadLogs = (container: string) => { @@ -139,45 +150,49 @@ const PodLogs = ({ pod }: PodLogsProps) => { }; }; - const handleViewPodLogs = async () => { - axios - .get( - "/api/resources/pods/" + - pod.namespace + - "/" + - pod.name + - "/" + - pod.containers[0].name + - "/logs", - ) - .then((res) => { - if (res.data) { - let log = ""; - res.data.forEach((s: string) => { - log += s; - log += "\n"; - }); - setLogs(log); - } else { - setLogs("No logs available"); - } - }) - .catch((error) => { - setError(mapResponseError(error)); - }); - - setLogsModal({ - on: true, - namespace: pod.namespace, - pod: pod.name, - containers: pod.containers, - initContainers: pod.initContainers, - }); - }; - return ( <> -