[Server] support agent modify daemonset

cli/ctl/agent.go

@@ -434,21 +434,15 @@ func upgadeAgent(cmd *cobra.Command, args []string) {
 	var (
 		vtapController string
 		vtapLcuuid     string
-		vtapType       int
 	)
 
 	if len(response.Get("DATA").MustArray()) > 0 {
 		vtapLcuuid = response.Get("DATA").GetIndex(0).Get("LCUUID").MustString()
 		vtapController = response.Get("DATA").GetIndex(0).Get("CONTROLLER_IP").MustString()
-		vtapType = response.Get("DATA").GetIndex(0).Get("TYPE").MustInt()
 	} else {
 		fmt.Printf("get agent(%s) info failed, url: %s\n", vtapName, vtapURL)
 		return
 	}
-	if vtapType == int(common.VTAP_TYPE_POD_VM) || vtapType == int(common.VTAP_TYPE_POD_HOST) || vtapType == int(common.VTAP_TYPE_K8S_SIDECAR) {
-		fmt.Printf("agent (%s) type is %v, not supported upgrade by cli\n", vtapName, common.VtapType(vtapType))
-		return
-	}
 	if vtapController == "" || vtapLcuuid == "" {
 		fmt.Printf("get agent(%s) info failed, url: %s\n", vtapName, vtapURL)
 		return

cli/ctl/agent_repo.go

@@ -54,6 +54,10 @@ func RegisterRepoCommand() *cobra.Command {
 	return repo
 }
 
+var repoAgentCreateExample = `deepflow-ctl repo agent create --arch x86 --image deepflow-agent
+deepflow-ctl repo agent create --arch x86 --version-image /root/deepflow-agent --image deepflow-agent.exe 
+deepflow-ctl repo agent create --arch x86 --version-image /root/deepflow-agent --k8s-image registry.cn-beijing.aliyuncs.com/deepflow-ce/deepflowio-agent:latest`
+
 func registerAgentCommand() *cobra.Command {
 	agent := &cobra.Command{
 		Use:   "agent",
@@ -63,38 +67,42 @@ func registerAgentCommand() *cobra.Command {
 		},
 	}
 
-	var arch, image, versionImage string
+	var arch, image, versionImage, k8sImage string
 	timeout := common.DefaultTimeout
 	create := &cobra.Command{
 		Use:     "create",
 		Short:   "create repo agent",
-		Example: "deepflow-ctl repo agent create --arch x86 --image deepflow-agent",
+		Example: repoAgentCreateExample,
 		Run: func(cmd *cobra.Command, args []string) {
-			if _, err := os.Stat(image); errors.Is(err, os.ErrNotExist) {
-				fmt.Printf("file %s not found\n", image)
+			if len(k8sImage) != 0 && len(image) != 0 {
+				printutil.ErrorfWithColor("only one flag is supported: 'image' or 'k8s-image'", image, versionImage)
 				return
 			}
-			if strings.HasSuffix(image, ".exe") {
+			if strings.HasSuffix(image, ".exe") || len(k8sImage) != 0 {
 				if versionImage == "" {
-					printutil.ErrorWithColor("version-image must be set when uploading a window image")
+					printutil.ErrorWithColor("'version-image' must be specified when uploading an image to retrieve its version")
 					return
 				}
 				if _, err := os.Stat(versionImage); errors.Is(err, os.ErrNotExist) {
 					fmt.Printf("file %s not found\n", versionImage)
 					return
 				}
-				printutil.WarnfWithColor("make sure %s and %s have the same version", image, versionImage)
+				imageName := image
+				if len(imageName) == 0 {
+					imageName = k8sImage
+				}
+				printutil.WarnfWithColor("make sure %s and %s have the same version", imageName, versionImage)
 			}
-			if err := createRepoAgent(cmd, arch, image, versionImage); err != nil {
+			if err := createRepoAgent(cmd, arch, image, versionImage, k8sImage); err != nil {
 				fmt.Println(err)
 			}
 		},
 	}
 	create.Flags().StringVarP(&arch, "arch", "", "", "arch of deepflow-agent")
 	create.Flags().StringVarP(&image, "image", "", "", "deepflow-agent image to upload")
 	create.Flags().StringVarP(&versionImage, "version-image", "", "", "deepflow-agent image to get branch, rev_count and commit_id")
+	create.Flags().StringVarP(&k8sImage, "k8s-image", "", "", "deepflow-agent Kubernetes image: if k8s-image is not empty, the image flag will be ignored.")
 	create.Flags().DurationVar(&timeout, "timeout", 0, "timeout duration(default: 30s), e.g., 1s 1m 1h")
-	create.MarkFlagsRequiredTogether("arch", "image")
 
 	list := &cobra.Command{
 		Use:     "list",
@@ -122,9 +130,9 @@ func registerAgentCommand() *cobra.Command {
 	return agent
 }
 
-func createRepoAgent(cmd *cobra.Command, arch, image, versionImage string) error {
+func createRepoAgent(cmd *cobra.Command, arch, image, versionImage, k8sImage string) error {
 	execImage := image
-	if versionImage != "" {
+	if versionImage != "" || len(k8sImage) != 0 {
 		execImage = versionImage
 	}
 	agentOutput, err := getAgentOutput(execImage)
@@ -135,27 +143,34 @@ func createRepoAgent(cmd *cobra.Command, arch, image, versionImage string) error
 
 	bodyBuf := &bytes.Buffer{}
 	bodyWriter := multipart.NewWriter(bodyBuf)
-	bodyWriter.WriteField("NAME", path.Base(image))
+	if len(k8sImage) != 0 {
+		bodyWriter.WriteField("NAME", k8sImage)
+	} else {
+		bodyWriter.WriteField("NAME", path.Base(image))
+	}
 	bodyWriter.WriteField("ARCH", arch)
 	bodyWriter.WriteField("BRANCH", branch)
 	bodyWriter.WriteField("REV_COUNT", revCount)
 	bodyWriter.WriteField("COMMIT_ID", commitID)
+	bodyWriter.WriteField("K8S_IMAGE", k8sImage)
 	osStr := "Linux"
 	if strings.HasSuffix(image, ".exe") {
 		osStr = "Windows"
 	}
 	bodyWriter.WriteField("OS", osStr)
 
-	fileWriter, err := bodyWriter.CreateFormFile("IMAGE", path.Base(image))
-	f, err := os.Open(image)
-	if err != nil {
-		return err
-	}
-	defer f.Close()
-	if _, err = io.Copy(fileWriter, f); err != nil {
-		return err
-	}
 	contentType := bodyWriter.FormDataContentType()
+	if len(image) > 0 {
+		fileWriter, err := bodyWriter.CreateFormFile("IMAGE", path.Base(image))
+		f, err := os.Open(image)
+		if err != nil {
+			return err
+		}
+		defer f.Close()
+		if _, err = io.Copy(fileWriter, f); err != nil {
+			return err
+		}
+	}
 	bodyWriter.Close()
 
 	server := common.GetServerInfo(cmd)
@@ -216,10 +231,11 @@ func listRepoAgent(cmd *cobra.Command) {
 		branchMaxSize   = jsonparser.GetTheMaxSizeOfAttr(data, "BRANCH")
 		revCountMaxSize = jsonparser.GetTheMaxSizeOfAttr(data, "REV_COUNT")
 		commitIDMaxSize = jsonparser.GetTheMaxSizeOfAttr(data, "COMMIT_ID")
+		k8sImageMaxSize = jsonparser.GetTheMaxSizeOfAttr(data, "K8S_IMAGE")
 	)
-	cmdFormat := "%-*s %-*s %-*s %-*s %-*s %-19s %-*s\n"
+	cmdFormat := "%-*s %-*s %-*s %-*s %-*s %-19s %-*s %-*s\n"
 	fmt.Printf(cmdFormat, nameMaxSize, "NAME", archMaxSize, "ARCH", osMaxSize, "OS", branchMaxSize, "BRANCH",
-		revCountMaxSize, "REV_COUNT", "UPDATED_AT", commitIDMaxSize, "COMMIT_ID")
+		revCountMaxSize, "REV_COUNT", "UPDATED_AT", commitIDMaxSize, "COMMIT_ID", k8sImageMaxSize, "K8S_IMAGE")
 	for i := range data.MustArray() {
 		d := data.GetIndex(i)
 		fmt.Printf(cmdFormat,
@@ -230,6 +246,7 @@ func listRepoAgent(cmd *cobra.Command) {
 			revCountMaxSize, d.Get("REV_COUNT").MustString(),
 			d.Get("UPDATED_AT").MustString(),
 			commitIDMaxSize, d.Get("COMMIT_ID").MustString(),
+			k8sImageMaxSize, d.Get("K8S_IMAGE").MustString(),
 		)
 	}
 }
@@ -241,9 +258,13 @@ func deleteRepoAgent(cmd *cobra.Command, args []string) error {
 		return fmt.Errorf("must specify one name.\nExample: %s", cmd.Example)
 	}
 
+	body := map[string]interface{}{
+		"image_name": args[0],
+	}
+
 	server := common.GetServerInfo(cmd)
-	url := fmt.Sprintf("http://%s:%d/v1/vtap-repo/%s/", server.IP, server.Port, args[0])
-	_, err := common.CURLPerform("DELETE", url, nil, "", []common.HTTPOption{common.WithTimeout(common.GetTimeout(cmd))}...)
+	url := fmt.Sprintf("http://%s:%d/v1/vtap-repo/", server.IP, server.Port)
+	_, err := common.CURLPerform("DELETE", url, body, "", []common.HTTPOption{common.WithTimeout(common.GetTimeout(cmd))}...)
 	if err != nil {
 		return err
 	}

cli/go.mod

@@ -4,7 +4,7 @@ go 1.18
 
 require (
 	github.com/bitly/go-simplejson v0.5.0
-	github.com/deepflowio/deepflow/message v0.0.0-20240423024840-ece29545d0ac
+	github.com/deepflowio/deepflow/message v0.0.0-20240507083143-eaca2bed10f2
 	github.com/deepflowio/deepflow/server v0.0.0-20240423024840-ece29545d0ac
 	github.com/golang/protobuf v1.5.4
 	github.com/mattn/go-runewidth v0.0.14

cli/go.sum

@@ -75,8 +75,8 @@ github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ3
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/deepflowio/deepflow/message v0.0.0-20240423024840-ece29545d0ac h1:3qfsNktuiO8lyC5J5RF3Gf0sFz57c9a6bhCXk1/4tTE=
-github.com/deepflowio/deepflow/message v0.0.0-20240423024840-ece29545d0ac/go.mod h1:e+1lUMMlycCvFRKvlwt/y/0vxJnF8wVss3GyR1ARXY0=
+github.com/deepflowio/deepflow/message v0.0.0-20240507083143-eaca2bed10f2 h1:RaWabWp/uS3/uDCwsK7arzJ7/m5x4LeXC6EXpnl6ylI=
+github.com/deepflowio/deepflow/message v0.0.0-20240507083143-eaca2bed10f2/go.mod h1:e+1lUMMlycCvFRKvlwt/y/0vxJnF8wVss3GyR1ARXY0=
 github.com/deepflowio/deepflow/server v0.0.0-20240423024840-ece29545d0ac h1:5gysrosvC5kwK0gwH9R/1dbgeyJRKcQj371eB1tXqCk=
 github.com/deepflowio/deepflow/server v0.0.0-20240423024840-ece29545d0ac/go.mod h1:EMICsEChD3sF/62DhAsGJ/uDUEJDqEMcZjtanDH+C2o=
 github.com/docker/go-units v0.4.0 h1:3uh0PgVws3nIA0Q+MwDC8yjEPf9zjRfZZWXZYDct3Tw=

server/agent_config/config.go

@@ -27,6 +27,7 @@ type AgentGroupConfig struct {
 	MaxCollectPps                     *int          `json:"MAX_COLLECT_PPS" yaml:"max_collect_pps,omitempty"`
 	MaxNpbBps                         *int64        `json:"MAX_NPB_BPS" yaml:"max_npb_bps,omitempty"` // unit: bps
 	MaxCPUs                           *int          `json:"MAX_CPUS" yaml:"max_cpus,omitempty"`
+	MaxMilliCPUs                      *int          `json:"MAX_MILLICPUS" yaml:"max_millicpus,omitempty"`
 	MaxMemory                         *int          `json:"MAX_MEMORY" yaml:"max_memory,omitempty"` // unit: M
 	SyncInterval                      *int          `json:"SYNC_INTERVAL" yaml:"sync_interval,omitempty"`
 	PlatformSyncInterval              *int          `json:"PLATFORM_SYNC_INTERVAL" yaml:"platform_sync_interval,omitempty"`

server/agent_config/db.go

@@ -5,6 +5,7 @@ type AgentGroupConfigModel struct {
 	MaxCollectPps                     *int     `gorm:"column:max_collect_pps;type:int;default:null" json:"MAX_COLLECT_PPS"`
 	MaxNpbBps                         *int64   `gorm:"column:max_npb_bps;type:bigint;default:null" json:"MAX_NPB_BPS"` // unit: bps
 	MaxCPUs                           *int     `gorm:"column:max_cpus;type:int;default:null" json:"MAX_CPUS"`
+	MaxMilliCPUs                      *int     `gorm:"column:max_millicpus;type:int;default:null" json:"MAX_MILLICPUS"`
 	MaxMemory                         *int     `gorm:"column:max_memory;type:int;default:null" json:"MAX_MEMORY"` // unit: M
 	PlatformSyncInterval              *int     `gorm:"column:platform_sync_interval;type:int;default:null" json:"PLATFORM_SYNC_INTERVAL"`
 	SyncInterval                      *int     `gorm:"column:sync_interval;type:int;default:null" json:"SYNC_INTERVAL"`
@@ -86,6 +87,7 @@ type RAgentGroupConfigModel struct {
 	MaxCollectPps                     int     `gorm:"column:max_collect_pps;type:int;default:null" json:"MAX_COLLECT_PPS"`
 	MaxNpbBps                         int64   `gorm:"column:max_npb_bps;type:bigint;default:null" json:"MAX_NPB_BPS"` // unit: bps
 	MaxCPUs                           int     `gorm:"column:max_cpus;type:int;default:null" json:"MAX_CPUS"`
+	MaxMilliCPUs                      int     `gorm:"column:max_millicpus;type:int;default:null" json:"MAX_MILLICPUS"`
 	MaxMemory                         int     `gorm:"column:max_memory;type:int;default:null" json:"MAX_MEMORY"` // unit: M
 	PlatformSyncInterval              int     `gorm:"column:platform_sync_interval;type:int;default:null" json:"PLATFORM_SYNC_INTERVAL"`
 	SyncInterval                      int     `gorm:"column:sync_interval;type:int;default:null" json:"SYNC_INTERVAL"`

server/agent_config/json.go

@@ -20,6 +20,7 @@ type AgentGroupConfigResponse struct {
 	MaxCollectPps                     *int           `json:"MAX_COLLECT_PPS"`
 	MaxNpbBps                         *int64         `json:"MAX_NPB_BPS"` // unit: bps
 	MaxCPUs                           *int           `json:"MAX_CPUS"`
+	MaxMilliCPUs                      *int           `json:"MAX_MILLICPUS"`
 	MaxMemory                         *int           `json:"MAX_MEMORY"` // unit: M
 	SyncInterval                      *int           `json:"SYNC_INTERVAL"`
 	PlatformSyncInterval              *int           `json:"PLATFORM_SYNC_INTERVAL"`

server/controller/common/vtap_config.go

@@ -24,6 +24,7 @@ var DefaultVTapGroupConfig = &agent_config.AgentGroupConfigModel{
 	MaxCollectPps:                 &DefaultMaxCollectPps,
 	MaxNpbBps:                     &DefaultMaxNpbBps,
 	MaxCPUs:                       &DefaultMaxCPUs,
+	MaxMilliCPUs:                  &DefaultMaxMilliCPUs,
 	MaxMemory:                     &DefaultMaxMemory,
 	PlatformSyncInterval:          &DefaultPlatformSyncInterval,
 	SyncInterval:                  &DefaultSyncInterval,
@@ -95,6 +96,7 @@ var (
 	DefaultMaxCollectPps                 = 200000
 	DefaultMaxNpbBps                     = int64(1000000000)
 	DefaultMaxCPUs                       = 1
+	DefaultMaxMilliCPUs                  = 1000
 	DefaultMaxMemory                     = 768
 	DefaultPlatformSyncInterval          = 10
 	DefaultSyncInterval                  = 60

server/controller/db/mysql/migration/rawsql/init.sql

@@ -18,13 +18,14 @@ TRUNCATE TABLE plugin;
 
 CREATE TABLE IF NOT EXISTS vtap_repo (
     id                  INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
-    name                CHAR(64),
+    name                VCHAR(512),
     arch                VARCHAR(256) DEFAULT '',
     os                  VARCHAR(256) DEFAULT '',
     branch              VARCHAR(256) DEFAULT '',
     rev_count           VARCHAR(256) DEFAULT '',
     commit_id           VARCHAR(256) DEFAULT '',
-    image               LONGBLOB NOT NULL,
+    image               LONGBLOB,
+    k8s_image           VARCHAR(512) DEFAULT '',
     created_at          DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
     updated_at          DATETIME NOT NULL ON UPDATE CURRENT_TIMESTAMP DEFAULT CURRENT_TIMESTAMP
 )ENGINE=innodb AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='store deepflow-agent for easy upgrade';
@@ -1045,6 +1046,7 @@ CREATE TABLE IF NOT EXISTS vtap (
     os                      VARCHAR(256),
     kernel_version          VARCHAR(256),
     process_name            VARCHAR(256),
+    current_k8s_image       VARCHAR(512),
     license_type            INTEGER COMMENT '1: A类 2: B类 3: C类',
     license_functions       CHAR(64) COMMENT 'separated by ,; 1: 流量分发 2: 网络监控 3: 应用监控',
     tap_mode                INTEGER,
@@ -1623,6 +1625,7 @@ CREATE TABLE IF NOT EXISTS vtap_group_configuration(
     max_collect_pps                         INTEGER        DEFAULT NULL,
     max_npb_bps                             BIGINT         DEFAULT NULL     COMMENT 'unit: bps',
     max_cpus                                INTEGER        DEFAULT NULL,
+    max_millicpus                           INTEGER        DEFAULT NULL,
     max_memory                              INTEGER        DEFAULT NULL     COMMENT 'unit: M',
     platform_sync_interval                  INTEGER        DEFAULT NULL,
     sync_interval                           INTEGER        DEFAULT NULL,

server/controller/db/mysql/migration/rawsql/issu/6.5.1.28.sql

@@ -0,0 +1,41 @@
+-- modify start, add upgrade sql
+DROP PROCEDURE IF EXISTS AddColumnIfNotExists;
+
+CREATE PROCEDURE AddColumnIfNotExists(
+    IN tableName VARCHAR(255),
+    IN colName VARCHAR(255),
+    IN colType VARCHAR(255),
+    IN afterCol VARCHAR(255)
+)
+BEGIN
+    DECLARE column_count INT;
+
+    -- 检查列是否存在
+    SELECT COUNT(*)
+    INTO column_count
+    FROM information_schema.columns
+    WHERE TABLE_SCHEMA = DATABASE()
+    AND TABLE_NAME = tableName
+    AND column_name = colName;
+
+    -- 如果列不存在，则添加列
+    IF column_count = 0 THEN
+        SET @sql = CONCAT('ALTER TABLE ', tableName, ' ADD COLUMN ', colName, ' ', colType, ' AFTER ', afterCol);
+        PREPARE stmt FROM @sql;
+        EXECUTE stmt;
+        DEALLOCATE PREPARE stmt;
+    END IF;
+END;
+
+CALL AddColumnIfNotExists('vtap', 'current_k8s_image', 'VARCHAR(512)', 'process_name');
+CALL AddColumnIfNotExists('vtap_repo', 'k8s_image', 'VARCHAR(512)', 'image');
+CALL AddColumnIfNotExists('vtap_group_configuration', 'max_millicpus', 'VARCHAR(512)', 'max_cpus');
+
+DROP PROCEDURE AddColumnIfNotExists;
+
+ALTER TABLE vtap_repo MODIFY COLUMN image LONGBLOB;
+ALTER TABLE vtap_repo MODIFY COLUMN name VARCHAR(512);
+
+-- update db_version to latest, remeber update DB_VERSION_EXPECT in migrate/init.go
+UPDATE db_version SET version='6.5.1.28';
+-- modify end

server/controller/db/mysql/migration/version.go

@@ -18,5 +18,5 @@ package migration
 
 const (
 	DB_VERSION_TABLE    = "db_version"
-	DB_VERSION_EXPECTED = "6.5.1.27"
+	DB_VERSION_EXPECTED = "6.5.1.28"
 )

server/controller/db/mysql/model.go

@@ -191,6 +191,7 @@ type VTap struct {
 	Os                 string    `gorm:"column:os;type:varchar(256);default:null" json:"OS"`
 	KernelVersion      string    `gorm:"column:kernel_version;type:varchar(256);default:null" json:"KERNEL_VERSION"`
 	ProcessName        string    `gorm:"column:process_name;type:varchar(256);default:null" json:"PROCESS_NAME"`
+	CurrentK8sImage    string    `gorm:"column:current_k8s_image;type:varchar(512);default:null" json:"CURRENT_K8S_IMAGE"`
 	LicenseType        int       `gorm:"column:license_type;type:int;default:null" json:"LICENSE_TYPE"`   // 1: A类 2: B类 3: C类
 	LicenseFunctions   string    `gorm:"column:license_functions;type:char(64)" json:"LICENSE_FUNCTIONS"` // separated by ,; 1: 流量分发 2: 网络监控 3: 应用监控
 	TapMode            int       `gorm:"column:tap_mode;type:int;default:null" json:"TAP_MODE"`
@@ -396,7 +397,8 @@ type VTapRepo struct {
 	Branch    string          `gorm:"column:branch;type:varchar(256);default:''" json:"BRANCH"`
 	RevCount  string          `gorm:"column:rev_count;type:varchar(256);default:''" json:"REV_COUNT"`
 	CommitID  string          `gorm:"column:commit_id;type:varchar(256);default:''" json:"COMMIT_ID"`
-	Image     compressedBytes `gorm:"column:image;type:logblob;not null" json:"IMAGE"`
+	Image     compressedBytes `gorm:"column:image;type:logblob" json:"IMAGE"`
+	K8sImage  string          `gorm:"column:k8s_image;type:varchar(512);default:''" json:"K8S_IMAGE"`
 	CreatedAt time.Time       `gorm:"column:created_at;type:timestamp;not null;default:CURRENT_TIMESTAMP" json:"CREATED_AT"`
 	UpdatedAt time.Time       `gorm:"column:updated_at;type:timestamp;not null;default:CURRENT_TIMESTAMP" json:"UPDATED_AT"`
 }