feat: agent group config gets l7 protocols from ck

server/controller/config/common/config.go

@@ -17,7 +17,7 @@
 package common
 
 type Swagger struct {
-	Enabled bool `default:"false" yaml:"enabled"`
+	Enabled bool `default:"true" yaml:"enabled"`
 }
 
 // TODO use this

server/controller/db/metadb/db.go

@@ -41,6 +41,10 @@ func GetDB(orgID int) (*DB, error) {
 	return GetDBs().NewDBIfNotExists(orgID)
 }
 
+func RemoveDB(orgID int) {
+	GetDBs().Delete(orgID)
+}
+
 func GetConfig() config.Config {
 	return GetDBs().GetConfig()
 }
@@ -198,15 +202,6 @@ func (c *DBs) check(db *DB) error {
 	return edition.CheckDBVersion(db.DB, db.Config)
 }
 
-func (c *DBs) DoOnAllDBs(execFunc func(db *DB) error) error {
-	for _, db := range dbs.orgIDToDB {
-		if err := execFunc(db); err != nil {
-			return fmt.Errorf("org(id:%d, name:%s) %s", db.ORGID, db.Name, err.Error())
-		}
-	}
-	return nil
-}
-
 type DBNameLogPrefix struct {
 	Name string
 }

server/controller/db/metadb/org.go

@@ -216,3 +216,22 @@ func getTagNameFromTag(tag, prefix string) string {
 func GetColumnNameFromTag(tag string) string {
 	return getTagNameFromTag(tag, "column:")
 }
+
+func DoOnAllDBs(execFunc func(*DB) error) error {
+	orgIDs, err := GetORGIDs()
+	if err != nil {
+		return err
+	}
+	for _, orgID := range orgIDs {
+		db, err := GetDB(orgID)
+		if err != nil {
+			log.Errorf("failed to get db info: %v", err, db.LogPrefixORGID, db.LogPrefixName)
+			continue
+		}
+		if err := execFunc(db); err != nil {
+			log.Errorf("failed to execute function: %v", err, db.LogPrefixORGID, db.LogPrefixName)
+			return err
+		}
+	}
+	return nil
+}

server/controller/http/server.go

@@ -105,10 +105,6 @@ func (s *Server) RegisterRouters() {
 		i.RegisterTo(s.engine)
 	}
 	trouter.RegisterTo(s.engine)
-
-	for _, route := range s.engine.Routes() {
-		log.Infof(" TODO method: %s, path: %s", route.Method, route.Path)
-	}
 }
 
 func (s *Server) appendRegistrant() []registrant.Registrant {

server/controller/http/service/agent_group_config.go

@@ -20,6 +20,8 @@ import (
 	"bytes"
 	"errors"
 	"fmt"
+	"net/http"
+	"net/url"
 	"strconv"
 
 	"github.com/google/uuid"
@@ -33,6 +35,7 @@ import (
 	"github.com/deepflowio/deepflow/server/controller/db/metadb/model"
 	httpcommon "github.com/deepflowio/deepflow/server/controller/http/common"
 	"github.com/deepflowio/deepflow/server/controller/trisolaris/refresh"
+	querierConfig "github.com/deepflowio/deepflow/server/querier/config"
 )
 
 var (
@@ -141,11 +144,11 @@ func (a *AgentGroupConfig) GetAgentGroupConfigTemplateJson() ([]byte, error) {
 	if err := yaml.Unmarshal(domainKeyToInfoYamlBytes, &domainKeyToInfoNode); err != nil {
 		return nil, err
 	}
-	// TODO get from ck
-	l7Protocols := []string{
-		"HTTP", "HTTP2", "Dubbo", "gRPC", "SOFARPC", "FastCGI", "bRPC", "Tars", "Some/IP", "MySQL", "PostgreSQL",
-		"Oracle", "Redis", "MongoDB", "Memcached", "Kafka", "MQTT", "AMQP", "OpenWire", "NATS", "Pulsar", "ping", "ZMTP",
-		"RocketMQ", "DNS", "TLS", "Custom"}
+
+	l7Protocols, err := a.getL7ProtocolsFromCK(dbInfo)
+	if err != nil {
+		return nil, err
+	}
 	l7ProtocolsYamlBytes, err := yaml.Marshal(l7Protocols)
 	if err != nil {
 		return nil, err
@@ -170,6 +173,39 @@ func (a *AgentGroupConfig) GetAgentGroupConfigTemplateJson() ([]byte, error) {
 	return agentconf.ConvertTemplateYAMLToJSON(dynamicOptions)
 }
 
+func (a *AgentGroupConfig) getL7ProtocolsFromCK(db *metadb.DB) ([]string, error) {
+	reqBody := url.Values{
+		"db":  {"flow_log"},
+		"sql": {"show tag l7_protocol values from l7_flow_log"},
+	}
+	url := fmt.Sprintf("http://%s:%d/v1/query", common.GetPodIP(), querierConfig.Cfg.ListenPort)
+	resp, err := common.CURLForm(
+		http.MethodPost,
+		url,
+		reqBody,
+		common.WithORGHeader(strconv.Itoa(db.GetORGID())),
+	)
+	if err != nil {
+		log.Errorf("failed to get l7 protocols from ck: %v", err, db.LogPrefixORGID)
+		return nil, err
+	}
+	result := resp.Get("result")
+	var displayNameIdx int
+	cols := result.Get("columns").MustArray()
+	for i, col := range cols {
+		if col == "display_name" {
+			displayNameIdx = i
+		}
+	}
+
+	l7Protocols := make([]string, 0)
+	for i := range resp.Get("result").Get("values").MustArray() {
+		value := resp.Get("result").Get("values").GetIndex(i)
+		l7Protocols = append(l7Protocols, value.GetIndex(displayNameIdx).MustString())
+	}
+	return l7Protocols, nil
+}
+
 func (a *AgentGroupConfig) GetAgentGroupConfig(groupLcuuid string, dataType int) ([]byte, error) {
 	dbInfo, err := metadb.GetDB(a.resourceAccess.UserInfo.ORGID)
 	if err != nil {

server/controller/monitor/analyzer.go

@@ -82,7 +82,7 @@ func (c *AnalyzerCheck) Start(sCtx context.Context) {
 		for {
 			select {
 			case <-ticker.C:
-				if err := metadb.GetDBs().DoOnAllDBs(func(db *metadb.DB) error {
+				if err := metadb.DoOnAllDBs(func(db *metadb.DB) error {
 					// 数据节点健康检查
 					c.healthCheck(db)
 					// 检查没有分配数据节点的采集器，并进行分配

server/controller/monitor/controller.go

@@ -87,7 +87,7 @@ func (c *ControllerCheck) Start(sCtx context.Context) {
 		for {
 			select {
 			case <-ticker.C:
-				if err := metadb.GetDBs().DoOnAllDBs(func(db *metadb.DB) error {
+				if err := metadb.DoOnAllDBs(func(db *metadb.DB) error {
 					// 控制器健康检查
 					c.healthCheck(db)
 					// 检查没有分配控制器的采集器，并进行分配

server/controller/monitor/license/pseudo_license.go

@@ -65,7 +65,7 @@ func (v *VTapLicenseAllocation) Start(sCtx context.Context) {
 		for {
 			select {
 			case <-ticker.C:
-				if err := metadb.GetDBs().DoOnAllDBs(func(db *metadb.DB) error {
+				if err := metadb.DoOnAllDBs(func(db *metadb.DB) error {
 					v.allocLicense(db)
 					return nil
 				}); err != nil {

server/controller/monitor/vtap/vtap.go

@@ -56,7 +56,7 @@ func (v *VTapCheck) Start(sCtx context.Context) {
 		for {
 			select {
 			case <-ticker.C:
-				metadb.GetDBs().DoOnAllDBs(func(db *metadb.DB) error {
+				metadb.DoOnAllDBs(func(db *metadb.DB) error {
 					// check launch_server resource if exist
 					v.launchServerCheck(db)
 					// check vtap type

server/controller/tagrecorder/ch_app_label.go

@@ -58,9 +58,13 @@ func (l *ChAPPLabel) generateNewData(db *metadb.DB) (map[PrometheusAPPLabelKey]m
 	for _, prometheusLabel := range prometheusLabels {
 		labelName := prometheusLabel.Name
 		if slices.Contains(appLabelSlice, labelName) {
-			labelNameID := labelNameIDMap[labelName]
+			labelNameID, nameOK := labelNameIDMap[labelName]
 			labelValue := prometheusLabel.Value
-			labelValueID := valueNameIDMap[labelValue]
+			labelValueID, valueOK := valueNameIDMap[labelValue]
+			if !nameOK || !valueOK {
+				log.Warningf("label name or value not found in db, labelName: %s, labelValue: %s", labelName, labelValue)
+				continue
+			}
 			keyToItem[PrometheusAPPLabelKey{LabelNameID: labelNameID, LabelValueID: labelValueID}] = metadbmodel.ChAPPLabel{
 				LabelNameID:  labelNameID,
 				LabelValue:   labelValue,

server/controller/tagrecorder/ch_device.go

@@ -1023,7 +1023,7 @@ func (c *ChCustomServiceDevice) sourceToTarget(md *message.Metadata, source *met
 // onResourceUpdated implements SubscriberDataGenerator
 func (c *ChCustomServiceDevice) onResourceUpdated(sourceID int, fieldsUpdate *message.CustomServiceFieldsUpdate, db *metadb.DB) {
 	updateInfo := make(map[string]interface{})
-
+	log.Infof("TODO fieldsUpdate.Name: %#v", fieldsUpdate.Name, db.LogPrefixORGID)
 	if fieldsUpdate.Name.IsDifferent() {
 		updateInfo["name"] = fieldsUpdate.Name.GetNew()
 	}

server/controller/tagrecorder/check/tagrecorder.go

@@ -64,7 +64,7 @@ type TagRecorder struct {
 
 func (c *TagRecorder) Check() {
 	go func() {
-		if err := metadb.GetDBs().DoOnAllDBs(func(db *metadb.DB) error {
+		if err := metadb.DoOnAllDBs(func(db *metadb.DB) error {
 			t := time.Now()
 			log.Infof("database=%s tagrecorder health check data run", db.Name, db.LogPrefixORGID)
 			tagrecorder.GetTeamInfo(db)

server/controller/tagrecorder/db_operator.go

@@ -24,10 +24,10 @@ import (
 )
 
 type operator[MT MySQLChModel, KT ChModelKey] interface {
-	batchPage(keys []KT, items []MT, operateFunc func([]KT, []MT, *metadb.DB), db *metadb.DB)
-	add(keys []KT, dbItems []MT, db *metadb.DB)
-	update(oldDBItem MT, updateInfo map[string]interface{}, key KT, db *metadb.DB)
-	delete(keys []KT, dbItems []MT, db *metadb.DB)
+	batchPage(keys []KT, items []MT, operateFunc func([]KT, []MT, *metadb.DB) error, db *metadb.DB) error
+	add(keys []KT, dbItems []MT, db *metadb.DB) error
+	update(oldDBItem MT, updateInfo map[string]interface{}, key KT, db *metadb.DB) error
+	delete(keys []KT, dbItems []MT, db *metadb.DB) error
 	setConfig(config.ControllerConfig)
 }
 
@@ -46,7 +46,7 @@ func (b *operatorComponent[MT, KT]) setConfig(cfg config.ControllerConfig) {
 	b.cfg = cfg
 }
 
-func (b *operatorComponent[MT, KT]) batchPage(keys []KT, items []MT, operateFunc func([]KT, []MT, *metadb.DB), db *metadb.DB) {
+func (b *operatorComponent[MT, KT]) batchPage(keys []KT, items []MT, operateFunc func([]KT, []MT, *metadb.DB) error, db *metadb.DB) error {
 	count := len(items)
 	offset := b.cfg.TagRecorderCfg.MySQLBatchSize
 	var pages int
@@ -61,35 +61,42 @@ func (b *operatorComponent[MT, KT]) batchPage(keys []KT, items []MT, operateFunc
 		if end > count {
 			end = count
 		}
-		operateFunc(keys[start:end], items[start:end], db)
+		err := operateFunc(keys[start:end], items[start:end], db)
+		if err != nil {
+			return err
+		}
 	}
+	return nil
 }
 
-func (b *operatorComponent[MT, KT]) add(keys []KT, dbItems []MT, db *metadb.DB) {
+func (b *operatorComponent[MT, KT]) add(keys []KT, dbItems []MT, db *metadb.DB) error {
 	err := db.Clauses(clause.OnConflict{
 		UpdateAll: true,
 	}).Create(&dbItems).Error
 	if err != nil {
 		log.Errorf("add %s (keys: %+v values: %+v) failed: %s", b.resourceTypeName, keys, dbItems, err.Error(), db.LogPrefixORGID) // TODO is key needed?
-		return
+		return err
 	}
 	log.Infof("add %s (keys: %+v values: %+v) success", b.resourceTypeName, keys, dbItems, db.LogPrefixORGID)
+	return nil
 }
 
-func (b *operatorComponent[MT, KT]) update(oldDBItem MT, updateInfo map[string]interface{}, key KT, db *metadb.DB) {
+func (b *operatorComponent[MT, KT]) update(oldDBItem MT, updateInfo map[string]interface{}, key KT, db *metadb.DB) error {
 	err := db.Model(&oldDBItem).Updates(updateInfo).Error
 	if err != nil {
 		log.Errorf("update %s (key: %+v value: %+v) failed: %s", b.resourceTypeName, key, oldDBItem, err.Error(), db.LogPrefixORGID)
-		return
+		return err
 	}
 	log.Infof("update %s (key: %+v value: %+v, update info: %v) success", b.resourceTypeName, key, oldDBItem, updateInfo, db.LogPrefixORGID)
+	return nil
 }
 
-func (b *operatorComponent[MT, KT]) delete(keys []KT, dbItems []MT, db *metadb.DB) {
+func (b *operatorComponent[MT, KT]) delete(keys []KT, dbItems []MT, db *metadb.DB) error {
 	err := db.Delete(&dbItems).Error
 	if err != nil {
 		log.Errorf("delete %s (keys: %+v values: %+v) failed: %s", b.resourceTypeName, keys, dbItems, err.Error(), db.LogPrefixORGID)
-		return
+		return err
 	}
 	log.Infof("delete %s (keys: %+v values: %+v) success", b.resourceTypeName, keys, dbItems, db.LogPrefixORGID)
+	return nil
 }

server/controller/tagrecorder/updater.go

@@ -186,13 +186,19 @@ func (b *UpdaterComponent[MT, KT]) Refresh() bool {
 				} else {
 					updateInfo, ok := b.updaterDG.generateUpdateInfo(oldDBItem, newDBItem)
 					if ok {
-						b.dbOperator.update(oldDBItem, updateInfo, key, db)
+						err := b.dbOperator.update(oldDBItem, updateInfo, key, db)
+						if err != nil {
+							log.Errorf("failed to update %s: %s", b.resourceTypeName, err, db.LogPrefixORGID)
+						}
 						isUpdate = true
 					}
 				}
 			}
 			if len(itemsToAdd) > 0 {
-				b.dbOperator.batchPage(keysToAdd, itemsToAdd, b.dbOperator.add, db) // 1是个占位符
+				err := b.dbOperator.batchPage(keysToAdd, itemsToAdd, b.dbOperator.add, db) // 1是个占位符
+				if err != nil {
+					log.Errorf("failed to add %s: %s", b.resourceTypeName, err, db.LogPrefixORGID)
+				}
 			}
 
 			for key, oldDBItem := range oldKeyToDBItem {
@@ -203,7 +209,10 @@ func (b *UpdaterComponent[MT, KT]) Refresh() bool {
 				}
 			}
 			if len(itemsToDelete) > 0 {
-				b.dbOperator.batchPage(keysToDelete, itemsToDelete, b.dbOperator.delete, db) // 1是个占位符
+				err := b.dbOperator.batchPage(keysToDelete, itemsToDelete, b.dbOperator.delete, db) // 1是个占位符
+				if err != nil {
+					log.Errorf("failed to delete %s: %s", b.resourceTypeName, err, db.LogPrefixORGID)
+				}
 			}
 
 			if len(itemsToDelete) > 0 && len(itemsToAdd) == 0 && !isUpdate {

server/controller/trisolaris/dbcache/db_data.go

@@ -75,6 +75,7 @@ type DBDataCache struct {
 	cens                    []*models.CEN
 	processes               []*models.Process
 	vips                    []*models.VIP
+	customServices          []*models.CustomService
 
 	podNSs     []*models.PodNamespace
 	vtaps      []*models.VTap
@@ -290,6 +291,10 @@ func (d *DBDataCache) GetVIPs() []*models.VIP {
 	return d.vips
 }
 
+func (d *DBDataCache) GetCustomServices() []*models.CustomService {
+	return d.customServices
+}
+
 func GetTapTypesFromDB(db *gorm.DB) []*models.TapType {
 	tapTypes, err := dbmgr.DBMgr[models.TapType](db).Gets()
 	if err != nil {
@@ -639,4 +644,11 @@ func (d *DBDataCache) GetDataCacheFromDB(db *gorm.DB) {
 	} else {
 		log.Error(d.Log(err.Error()))
 	}
+
+	customServices, err := dbmgr.DBMgr[models.CustomService](db).Gets()
+	if err == nil {
+		d.customServices = customServices
+	} else {
+		log.Error(d.Log(err.Error()))
+	}
 }