feat: rbd log and cloud shell (#1379)

* feat: rainbond api chaos worker gateway logs

* feat: cloud shell

* feat: cloud shell and rbd logs

* fix: cancel toleration and set grctl image

Co-authored-by: 曲源成 <quyc@goodrain.com>

api/api/api_interface.go

@@ -41,6 +41,11 @@ type ClusterInterface interface {
 	YamlResourceName(w http.ResponseWriter, r *http.Request)
 	YamlResourceDetailed(w http.ResponseWriter, r *http.Request)
 	YamlResourceImport(w http.ResponseWriter, r *http.Request)
+	CreateShellPod(w http.ResponseWriter, r *http.Request)
+	DeleteShellPod(w http.ResponseWriter, r *http.Request)
+	RbdLog(w http.ResponseWriter, r *http.Request)
+	GetRbdPods(w http.ResponseWriter, r *http.Request)
+	HistoryRbdLogs(w http.ResponseWriter, r *http.Request)
 }
 
 //TenantInterface interface

api/api_routers/version2/v2Routers.go

@@ -105,6 +105,12 @@ func (v2 *V2) clusterRouter() chi.Router {
 	r.Get("/yaml_resource_name", controller.GetManager().YamlResourceName)
 	r.Get("/yaml_resource_detailed", controller.GetManager().YamlResourceDetailed)
 	r.Post("/yaml_resource_import", controller.GetManager().YamlResourceImport)
+	r.Get("/rbd-resource/log", controller.GetManager().RbdLog)
+	r.Get("/rbd-resource/pods", controller.GetManager().GetRbdPods)
+	r.Get("/rbd-name/{serviceID}/logs", controller.GetManager().HistoryRbdLogs)
+	r.Get("/log-file", controller.GetManager().LogList)
+	r.Post("/shell-pod", controller.GetManager().CreateShellPod)
+	r.Delete("/shell-pod", controller.GetManager().DeleteShellPod)
 	return r
 }
 

api/controller/cluster.go

@@ -19,12 +19,13 @@
 package controller
 
 import (
-	"github.com/goodrain/rainbond/api/model"
-	"net/http"
+	"strconv"
 
 	"github.com/go-chi/chi"
 	"github.com/goodrain/rainbond/api/handler"
+	"github.com/goodrain/rainbond/api/model"
 	"github.com/sirupsen/logrus"
+	"net/http"
 
 	httputil "github.com/goodrain/rainbond/util/http"
 )
@@ -268,3 +269,54 @@ func (t *ClusterController) YamlResourceImport(w http.ResponseWriter, r *http.Re
 	}
 	httputil.ReturnSuccess(r, w, ac)
 }
+
+// CreateShellPod -
+func (t *ClusterController) CreateShellPod(w http.ResponseWriter, r *http.Request) {
+	var sp model.ShellPod
+	if ok := httputil.ValidatorRequestStructAndErrorResponse(r, w, &sp, nil); !ok {
+		return
+	}
+	pod, err := handler.GetClusterHandler().CreateShellPod(sp.RegionName)
+	if err !=nil{
+		logrus.Error("create shell pod error:", err)
+		return
+	}
+	httputil.ReturnSuccess(r, w, pod)
+}
+
+// DeleteShellPod -
+func (t *ClusterController) DeleteShellPod(w http.ResponseWriter, r *http.Request) {
+	var sp model.ShellPod
+	if ok := httputil.ValidatorRequestStructAndErrorResponse(r, w, &sp, nil); !ok {
+		return
+	}
+	err := handler.GetClusterHandler().DeleteShellPod(sp.PodName)
+	if err !=nil{
+		logrus.Error("delete shell pod error:", err)
+		return
+	}
+	httputil.ReturnSuccess(r, w, "")
+}
+
+// RbdLog -
+func (t *ClusterController) RbdLog(w http.ResponseWriter, r *http.Request) {
+	podName := r.URL.Query().Get("pod_name")
+	follow, _ := strconv.ParseBool(r.URL.Query().Get("follow"))
+	err := handler.GetClusterHandler().RbdLog(w, r, podName, follow)
+
+	if err != nil {
+		httputil.ReturnBcodeError(r, w, err)
+		return
+	}
+	return
+}
+
+// GetRbdPods -
+func (t *ClusterController) GetRbdPods(w http.ResponseWriter, r *http.Request) {
+	res, err := handler.GetClusterHandler().GetRbdPods()
+	if err != nil {
+		httputil.ReturnBcodeError(r, w, err)
+		return
+	}
+	httputil.ReturnSuccess(r, w, res)
+}

api/controller/event_log.go

@@ -57,6 +57,13 @@ func (e *EventLogStruct) HistoryLogs(w http.ResponseWriter, r *http.Request) {
 	e.EventlogServerProxy.Proxy(w, r)
 }
 
+//HistoryLogs get rbd history logs
+//proxy
+func (e *EventLogStruct) HistoryRbdLogs(w http.ResponseWriter, r *http.Request) {
+	r.URL.Path = strings.Replace(r.URL.Path, "/v2/cluster/", "/", 1)
+	e.EventlogServerProxy.Proxy(w, r)
+}
+
 //LogList GetLogList
 func (e *EventLogStruct) LogList(w http.ResponseWriter, r *http.Request) {
 	// swagger:operation GET  /v2/tenants/{tenant_name}/services/{service_alias}/log-file v2 logList
@@ -75,7 +82,13 @@ func (e *EventLogStruct) LogList(w http.ResponseWriter, r *http.Request) {
 	//     schema:
 	//       "$ref": "#/responses/commandResponse"
 	//     description: 统一返回格式
-	serviceID := r.Context().Value(ctxutil.ContextKey("service_id")).(string)
+	rbdName := r.URL.Query().Get("rbd_name")
+	var serviceID string
+	if rbdName != ""{
+		serviceID = rbdName
+	}else {
+		serviceID = r.Context().Value(ctxutil.ContextKey("service_id")).(string)
+	}
 	fileList, err := handler.GetEventHandler().GetLogList(GetServiceAliasID(serviceID))
 	if err != nil {
 		if os.IsNotExist(err) {

api/handler/cluster.go

@@ -5,18 +5,23 @@ import (
 	"fmt"
 	"github.com/goodrain/rainbond/api/model"
 	"github.com/goodrain/rainbond/api/util"
+	"github.com/goodrain/rainbond/api/util/bcode"
 	dbmodel "github.com/goodrain/rainbond/db/model"
 	"github.com/goodrain/rainbond/util/constants"
+	"github.com/pkg/errors"
 	"github.com/shirou/gopsutil/disk"
 	"github.com/sirupsen/logrus"
+	"io"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apiserver/pkg/util/flushwriter"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"net"
+	"net/http"
 	"os"
 	"runtime"
 	"strconv"
@@ -43,15 +48,20 @@ type ClusterHandler interface {
 	AppYamlResourceName(yamlResource model.YamlResource) (map[string]model.LabelResource, *util.APIHandleError)
 	AppYamlResourceDetailed(yamlResource model.YamlResource, yamlImport bool) (model.ApplicationResource, *util.APIHandleError)
 	AppYamlResourceImport(yamlResource model.YamlResource, components model.ApplicationResource) (model.AppComponent, *util.APIHandleError)
+	RbdLog(w http.ResponseWriter, r *http.Request, podName string, follow bool) error
+	GetRbdPods() (rbds []model.RbdResp, err error)
+	CreateShellPod(regionName string) (pod *corev1.Pod, err error)
+	DeleteShellPod(podName string) error
 }
 
 // NewClusterHandler -
-func NewClusterHandler(clientset *kubernetes.Clientset, RbdNamespace string, config *rest.Config, mapper meta.RESTMapper) ClusterHandler {
+func NewClusterHandler(clientset *kubernetes.Clientset, RbdNamespace, grctlImage string, config *rest.Config, mapper meta.RESTMapper) ClusterHandler {
 	return &clusterAction{
-		namespace: RbdNamespace,
-		clientset: clientset,
-		config:    config,
-		mapper:    mapper,
+		namespace:  RbdNamespace,
+		clientset:  clientset,
+		config:     config,
+		mapper:     mapper,
+		grctlImage: grctlImage,
 	}
 }
 
@@ -62,6 +72,7 @@ type clusterAction struct {
 	cacheTime        time.Time
 	config           *rest.Config
 	mapper           meta.RESTMapper
+	grctlImage       string
 }
 
 //GetClusterInfo -
@@ -393,3 +404,135 @@ func MergeMap(map1 map[string][]string, map2 map[string][]string) map[string][]s
 	}
 	return map2
 }
+
+// CreateShellPod -
+func (c *clusterAction) CreateShellPod(regionName string) (pod *corev1.Pod, err error) {
+	ctx := context.Background()
+	volumes := []corev1.Volume{
+		{
+			Name: "grctl-config",
+			VolumeSource: corev1.VolumeSource{
+				EmptyDir: &corev1.EmptyDirVolumeSource{},
+			},
+		},
+	}
+	volumeMounts := []corev1.VolumeMount{
+		{
+			Name:      "grctl-config",
+			MountPath: "/root/.rbd",
+		},
+	}
+	shellPod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			GenerateName: fmt.Sprintf("shell-%v-", regionName),
+			Namespace:    c.namespace,
+		},
+		Spec: corev1.PodSpec{
+			TerminationGracePeriodSeconds: new(int64),
+			RestartPolicy:                 corev1.RestartPolicyNever,
+			NodeSelector: map[string]string{
+				"kubernetes.io/os": "linux",
+			},
+			ServiceAccountName: "rainbond-operator",
+			Containers: []corev1.Container{
+				{
+					Name:            "shell",
+					TTY:             true,
+					Stdin:           true,
+					StdinOnce:       true,
+					Image:           c.grctlImage,
+					ImagePullPolicy: corev1.PullIfNotPresent,
+					VolumeMounts:    volumeMounts,
+				},
+			},
+			InitContainers: []corev1.Container{
+				{
+					Name:            "init-shell",
+					TTY:             true,
+					Stdin:           true,
+					StdinOnce:       true,
+					Image:           c.grctlImage,
+					ImagePullPolicy: corev1.PullIfNotPresent,
+					Command:         []string{"grctl", "install"},
+					VolumeMounts:    volumeMounts,
+				},
+			},
+			Volumes: volumes,
+		},
+	}
+	pod, err = c.clientset.CoreV1().Pods("rbd-system").Create(ctx, shellPod, metav1.CreateOptions{})
+	if err != nil {
+		logrus.Error("create shell pod error:", err)
+		return nil, err
+	}
+	return pod, nil
+}
+
+// DeleteShellPod -
+func (c *clusterAction) DeleteShellPod(podName string) (err error) {
+	err = c.clientset.CoreV1().Pods("rbd-system").Delete(context.Background(), podName, metav1.DeleteOptions{})
+	if err != nil {
+		logrus.Error("delete shell pod error:", err)
+		return err
+	}
+	return nil
+}
+
+// RbdLog returns the logs reader for a container in a pod, a pod or a component.
+func (c *clusterAction) RbdLog(w http.ResponseWriter, r *http.Request, podName string, follow bool) error {
+	if podName == "" {
+		// Only support return the logs reader for a container now.
+		return errors.WithStack(bcode.NewBadRequest("the field 'podName' and 'containerName' is required"))
+	}
+	request := c.clientset.CoreV1().Pods("rbd-system").GetLogs(podName, &corev1.PodLogOptions{
+		Follow: follow,
+	})
+	out, err := request.Stream(context.TODO())
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			return errors.Wrap(bcode.ErrPodNotFound, "get pod log")
+		}
+		return errors.Wrap(err, "get stream from request")
+	}
+	defer out.Close()
+
+	w.Header().Set("Transfer-Encoding", "chunked")
+	w.WriteHeader(http.StatusOK)
+
+	// Flush headers, if possible
+	if flusher, ok := w.(http.Flusher); ok {
+		flusher.Flush()
+	}
+
+	writer := flushwriter.Wrap(w)
+
+	_, err = io.Copy(writer, out)
+	if err != nil {
+		if strings.HasSuffix(err.Error(), "write: broken pipe") {
+			return nil
+		}
+		logrus.Warningf("write stream to response: %v", err)
+	}
+	return nil
+}
+
+// GetRbdPods -
+func (c *clusterAction) GetRbdPods() (rbds []model.RbdResp, err error) {
+	pods, err := c.clientset.CoreV1().Pods("rbd-system").List(context.Background(), metav1.ListOptions{})
+	if err != nil {
+		logrus.Error("get rbd pod list error:", err)
+		return nil, err
+	}
+	var rbd model.RbdResp
+	for _, pod := range pods.Items {
+		if strings.Contains(pod.Name, "rbd-chaos") || strings.Contains(pod.Name, "rbd-api") || strings.Contains(pod.Name, "rbd-worker") || strings.Contains(pod.Name, "rbd-gateway") {
+			rbdSplit := strings.Split(pod.Name, "-")
+			rbdName := fmt.Sprintf("%s-%s", rbdSplit[0], rbdSplit[1])
+			rbd.RbdName = rbdName
+			rbd.PodName = pod.Name
+			rbd.NodeName = pod.Spec.NodeName
+			rbds = append(rbds, rbd)
+		}
+	}
+	return rbds, nil
+}

api/handler/eventLog.go

@@ -101,6 +101,7 @@ func (l *LogAction) GetLogInstance(serviceID string) (string, error) {
 	if err != nil {
 		return "", err
 	}
+
 	if len(value.Kvs) > 0 {
 		return string(value.Kvs[0].Value), nil
 	}

api/handler/handler.go

@@ -84,7 +84,7 @@ func InitHandle(conf option.Config,
 	batchOperationHandler = CreateBatchOperationHandler(mqClient, statusCli, operationHandler)
 	defaultAppRestoreHandler = NewAppRestoreHandler()
 	defPodHandler = NewPodHandler(statusCli)
-	defClusterHandler = NewClusterHandler(kubeClient, conf.RbdNamespace, config, mapper)
+	defClusterHandler = NewClusterHandler(kubeClient, conf.RbdNamespace, conf.GrctlImage, config, mapper)
 	defaultVolumeTypeHandler = CreateVolumeTypeManger(statusCli)
 	defaultEtcdHandler = NewEtcdHandler(etcdcli)
 	defaultmonitorHandler = NewMonitorHandler(prometheusCli)

api/model/model.go

@@ -2064,3 +2064,16 @@ type SyncAppConfigGroup struct {
 type AppStatusesReq struct {
 	AppIDs []string `json:"app_ids"`
 }
+
+// RbdResp -
+type RbdResp struct {
+	RbdName  string `json:"rbd_name"`
+	NodeName string `json:"node_name"`
+	PodName  string `json:"pod_name"`
+}
+
+// ShellPod -
+type ShellPod struct {
+	RegionName string `json:"region_name"`
+	PodName    string `json:"pod_name"`
+}

cmd/api/option/option.go

@@ -62,6 +62,7 @@ type Config struct {
 	PrometheusEndpoint     string
 	RbdNamespace           string
 	ShowSQL                bool
+	GrctlImage             string
 }
 
 //APIServer  apiserver server
@@ -115,6 +116,7 @@ func (a *APIServer) AddFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&a.PrometheusEndpoint, "prom-api", "rbd-monitor:9999", "The service DNS name of Prometheus api. Default to rbd-monitor:9999")
 	fs.StringVar(&a.RbdNamespace, "rbd-namespace", "rbd-system", "rbd component namespace")
 	fs.BoolVar(&a.ShowSQL, "show-sql", false, "The trigger for showing sql.")
+	fs.StringVar(&a.GrctlImage, "grctl-image", "registry.cn-hangzhou.aliyuncs.com/goodrain/rbd-grctl:v5.10.0-release", "use grctl image")
 }
 
 //SetLog 设置log

eventlog/exit/web/manager.go

@@ -488,6 +488,7 @@ func (s *SocketServer) listen() {
 	// new websocket pubsub
 	r.Get("/services/{serviceID}/pubsub", s.pubsub)
 	r.Get("/tenants/{tenantName}/services/{serviceID}/logs", s.getDockerLogs)
+	r.Get("/rbd-name/{serviceID}/logs", s.getDockerLogs)
 	//monitor setting
 	s.prometheus(r)
 	//pprof debug

eventlog/store/manager.go

@@ -468,6 +468,11 @@ loop:
 			}
 			containerID := m[0:12]        //0-12
 			serviceID := string(m[13:45]) //13-45
+			// rbd logs
+			if strings.Contains(serviceID, "rbd-"){
+				nameSlice := strings.Split(serviceID,"time")
+				serviceID = nameSlice[0]
+			}
 			log := m[45:]
 			logrus.Debugf("containerID [%s] serviceID [%s] log [%s]", containerID, serviceID, string(log))
 			buffer := bytes.NewBuffer(containerID)

grctl/clients/kubernetes.go

@@ -20,13 +20,14 @@ package clients
 
 import (
 	"fmt"
+	k8sutil "github.com/goodrain/rainbond/util/k8s"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	"k8s.io/client-go/rest"
 	"os"
 	"path"
 
 	rainbondv1alpha1 "github.com/goodrain/rainbond-operator/api/v1alpha1"
 	"github.com/goodrain/rainbond/builder/sources"
-	k8sutil "github.com/goodrain/rainbond/util/k8s"
 	"github.com/sirupsen/logrus"
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
@@ -58,15 +59,20 @@ func InitClient(kubeconfig string) error {
 		homePath, _ := sources.Home()
 		kubeconfig = path.Join(homePath, ".kube/config")
 	}
+	var config *rest.Config
 	_, err := os.Stat(kubeconfig)
 	if err != nil {
-		fmt.Printf("Please make sure the kube-config file(%s) exists\n", kubeconfig)
-		os.Exit(1)
-	}
-	// use the current context in kubeconfig
-	config, err := k8sutil.NewRestConfig(kubeconfig)
-	if err != nil {
-		return err
+		fmt.Printf("Not find kube-config file(%s)\n", kubeconfig)
+		if config, err = rest.InClusterConfig(); err != nil{
+			logrus.Error("get cluster config error:", err)
+			return err
+		}
+	} else {
+		// use the current context in kubeconfig
+		config, err = k8sutil.NewRestConfig(kubeconfig)
+		if err != nil {
+			return err
+		}
 	}
 	config.QPS = 50
 	config.Burst = 100