feat: link AI event data source retention updates

Author: kylewanginchina

server/controller/db/metadb/migrator/schema/schema_regression_test.go

@@ -34,6 +34,27 @@ func TestDBVersionExpectedMatchesHighestMySQLIssue(t *testing.T) {
 	}
 }
 
+func TestMySQLDMLInsert_AIAgentEventDataSourcesShareDefaultRetention(t *testing.T) {
+	sqlPath := filepath.Join("rawsql", "mysql", "dml_insert.sql")
+	content, err := os.ReadFile(sqlPath)
+	if err != nil {
+		t.Fatalf("read dml_insert sql failed: %v", err)
+	}
+	sql := string(content)
+
+	required := []string{
+		"VALUES (27, '事件-文件读写聚合事件', 'event.file_agg_event', 0, 7*24, @lcuuid);",
+		"VALUES (28, '事件-文件管理事件', 'event.file_mgmt_event', 0, 7*24, @lcuuid);",
+		"VALUES (29, '事件-进程权限事件', 'event.proc_perm_event', 0, 7*24, @lcuuid);",
+		"VALUES (30, '事件-进程操作事件', 'event.proc_ops_event', 0, 7*24, @lcuuid);",
+	}
+	for _, item := range required {
+		if !strings.Contains(sql, item) {
+			t.Fatalf("missing AI event data source default retention entry: %s", item)
+		}
+	}
+}
+
 func versionGreater(left, right string) bool {
 	leftParts := strings.Split(left, ".")
 	rightParts := strings.Split(right, ".")

server/controller/http/service/data_source.go

@@ -108,6 +108,37 @@ var DEFAULT_DATA_SOURCE_DISPLAY_NAMES = []string{
 	"日志-日志数据",       // application_log.log
 }
 
+var aiAgentRetentionCollections = []string{
+	"event.file_agg_event",
+	"event.file_mgmt_event",
+	"event.proc_perm_event",
+	"event.proc_ops_event",
+}
+
+func linkedRetentionCollections(collection string) []string {
+	for _, item := range aiAgentRetentionCollections {
+		if item == collection {
+			return append([]string(nil), aiAgentRetentionCollections...)
+		}
+	}
+	return nil
+}
+
+func resolveRetentionTargets(current metadbmodel.DataSource, all []metadbmodel.DataSource) []metadbmodel.DataSource {
+	linkedCollections := linkedRetentionCollections(current.DataTableCollection)
+	if len(linkedCollections) == 0 {
+		return []metadbmodel.DataSource{current}
+	}
+
+	targets := make([]metadbmodel.DataSource, 0, len(linkedCollections))
+	for _, dataSource := range all {
+		if utils.Find(linkedCollections, dataSource.DataTableCollection) {
+			targets = append(targets, dataSource)
+		}
+	}
+	return targets
+}
+
 func (d *DataSource) GetDataSources(orgID int, filter map[string]interface{}, specCfg *config.Specification) (resp []model.DataSource, err error) {
 	dbInfo, err := metadb.GetDB(orgID)
 	if err != nil {
@@ -409,8 +440,29 @@ func (d *DataSource) UpdateDataSource(orgID int, lcuuid string, dataSourceUpdate
 		return response[0], nil
 	}
 
-	oldRetentionTime := dataSource.RetentionTime
+	targets := []metadbmodel.DataSource{dataSource}
+	if linkedCollections := linkedRetentionCollections(dataSource.DataTableCollection); len(linkedCollections) > 0 {
+		var grouped []metadbmodel.DataSource
+		if err := db.Where("data_table_collection IN ?", linkedCollections).
+			Order("id ASC").
+			Find(&grouped).Error; err != nil {
+			return model.DataSource{}, err
+		}
+		targets = resolveRetentionTargets(dataSource, grouped)
+		if len(targets) != len(linkedCollections) {
+			return model.DataSource{}, response.ServiceError(
+				httpcommon.SERVER_ERROR,
+				fmt.Sprintf("linked retention data sources incomplete for %s", dataSource.DataTableCollection),
+			)
+		}
+	}
+
+	oldRetentionTimes := make(map[string]int, len(targets))
 	if dataSourceUpdate.RetentionTime != nil {
+		for i := range targets {
+			oldRetentionTimes[targets[i].Lcuuid] = targets[i].RetentionTime
+			targets[i].RetentionTime = *dataSourceUpdate.RetentionTime
+		}
 		dataSource.RetentionTime = *dataSourceUpdate.RetentionTime
 	}
 
@@ -421,29 +473,33 @@ func (d *DataSource) UpdateDataSource(orgID int, lcuuid string, dataSourceUpdate
 	}
 	var errs []error
 	for _, analyzer := range analyzers {
-		err = d.CallIngesterAPIModRP(orgID, analyzer.IP, dataSource)
-		if err != nil {
-			errs = append(errs, fmt.Errorf(
-				"failed to config analyzer (name: %s, ip:%s) update data_source (%s) error: %w",
-				analyzer.Name, analyzer.IP, dataSource.DisplayName, err,
-			))
-			continue
+		for _, target := range targets {
+			err = d.CallIngesterAPIModRP(orgID, analyzer.IP, target)
+			if err != nil {
+				errs = append(errs, fmt.Errorf(
+					"failed to config analyzer (name: %s, ip:%s) update data_source (%s) error: %w",
+					analyzer.Name, analyzer.IP, target.DisplayName, err,
+				))
+				continue
+			}
+			log.Infof("config analyzer (%s) mod data_source (%s) complete, retention time change: %ds -> %ds",
+				analyzer.IP, target.DisplayName, oldRetentionTimes[target.Lcuuid], target.RetentionTime, dbInfo.LogPrefixORGID)
 		}
-		log.Infof("config analyzer (%s) mod data_source (%s) complete, retention time change: %ds -> %ds",
-			analyzer.IP, dataSource.DisplayName, oldRetentionTime, dataSource.RetentionTime, dbInfo.LogPrefixORGID)
 	}
 
 	if len(errs) == 0 {
-		if err := db.Model(&dataSource).Updates(
-			map[string]interface{}{
-				"state":          common.DATA_SOURCE_STATE_NORMAL,
-				"retention_time": dataSource.RetentionTime,
-			},
-		).Error; err != nil {
-			return model.DataSource{}, err
+		for _, target := range targets {
+			if err := db.Model(&target).Updates(
+				map[string]interface{}{
+					"state":          common.DATA_SOURCE_STATE_NORMAL,
+					"retention_time": target.RetentionTime,
+				},
+			).Error; err != nil {
+				return model.DataSource{}, err
+			}
+			log.Infof("update data_source (%s), retention time change: %ds -> %ds",
+				target.DisplayName, oldRetentionTimes[target.Lcuuid], target.RetentionTime, dbInfo.LogPrefixORGID)
 		}
-		log.Infof("update data_source (%s), retention time change: %ds -> %ds",
-			dataSource.DisplayName, oldRetentionTime, dataSource.RetentionTime, dbInfo.LogPrefixORGID)
 	}
 	var errStrs []string
 	for _, e := range errs {
@@ -453,10 +509,12 @@ func (d *DataSource) UpdateDataSource(orgID int, lcuuid string, dataSourceUpdate
 
 	for _, e := range errs {
 		if errors.Is(e, httpcommon.ErrorFail) {
-			if err := db.Model(&dataSource).Updates(
-				map[string]interface{}{"state": common.DATA_SOURCE_STATE_EXCEPTION},
-			).Error; err != nil {
-				return model.DataSource{}, err
+			for _, target := range targets {
+				if err := db.Model(&target).Updates(
+					map[string]interface{}{"state": common.DATA_SOURCE_STATE_EXCEPTION},
+				).Error; err != nil {
+					return model.DataSource{}, err
+				}
 			}
 			err = response.ServiceError(httpcommon.PARTIAL_CONTENT, errMsg)
 			break