gin event source实现k8s pod logs 实时刷新

【代码】gin event source实现k8s pod logs 实时刷新。

wangzifei1234

477人浏览 · 2023-06-25 10:18:46

wangzifei1234 · 2023-06-25 10:18:46 发布

前端js

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>
<div id="result">

</div>

<div>
    <button type="button" onclick="submit()">提交!</button>
</div>
</body>
<script>
    function submit() {

        const stream = new EventSource("http://localhost:8080/stream?namespace=dev&podName=xxxx&containerName=xxxxxx")
        stream.addEventListener("message", function (e) {
           console.log(e)
           document.getElementById("result").innerHTML += e.data
        });
        stream.addEventListener("stop", function (e) {
            stream.close()
            console.log(e)
        });
        stream.onerror = function (event) {
            stream.close()
            console.log(event)
        }
    }

</script>
</html>

后端go代码

package main

import (
	"bufio"
	"context"
	"errors"
	"fmt"
	"github.com/gin-gonic/gin"
	cors "github.com/rs/cors/wrapper/gin"
	"io"
	v1 "k8s.io/api/core/v1"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/clientcmd"
	"log"
	"net/http"
	"os"
	"os/signal"
	"time"
)

const kubeconfig = `
`

type SteamRequest struct {
	Namespace     string `form:"namespace" binding:"required"`
	PodName       string `form:"podName" binding:"required"`
	ContainerName string `form:"containerName" binding:"required"`
}

func setupRouter() *gin.Engine {
	r := gin.Default()

	r.Use(cors.New(cors.Options{
		AllowedOrigins: []string{"*"},
		AllowedMethods: []string{
			http.MethodHead,
			http.MethodGet,
			http.MethodPost,
			http.MethodPut,
			http.MethodPatch,
			http.MethodDelete,
		},
		AllowedHeaders:     []string{"*"},
		AllowCredentials:   true,
		OptionsPassthrough: true,
	}))

	r.GET("/stream", func(c *gin.Context) {
		var request SteamRequest
		err := c.ShouldBindQuery(&request)
		if err != nil {
			log.Fatal(err.Error())
		}
		log.Printf("request: %v \n", request)

		clientConfig, err := clientcmd.NewClientConfigFromBytes([]byte(kubeconfig))
		if err != nil {
			panic(err.Error())
		}
		config, err := clientConfig.ClientConfig()
		if err != nil {
			panic(err.Error())
		}
		// create the clientset
		clientset, err := kubernetes.NewForConfig(config)
		if err != nil {
			panic(err.Error())
		}
		var lines int64 = 100
		stream, err := clientset.CoreV1().Pods(request.Namespace).GetLogs(request.PodName, &v1.PodLogOptions{
			Container:    request.ContainerName,
			Follow:       true,
			Timestamps:   false,
			SinceSeconds: nil,
			SinceTime:    nil,
			TailLines:    &lines,
			Previous:     false,
		}).Stream(context.Background())

		if err != nil {
			return
		}

		chanStream := make(chan string, 100)
		go func() {
			defer stream.Close()
			defer close(chanStream)
			bufReader := bufio.NewReader(stream)
			for {
				line, err := bufReader.ReadString('\n')

				if errors.Is(err, io.EOF) {
					fmt.Println("Stream finished")
					chanStream <- "<!finish>"
					return
				}

				if err != nil {
					fmt.Printf("Stream error: %v\n", err)
					chanStream <- "<!error>"
					return
				}
				chanStream <- line
				fmt.Printf("Stream response: %v\n", line)
			}
		}()

		c.Stream(func(w io.Writer) bool {
			if msg, ok := <-chanStream; ok {
				if msg == "<!finish>" {
					c.SSEvent("stop", "finish")
				}
				if msg == "<!error>" {
					c.SSEvent("stop", "error")
				}
				c.SSEvent("message", msg)
				fmt.Printf("message: %v\n", msg)

				return true
			}
			return false
		})

	})
	return r
}

func main() {
	gin.SetMode(gin.ReleaseMode)
	router := setupRouter()

	srv := &http.Server{
		Addr:    ":8080",
		Handler: router,
	}
	go func() {
		// 服务连接
		if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
			log.Fatalf("listen: %s\n", err)
		}
	}()

	// 等待中断信号以优雅地关闭服务器（设置 5 秒的超时时间）
	quit := make(chan os.Signal)
	signal.Notify(quit, os.Interrupt)
	<-quit
	log.Println("Shutdown Server ...")

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if err := srv.Shutdown(ctx); err != nil {
		log.Fatal("Server Shutdown:", err)
	}
	log.Println("Server exiting")
}

K8S/Kubernetes

K8S/Kubernetes社区为您提供最前沿的新闻资讯和知识内容

更多推荐

【深度】阿里巴巴万级规模 K8s 集群全局高可用体系之美

作者 | 韩堂、柘远、沉醉来源 | 阿里巴巴云原生公众号前言台湾作家林清玄在接受记者采访的时候，如此评价自己 30 多年写作生涯：“第一个十年我才华横溢，‘贼光闪现’，令周边黯然失色；第二个十年，我终于‘宝光现形’，不再去抢风头，反而与身边的美丽相得益彰；进入第三个十年，繁华落尽见真醇，我进入了‘醇光初现’的阶段，真正体味到了境界之美”。长夜有穷，真水无香。领略过了 K8s“身在江

K8S/Kubernetes

如何基于 K8s 构建下一代 DevOps 平台？

作者 | 孙健波（天元）导读：当前云原生 DevOps 体系现状如何？面临哪些挑战？如何通过 OAM 解决云原生 DevOps 场景下的诸多问题？云原生开发应用模型 OAM(Open Application Model) 社区核心成员孙健波将为大家一一解答，并分享如何基于 OAM 和 Kubernetes 打造无限能力的下一代 DevOps 平台。什么是 DevOps？为什么基于 Kub