[Server] support agent remote command

server/controller/http/router/agent_cmd.go

@@ -0,0 +1,189 @@
+/*
+ * Copyright (c) 2024 Yunshan Networks
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package router
+
+import (
+ "bytes"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "net/http"
+ "strconv"
+
+ "github.com/deepflowio/deepflow/message/trident"
+ httpcommon "github.com/deepflowio/deepflow/server/controller/http/common"
+ . "github.com/deepflowio/deepflow/server/controller/http/router/common"
+ "github.com/deepflowio/deepflow/server/controller/model"
+ "github.com/gin-gonic/gin"
+ "github.com/golang/protobuf/proto"
+)
+
+type AgentCMD struct{}
+
+func NewAgentCMD() *AgentCMD {
+ return new(AgentCMD)
+}
+
+func (c *AgentCMD) RegisterTo(e *gin.Engine) {
+ e.GET("/v1/agent/:id/cmd", getCMDAndNamespaceHandler)
+ e.GET("/v1/agent/:id/cmd/run", cmdRunHandler)
+}
+
+func getCMDAndNamespaceHandler(c *gin.Context) {
+ agentID, err := getAgentID(c)
+ if err != nil {
+ BadRequestResponse(c, httpcommon.INVALID_PARAMETERS, err.Error())
+ return
+ }
+ log.Infof("agent id: %d", agentID)
+
+ // TODO:
+ // 1. 请求获取采集器支持的命令
+ // 2. 请求获取 namespace
+ // 3. 构造返回值
+
+ resp := &model.RemoteExecResp{
+ RemoteCommand: []trident.RemoteCommand{
+ {
+ Id: proto.Uint32(1),
+ Cmd: proto.String("top"),
+ ParamNames: []string{"-b", "-c"},
+ CmdType: (*trident.CommandType)(trident.OutputFormat_TEXT.Enum()),
+ OutputFormat: trident.OutputFormat_TEXT.Enum(),
+ },
+ },
+ LinuxNamespace: []trident.LinuxNamespace{
+ {
+ Id: proto.Uint64(1),
+ NsType: proto.String("net"),
+ User: proto.String("Root"),
+ Pid: proto.Uint32(1),
+ Cmd: proto.String("/usr/lib/systemd/systemd --switched-root --system --deserialize 22"),
+ },
+ },
+ }
+ JsonResponse(c, resp, nil)
+}
+
+func getAgentID(c *gin.Context) (int, error) {
+ agentIDStr := c.Param("id")
+ if agentIDStr == "" {
+ return 0, errors.New("id can not be empty")
+ }
+ agentID, err := strconv.Atoi(agentIDStr)
+ if err != nil {
+ return 0, fmt.Errorf("agent id(%s) can not convert to int", agentIDStr)
+ }
+ return agentID, nil
+}
+
+func cmdRunHandler(c *gin.Context) {
+ agentID, err := getAgentID(c)
+ if err != nil {
+ BadRequestResponse(c, httpcommon.INVALID_PARAMETERS, err.Error())
+ return
+ }
+
+ req := model.RemoteExecReq{}
+ if err := c.ShouldBindJSON(&req); err != nil {
+ BadRequestResponse(c, httpcommon.INVALID_PARAMETERS, err.Error())
+ return
+ }
+ agentReq := trident.RemoteExecRequest{
+ ExecType: trident.ExecutionType_RUN_COMMAND.Enum(),
+ CommandId: req.CommandId,
+ Params: req.Params,
+ LinuxNsPid: req.LinuxNsPid,
+ }
+ log.Infof("agentReq %#v", agentReq)
+
+ log.Infof("req.CommandId %v", *req.CommandId)
+ log.Infof("agent id %v", agentID)
+ log.Infof("OutputFormat %v", req.OutputFormat)
+ b, _ := json.Marshal(req)
+ log.Infof("req: %s", string(b))
+ // log.Infof("OutputFilename %v", req.OutputFilename)
+
+ // 1. 获取采集器执行命令的返回
+ content := bytes.NewBufferString(topCMDResp)
+ if req.OutputFormat.String() == trident.OutputFormat_TEXT.String() {
+ sendAsText(c, content)
+ return
+ } else {
+ sendAsFile(c, req.OutputFilename, content)
+ }
+
+}
+
+var topCMDResp = `top - 15:43:46 up 3 days, 6:41, 25 users, load average: 3.16, 3.72, 3.48
+Tasks: 629 total, 2 running, 508 sleeping, 2 stopped, 0 zombie
+%Cpu(s): 6.6 us, 2.8 sy, 0.0 ni, 88.4 id, 0.0 wa, 0.6 hi, 1.6 si, 0.0 st
+KiB Mem : 65699328 total, 18650788 free, 28110492 used, 18938048 buff/cache
+KiB Swap: 0 total, 0 free, 0 used. 37152692 avail Mem
+
+ PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND
+ 2454 root 20 0 894364 201400 68632 S 50.0 0.3 158:47.45 kube-controller-manager --allocate-node-cidrs=true --authentication-kubeconfig=/etc/kubernetes/controll+
+10363 root 20 0 1978152 17124 7284 S 33.3 0.0 3108:56 ./golang-cp-exec
+16208 root 20 0 162688 5044 3856 R 16.7 0.0 0:00.04 top -c -n 1
+ 2461 root 20 0 2018796 1.1g 78112 S 11.1 1.7 647:09.17 kube-apiserver --advertise-address=10.1.4.1 --allow-privileged=true --anonymous-auth=True --apiserver-c+
+18132 root 20 0 2471492 191972 69684 S 11.1 0.3 8:54.26 /usr/local/bin/kubelet --logtostderr=true --v=2 --node-ip=10.1.4.1 --hostname-override=node1 --bootstra+
+30294 root 20 0 18.4g 593836 40068 S 11.1 0.9 498:09.03 java -jar /otel-simple-webflux.jar
+ 1537 mysql 20 0 2082924 374400 29560 S 5.6 0.6 6:57.92 /usr/local/mysql/mysql-8.0/bin/mysqld --basedir=/usr/local/mysql/mysql-8.0 --datadir=/usr/local/mysql/m+
+ 1619 root 10 -10 248632 67844 25184 R 5.6 0.1 228:25.00 /usr/local/aegis/aegis_client/aegis_11_91/AliYunDunMonitor
+ 3066 root 20 0 1428152 5232 4500 S 5.6 0.0 112:24.75 ./ticker-ok -ticker-leak-count 2
+ 9153 root 20 0 18.4g 608356 40376 S 5.6 0.9 503:12.90 java -jar /otel-simple-webflux.jar
+25771 root 20 0 1966744 75160 5872 S 5.6 0.1 97:41.70 /root/venv/bin/python3 -u /root/df-llm-agent/app.py
+ 1 root 20 0 52044 13424 5592 S 0.0 0.0 54:34.25 /usr/lib/systemd/systemd --switched-root --system --deserialize 22
+ 2 root 20 0 0 0 0 S 0.0 0.0 0:00.77 [kthreadd]
+ 3 root 0 -20 0 0 0 I 0.0 0.0 0:00.00 [rcu_gp]
+ 4 root 0 -20 0 0 0 I 0.0 0.0 0:00.00 [rcu_par_gp]
+ 6 root 0 -20 0 0 0 I 0.0 0.0 0:00.00 [kworker/0:0H-kb]
+ 8 root 0 -20 0 0 0 I 0.0 0.0 0:00.00 [mm_percpu_wq]
+ 9 root 20 0 0 0 0 S 0.0 0.0 1:21.57 [ksoftirqd/0]
+ 10 root 20 0 0 0 0 I 0.0 0.0 6:29.96 [rcu_sched]
+ 11 root 20 0 0 0 0 I 0.0 0.0 0:00.00 [rcu_bh]
+ 12 root rt 0 0 0 0 S 0.0 0.0 1:17.22 [migration/0]
+ 14 root 20 0 0 0 0 S 0.0 0.0 0:00.00 [cpuhp/0]
+ 15 root 20 0 0 0 0 S 0.0 0.0 0:00.00 [cpuhp/1]
+ 16 root rt 0 0 0 0 S 0.0 0.0 1:42.71 [migration/1]
+ 17 root 20 0 0 0 0 S 0.0 0.0 0:39.80 [ksoftirqd/1]
+ 19 root 0 -20 0 0 0 I 0.0 0.0 0:00.00 [kworker/1:0H-kb]
+ 20 root 20 0 0 0 0 S 0.0 0.0 0:00.00 [cpuhp/2]
+ 21 root rt 0 0 0 0 S 0.0 0.0 1:51.45 [migration/2]
+ 22 root 20 0 0 0 0 S 0.0 0.0 17:12.36 [ksoftirqd/2]
+ 24 root 0 -20 0 0 0 I 0.0 0.0 0:00.00 [kworker/2:0H-kb]
+ 25 root 20 0 0 0 0 S 0.0 0.0 0:00.00 [cpuhp/3]
+ 26 root rt 0 0 0 0 S 0.0 0.0 2:04.07 [migration/3]
+ 27 root 20 0 0 0 0 S 0.0 0.0 0:32.96 [ksoftirqd/3]`
+
+func sendAsFile(c *gin.Context, fileName string, content *bytes.Buffer) {
+ c.Writer.Header().Set("Content-Type", "application/octet-stream")
+ if fileName != "" {
+ c.Writer.Header().Set("Content-Disposition", fmt.Sprintf("attachment; filename*=utf-8''%s", fileName))
+ }
+
+ if _, err := io.Copy(c.Writer, content); err != nil {
+ c.AbortWithStatus(http.StatusInternalServerError)
+ log.Error(err)
+ return
+ }
+}
+
+func sendAsText(c *gin.Context, content *bytes.Buffer) {
+ JsonResponse(c, content.String(), nil)
+}

server/controller/http/router/common/controller.go

@@ -54,3 +54,7 @@ func ForwardMasterController(c *gin.Context, masterControllerName string, port i
 
  c.DataFromReader(resp.StatusCode, resp.ContentLength, resp.Header.Get("Content-Type"), resp.Body, make(map[string]string))
 }
+
+func ForwardAgentConectionController(c *gin.Context) {
+
+}

server/controller/http/server.go

@@ -115,6 +115,7 @@ func (s *Server) appendRegistrant() []registrant.Registrant {
  router.NewMail(),
  router.NewPrometheus(),
  router.NewDatabase(s.controllerConfig),
+ router.NewAgentCMD(),
  // icon
  router.NewIcon(s.controllerConfig),
 

server/controller/model/model.go

@@ -19,6 +19,7 @@ package model
 import (
  "time"
 
+ "github.com/deepflowio/deepflow/message/trident"
  "github.com/deepflowio/deepflow/server/agent_config"
 )
 
@@ -741,3 +742,16 @@ type MailServer struct {
  NtlmPassword string `json:"NTLM_PASSWORD"`
  Lcuuid string `json:"LCUUID"`
 }
+
+type RemoteExecReq struct {
+ trident.RemoteExecRequest
+
+ OutputFormat *trident.OutputFormat `json:"output_format" binding:"required,oneof=0 1"` // 0: "TEXT", 1: "BINARY"
+ OutputFilename string `json:"output_filename"`
+}
+
+type RemoteExecResp struct {
+ Content string `json:"content,omitempty"` // RUN_COMMAND
+ RemoteCommand []trident.RemoteCommand `json:"remote_commands,omitempty"` // LIST_COMMAND
+ LinuxNamespace []trident.LinuxNamespace `json:"linux_namespace,omitempty"` // LIST_NAMESPACE
+}