Merge pull request #167 from iutx/feat/remove-erda-cmp-steve-dep

refactor: replace steve aggregator with proxy manager for cluster console

Author: iutx

.dockerignore

@@ -0,0 +1,9 @@
+dist/
+
+# build output
+/bin
+bin
+
+# go cache
+.gocache/
+.gomodcache/

.gitignore

@@ -40,3 +40,7 @@ go.sum
 # for local debug
 /.env
 /.env.*
+
+# go cache
+.gocache/
+.gomodcache/

Dockerfile

@@ -1,7 +1,5 @@
-# syntax = docker/dockerfile:1.2
-
 # Build the manager binary
-FROM golang:1.16 as builder
+FROM m.daocloud.io/docker.io/golang:1.16 AS builder
 
 ARG APP
 WORKDIR /workspace
@@ -14,8 +12,7 @@ RUN --mount=type=cache,target=/root/.cache/go-build \
     --mount=type=cache,target=/go/pkg/mod \
     CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -mod readonly -a  -o ${APP} ./cmd/${APP}/${APP}.go
 
-#FROM centos:7
-FROM kubeprober/alpine:v3.9
+FROM registry.erda.cloud/retag/kubeprober/alpine:v3.9
 
 ARG ARCH=amd64
 ARG APP

pkg/probe-master/alert/ticket/constants.go

@@ -0,0 +1,14 @@
+package ticket
+
+import erda_api "github.com/erda-project/erda/apistructs"
+
+type IssuePriority = erda_api.IssuePriority
+type IssueType = erda_api.IssueType
+
+const (
+	IssueTypeTicket     = erda_api.IssueTypeTicket
+	IssuePriorityUrgent = erda_api.IssuePriorityUrgent
+	IssuePriorityHigh   = erda_api.IssuePriorityHigh
+	IssuePriorityNormal = erda_api.IssuePriorityNormal
+	IssuePriorityLow    = erda_api.IssuePriorityLow
+)

pkg/probe-master/tunnel-server/handler/cluster_proxy.go

@@ -0,0 +1,164 @@
+package handler
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"net/url"
+	"strings"
+
+	"github.com/sirupsen/logrus"
+	"k8s.io/apimachinery/pkg/util/proxy"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/transport"
+
+	kubeproberv1 "github.com/erda-project/kubeprober/apis/v1"
+	"github.com/erda-project/kubeprober/pkg/probe-master/k8sclient"
+	dialclient "github.com/erda-project/kubeprober/pkg/probe-master/tunnel-client"
+)
+
+type ProxyManager struct {
+	Ctx context.Context
+}
+
+var proxyManager *ProxyManager
+
+// NewProxyManager initializes the proxy manager used by console/exec proxying.
+func NewProxyManager(ctx context.Context) *ProxyManager {
+	proxyManager = &ProxyManager{Ctx: ctx}
+	return proxyManager
+}
+
+func (p *ProxyManager) ProxyRequest(rw http.ResponseWriter, req *http.Request, clusterName string) {
+	if clusterName == "" {
+		http.Error(rw, "cluster name is required", http.StatusNotFound)
+		return
+	}
+
+	cluster, err := k8sclient.GetCluster(clusterName)
+	if err != nil {
+		logrus.Errorf("failed to get cluster %s, %v", clusterName, err)
+		http.Error(rw, "Internal server error", http.StatusInternalServerError)
+		return
+	}
+	if cluster == nil {
+		http.Error(rw, fmt.Sprintf("cluster %s not found", clusterName), http.StatusNotFound)
+		return
+	}
+
+	handler, err := newClusterProxyHandler(cluster)
+	if err != nil {
+		logrus.Errorf("failed to create proxy handler for cluster %s, %v", clusterName, err)
+		http.Error(rw, "Internal server error", http.StatusInternalServerError)
+		return
+	}
+
+	handler.ServeHTTP(rw, req)
+}
+
+func newClusterProxyHandler(cluster *kubeproberv1.Cluster) (http.Handler, error) {
+	restConfig, err := dialclient.GenerateProbeClientConf(cluster)
+	if err != nil {
+		return nil, err
+	}
+
+	return newK8sProxyHandler(cluster.Name, restConfig)
+}
+
+func newK8sProxyHandler(clusterName string, cfg *rest.Config) (http.Handler, error) {
+	host := cfg.Host
+	if !strings.HasPrefix(host, "http://") && !strings.HasPrefix(host, "https://") {
+		host = "https://" + host
+	}
+	if !strings.HasSuffix(host, "/") {
+		host = host + "/"
+	}
+	target, err := url.Parse(host)
+	if err != nil {
+		return nil, err
+	}
+
+	transportRT, err := rest.TransportFor(cfg)
+	if err != nil {
+		return nil, err
+	}
+
+	upgradeTransport, err := makeUpgradeTransport(cfg, transportRT)
+	if err != nil {
+		return nil, err
+	}
+
+	proxyHandler := proxy.NewUpgradeAwareHandler(target, transportRT, false, false, &proxyResponder{})
+	proxyHandler.UpgradeTransport = upgradeTransport
+	proxyHandler.UseRequestLocation = true
+	proxyHandler.UseLocationHost = true
+
+	handler := http.Handler(proxyHandler)
+	if len(target.Path) > 1 {
+		handler = prependPath(target.Path[:len(target.Path)-1], handler)
+	}
+
+	prefix := clusterURLPrefix(clusterName)
+	if len(prefix) > 2 {
+		handler = stripLeaveSlash(prefix, handler)
+	}
+
+	return proxyHeaders(handler), nil
+}
+
+type proxyResponder struct{}
+
+func (r *proxyResponder) Error(w http.ResponseWriter, req *http.Request, err error) {
+	logrus.Errorf("error while proxying request: %v", err)
+	http.Error(w, err.Error(), http.StatusInternalServerError)
+}
+
+func clusterURLPrefix(clusterName string) string {
+	return "/api/k8s/clusters/" + clusterName
+}
+
+func proxyHeaders(handler http.Handler) http.Handler {
+	return http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
+		req.Header.Del("Authorization")
+		if req.Header.Get("X-Forwarded-Proto") == "" && req.TLS != nil {
+			req.Header.Set("X-Forwarded-Proto", "https")
+		}
+		handler.ServeHTTP(rw, req)
+	})
+}
+
+func prependPath(prefix string, handler http.Handler) http.Handler {
+	return http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
+		if len(req.URL.Path) > 1 {
+			req.URL.Path = prefix + req.URL.Path
+		} else {
+			req.URL.Path = prefix
+		}
+		handler.ServeHTTP(rw, req)
+	})
+}
+
+func stripLeaveSlash(prefix string, handler http.Handler) http.Handler {
+	return http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
+		path := strings.TrimPrefix(req.URL.Path, prefix)
+		if len(path) > 0 && path[:1] != "/" {
+			path = "/" + path
+		}
+		req.URL.Path = path
+		handler.ServeHTTP(rw, req)
+	})
+}
+
+func makeUpgradeTransport(cfg *rest.Config, rt http.RoundTripper) (proxy.UpgradeRequestRoundTripper, error) {
+	transportConfig, err := cfg.TransportConfig()
+	if err != nil {
+		return nil, err
+	}
+
+	upgrader, err := transport.HTTPWrappersForConfig(transportConfig, proxy.MirrorRequest)
+	if err != nil {
+		return nil, err
+	}
+
+	return proxy.NewUpgradeRequestRoundTripper(rt, upgrader), nil
+}

pkg/probe-master/tunnel-server/handler/console.go

@@ -25,93 +25,115 @@ import (
 	v1 "k8s.io/api/core/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
-	"github.com/erda-project/erda/apistructs"
 	"github.com/erda-project/kubeprober/pkg/probe-master/k8sclient"
 	dialclient "github.com/erda-project/kubeprober/pkg/probe-master/tunnel-client"
 )
 
+const (
+	probeAgentLabelKey   = "app"
+	probeAgentLabelValue = "probe-agent"
+	probeAgentContainer  = "probe-agent"
+	execCommandScript    = "kubectl-shell.sh"
+)
+
 func ClusterConsole(rw http.ResponseWriter, req *http.Request) {
-	// TODO make blow correct
-	vars := mux.Vars(req)
-	clusterName := vars["clusterName"]
+	clusterName := mux.Vars(req)["clusterName"]
+	if clusterName == "" {
+		http.Error(rw, "cluster name is required", http.StatusNotFound)
+		return
+	}
 
 	cluster, err := k8sclient.GetCluster(clusterName)
 	if err != nil {
-		errMsg := fmt.Sprintf("[cluster console] failed to list cluster with name: %s", clusterName)
-		logrus.Errorf(errMsg)
-		rw.Write([]byte(errMsg))
-		rw.WriteHeader(http.StatusInternalServerError)
+		logrus.Errorf("[cluster console] failed to get cluster %s: %v", clusterName, err)
+		http.Error(rw, "Internal server error", http.StatusInternalServerError)
 		return
 	}
 	if cluster == nil {
-		errMsg := fmt.Sprintf("[cluster console] failed to find cluster with name: %s\n", clusterName)
-		rw.Write([]byte(errMsg))
-		rw.WriteHeader(http.StatusBadRequest)
-		return
-	}
-
-	token := cluster.Spec.ClusterConfig.Token
-	if token == "" {
-		errMsg := fmt.Sprintf("[cluster console] invalid token for cluster with name: %s\n", clusterName)
-		rw.Write([]byte(errMsg))
-		rw.WriteHeader(http.StatusInternalServerError)
+		http.Error(rw, fmt.Sprintf("cluster %s not found", clusterName), http.StatusBadRequest)
 		return
 	}
 
-	t, err := base64.StdEncoding.DecodeString(cluster.Spec.ClusterConfig.Token)
+	token, err := decodeClusterToken(clusterName, cluster.Spec.ClusterConfig.Token)
 	if err != nil {
-		errMsg := fmt.Sprintf("[cluster console] invalid token for cluster: %s\n", clusterName)
-		rw.Write([]byte(errMsg))
-		rw.WriteHeader(http.StatusInternalServerError)
+		logrus.Errorf("[cluster console] invalid token for cluster %s: %v", clusterName, err)
+		http.Error(rw, "Internal server error", http.StatusInternalServerError)
 		return
 	}
-	token = string(t)
 
 	clusterclient, err := dialclient.GenerateProbeClient(cluster)
 	if err != nil {
-		errMsg := fmt.Sprintf("[cluster console] invalid token for cluster with name: %s\n", clusterName)
-		rw.Write([]byte(errMsg))
-		rw.WriteHeader(http.StatusInternalServerError)
+		logrus.Errorf("[cluster console] failed to build k8s client for cluster %s: %v", clusterName, err)
+		http.Error(rw, "Internal server error", http.StatusInternalServerError)
 		return
 	}
-	podList := &v1.PodList{}
 
-	err = clusterclient.List(context.Background(), podList,
-		client.InNamespace(cluster.Spec.ClusterConfig.ProbeNamespaces),
-		client.MatchingLabels{"app": "probe-agent"})
+	pod, err := findRunningProbeAgent(req.Context(), clusterclient, cluster.Spec.ClusterConfig.ProbeNamespaces)
 	if err != nil {
-		errMsg := fmt.Sprintf("[cluster console] failed to find probe-agent pod for cluster with name: %s\n", clusterName)
-		rw.Write([]byte(errMsg))
-		rw.WriteHeader(http.StatusInternalServerError)
+		logrus.Errorf("[cluster console] failed to list probe-agent pods for cluster %s: %v", clusterName, err)
+		http.Error(rw, "Internal server error", http.StatusInternalServerError)
+		return
+	}
+	if pod == nil {
+		logrus.Errorf("failed to find a ready probe-agent pod for cluster %s", clusterName)
+		http.Error(rw, fmt.Sprintf("cluster %s does not have a ready probe-agent pod", clusterName), http.StatusInternalServerError)
 		return
 	}
 
-	for _, pod := range podList.Items {
-		if pod.Status.Phase != v1.PodRunning {
-			continue
-		}
+	req.URL.Path = execURLPath(clusterName, pod.Namespace, pod.Name)
+	req.URL.RawQuery = execQuery(token).Encode()
+
+	if proxyManager == nil {
+		logrus.Errorf("proxy manager not initialized for cluster %s", clusterName)
+		http.Error(rw, "Internal server error", http.StatusInternalServerError)
+		return
+	}
 
-		vars := url.Values{}
-		vars.Add("container", "probe-agent")
-		vars.Add("stdout", "1")
-		vars.Add("stdin", "1")
-		vars.Add("stderr", "1")
-		vars.Add("tty", "1")
-		vars.Add("command", "kubectl-shell.sh")
-		vars.Add("command", token)
+	proxyManager.ProxyRequest(rw, req, clusterName)
+}
 
-		path := fmt.Sprintf("/api/k8s/clusters/%s/api/v1/namespaces/%s/pods/%s/exec", clusterName, pod.Namespace, pod.Name)
+func decodeClusterToken(clusterName, encoded string) (string, error) {
+	if encoded == "" {
+		return "", fmt.Errorf("empty token for cluster %s", clusterName)
+	}
+	decoded, err := base64.StdEncoding.DecodeString(encoded)
+	if err != nil {
+		return "", err
+	}
+	return string(decoded), nil
+}
 
-		req.URL.Path = path
-		req.URL.RawQuery = vars.Encode()
+func findRunningProbeAgent(ctx context.Context, c client.Client, namespace string) (*v1.Pod, error) {
+	podList := &v1.PodList{}
+	err := c.List(ctx, podList,
+		client.InNamespace(namespace),
+		client.MatchingLabels{probeAgentLabelKey: probeAgentLabelValue})
+	if err != nil {
+		return nil, err
+	}
 
-		a.ServeHTTP(rw, req)
-		return
+	for i := range podList.Items {
+		pod := &podList.Items[i]
+		if pod.Status.Phase == v1.PodRunning {
+			return pod, nil
+		}
 	}
 
-	logrus.Errorf("failed to find a ready probe-agent pod for cluster %s", clusterName)
-	rw.WriteHeader(http.StatusInternalServerError)
-	rw.Write(apistructs.NewSteveError(apistructs.ServerError,
-		fmt.Sprintf("cluster %s does not have a ready probe-agent pod", clusterName)).JSON())
-	return
+	return nil, nil
+}
+
+func execQuery(token string) url.Values {
+	query := url.Values{}
+	query.Add("container", probeAgentContainer)
+	query.Add("stdout", "1")
+	query.Add("stdin", "1")
+	query.Add("stderr", "1")
+	query.Add("tty", "1")
+	query.Add("command", execCommandScript)
+	query.Add("command", token)
+	return query
+}
+
+func execURLPath(clusterName, namespace, podName string) string {
+	return fmt.Sprintf("/api/k8s/clusters/%s/api/v1/namespaces/%s/pods/%s/exec", clusterName, namespace, podName)
 }