perf: remove mq etcd dependency, export log api

Signed-off-by: 逆流而上 <[email protected]>

api/api/api_interface.go

@@ -281,6 +281,7 @@ type AppRestoreInterface interface {
 // PodInterface defines api methods about k8s pods.
 type PodInterface interface {
 	PodDetail(w http.ResponseWriter, r *http.Request)
+	PodLogs(w http.ResponseWriter, r *http.Request)
 }
 
 // RegistryAuthSecretInterface registry auth secret interface

api/api_routers/version2/v2Routers.go

@@ -64,6 +64,7 @@ func (v2 *V2) Routes() chi.Router {
 	r.Mount("/monitor", v2.monitorRouter())
 	r.Mount("/helm", v2.helmRouter())
 	r.Mount("/proxy-pass", v2.proxyRoute())
+	r.Get("/pods/logs", controller.GetManager().PodLogs)
 
 	return r
 }

api/controller/pods.go

@@ -19,18 +19,21 @@
 package controller
 
 import (
+	"bufio"
 	"fmt"
-	"net/http"
-	"strings"
-
 	"github.com/go-chi/chi"
 	"github.com/goodrain/rainbond/api/handler"
 	ctxutil "github.com/goodrain/rainbond/api/util/ctx"
 	"github.com/goodrain/rainbond/db"
 	"github.com/goodrain/rainbond/db/model"
+	"github.com/goodrain/rainbond/pkg/component/k8s"
 	httputil "github.com/goodrain/rainbond/util/http"
 	"github.com/goodrain/rainbond/worker/server"
 	"github.com/sirupsen/logrus"
+	corev1 "k8s.io/api/core/v1"
+	"net/http"
+	"strings"
+	"time"
 )
 
 // PodController is an implementation of PodInterface
@@ -114,3 +117,57 @@ func (p *PodController) PodDetail(w http.ResponseWriter, r *http.Request) {
 	}
 	httputil.ReturnSuccess(r, w, pd)
 }
+
+// PodLogs -
+func (p *PodController) PodLogs(w http.ResponseWriter, r *http.Request) {
+	podNamespace := r.URL.Query().Get("pod_ns")
+	podName := r.URL.Query().Get("pod_name")
+	// Get Kubernetes pod logs
+	lines := int64(1280)
+	req := k8s.Default().Clientset.CoreV1().Pods(podNamespace).GetLogs(podName, &corev1.PodLogOptions{
+		Follow:     true,
+		Timestamps: true,
+		TailLines:  &lines,
+	})
+	stream, err := req.Stream(r.Context())
+	if err != nil {
+		logrus.Errorf("Error opening log stream: %v", err)
+		http.Error(w, "Error opening log stream", http.StatusInternalServerError)
+		return
+	}
+	defer stream.Close()
+	// Use Flusher to send headers to the client
+	flusher, ok := w.(http.Flusher)
+	if !ok {
+		logrus.Errorf("Streaming not supported")
+		http.Error(w, "Streaming not supported", http.StatusInternalServerError)
+		return
+	}
+
+	// Set headers for SSE
+	w.Header().Set("Content-Type", "text/event-stream")
+	w.Header().Set("Cache-Control", "no-cache")
+	w.Header().Set("Connection", "keep-alive")
+
+	scanner := bufio.NewScanner(stream)
+	ticker := time.NewTicker(50 * time.Millisecond)
+	defer ticker.Stop()
+
+	var messages []string
+
+	for {
+		select {
+		case <-ticker.C:
+			fmt.Fprintf(w, strings.Join(messages, "\n\n"))
+			messages = messages[:0] // Clear the slice
+			flusher.Flush()
+		case <-r.Context().Done():
+			logrus.Warningf("Request context done: %v", r.Context().Err())
+			return
+		default:
+			if len(messages) < 128 && scanner.Scan() {
+				messages = append(messages, "data: "+scanner.Text())
+			}
+		}
+	}
+}

api/server/api.go

@@ -23,18 +23,16 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 	"github.com/coreos/etcd/clientv3"
+	"github.com/goodrain/rainbond/api/handler"
 	"github.com/goodrain/rainbond/pkg/interceptors"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"github.com/prometheus/common/version"
 	"io/ioutil"
 	"log"
 	"net/http"
 	"os"
 	"strings"
-	"time"
-
-	"github.com/goodrain/rainbond/api/handler"
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/promhttp"
-	"github.com/prometheus/common/version"
 
 	"github.com/goodrain/rainbond/util"
 
@@ -108,7 +106,7 @@ func (m *Manager) SetMiddleware() {
 	//Gracefully absorb panics and prints the stack trace
 	r.Use(interceptors.Recoverer)
 	//request time out
-	r.Use(middleware.Timeout(time.Second * 5))
+	//r.Use(middleware.Timeout(time.Second * 5))
 	//simple authz
 	if os.Getenv("TOKEN") != "" {
 		r.Use(apimiddleware.FullToken)

cmd/api/main.go

@@ -47,7 +47,8 @@ func main() {
 		APIConfig: s.Config,
 	})
 	// 启动 rbd-api
-	err := rainbond.New(context.Background(), configs.Default()).Registry(component.Database()).
+	err := rainbond.New(context.Background(), configs.Default()).
+		Registry(component.Database()).
 		Registry(component.Grpc()).
 		Registry(component.Event()).
 		Registry(component.K8sClient()).

cmd/mq/mq.go

@@ -19,13 +19,15 @@
 package main
 
 import (
-	"fmt"
+	"context"
+	"github.com/goodrain/rainbond/config/configs"
+	"github.com/goodrain/rainbond/pkg/component"
+	"github.com/goodrain/rainbond/pkg/rainbond"
+	"github.com/sirupsen/logrus"
 	"os"
 
 	"github.com/goodrain/rainbond/cmd"
 	"github.com/goodrain/rainbond/cmd/mq/option"
-	"github.com/goodrain/rainbond/cmd/mq/server"
-
 	"github.com/spf13/pflag"
 )
 
@@ -37,8 +39,17 @@ func main() {
 	s.AddFlags(pflag.CommandLine)
 	pflag.Parse()
 	s.SetLog()
-	if err := server.Run(s); err != nil {
-		fmt.Fprintf(os.Stderr, "error: %v\n", err)
-		os.Exit(1)
+
+	configs.SetDefault(&configs.Config{
+		AppName:  "rbd-mq",
+		MQConfig: s.Config,
+	})
+	err := rainbond.New(context.Background(), configs.Default()).
+		Registry(component.MQClient()).
+		RegistryCancel(component.MQGrpcServer()).
+		RegistryCancel(component.MQHealthServer()).
+		Start()
+	if err != nil {
+		logrus.Errorf("start rbd mq error %s", err.Error())
 	}
 }

config/configs/config.go

@@ -1,6 +1,9 @@
 package configs
 
-import "github.com/goodrain/rainbond/cmd/api/option"
+import (
+	apiconfig "github.com/goodrain/rainbond/cmd/api/option"
+	mqconfig "github.com/goodrain/rainbond/cmd/mq/option"
+)
 
 // Env -
 type Env string
@@ -11,7 +14,8 @@ type Config struct {
 	Version   string
 	Env       Env
 	Debug     bool
-	APIConfig option.Config
+	APIConfig apiconfig.Config
+	MQConfig  mqconfig.Config
 }
 
 var defaultConfig *Config