feat: data plane transmission supports encryption

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-20240718034203-09d6b7d4e926
+	github.com/deepflowio/deepflow/message v0.0.0-20240819073639-b32b909abd7d
 	github.com/deepflowio/deepflow/server v0.0.0-20240722004820-198136782148
 	github.com/golang/protobuf v1.5.4
 	github.com/mattn/go-runewidth v0.0.14

cli/go.sum

@@ -74,8 +74,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-20240718034203-09d6b7d4e926 h1:MPz1PPVG5GVEIdIYxdxeOqiI+kmr2zGpTetVao4LuJo=
-github.com/deepflowio/deepflow/message v0.0.0-20240718034203-09d6b7d4e926/go.mod h1:e+1lUMMlycCvFRKvlwt/y/0vxJnF8wVss3GyR1ARXY0=
+github.com/deepflowio/deepflow/message v0.0.0-20240819073639-b32b909abd7d h1:QyiXuNj9wvLSGeidpY8oGhbrXq4HTsjGwVjUHOKLM1o=
+github.com/deepflowio/deepflow/message v0.0.0-20240819073639-b32b909abd7d/go.mod h1:e+1lUMMlycCvFRKvlwt/y/0vxJnF8wVss3GyR1ARXY0=
 github.com/deepflowio/deepflow/server v0.0.0-20240722004820-198136782148 h1:wbH8R0w9d7OdF+DaxstoHSmFbT2OhLIsI2GDF+1au2E=
 github.com/deepflowio/deepflow/server v0.0.0-20240722004820-198136782148/go.mod h1:sVACr39AxYb+iivqCDCtDN8LxA0yDBVvnHefgtBYV4Y=
 github.com/docker/go-units v0.4.0 h1:3uh0PgVws3nIA0Q+MwDC8yjEPf9zjRfZZWXZYDct3Tw=

server/common/module_shared.go

@@ -124,3 +124,20 @@ func DropOrg(orgId uint16) error {
 	}
 	return nil
 }
+
+var agentKeyOP AgentKeyOP
+
+type AgentKeyOP interface {
+	GetAgentKey(orgID int, AgentID int) ([]byte, error)
+}
+
+func SetAgentKeyOP(akOP AgentKeyOP) {
+	agentKeyOP = akOP
+}
+
+func GetAgentKey(orgID int, AgentID int) ([]byte, error) {
+	if agentKeyOP == nil {
+		return nil, fmt.Errorf("agentKeyOP is nil, get agent(orgid=%d, id=%d) failed", orgID, AgentID)
+	}
+	return agentKeyOP.GetAgentKey(orgID, AgentID)
+}

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

@@ -2339,6 +2339,15 @@ CREATE TABLE IF NOT EXISTS license (
 ) ENGINE=innodb DEFAULT CHARSET=utf8 AUTO_INCREMENT=1;
 TRUNCATE TABLE license;
 
+CREATE TABLE IF NOT EXISTS agent_key (
+    id                  INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
+    agent_id            INTEGER,
+    status              INTEGER DEFAULT 0,
+    value               blob,
+    created_at          DATETIME DEFAULT CURRENT_TIMESTAMP,
+) ENGINE=innodb DEFAULT CHARSET=utf8 AUTO_INCREMENT=1;
+TRUNCATE TABLE agent_key;
+
 CREATE TABLE IF NOT EXISTS sys_event_alarm (
     id                  INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
     process_name        VARCHAR(256),

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

@@ -0,0 +1,13 @@
+CREATE TABLE IF NOT EXISTS agent_key (
+    id                  INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
+    agent_id            INTEGER,
+    status              INTEGER DEFAULT 0,
+    value               blob,
+    created_at          DATETIME DEFAULT CURRENT_TIMESTAMP
+) ENGINE=innodb DEFAULT CHARSET=utf8 AUTO_INCREMENT=1;
+
+-- update db_version to latest, remember update DB_VERSION_EXPECTED in migration/version.go
+UPDATE db_version SET version='6.6.1.9';
+-- modify end
+
+COMMIT;

server/controller/db/mysql/model.go

@@ -245,6 +245,18 @@ func (LicenseFuncLog) TableName() string {
 	return "license_func_log"
 }
 
+type AgentKey struct {
+	ID        int       `gorm:"primaryKey;column:id;type:int;not null" json:"ID"`
+	AgentID   int       `gorm:"column:agent_id;type:int" json:"AGENT_ID"`
+	Status    int       `gorm:"column:status;type:int" json:"STATUS"`
+	Value     []byte    `gorm:"column:value;type:blob" json:"VALUE"`
+	CreatedAt time.Time `gorm:"column:created_at;type:datetime;default:CURRENT_TIMESTAMP" json:"CREATED_AT"`
+}
+
+func (AgentKey) TableName() string {
+	return "agent_key"
+}
+
 type DataSource struct {
 	ID                        int       `gorm:"primaryKey;column:id;type:int;not null" json:"ID"`
 	DisplayName               string    `gorm:"column:display_name;type:char(64);default:''" json:"DISPLAY_NAME"`

server/controller/grpc/server.go

@@ -98,6 +98,8 @@ func RunTLS(ctx context.Context, cfg *config.ControllerConfig) {
 	go sslServer.Serve(lis)
 	log.Infof("listening and serving SSL GRPC on: %s", cfg.SSLGrpcPort)
 
+	cfg.TrisolarisCfg.SetDataPlaneEncryption(true)
+
 	wg := utils.GetWaitGroupInCtx(ctx)
 	wg.Add(1)
 	defer wg.Done()

server/controller/trisolaris/config/config.go

@@ -19,6 +19,7 @@ package config
 import (
 	"net"
 	"os"
+	"time"
 
 	"github.com/deepflowio/deepflow/server/controller/common"
 	"github.com/deepflowio/deepflow/server/libs/logger"
@@ -45,13 +46,15 @@ type Config struct {
 	RegionDomainPrefix             string   `yaml:"region-domain-prefix"`
 	ClearKubernetesTime            int      `default:"600" yaml:"clear-kubernetes-time"`
 	NodeIP                         string
-	VTapCacheRefreshInterval       int  `default:"300" yaml:"vtapcache-refresh-interval"`
-	MetaDataRefreshInterval        int  `default:"60" yaml:"metadata-refresh-interval"`
-	NodeRefreshInterval            int  `default:"60" yaml:"node-refresh-interval"`
-	GPIDRefreshInterval            int  `default:"9" yaml:"gpid-refresh-interval"`
-	VTapAutoRegister               bool `default:"true" yaml:"vtap-auto-register"`
-	DomainAutoRegister             bool `default:"true" yaml:"domain-auto-register"`
-	DefaultTapMode                 int  `yaml:"default-tap-mode"`
+	VTapCacheRefreshInterval       int           `default:"300" yaml:"vtapcache-refresh-interval"`
+	MetaDataRefreshInterval        int           `default:"60" yaml:"metadata-refresh-interval"`
+	NodeRefreshInterval            int           `default:"60" yaml:"node-refresh-interval"`
+	GPIDRefreshInterval            int           `default:"9" yaml:"gpid-refresh-interval"`
+	VTapAutoRegister               bool          `default:"true" yaml:"vtap-auto-register"`
+	DomainAutoRegister             bool          `default:"true" yaml:"domain-auto-register"`
+	AgentKeyAgingTime              time.Duration `default:"720h" yaml:"agent-key-aging-time"`
+	AgentKeyLength                 int           `default:"16" yaml:"agent-key-length"`
+	DefaultTapMode                 int           `yaml:"default-tap-mode"`
 	BillingMethod                  string
 	GrpcPort                       int
 	IngesterPort                   int
@@ -65,6 +68,7 @@ type Config struct {
 	IngesterAPI                    common.IngesterApi // data source
 	AllAgentConnectToNatIP         bool
 	NoIPOverlapping                bool
+	DataPlaneEncryption            bool
 }
 
 func (c *Config) Convert() {
@@ -170,3 +174,12 @@ func (c *Config) SetFPermitConfig(fpermit common.FPermit) {
 func (c *Config) GetFPermitConfig() common.FPermit {
 	return c.FPermit
 }
+
+func (c *Config) SetDataPlaneEncryption(enabled bool) {
+	log.Infof("set data plane encryption enabled(%v)", enabled)
+	c.DataPlaneEncryption = enabled
+}
+
+func (c *Config) GetDataPlaneEncryption() bool {
+	return c.DataPlaneEncryption
+}

server/controller/trisolaris/dbmgr/dbmgr.go

@@ -206,6 +206,11 @@ func (obj *_DBMgr[M]) DeleteBatchFromID(ids []int) (err error) {
 	return
 }
 
+func (obj *_DBMgr[M]) DeleteBatchFromAgentID(ids []int, data *M) (err error) {
+	err = obj.DB.WithContext(obj.ctx).Where("`agent_id` in (?)", ids).Delete(data).Error
+	return
+}
+
 func (obj *_DBMgr[M]) Insert(data *M) (err error) {
 	err = obj.DB.WithContext(obj.ctx).Create(data).Error
 	return

server/controller/trisolaris/services/grpc/synchronize/vtap.go

@@ -108,6 +108,7 @@ func (e *VTapEvent) generateConfigInfo(c *vtap.VTapCache, clusterID string, gVTa
 	tapMode := api.TapMode(*vtapConfig.TapMode)
 	breakerMetricStr := convertBreakerMetric(*vtapConfig.SystemLoadCircuitBreakerMetric)
 	loadMetric := api.SystemLoadMetric(api.SystemLoadMetric_value[breakerMetricStr])
+	agentKey, _ := gVTapInfo.GetAgentKeyManager().GetAgentKey(int(c.GetVTapID()))
 	configure := &api.Config{
 		CollectorEnabled:              proto.Bool(Int2Bool(*vtapConfig.CollectorEnabled)),
 		CollectorSocketType:           &collectorSocketType,
@@ -187,6 +188,7 @@ func (e *VTapEvent) generateConfigInfo(c *vtap.VTapCache, clusterID string, gVTa
 
 		TeamId:     proto.Uint32(uint32(c.GetTeamID())),
 		OrganizeId: proto.Uint32(uint32(c.GetOrganizeID())),
+		SecretKey:  proto.String(string(agentKey)),
 	}
 
 	cacheTSBIP := c.GetTSDBIP()

server/controller/trisolaris/trisolaris.go

@@ -18,12 +18,14 @@ package trisolaris
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/golang/protobuf/proto"
 	"gorm.io/gorm"
 
 	"github.com/deepflowio/deepflow/message/trident"
+	scommon "github.com/deepflowio/deepflow/server/common"
 	. "github.com/deepflowio/deepflow/server/controller/common"
 	"github.com/deepflowio/deepflow/server/controller/db/mysql"
 	"github.com/deepflowio/deepflow/server/controller/election"
@@ -390,6 +392,7 @@ func (m *TrisolarisManager) Start() error {
 	}
 	go m.TimedCheckORG()
 	m.getTeamData(orgIDs)
+	scommon.SetAgentKeyOP(m)
 	log.Infof("finish orgdata init %v", orgIDs)
 	return nil
 }
@@ -467,13 +470,22 @@ func (m *TrisolarisManager) GetVTapInfo(orgID int) *vtap.VTapInfo {
 }
 
 func (m *TrisolarisManager) GetVTapCache(orgID int, key string) *vtap.VTapCache {
-	vttridentnfo := m.GetVTapInfo(orgID)
-	if vttridentnfo != nil {
-		return vttridentnfo.GetVTapCache(key)
+	vtapInfo := m.GetVTapInfo(orgID)
+	if vtapInfo != nil {
+		return vtapInfo.GetVTapCache(key)
 	}
 	return nil
 }
 
+func (m *TrisolarisManager) GetAgentKey(orgID int, agentID int) ([]byte, error) {
+	vtapInfo := m.GetVTapInfo(orgID)
+	if vtapInfo != nil {
+		return vtapInfo.GetAgentKeyManager().GetAgentKey(agentID)
+	}
+
+	return nil, fmt.Errorf("org(id=%d) not foud vtapInfo", orgID)
+}
+
 func (m *TrisolarisManager) checkORG() {
 	orgIDs, err := mysql.GetORGIDs()
 	if err != nil {
@@ -486,7 +498,8 @@ func (m *TrisolarisManager) checkORG() {
 		orgIDsUint32[index] = uint32(orgID)
 	}
 	m.orgIDData = &trident.OrgIDsResponse{
-		OrgIds: orgIDsUint32,
+		OrgIds:     orgIDsUint32,
+		UpdateTime: proto.Uint32(uint32(time.Now().Unix())),
 	}
 
 	for orgID, trisolaris := range m.orgToTrisolaris {

server/controller/trisolaris/vtap/agent_key/agent_key.go

@@ -0,0 +1,42 @@
+/*
+ * 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 agent_key
+
+import (
+	"fmt"
+
+	"gorm.io/gorm"
+
+	"github.com/deepflowio/deepflow/server/controller/trisolaris/config"
+)
+
+type AgentKeyManager struct{}
+
+func NewAgentKeyManager(db *gorm.DB, cfg *config.Config, orgID int) *AgentKeyManager {
+	return &AgentKeyManager{}
+
+}
+
+func (a *AgentKeyManager) InitData() {
+}
+
+func (a *AgentKeyManager) GetAgentKey(AgentID int) ([]byte, error) {
+	return nil, fmt.Errorf("Community Edition does not support data-side encryption")
+}
+
+func (a *AgentKeyManager) MonitorAgentKey() {
+}

server/controller/trisolaris/vtap/agent_key/go.mod

@@ -0,0 +1,3 @@
+module github.com/deepflowio/deepflow/server/controller/trisolaris/vtap/agent_key
+
+go 1.18

server/controller/trisolaris/vtap/vtap.go

@@ -36,14 +36,15 @@ import (
 	"github.com/deepflowio/deepflow/server/agent_config"
 	"github.com/deepflowio/deepflow/server/controller/common"
 	. "github.com/deepflowio/deepflow/server/controller/common"
-	mysql_model "github.com/deepflowio/deepflow/server/controller/db/mysql" // FIXME: To avoid ambiguity, name the package either mysql_model or db_model.
+	mysql_model "github.com/deepflowio/deepflow/server/controller/db/mysql"
 	. "github.com/deepflowio/deepflow/server/controller/trisolaris/common"
 	"github.com/deepflowio/deepflow/server/controller/trisolaris/config"
 	"github.com/deepflowio/deepflow/server/controller/trisolaris/dbmgr"
 	"github.com/deepflowio/deepflow/server/controller/trisolaris/metadata"
 	"github.com/deepflowio/deepflow/server/controller/trisolaris/pushmanager"
 	. "github.com/deepflowio/deepflow/server/controller/trisolaris/utils"
 	"github.com/deepflowio/deepflow/server/controller/trisolaris/utils/atomicbool"
+	"github.com/deepflowio/deepflow/server/controller/trisolaris/vtap/agent_key"
 	"github.com/deepflowio/deepflow/server/libs/logger"
 )
 
@@ -112,6 +113,8 @@ type VTapInfo struct {
 	processInfo *ProcessInfo
 	dbVTapIDs   mapset.Set
 
+	agentKeyManager *agent_key.AgentKeyManager
+
 	ctx    context.Context
 	cancel context.CancelFunc
 	ORGID
@@ -154,6 +157,7 @@ func NewVTapInfo(db *gorm.DB, metaData *metadata.MetaData, cfg *config.Config, o
 		config:                         cfg,
 		vTapIPs:                        &atomic.Value{},
 		dbVTapIDs:                      mapset.NewSet(),
+		agentKeyManager:                agent_key.NewAgentKeyManager(db, cfg, orgID),
 		ORGID:                          ORGID(orgID),
 		ctx:                            ctx,
 		cancel:                         cancel,
@@ -179,6 +183,14 @@ func (v *VTapInfo) AddVTapCache(vtap *mysql_model.VTap) {
 	log.Infof(v.Logf("add cache ctrl_ip: %s ctrl_mac: %s", vTapCache.GetCtrlIP(), vTapCache.GetCtrlMac()))
 }
 
+func (v *VTapInfo) GetAgentKeyManager() *agent_key.AgentKeyManager {
+	if v == nil {
+		return nil
+	}
+
+	return v.agentKeyManager
+}
+
 func (v *VTapInfo) GetVTapCache(key string) *VTapCache {
 	if v == nil {
 		return nil
@@ -875,6 +887,7 @@ func (v *VTapInfo) InitData() {
 	v.generateAllVTapRemoteSegements()
 	v.generateVTapIP()
 	v.generateLocalClusterID()
+	v.agentKeyManager.InitData()
 	v.isReady.Set()
 }
 
@@ -1255,6 +1268,7 @@ func (v *VTapInfo) monitorVTapRegister() {
 			select {
 			case <-v.chRegisterSuccess:
 				v.putChVTapChangedForSegment()
+				v.agentKeyManager.MonitorAgentKey()
 			default:
 			}
 			log.Info(v.Logf("end vtap register"))
@@ -1274,11 +1288,13 @@ func (v *VTapInfo) timedRefreshVTapCache() {
 		case <-tickerVTapCache:
 			log.Info(v.Logf("start generate vtap cache data from timed"))
 			v.GenerateVTapCache()
+			v.agentKeyManager.MonitorAgentKey()
 			v.processInfo.DeleteAgentExpiredData(v.dbVTapIDs)
 			log.Info(v.Logf("end generate vtap cache data from timed"))
 		case <-v.chVTapCacheRefresh:
 			log.Info(v.Logf("start generate vtap cache data from rpc"))
 			v.GenerateVTapCache()
+			v.agentKeyManager.MonitorAgentKey()
 			pushmanager.Broadcast(v.GetORGID())
 			log.Info(v.Logf("end generate vtap cache data from rpc"))
 		case <-v.ctx.Done():

server/go.mod

@@ -12,6 +12,7 @@ replace (
 	github.com/deepflowio/deepflow/server/controller/http/service/agentlicense => ./controller/http/service/agentlicense
 	github.com/deepflowio/deepflow/server/controller/http/service/configuration => ./controller/http/service/configuration
 	github.com/deepflowio/deepflow/server/controller/monitor/license => ./controller/monitor/license
+	github.com/deepflowio/deepflow/server/controller/trisolaris/vtap/agent_key => ./controller/trisolaris/vtap/agent_key
 	github.com/deepflowio/deepflow/server/ingester/config/configdefaults => ./ingester/config/configdefaults
 	github.com/deepflowio/deepflow/server/libs/logger/blocker => ./libs/logger/blocker
 	github.com/deepflowio/deepflow/server/querier/app/distributed_tracing/service/tracemap => ./querier/app/distributed_tracing/service/tracemap
@@ -40,7 +41,7 @@ require (
 	github.com/cornelk/hashmap v1.0.8
 	github.com/deckarep/golang-set v1.8.0
 	github.com/deckarep/golang-set/v2 v2.1.0
-	github.com/deepflowio/deepflow/message v0.0.0-20240725065348-535fb6f1efdc
+	github.com/deepflowio/deepflow/message v0.0.0-20240819073639-b32b909abd7d
 	github.com/deepflowio/deepflow/server/controller/cloud/kubernetes_gather/expand v0.0.0-00010101000000-000000000000
 	github.com/deepflowio/deepflow/server/controller/cloud/platform v0.0.0-00010101000000-000000000000
 	github.com/deepflowio/deepflow/server/controller/cloud/tencent/expand v0.0.0-00010101000000-000000000000
@@ -103,6 +104,7 @@ require (
 	github.com/bytedance/sonic v1.11.8
 	github.com/deepflowio/deepflow/server/controller/http/appender v0.0.0-00010101000000-000000000000
 	github.com/deepflowio/deepflow/server/controller/http/service/agentlicense v0.0.0-00010101000000-000000000000
+	github.com/deepflowio/deepflow/server/controller/trisolaris/vtap/agent_key v0.0.0-00010101000000-000000000000
 	github.com/deepflowio/deepflow/server/libs/logger/blocker v0.0.0-20240822020041-cdaf0f82ce6f
 	github.com/deepflowio/deepflow/server/querier/app/distributed_tracing/service/tracemap v0.0.0-00010101000000-000000000000
 	github.com/deepflowio/deepflow/server/querier/app/prometheus/router/packet_adapter v0.0.0-00010101000000-000000000000

server/go.sum

@@ -167,8 +167,8 @@ github.com/deckarep/golang-set v1.8.0 h1:sk9/l/KqpunDwP7pSjUg0keiOOLEnOBHzykLrsP
 github.com/deckarep/golang-set v1.8.0/go.mod h1:5nI87KwE7wgsBU1F4GKAw2Qod7p5kyS383rP6+o6qqo=
 github.com/deckarep/golang-set/v2 v2.1.0 h1:g47V4Or+DUdzbs8FxCCmgb6VYd+ptPAngjM6dtGktsI=
 github.com/deckarep/golang-set/v2 v2.1.0/go.mod h1:VAky9rY/yGXJOLEDv3OMci+7wtDpOF4IN+y82NBOac4=
-github.com/deepflowio/deepflow/message v0.0.0-20240725065348-535fb6f1efdc h1:B5hQ+ItZ4+dy4APMWZQeSV187j4dzDsxkAJfuEg1BLA=
-github.com/deepflowio/deepflow/message v0.0.0-20240725065348-535fb6f1efdc/go.mod h1:e+1lUMMlycCvFRKvlwt/y/0vxJnF8wVss3GyR1ARXY0=
+github.com/deepflowio/deepflow/message v0.0.0-20240819073639-b32b909abd7d h1:QyiXuNj9wvLSGeidpY8oGhbrXq4HTsjGwVjUHOKLM1o=
+github.com/deepflowio/deepflow/message v0.0.0-20240819073639-b32b909abd7d/go.mod h1:e+1lUMMlycCvFRKvlwt/y/0vxJnF8wVss3GyR1ARXY0=
 github.com/deepflowio/tempopb v0.0.0-20230215110519-15853baf3a79 h1:erRwXyZiUZxxZX/Q1QHesZXgxjiunZUFy+ggCRimkD4=
 github.com/deepflowio/tempopb v0.0.0-20230215110519-15853baf3a79/go.mod h1:h2rkZ319TExIUGuK8a2dlcGd8qc6wdhsrcpXWaJQaQE=
 github.com/dennwc/varint v1.0.0 h1:kGNFFSSw8ToIy3obO/kKr8U9GZYUAxQEVuix4zfDWzE=