Merge branch 'main' into change-workflow

alert/eval/eval.go

@@ -844,7 +844,7 @@ func (arw *AlertRuleWorker) GetHostAnomalyPoint(ruleConfig string) ([]models.Ano
 				}
 				m["ident"] = target.Ident
 
-				lst = append(lst, models.NewAnomalyPoint(trigger.Type, m, now, float64(now-target.BeatTime), trigger.Severity))
+				lst = append(lst, models.NewAnomalyPoint(trigger.Type, m, now, float64(now-target.UpdateAt), trigger.Severity))
 			}
 		case "offset":
 			idents, exists := arw.Processor.TargetsOfAlertRuleCache.Get(arw.Processor.EngineName, arw.Rule.Id)
@@ -873,7 +873,7 @@ func (arw *AlertRuleWorker) GetHostAnomalyPoint(ruleConfig string) ([]models.Ano
 					continue
 				}
 				if target, exists := targetMap[ident]; exists {
-					if now-target.BeatTime > 120 {
+					if now-target.UpdateAt > 120 {
 						// means this target is not a active host, do not check offset
 						continue
 					}

center/metas/metas.go

@@ -118,7 +118,7 @@ func (s *Set) updateTargets(m map[string]models.HostMeta) error {
 	}
 
 	start := time.Now()
-	err := storage.MSet(context.Background(), s.redis, newMap, 7*24*time.Hour)
+	err := storage.MSet(context.Background(), s.redis, newMap)
 	if err != nil {
 		cstats.RedisOperationLatency.WithLabelValues("mset_target_meta", "fail").Observe(time.Since(start).Seconds())
 		return err
@@ -127,7 +127,7 @@ func (s *Set) updateTargets(m map[string]models.HostMeta) error {
 	}
 
 	if len(extendMap) > 0 {
-		err = storage.MSet(context.Background(), s.redis, extendMap, 7*24*time.Hour)
+		err = storage.MSet(context.Background(), s.redis, extendMap)
 		if err != nil {
 			cstats.RedisOperationLatency.WithLabelValues("mset_target_extend", "fail").Observe(time.Since(start).Seconds())
 			return err

center/router/router.go

@@ -391,8 +391,8 @@ func (rt *Router) Config(r *gin.Engine) {
 		pages.GET("/busi-group/:id/recording-rules", rt.auth(), rt.user(), rt.perm("/recording-rules"), rt.recordingRuleGets)
 		pages.POST("/busi-group/:id/recording-rules", rt.auth(), rt.user(), rt.perm("/recording-rules/add"), rt.bgrw(), rt.recordingRuleAddByFE)
 		pages.DELETE("/busi-group/:id/recording-rules", rt.auth(), rt.user(), rt.perm("/recording-rules/del"), rt.bgrw(), rt.recordingRuleDel)
+		pages.PUT("/busi-group/:id/recording-rule/:rrid", rt.auth(), rt.user(), rt.perm("/recording-rules/put"), rt.bgrw(), rt.recordingRulePutByFE)
 		pages.GET("/recording-rule/:rrid", rt.auth(), rt.user(), rt.perm("/recording-rules"), rt.recordingRuleGet)
-		pages.PUT("/recording-rule/:rrid", rt.auth(), rt.user(), rt.perm("/recording-rules"), rt.recordingRulePutByFE)
 		pages.PUT("/busi-group/:id/recording-rules/fields", rt.auth(), rt.user(), rt.perm("/recording-rules/put"), rt.recordingRulePutFields)
 
 		pages.GET("/busi-groups/alert-mutes", rt.auth(), rt.user(), rt.perm("/alert-mutes"), rt.alertMuteGetsByGids)

center/router/router_datasource.go

@@ -276,7 +276,7 @@ func (rt *Router) datasourceUpsert(c *gin.Context) {
 		}
 		err = req.Add(rt.Ctx)
 	} else {
-		err = req.Update(rt.Ctx, "name", "identifier", "description", "cluster_name", "settings", "http", "auth", "updated_by", "updated_at", "is_default", "weight")
+		err = req.Update(rt.Ctx, "name", "identifier", "description", "cluster_name", "settings", "http", "auth", "updated_by", "updated_at", "is_default")
 	}
 
 	Render(c, nil, err)

center/router/router_login.go

@@ -571,19 +571,12 @@ func (rt *Router) loginCallbackFeiShu(c *gin.Context) {
 	} else {
 		user = new(models.User)
 		defaultRoles := []string{}
-		defaultUserGroups := []int64{}
 		if rt.Sso.FeiShu != nil && rt.Sso.FeiShu.FeiShuConfig != nil {
 			defaultRoles = rt.Sso.FeiShu.FeiShuConfig.DefaultRoles
-			defaultUserGroups = rt.Sso.FeiShu.FeiShuConfig.DefaultUserGroups
 		}
-
 		user.FullSsoFields(feishu.SsoTypeName, ret.Username, ret.Nickname, ret.Phone, ret.Email, defaultRoles)
+		// create user from feishu
 		ginx.Dangerous(user.Add(rt.Ctx))
-
-		if len(defaultUserGroups) > 0 {
-			ginx.Dangerous(user.UpdateUserGroup(rt.Ctx, defaultUserGroups))
-		}
-
 	}
 
 	// set user login state

center/router/router_query.go

@@ -1,16 +1,13 @@
 package router
 
 import (
-	"context"
 	"fmt"
 	"sort"
 	"sync"
 
 	"github.com/ccfos/nightingale/v6/alert/eval"
 	"github.com/ccfos/nightingale/v6/dscache"
-	"github.com/ccfos/nightingale/v6/dskit/doris"
 	"github.com/ccfos/nightingale/v6/models"
-
 	"github.com/gin-gonic/gin"
 	"github.com/toolkits/pkg/ginx"
 	"github.com/toolkits/pkg/logger"
@@ -120,13 +117,10 @@ func (rt *Router) QueryLogBatch(c *gin.Context) {
 }
 
 func QueryDataConcurrently(anonymousAccess bool, ctx *gin.Context, f models.QueryParam) ([]models.DataResp, error) {
-	var (
-		resp []models.DataResp
-		mu   sync.Mutex
-		wg   sync.WaitGroup
-		errs []error
-		rCtx = ctx.Request.Context()
-	)
+	var resp []models.DataResp
+	var mu sync.Mutex
+	var wg sync.WaitGroup
+	var errs []error
 
 	for _, q := range f.Queries {
 		if !anonymousAccess && !CheckDsPerm(ctx, f.DatasourceId, f.Cate, q) {
@@ -138,17 +132,12 @@ func QueryDataConcurrently(anonymousAccess bool, ctx *gin.Context, f models.Quer
 			logger.Warningf("cluster:%d not exists", f.DatasourceId)
 			return nil, fmt.Errorf("cluster not exists")
 		}
-		
-		vCtx := rCtx
-		if f.Cate == models.DORIS {
-			vCtx = context.WithValue(vCtx, doris.NoNeedCheckMaxRow, true)
-		}
 
 		wg.Add(1)
 		go func(query interface{}) {
 			defer wg.Done()
 
-			data, err := plug.QueryData(vCtx, query)
+			data, err := plug.QueryData(ctx.Request.Context(), query)
 			if err != nil {
 				logger.Warningf("query data error: req:%+v err:%v", query, err)
 				mu.Lock()

center/router/router_recording_rule.go

@@ -112,7 +112,6 @@ func (rt *Router) recordingRulePutByFE(c *gin.Context) {
 	}
 
 	rt.bgrwCheck(c, ar.GroupId)
-	rt.bgroCheck(c, f.GroupId)
 
 	f.UpdateBy = c.MustGet("username").(string)
 	ginx.NewRender(c).Message(ar.Update(rt.Ctx, f))

center/router/router_target.go

@@ -38,16 +38,6 @@ func (rt *Router) targetGetsByHostFilter(c *gin.Context) {
 	total, err := models.TargetCountByFilter(rt.Ctx, query)
 	ginx.Dangerous(err)
 
-	models.FillTargetsBeatTime(rt.Redis, hosts)
-	now := time.Now().Unix()
-	for i := 0; i < len(hosts); i++ {
-		if now-hosts[i].BeatTime < 60 {
-			hosts[i].TargetUp = 2
-		} else if now-hosts[i].BeatTime < 180 {
-			hosts[i].TargetUp = 1
-		}
-	}
-
 	ginx.NewRender(c).Data(gin.H{
 		"list":  hosts,
 		"total": total,
@@ -91,24 +81,9 @@ func (rt *Router) targetGets(c *gin.Context) {
 		models.BuildTargetWhereWithBgids(bgids),
 		models.BuildTargetWhereWithDsIds(dsIds),
 		models.BuildTargetWhereWithQuery(query),
+		models.BuildTargetWhereWithDowntime(downtime),
 		models.BuildTargetWhereWithHosts(hosts),
 	}
-
-	// downtime 筛选：从缓存获取心跳时间，选择较小的集合用 IN 或 NOT IN 过滤
-	if downtime != 0 {
-		downtimeOpt, hasMatch := rt.downtimeFilter(downtime)
-		if !hasMatch {
-			ginx.NewRender(c).Data(gin.H{
-				"list":  []*models.Target{},
-				"total": 0,
-			}, nil)
-			return
-		}
-		if downtimeOpt != nil {
-			options = append(options, downtimeOpt)
-		}
-	}
-
 	total, err := models.TargetTotal(rt.Ctx, options...)
 	ginx.Dangerous(err)
 
@@ -127,17 +102,14 @@ func (rt *Router) targetGets(c *gin.Context) {
 		now := time.Now()
 		cache := make(map[int64]*models.BusiGroup)
 
-		// 从 Redis 补全 BeatTime
-		models.FillTargetsBeatTime(rt.Redis, list)
-
 		var keys []string
 		for i := 0; i < len(list); i++ {
 			ginx.Dangerous(list[i].FillGroup(rt.Ctx, cache))
 			keys = append(keys, models.WrapIdent(list[i].Ident))
 
-			if now.Unix()-list[i].BeatTime < 60 {
+			if now.Unix()-list[i].UpdateAt < 60 {
 				list[i].TargetUp = 2
-			} else if now.Unix()-list[i].BeatTime < 180 {
+			} else if now.Unix()-list[i].UpdateAt < 180 {
 				list[i].TargetUp = 1
 			}
 		}
@@ -176,43 +148,6 @@ func (rt *Router) targetGets(c *gin.Context) {
 	}, nil)
 }
 
-// downtimeFilter 从缓存获取心跳时间，生成 downtime 筛选条件
-// 选择匹配集和非匹配集中较小的一方，用 IN 或 NOT IN 来减少 SQL 参数量
-// 返回值：
-//   - option: 筛选条件，nil 表示所有 target 都符合条件（无需过滤）
-//   - hasMatch: 是否有符合条件的 target，false 表示无匹配应返回空结果
-func (rt *Router) downtimeFilter(downtime int64) (option models.BuildTargetWhereOption, hasMatch bool) {
-	now := time.Now().Unix()
-	targets := rt.TargetCache.GetAll()
-	var matchIdents, nonMatchIdents []string
-	for _, target := range targets {
-		matched := false
-		if downtime > 0 {
-			matched = target.BeatTime < now-downtime
-		} else if downtime < 0 {
-			matched = target.BeatTime > now+downtime
-		}
-		if matched {
-			matchIdents = append(matchIdents, target.Ident)
-		} else {
-			nonMatchIdents = append(nonMatchIdents, target.Ident)
-		}
-	}
-
-	if len(matchIdents) == 0 {
-		return nil, false
-	}
-
-	if len(nonMatchIdents) == 0 {
-		return nil, true
-	}
-
-	if len(matchIdents) <= len(nonMatchIdents) {
-		return models.BuildTargetWhereWithIdents(matchIdents), true
-	}
-	return models.BuildTargetWhereExcludeIdents(nonMatchIdents), true
-}
-
 func (rt *Router) targetExtendInfoByIdent(c *gin.Context) {
 	ident := ginx.QueryStr(c, "ident", "")
 	key := models.WrapExtendIdent(ident)

cli/upgrade/upgrade.sql

@@ -71,10 +71,7 @@ CREATE TABLE `datasource`
     `updated_at` bigint not null default 0,
     `updated_by` varchar(64) not null default '',
     PRIMARY KEY (`id`)
-) ENGINE = InnoDB DEFAULT CHARSET = utf8mb4;
-
--- datasource add weight field
-alter table `datasource` add `weight` int not null default 0;
+) ENGINE = InnoDB DEFAULT CHARSET = utf8mb4; 
 
 CREATE TABLE `builtin_cate` (
     `id` bigint unsigned not null auto_increment,

datasource/commons/eslike/eslike.go

@@ -565,14 +565,6 @@ func QueryData(ctx context.Context, queryParam interface{}, cliTimeout int64, ve
 		return nil, err
 	}
 
-	// 检查是否有 shard failures，有部分数据时仅记录警告继续处理
-	if shardErr := checkShardFailures(result.Shards, "query_data", searchSourceString); shardErr != nil {
-		if len(result.Aggregations["ts"]) == 0 {
-			return nil, shardErr
-		}
-		// 有部分数据，checkShardFailures 已记录警告，继续处理
-	}
-
 	logger.Debugf("query_data searchSource:%s resp:%s", string(jsonSearchSource), string(result.Aggregations["ts"]))
 
 	js, err := simplejson.NewJson(result.Aggregations["ts"])
@@ -610,40 +602,6 @@ func QueryData(ctx context.Context, queryParam interface{}, cliTimeout int64, ve
 	return items, nil
 }
 
-// checkShardFailures 检查 ES 查询结果中的 shard failures，返回格式化的错误信息
-func checkShardFailures(shards *elastic.ShardsInfo, logPrefix string, queryContext interface{}) error {
-	if shards == nil || shards.Failed == 0 || len(shards.Failures) == 0 {
-		return nil
-	}
-
-	var failureReasons []string
-	for _, failure := range shards.Failures {
-		reason := ""
-		if failure.Reason != nil {
-			if reasonType, ok := failure.Reason["type"].(string); ok {
-				reason = reasonType
-			}
-			if reasonMsg, ok := failure.Reason["reason"].(string); ok {
-				if reason != "" {
-					reason += ": " + reasonMsg
-				} else {
-					reason = reasonMsg
-				}
-			}
-		}
-		if reason != "" {
-			failureReasons = append(failureReasons, fmt.Sprintf("index=%s shard=%d: %s", failure.Index, failure.Shard, reason))
-		}
-	}
-
-	if len(failureReasons) > 0 {
-		errMsg := fmt.Sprintf("elasticsearch shard failures (%d/%d failed): %s", shards.Failed, shards.Total, strings.Join(failureReasons, "; "))
-		logger.Warningf("%s query:%v %s", logPrefix, queryContext, errMsg)
-		return fmt.Errorf("%s", errMsg)
-	}
-	return nil
-}
-
 func HitFilter(typ string) bool {
 	switch typ {
 	case "keyword", "date", "long", "integer", "short", "byte", "double", "float", "half_float", "scaled_float", "unsigned_long":
@@ -720,27 +678,21 @@ func QueryLog(ctx context.Context, queryParam interface{}, timeout int64, versio
 	} else {
 		source = source.From(param.P).Sort(param.DateField, param.Ascending)
 	}
-	sourceBytes, _ := json.Marshal(source)
 	result, err := search(ctx, indexArr, source, param.Timeout, param.MaxShard)
 	if err != nil {
-		logger.Warningf("query_log source:%s error:%v", string(sourceBytes), err)
+		logger.Warningf("query data error:%v", err)
 		return nil, 0, err
 	}
 
-	// 检查是否有 shard failures，有部分数据时仅记录警告继续处理
-	if shardErr := checkShardFailures(result.Shards, "query_log", string(sourceBytes)); shardErr != nil {
-		if len(result.Hits.Hits) == 0 {
-			return nil, 0, shardErr
-		}
-		// 有部分数据，checkShardFailures 已记录警告，继续处理
-	}
-
 	total := result.TotalHits()
+
 	var ret []interface{}
-	logger.Debugf("query_log source:%s len:%d total:%d", string(sourceBytes), len(result.Hits.Hits), total)
+
+	b, _ := json.Marshal(source)
+	logger.Debugf("query data result query source:%s len:%d total:%d", string(b), len(result.Hits.Hits), total)
 
 	resultBytes, _ := json.Marshal(result)
-	logger.Debugf("query_log source:%s result:%s", string(sourceBytes), string(resultBytes))
+	logger.Debugf("query data result query source:%s result:%s", string(b), string(resultBytes))
 
 	if strings.HasPrefix(version, "6") {
 		for i := 0; i < len(result.Hits.Hits); i++ {

datasource/datasource.go

@@ -133,5 +133,4 @@ type DatasourceInfo struct {
 	CreatedAt      int64                  `json:"created_at"`
 	UpdatedAt      int64                  `json:"updated_at"`
 	IsDefault      bool                   `json:"is_default"`
-	Weight         int                    `json:"weight"`
 }

datasource/doris/doris.go

@@ -79,19 +79,52 @@ func (d *Doris) Equal(p datasource.Datasource) bool {
 		return false
 	}
 
-	return d.Addr == newest.Addr &&
-		d.FeAddr == newest.FeAddr &&
-		d.User == newest.User &&
-		d.Password == newest.Password &&
-		d.EnableWrite == newest.EnableWrite &&
-		d.UserWrite == newest.UserWrite &&
-		d.PasswordWrite == newest.PasswordWrite &&
-		d.MaxQueryRows == newest.MaxQueryRows &&
-		d.Timeout == newest.Timeout &&
-		d.MaxIdleConns == newest.MaxIdleConns &&
-		d.MaxOpenConns == newest.MaxOpenConns &&
-		d.ConnMaxLifetime == newest.ConnMaxLifetime &&
-		d.ClusterName == newest.ClusterName
+	// only compare first shard
+	if d.Addr != newest.Addr {
+		return false
+	}
+
+	if d.User != newest.User {
+		return false
+	}
+
+	if d.Password != newest.Password {
+		return false
+	}
+
+	if d.EnableWrite != newest.EnableWrite {
+		return false
+	}
+
+	if d.FeAddr != newest.FeAddr {
+		return false
+	}
+
+	if d.MaxQueryRows != newest.MaxQueryRows {
+		return false
+	}
+
+	if d.Timeout != newest.Timeout {
+		return false
+	}
+
+	if d.MaxIdleConns != newest.MaxIdleConns {
+		return false
+	}
+
+	if d.MaxOpenConns != newest.MaxOpenConns {
+		return false
+	}
+
+	if d.ConnMaxLifetime != newest.ConnMaxLifetime {
+		return false
+	}
+
+	if d.ClusterName != newest.ClusterName {
+		return false
+	}
+
+	return true
 }
 
 func (d *Doris) MakeLogQuery(ctx context.Context, query interface{}, eventTags []string, start, end int64) (interface{}, error) {
@@ -148,7 +181,7 @@ func (d *Doris) QueryData(ctx context.Context, query interface{}) ([]models.Data
 		}
 	}
 
-	items, err := d.QueryTimeseries(ctx, &doris.QueryParam{
+	items, err := d.QueryTimeseries(context.TODO(), &doris.QueryParam{
 		Database: dorisQueryParam.Database,
 		Sql:      dorisQueryParam.SQL,
 		Keys: types.Keys{

docker/compose-postgres/initsql_for_postgres/a-n9e-for-Postgres.sql

@@ -738,7 +738,6 @@ CREATE TABLE datasource
     http varchar(4096) not null default '',
     auth varchar(8192) not null default '',
     is_default boolean not null default false,
-    weight int not null default 0,
     created_at bigint not null default 0,
     created_by varchar(64) not null default '',
     updated_at bigint not null default 0,

docker/initsql/a-n9e.sql

@@ -655,7 +655,6 @@ CREATE TABLE `datasource`
     `http` varchar(4096) not null default '',
     `auth` varchar(8192) not null default '',
     `is_default` boolean COMMENT 'is default datasource',
-    `weight` int not null default 0,
     `created_at` bigint not null default 0,
     `created_by` varchar(64) not null default '',
     `updated_at` bigint not null default 0,

docker/migratesql/migrate.sql

@@ -358,6 +358,3 @@ CREATE TABLE `user_view_favorite` (
     KEY `idx_view_id` (`view_id`),
     KEY `idx_user_id` (`user_id`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='user favorite views';
-
-/* v9 2026-01-20 datasource weight */
-ALTER TABLE `datasource` ADD COLUMN `weight` int not null default 0 COMMENT 'weight for sorting';

docker/sqlite.sql

@@ -589,7 +589,6 @@ CREATE TABLE `datasource`
     `http` varchar(4096) not null default '',
     `auth` varchar(8192) not null default '',
     `is_default` tinyint not null default 0,
-    `weight` int not null default 0,
     `created_at` bigint not null default 0,
     `created_by` varchar(64) not null default '',
     `updated_at` bigint not null default 0,

dscache/sync.go

@@ -90,7 +90,7 @@ func getDatasourcesFromDBLoop(ctx *ctx.Context, fromAPI bool) {
 					foundDefaultDatasource = true
 				}
 
-				// logger.Debugf("get datasource: %+v", item)
+				logger.Debugf("get datasource: %+v", item)
 				ds := datasource.DatasourceInfo{
 					Id:             item.Id,
 					Name:           item.Name,
@@ -104,7 +104,6 @@ func getDatasourcesFromDBLoop(ctx *ctx.Context, fromAPI bool) {
 					AuthJson:       item.AuthJson,
 					Status:         item.Status,
 					IsDefault:      item.IsDefault,
-					Weight:         item.Weight,
 				}
 
 				if item.PluginType == "elasticsearch" {
@@ -237,5 +236,5 @@ func PutDatasources(items []datasource.DatasourceInfo) {
 		}
 	}
 
-	// logger.Debugf("get plugin by type success Ids:%v", ids)
+	logger.Debugf("get plugin by type success Ids:%v", ids)
 }

dskit/doris/doris.go

@@ -39,9 +39,6 @@ type Doris struct {
 	MaxQueryRows    int    `json:"doris.max_query_rows" mapstructure:"doris.max_query_rows"`
 	ClusterName     string `json:"doris.cluster_name" mapstructure:"doris.cluster_name"`
 	EnableWrite     bool   `json:"doris.enable_write" mapstructure:"doris.enable_write"`
-	// 写用户，用来区分读写用户，减少数据源
-	UserWrite     string `json:"doris.user_write" mapstructure:"doris.user_write"`
-	PasswordWrite string `json:"doris.password_write" mapstructure:"doris.password_write"`
 }
 
 // NewDorisWithSettings initializes a new Doris instance with the given settings
@@ -91,13 +88,13 @@ func (d *Doris) NewConn(ctx context.Context, database string) (*sql.DB, error) {
 
 	var keys []string
 	keys = append(keys, d.Addr)
-	keys = append(keys, d.User, d.Password)
+	keys = append(keys, d.Password, d.User)
 	if len(database) > 0 {
 		keys = append(keys, database)
 	}
-	cachedKey := strings.Join(keys, ":")
+	cachedkey := strings.Join(keys, ":")
 	// cache conn with database
-	conn, ok := pool.PoolClient.Load(cachedKey)
+	conn, ok := pool.PoolClient.Load(cachedkey)
 	if ok {
 		return conn.(*sql.DB), nil
 	}
@@ -105,7 +102,7 @@ func (d *Doris) NewConn(ctx context.Context, database string) (*sql.DB, error) {
 	var err error
 	defer func() {
 		if db != nil && err == nil {
-			pool.PoolClient.Store(cachedKey, db)
+			pool.PoolClient.Store(cachedkey, db)
 		}
 	}()
 
@@ -124,79 +121,6 @@ func (d *Doris) NewConn(ctx context.Context, database string) (*sql.DB, error) {
 	return db, nil
 }
 
-// NewWriteConn establishes a new connection to Doris for write operations
-// When EnableWrite is true and UserWrite is configured, it uses the write user credentials
-// Otherwise, it reuses the read connection from NewConn
-func (d *Doris) NewWriteConn(ctx context.Context, database string) (*sql.DB, error) {
-	// If write user is not configured, reuse the read connection
-	if !d.EnableWrite || len(d.UserWrite) == 0 {
-		return d.NewConn(ctx, database)
-	}
-
-	if len(d.Addr) == 0 {
-		return nil, errors.New("empty fe-node addr")
-	}
-
-	// Set default values similar to postgres implementation
-	if d.Timeout == 0 {
-		d.Timeout = 60000
-	}
-	if d.MaxIdleConns == 0 {
-		d.MaxIdleConns = 10
-	}
-	if d.MaxOpenConns == 0 {
-		d.MaxOpenConns = 100
-	}
-	if d.ConnMaxLifetime == 0 {
-		d.ConnMaxLifetime = 14400
-	}
-	if d.MaxQueryRows == 0 {
-		d.MaxQueryRows = 500
-	}
-
-	// Use write user credentials
-	user := d.UserWrite
-	password := d.PasswordWrite
-
-	var keys []string
-	keys = append(keys, d.Addr)
-	keys = append(keys, user, password)
-	if len(database) > 0 {
-		keys = append(keys, database)
-	}
-	cachedKey := strings.Join(keys, ":")
-	// cache conn with database
-	conn, ok := pool.PoolClient.Load(cachedKey)
-	if ok {
-		return conn.(*sql.DB), nil
-	}
-	var db *sql.DB
-	var err error
-	defer func() {
-		if db != nil && err == nil {
-			pool.PoolClient.Store(cachedKey, db)
-		}
-	}()
-
-	// Simplified connection logic for Doris using MySQL driver
-	dsn := fmt.Sprintf("%s:%s@tcp(%s)/%s?charset=utf8", user, password, d.Addr, database)
-	db, err = sql.Open("mysql", dsn)
-	if err != nil {
-		return nil, err
-	}
-
-	// Set connection pool configuration for write connections
-	// Use more conservative values since write operations are typically less frequent
-	writeMaxIdleConns := max(d.MaxIdleConns/5, 2)
-	writeMaxOpenConns := max(d.MaxOpenConns/10, 5)
-
-	db.SetMaxIdleConns(writeMaxIdleConns)
-	db.SetMaxOpenConns(writeMaxOpenConns)
-	db.SetConnMaxLifetime(time.Duration(d.ConnMaxLifetime) * time.Second)
-
-	return db, nil
-}
-
 // createTimeoutContext creates a context with timeout based on Doris configuration
 func (d *Doris) createTimeoutContext(ctx context.Context) (context.Context, context.CancelFunc) {
 	timeout := d.Timeout
@@ -548,7 +472,7 @@ func (d *Doris) ExecContext(ctx context.Context, database string, sql string) er
 	timeoutCtx, cancel := d.createTimeoutContext(ctx)
 	defer cancel()
 
-	db, err := d.NewWriteConn(timeoutCtx, database)
+	db, err := d.NewConn(timeoutCtx, database)
 	if err != nil {
 		return err
 	}

dskit/doris/logs.go

@@ -10,14 +10,13 @@ const (
 	TimeseriesAggregationTimestamp = "__ts__"
 )
 
-// QueryLogs 查询日志
 // TODO: 待测试, MAP/ARRAY/STRUCT/JSON 等类型能否处理
 func (d *Doris) QueryLogs(ctx context.Context, query *QueryParam) ([]map[string]interface{}, error) {
 	// 等同于 Query()
-	return d.Query(ctx, query, true)
+	return d.Query(ctx, query)
 }
 
-// QueryHistogram 本质是查询时序数据, 取第一组, SQL由上层封装, 不再做复杂的解析和截断
+// 本质是查询时序数据, 取第一组, SQL由上层封装, 不再做复杂的解析和截断
 func (d *Doris) QueryHistogram(ctx context.Context, query *QueryParam) ([][]float64, error) {
 	values, err := d.QueryTimeseries(ctx, query)
 	if err != nil {

dskit/doris/timeseries.go

@@ -15,10 +15,6 @@ const (
 	TimeFieldFormatDateTime    = "datetime"
 )
 
-type noNeedCheckMaxRowKey struct{}
-
-var NoNeedCheckMaxRow = noNeedCheckMaxRowKey{}
-
 // 不再拼接SQL, 完全信赖用户的输入
 type QueryParam struct {
 	Database string     `json:"database"`
@@ -43,7 +39,7 @@ var (
 )
 
 // Query executes a given SQL query in Doris and returns the results with MaxQueryRows check
-func (d *Doris) Query(ctx context.Context, query *QueryParam, checkMaxRow bool) ([]map[string]interface{}, error) {
+func (d *Doris) Query(ctx context.Context, query *QueryParam) ([]map[string]interface{}, error) {
 	// 校验SQL的合法性, 过滤掉 write请求
 	sqlItem := strings.Split(strings.ToUpper(query.Sql), " ")
 	for _, item := range sqlItem {
@@ -52,12 +48,10 @@ func (d *Doris) Query(ctx context.Context, query *QueryParam, checkMaxRow bool)
 		}
 	}
 
-	if checkMaxRow {
-		// 检查查询结果行数
-		err := d.CheckMaxQueryRows(ctx, query.Database, query.Sql)
-		if err != nil {
-			return nil, err
-		}
+	// 检查查询结果行数
+	err := d.CheckMaxQueryRows(ctx, query.Database, query.Sql)
+	if err != nil {
+		return nil, err
 	}
 
 	rows, err := d.ExecQuery(ctx, query.Database, query.Sql)
@@ -69,12 +63,8 @@ func (d *Doris) Query(ctx context.Context, query *QueryParam, checkMaxRow bool)
 
 // QueryTimeseries executes a time series data query using the given parameters with MaxQueryRows check
 func (d *Doris) QueryTimeseries(ctx context.Context, query *QueryParam) ([]types.MetricValues, error) {
-	// 默认需要检查，除非调用方声明不需要检查
-	checkMaxRow := true
-	if noCheck, ok := ctx.Value(NoNeedCheckMaxRow).(bool); ok && noCheck {
-		checkMaxRow = false
-	}
-	rows, err := d.Query(ctx, query, checkMaxRow)
+	// 使用 Query 方法执行查询，Query方法内部已包含MaxQueryRows检查
+	rows, err := d.Query(ctx, query)
 	if err != nil {
 		return nil, err
 	}

memsto/target_cache.go

@@ -27,8 +27,7 @@ type TargetCacheType struct {
 	redis           storage.Redis
 
 	sync.RWMutex
-	targets      map[string]*models.Target // key: ident
-	targetsIndex map[string][]string       // key: ip, value: ident list
+	targets map[string]*models.Target // key: ident
 }
 
 func NewTargetCache(ctx *ctx.Context, stats *Stats, redis storage.Redis) *TargetCacheType {
@@ -39,7 +38,6 @@ func NewTargetCache(ctx *ctx.Context, stats *Stats, redis storage.Redis) *Target
 		stats:           stats,
 		redis:           redis,
 		targets:         make(map[string]*models.Target),
-		targetsIndex:    make(map[string][]string),
 	}
 
 	tc.SyncTargets()
@@ -53,7 +51,6 @@ func (tc *TargetCacheType) Reset() {
 	tc.statTotal = -1
 	tc.statLastUpdated = -1
 	tc.targets = make(map[string]*models.Target)
-	tc.targetsIndex = make(map[string][]string)
 }
 
 func (tc *TargetCacheType) StatChanged(total, lastUpdated int64) bool {
@@ -65,17 +62,8 @@ func (tc *TargetCacheType) StatChanged(total, lastUpdated int64) bool {
 }
 
 func (tc *TargetCacheType) Set(m map[string]*models.Target, total, lastUpdated int64) {
-	idx := make(map[string][]string, len(m))
-	for ident, target := range m {
-		if _, ok := idx[target.HostIp]; !ok {
-			idx[target.HostIp] = []string{}
-		}
-		idx[target.HostIp] = append(idx[target.HostIp], ident)
-	}
-
 	tc.Lock()
 	tc.targets = m
-	tc.targetsIndex = idx
 	tc.Unlock()
 
 	// only one goroutine used, so no need lock
@@ -90,75 +78,6 @@ func (tc *TargetCacheType) Get(ident string) (*models.Target, bool) {
 	return val, has
 }
 
-func (tc *TargetCacheType) GetByIp(ip string) ([]*models.Target, bool) {
-	tc.RLock()
-	defer tc.RUnlock()
-	idents, has := tc.targetsIndex[ip]
-	if !has {
-		return nil, false
-	}
-	targs := make([]*models.Target, 0, len(idents))
-	for _, ident := range idents {
-		if val, has := tc.targets[ident]; has {
-			targs = append(targs, val)
-		}
-	}
-	return targs, len(targs) > 0
-}
-
-func (tc *TargetCacheType) GetAll() []*models.Target {
-	tc.RLock()
-	defer tc.RUnlock()
-	lst := make([]*models.Target, 0, len(tc.targets))
-	for _, target := range tc.targets {
-		lst = append(lst, target)
-	}
-	return lst
-}
-
-// GetAllBeatTime 返回所有 target 的心跳时间 map，key 为 ident，value 为 BeatTime
-func (tc *TargetCacheType) GetAllBeatTime() map[string]int64 {
-	tc.RLock()
-	defer tc.RUnlock()
-	beatTimeMap := make(map[string]int64, len(tc.targets))
-	for ident, target := range tc.targets {
-		beatTimeMap[ident] = target.BeatTime
-	}
-	return beatTimeMap
-}
-
-// refreshBeatTime 从 Redis 刷新缓存中所有 target 的 BeatTime
-func (tc *TargetCacheType) refreshBeatTime() {
-	if tc.redis == nil {
-		return
-	}
-
-	// 快照 ident 列表，避免持锁访问 Redis
-	tc.RLock()
-	idents := make([]string, 0, len(tc.targets))
-	for ident := range tc.targets {
-		idents = append(idents, ident)
-	}
-	tc.RUnlock()
-
-	if len(idents) == 0 {
-		return
-	}
-
-	beatTimes := models.FetchBeatTimesFromRedis(tc.redis, idents)
-	if len(beatTimes) == 0 {
-		return
-	}
-
-	tc.Lock()
-	for ident, ts := range beatTimes {
-		if target, ok := tc.targets[ident]; ok {
-			target.BeatTime = ts
-		}
-	}
-	tc.Unlock()
-}
-
 func (tc *TargetCacheType) Gets(idents []string) []*models.Target {
 	tc.RLock()
 	defer tc.RUnlock()
@@ -186,7 +105,7 @@ func (tc *TargetCacheType) GetOffsetHost(targets []*models.Target, now, offset i
 			continue
 		}
 
-		if now-target.BeatTime > 120 {
+		if now-target.UpdateAt > 120 {
 			// means this target is not a active host, do not check offset
 			continue
 		}
@@ -228,7 +147,6 @@ func (tc *TargetCacheType) syncTargets() error {
 	}
 
 	if !tc.StatChanged(stat.Total, stat.LastUpdated) {
-		tc.refreshBeatTime()
 		tc.stats.GaugeCronDuration.WithLabelValues("sync_targets").Set(0)
 		tc.stats.GaugeSyncNumber.WithLabelValues("sync_targets").Set(0)
 		dumper.PutSyncRecord("targets", start.Unix(), -1, -1, "not changed")
@@ -252,9 +170,6 @@ func (tc *TargetCacheType) syncTargets() error {
 		}
 	}
 
-	// 从 Redis 批量获取心跳时间填充 BeatTime
-	models.FillTargetsBeatTime(tc.redis, lst)
-
 	for i := 0; i < len(lst); i++ {
 		m[lst[i].Ident] = lst[i]
 	}
@@ -271,18 +186,57 @@ func (tc *TargetCacheType) syncTargets() error {
 
 // get host update time
 func (tc *TargetCacheType) GetHostUpdateTime(targets []string) map[string]int64 {
+	metaMap := make(map[string]int64)
 	if tc.redis == nil {
-		return make(map[string]int64)
+		return metaMap
 	}
 
-	metaMap := models.FetchBeatTimesFromRedis(tc.redis, targets)
+	num := 0
+	var keys []string
+	for i := 0; i < len(targets); i++ {
+		keys = append(keys, models.WrapIdentUpdateTime(targets[i]))
+		num++
+		if num == 100 {
+			vals := storage.MGet(context.Background(), tc.redis, keys)
+			for _, value := range vals {
+				var hostUpdateTime models.HostUpdateTime
+				if value == nil {
+					continue
+				}
+
+				err := json.Unmarshal(value, &hostUpdateTime)
+				if err != nil {
+					logger.Errorf("failed to unmarshal host meta: %s value:%v", err, value)
+					continue
+				}
+				metaMap[hostUpdateTime.Ident] = hostUpdateTime.UpdateTime
+			}
+			keys = keys[:0]
+			num = 0
+		}
+	}
+
+	vals := storage.MGet(context.Background(), tc.redis, keys)
+	for _, value := range vals {
+		var hostUpdateTime models.HostUpdateTime
+		if value == nil {
+			continue
+		}
+
+		err := json.Unmarshal(value, &hostUpdateTime)
+		if err != nil {
+			logger.Warningf("failed to unmarshal host err:%v value:%s", err, string(value))
+			continue
+		}
+		metaMap[hostUpdateTime.Ident] = hostUpdateTime.UpdateTime
+	}
 
 	for _, ident := range targets {
 		if _, ok := metaMap[ident]; !ok {
 			// if not exists, get from cache
 			target, exists := tc.Get(ident)
 			if exists {
-				metaMap[ident] = target.BeatTime
+				metaMap[ident] = target.UpdateAt
 			}
 		}
 	}

models/alert_rule.go

@@ -509,16 +509,10 @@ func (ar *AlertRule) Verify() error {
 
 	ar.AppendTags = strings.TrimSpace(ar.AppendTags)
 	arr := strings.Fields(ar.AppendTags)
-	appendTagKeys := make(map[string]struct{})
 	for i := 0; i < len(arr); i++ {
 		if !strings.Contains(arr[i], "=") {
 			return fmt.Errorf("AppendTags(%s) invalid", arr[i])
 		}
-		pair := strings.SplitN(arr[i], "=", 2)
-		if _, exists := appendTagKeys[pair[0]]; exists {
-			return fmt.Errorf("AppendTags has duplicate key: %s", pair[0])
-		}
-		appendTagKeys[pair[0]] = struct{}{}
 	}
 
 	gids := strings.Fields(ar.NotifyGroups)

models/datasource.go

@@ -45,7 +45,6 @@ type Datasource struct {
 	CreatedBy       string                 `json:"created_by"`
 	UpdatedBy       string                 `json:"updated_by"`
 	IsDefault       bool                   `json:"is_default"`
-	Weight          int                    `json:"weight"`
 	Transport       *http.Transport        `json:"-" gorm:"-"`
 	ForceSave       bool                   `json:"force_save" gorm:"-"`
 }
@@ -518,8 +517,7 @@ func (ds *Datasource) Encrypt(openRsa bool, publicKeyData []byte) error {
 // Decrypt 用于 edge 将从中心同步的数据源解密，中心不可调用
 func (ds *Datasource) Decrypt() error {
 	if rsaConfig == nil {
-		logger.Debugf("datasource %s rsa config is nil", ds.Name)
-		return nil
+		return errors.New("rsa config is nil")
 	}
 
 	if !rsaConfig.OpenRSA {

models/event_pipeline_execution.go

@@ -6,7 +6,7 @@ import (
 	"fmt"
 
 	"github.com/ccfos/nightingale/v6/pkg/ctx"
-	"github.com/ccfos/nightingale/v6/pkg/poster"
+  "github.com/ccfos/nightingale/v6/pkg/poster"
 
 	"gorm.io/gorm"
 )

models/migrate/migrate.go

@@ -234,7 +234,6 @@ type Target struct {
 type Datasource struct {
 	IsDefault  bool   `gorm:"column:is_default;type:boolean;comment:is default datasource"`
 	Identifier string `gorm:"column:identifier;type:varchar(255);default:'';comment:identifier"`
-	Weight     int    `gorm:"column:weight;type:int;default:0;comment:weight for sorting"`
 }
 
 type Configs struct {

models/recording_rule.go

@@ -212,6 +212,7 @@ func (re *RecordingRule) Update(ctx *ctx.Context, ref RecordingRule) error {
 
 	ref.FE2DB()
 	ref.Id = re.Id
+	ref.GroupId = re.GroupId
 	ref.CreateAt = re.CreateAt
 	ref.CreateBy = re.CreateBy
 	ref.UpdateAt = time.Now().Unix()

models/target.go

@@ -1,8 +1,6 @@
 package models
 
 import (
-	"context"
-	"encoding/json"
 	"log"
 	"sort"
 	"strings"
@@ -10,7 +8,6 @@ import (
 
 	"github.com/ccfos/nightingale/v6/pkg/ctx"
 	"github.com/ccfos/nightingale/v6/pkg/poster"
-	"github.com/ccfos/nightingale/v6/storage"
 	"golang.org/x/exp/slices"
 
 	"github.com/pkg/errors"
@@ -39,7 +36,6 @@ type Target struct {
 	OS           string            `json:"os" gorm:"column:os"`
 	HostTags     []string          `json:"host_tags" gorm:"serializer:json"`
 
-	BeatTime   int64    `json:"beat_time" gorm:"-"` // 实时心跳时间，从 Redis 获取
 	UnixTime   int64    `json:"unixtime" gorm:"-"`
 	Offset     int64    `json:"offset" gorm:"-"`
 	TargetUp   float64  `json:"target_up" gorm:"-"`
@@ -101,6 +97,12 @@ func (t *Target) MatchGroupId(gid ...int64) bool {
 }
 
 func (t *Target) AfterFind(tx *gorm.DB) (err error) {
+	delta := time.Now().Unix() - t.UpdateAt
+	if delta < 60 {
+		t.TargetUp = 2
+	} else if delta < 180 {
+		t.TargetUp = 1
+	}
 	t.FillTagsMap()
 	return
 }
@@ -180,24 +182,6 @@ func BuildTargetWhereWithHosts(hosts []string) BuildTargetWhereOption {
 	}
 }
 
-func BuildTargetWhereWithIdents(idents []string) BuildTargetWhereOption {
-	return func(session *gorm.DB) *gorm.DB {
-		if len(idents) > 0 {
-			session = session.Where("ident in (?)", idents)
-		}
-		return session
-	}
-}
-
-func BuildTargetWhereExcludeIdents(idents []string) BuildTargetWhereOption {
-	return func(session *gorm.DB) *gorm.DB {
-		if len(idents) > 0 {
-			session = session.Where("ident not in (?)", idents)
-		}
-		return session
-	}
-}
-
 func BuildTargetWhereWithQuery(query string) BuildTargetWhereOption {
 	return func(session *gorm.DB) *gorm.DB {
 		if query != "" {
@@ -219,6 +203,17 @@ func BuildTargetWhereWithQuery(query string) BuildTargetWhereOption {
 	}
 }
 
+func BuildTargetWhereWithDowntime(downtime int64) BuildTargetWhereOption {
+	return func(session *gorm.DB) *gorm.DB {
+		if downtime > 0 {
+			session = session.Where("target.update_at < ?", time.Now().Unix()-downtime)
+		} else if downtime < 0 {
+			session = session.Where("target.update_at > ?", time.Now().Unix()+downtime)
+		}
+		return session
+	}
+}
+
 func buildTargetWhere(ctx *ctx.Context, options ...BuildTargetWhereOption) *gorm.DB {
 	sub := DB(ctx).Model(&Target{}).Distinct("target.ident")
 	for _, opt := range options {
@@ -269,6 +264,21 @@ func TargetCountByFilter(ctx *ctx.Context, query []map[string]interface{}) (int6
 	return Count(session)
 }
 
+func MissTargetGetsByFilter(ctx *ctx.Context, query []map[string]interface{}, ts int64) ([]*Target, error) {
+	var lst []*Target
+	session := TargetFilterQueryBuild(ctx, query, 0, 0)
+	session = session.Where("update_at < ?", ts)
+
+	err := session.Order("ident").Find(&lst).Error
+	return lst, err
+}
+
+func MissTargetCountByFilter(ctx *ctx.Context, query []map[string]interface{}, ts int64) (int64, error) {
+	session := TargetFilterQueryBuild(ctx, query, 0, 0)
+	session = session.Where("update_at < ?", ts)
+	return Count(session)
+}
+
 func TargetFilterQueryBuild(ctx *ctx.Context, query []map[string]interface{}, limit, offset int) *gorm.DB {
 	sub := DB(ctx).Model(&Target{}).Distinct("target.ident").Joins("left join " +
 		"target_busi_group on target.ident = target_busi_group.target_ident")
@@ -609,66 +619,6 @@ func (t *Target) FillMeta(meta *HostMeta) {
 	t.RemoteAddr = meta.RemoteAddr
 }
 
-// FetchBeatTimesFromRedis 从 Redis 批量获取心跳时间，返回 ident -> updateTime 的映射
-func FetchBeatTimesFromRedis(redis storage.Redis, idents []string) map[string]int64 {
-	result := make(map[string]int64, len(idents))
-	if redis == nil || len(idents) == 0 {
-		return result
-	}
-
-	num := 0
-	var keys []string
-	for i := 0; i < len(idents); i++ {
-		keys = append(keys, WrapIdentUpdateTime(idents[i]))
-		num++
-		if num == 100 {
-			fetchBeatTimeBatch(redis, keys, result)
-			keys = keys[:0]
-			num = 0
-		}
-	}
-
-	if len(keys) > 0 {
-		fetchBeatTimeBatch(redis, keys, result)
-	}
-
-	return result
-}
-
-func fetchBeatTimeBatch(redis storage.Redis, keys []string, result map[string]int64) {
-	vals := storage.MGet(context.Background(), redis, keys)
-	for _, value := range vals {
-		if value == nil {
-			continue
-		}
-		var hut HostUpdateTime
-		if err := json.Unmarshal(value, &hut); err != nil {
-			logger.Warningf("failed to unmarshal host update time: %v", err)
-			continue
-		}
-		result[hut.Ident] = hut.UpdateTime
-	}
-}
-
-// FillTargetsBeatTime 从 Redis 批量获取心跳时间填充 target.BeatTime
-func FillTargetsBeatTime(redis storage.Redis, targets []*Target) {
-	if len(targets) == 0 {
-		return
-	}
-
-	idents := make([]string, len(targets))
-	for i, t := range targets {
-		idents[i] = t.Ident
-	}
-
-	beatTimes := FetchBeatTimesFromRedis(redis, idents)
-	for _, t := range targets {
-		if ts, ok := beatTimes[t.Ident]; ok {
-			t.BeatTime = ts
-		}
-	}
-}
-
 func TargetIdents(ctx *ctx.Context, ids []int64) ([]string, error) {
 	var ret []string
 

models/user.go

@@ -315,18 +315,6 @@ func (u *User) UpdatePassword(ctx *ctx.Context, password, updateBy string) error
 	}).Error
 }
 
-func (u *User) UpdateUserGroup(ctx *ctx.Context, userGroupIds []int64) error {
-
-	count := len(userGroupIds)
-	for i := 0; i < count; i++ {
-		err := UserGroupMemberAdd(ctx, userGroupIds[i], u.Id)
-		if err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
 func UpdateUserLastActiveTime(ctx *ctx.Context, userId int64, lastActiveTime int64) error {
 	return DB(ctx).Model(&User{}).Where("id = ?", userId).Updates(map[string]interface{}{
 		"last_active_time": lastActiveTime,

pkg/cfg/scan.go

@@ -1,7 +1,7 @@
 package cfg
 
 import (
-	"os"
+	"io/ioutil"
 )
 
 type scanner struct {
@@ -23,6 +23,6 @@ func (s *scanner) Data() []byte {
 
 func (s *scanner) Read(file string) {
 	if s.err == nil {
-		s.data, s.err = os.ReadFile(file)
+		s.data, s.err = ioutil.ReadFile(file)
 	}
 }

pkg/feishu/feishu.go

@@ -32,18 +32,17 @@ type SsoClient struct {
 }
 
 type Config struct {
-	Enable            bool     `json:"enable"`
-	AuthURL           string   `json:"auth_url"`
-	DisplayName       string   `json:"display_name"`
-	AppID             string   `json:"app_id"`
-	AppSecret         string   `json:"app_secret"`
-	RedirectURL       string   `json:"redirect_url"`
-	UsernameField     string   `json:"username_field"`  // name, email, phone
-	FeiShuEndpoint    string   `json:"feishu_endpoint"` // 飞书API端点，默认为 open.feishu.cn
-	Proxy             string   `json:"proxy"`
-	CoverAttributes   bool     `json:"cover_attributes"`
-	DefaultRoles      []string `json:"default_roles"`
-	DefaultUserGroups []int64  `json:"default_user_groups"`
+	Enable          bool     `json:"enable"`
+	AuthURL         string   `json:"auth_url"`
+	DisplayName     string   `json:"display_name"`
+	AppID           string   `json:"app_id"`
+	AppSecret       string   `json:"app_secret"`
+	RedirectURL     string   `json:"redirect_url"`
+	UsernameField   string   `json:"username_field"`  // name, email, phone
+	FeiShuEndpoint  string   `json:"feishu_endpoint"` // 飞书API端点，默认为 open.feishu.cn
+	Proxy           string   `json:"proxy"`
+	CoverAttributes bool     `json:"cover_attributes"`
+	DefaultRoles    []string `json:"default_roles"`
 }
 
 type CallbackOutput struct {
@@ -313,8 +312,6 @@ func (s *SsoClient) Callback(redis storage.Redis, ctx context.Context, code, sta
 
 	// 根据UsernameField配置确定username
 	switch s.FeiShuConfig.UsernameField {
-	case "userid":
-		callbackOutput.Username = username
 	case "name":
 		if nickname == "" {
 			return nil, errors.New("feishu user name is empty")

pushgw/idents/idents.go

@@ -106,7 +106,6 @@ func (s *Set) UpdateTargets(lst []string, now int64) error {
 		return nil
 	}
 
-	// 心跳时间只写入 Redis，不再写入 MySQL update_at
 	err := s.updateTargetsUpdateTs(lst, now, s.redis)
 	if err != nil {
 		logger.Errorf("update_ts: failed to update targets: %v error: %v", lst, err)
@@ -134,7 +133,12 @@ func (s *Set) UpdateTargets(lst []string, now int64) error {
 		return nil
 	}
 
-	// 新 target 仍需 INSERT 注册到 MySQL
+	if s.configs.UpdateDBTargetTimestampDisable {
+		// 如果 mysql 压力太大，关闭更新 db 的操作
+		return nil
+	}
+
+	// there are some idents not found in db, so insert them
 	var exists []string
 	err = s.ctx.DB.Table("target").Where("ident in ?", lst).Pluck("ident", &exists).Error
 	if err != nil {
@@ -149,9 +153,35 @@ func (s *Set) UpdateTargets(lst []string, now int64) error {
 		}
 	}
 
+	// 从批量更新一批机器的时间戳，改成逐台更新，是为了避免批量更新时，mysql的锁竞争问题
+	start := time.Now()
+	duration := time.Since(start).Seconds()
+	if len(exists) > 0 {
+		sema := semaphore.NewSemaphore(s.configs.UpdateDBTargetConcurrency)
+		wg := sync.WaitGroup{}
+		for i := 0; i < len(exists); i++ {
+			sema.Acquire()
+			wg.Add(1)
+			go func(ident string) {
+				defer sema.Release()
+				defer wg.Done()
+				s.updateDBTargetTs(ident, now)
+			}(exists[i])
+		}
+		wg.Wait()
+	}
+	pstat.DBOperationLatency.WithLabelValues("update_targets_ts").Observe(duration)
+
 	return nil
 }
 
+func (s *Set) updateDBTargetTs(ident string, now int64) {
+	err := s.ctx.DB.Exec("UPDATE target SET update_at = ? WHERE ident = ?", now, ident).Error
+	if err != nil {
+		logger.Error("update_target: failed to update target:", ident, "error:", err)
+	}
+}
+
 func (s *Set) updateTargetsUpdateTs(lst []string, now int64, redis storage.Redis) error {
 	if redis == nil {
 		logger.Debugf("update_ts: redis is nil")
@@ -218,7 +248,7 @@ func (s *Set) writeTargetTsInRedis(ctx context.Context, redis storage.Redis, con
 
 	for i := 0; i < retryCount; i++ {
 		start := time.Now()
-		err := storage.MSet(ctx, redis, content, 24*time.Hour)
+		err := storage.MSet(ctx, redis, content)
 		duration := time.Since(start).Seconds()
 
 		logger.Debugf("update_ts: write target ts in redis, keys: %v, retryCount: %d, retryInterval: %v, error: %v", keys, retryCount, retryInterval, err)

pushgw/pconf/conf.go

@@ -18,6 +18,8 @@ type Pushgw struct {
 	UpdateTargetRetryIntervalMills int64
 	UpdateTargetTimeoutMills       int64
 	UpdateTargetBatchSize          int
+	UpdateDBTargetConcurrency      int
+	UpdateDBTargetTimestampDisable bool
 	PushConcurrency                int
 	UpdateTargetByUrlConcurrency   int
 
@@ -127,6 +129,10 @@ func (p *Pushgw) PreCheck() {
 		p.UpdateTargetBatchSize = 20
 	}
 
+	if p.UpdateDBTargetConcurrency <= 0 {
+		p.UpdateDBTargetConcurrency = 16
+	}
+
 	if p.PushConcurrency <= 0 {
 		p.PushConcurrency = 16
 	}

pushgw/router/fns.go

@@ -109,30 +109,21 @@ func (rt *Router) debugSample(remoteAddr string, v *prompb.TimeSeries) {
 }
 
 func (rt *Router) DropSample(v *prompb.TimeSeries) bool {
-	// 快速路径：检查仅 __name__ 的过滤器 O(1)
-	if len(rt.dropByNameOnly) > 0 {
-		for i := 0; i < len(v.Labels); i++ {
-			if v.Labels[i].Name == "__name__" {
-				if _, ok := rt.dropByNameOnly[v.Labels[i].Value]; ok {
-					return true
-				}
-				break // __name__ 只会出现一次，找到后直接跳出
-			}
-		}
-	}
-
-	// 慢速路径：处理复杂的多条件过滤器
-	if len(rt.dropComplex) == 0 {
+	filters := rt.Pushgw.DropSample
+	if len(filters) == 0 {
 		return false
 	}
 
-	// 只有复杂过滤器存在时才创建 labelMap
-	labelMap := make(map[string]string, len(v.Labels))
+	labelMap := make(map[string]string)
 	for i := 0; i < len(v.Labels); i++ {
 		labelMap[v.Labels[i].Name] = v.Labels[i].Value
 	}
 
-	for _, filter := range rt.dropComplex {
+	for _, filter := range filters {
+		if len(filter) == 0 {
+			continue
+		}
+
 		if matchSample(filter, labelMap) {
 			return true
 		}

pushgw/router/router.go

@@ -6,7 +6,6 @@ import (
 
 	"github.com/gin-gonic/gin"
 	"github.com/prometheus/prometheus/prompb"
-	"github.com/toolkits/pkg/logger"
 
 	"github.com/ccfos/nightingale/v6/alert/aconf"
 	"github.com/ccfos/nightingale/v6/center/metas"
@@ -34,10 +33,6 @@ type Router struct {
 	Ctx            *ctx.Context
 	HandleTS       HandleTSFunc
 	HeartbeatApi   string
-
-	// 预编译的 DropSample 过滤器
-	dropByNameOnly map[string]struct{} // 仅 __name__ 条件的快速匹配
-	dropComplex    []map[string]string // 多条件的复杂匹配
 }
 
 func stat() gin.HandlerFunc {
@@ -56,7 +51,7 @@ func stat() gin.HandlerFunc {
 func New(httpConfig httpx.Config, pushgw pconf.Pushgw, aconf aconf.Alert, tc *memsto.TargetCacheType, bg *memsto.BusiGroupCacheType,
 	idents *idents.Set, metas *metas.Set,
 	writers *writer.WritersType, ctx *ctx.Context) *Router {
-	rt := &Router{
+	return &Router{
 		HTTP:           httpConfig,
 		Pushgw:         pushgw,
 		Aconf:          aconf,
@@ -68,38 +63,6 @@ func New(httpConfig httpx.Config, pushgw pconf.Pushgw, aconf aconf.Alert, tc *me
 		MetaSet:        metas,
 		HandleTS:       func(pt *prompb.TimeSeries) *prompb.TimeSeries { return pt },
 	}
-
-	// 预编译 DropSample 过滤器
-	rt.initDropSampleFilters()
-
-	return rt
-}
-
-// initDropSampleFilters 预编译 DropSample 过滤器，将单条件 __name__ 过滤器
-// 放入 map 实现 O(1) 查找，多条件过滤器保留原有逻辑
-func (rt *Router) initDropSampleFilters() {
-	rt.dropByNameOnly = make(map[string]struct{})
-	rt.dropComplex = make([]map[string]string, 0)
-
-	for _, filter := range rt.Pushgw.DropSample {
-		if len(filter) == 0 {
-			continue
-		}
-
-		// 如果只有一个条件且是 __name__，放入快速匹配 map
-		if len(filter) == 1 {
-			if name, ok := filter["__name__"]; ok {
-				rt.dropByNameOnly[name] = struct{}{}
-				continue
-			}
-		}
-
-		// 其他情况放入复杂匹配列表
-		rt.dropComplex = append(rt.dropComplex, filter)
-	}
-
-	logger.Infof("DropSample filters initialized: %d name-only, %d complex",
-		len(rt.dropByNameOnly), len(rt.dropComplex))
 }
 
 func (rt *Router) Config(r *gin.Engine) {

storage/redis.go

@@ -163,10 +163,10 @@ func MGet(ctx context.Context, r Redis, keys []string) [][]byte {
 	return vals
 }
 
-func MSet(ctx context.Context, r Redis, m map[string]interface{}, expiration time.Duration) error {
+func MSet(ctx context.Context, r Redis, m map[string]interface{}) error {
 	pipe := r.Pipeline()
 	for k, v := range m {
-		pipe.Set(ctx, k, v, expiration)
+		pipe.Set(ctx, k, v, 0)
 	}
 	_, err := pipe.Exec(ctx)
 	return err

storage/redis_test.go

@@ -30,7 +30,7 @@ func TestMiniRedisMGet(t *testing.T) {
 	mp["key2"] = "value2"
 	mp["key3"] = "value3"
 
-	err = MSet(context.Background(), rdb, mp, 0)
+	err = MSet(context.Background(), rdb, mp)
 	if err != nil {
 		t.Fatalf("failed to set miniredis value: %v", err)
 	}