feishu config add defaultUserGroup (#3076)

Co-authored-by: jie210 <huangjie@flashcat.com>

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.UpdateAt), trigger.Severity))
+				lst = append(lst, models.NewAnomalyPoint(trigger.Type, m, now, float64(now-target.BeatTime), 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.UpdateAt > 120 {
+					if now-target.BeatTime > 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)
+	err := storage.MSet(context.Background(), s.redis, newMap, 7*24*time.Hour)
 	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)
+		err = storage.MSet(context.Background(), s.redis, extendMap, 7*24*time.Hour)
 		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")
+		err = req.Update(rt.Ctx, "name", "identifier", "description", "cluster_name", "settings", "http", "auth", "updated_by", "updated_at", "is_default", "weight")
 	}
 
 	Render(c, nil, err)

center/router/router_login.go

@@ -571,12 +571,19 @@ 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,13 +1,16 @@
 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"
@@ -117,10 +120,13 @@ func (rt *Router) QueryLogBatch(c *gin.Context) {
 }
 
 func QueryDataConcurrently(anonymousAccess bool, ctx *gin.Context, f models.QueryParam) ([]models.DataResp, error) {
-	var resp []models.DataResp
-	var mu sync.Mutex
-	var wg sync.WaitGroup
-	var errs []error
+	var (
+		resp []models.DataResp
+		mu   sync.Mutex
+		wg   sync.WaitGroup
+		errs []error
+		rCtx = ctx.Request.Context()
+	)
 
 	for _, q := range f.Queries {
 		if !anonymousAccess && !CheckDsPerm(ctx, f.DatasourceId, f.Cate, q) {
@@ -132,12 +138,17 @@ 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(ctx.Request.Context(), query)
+			data, err := plug.QueryData(vCtx, query)
 			if err != nil {
 				logger.Warningf("query data error: req:%+v err:%v", query, err)
 				mu.Lock()

center/router/router_recording_rule.go

@@ -112,6 +112,7 @@ 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,6 +38,16 @@ 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,
@@ -81,9 +91,24 @@ 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)
 
@@ -102,14 +127,17 @@ 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].UpdateAt < 60 {
+			if now.Unix()-list[i].BeatTime < 60 {
 				list[i].TargetUp = 2
-			} else if now.Unix()-list[i].UpdateAt < 180 {
+			} else if now.Unix()-list[i].BeatTime < 180 {
 				list[i].TargetUp = 1
 			}
 		}
@@ -148,6 +176,43 @@ 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,7 +71,10 @@ 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; 
+) ENGINE = InnoDB DEFAULT CHARSET = utf8mb4;
+
+-- datasource add weight field
+alter table `datasource` add `weight` int not null default 0;
 
 CREATE TABLE `builtin_cate` (
     `id` bigint unsigned not null auto_increment,

datasource/commons/eslike/eslike.go

@@ -565,6 +565,14 @@ 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"])
@@ -602,6 +610,40 @@ 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":
@@ -678,21 +720,27 @@ 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 data error:%v", err)
+		logger.Warningf("query_log source:%s error:%v", string(sourceBytes), 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{}
-
-	b, _ := json.Marshal(source)
-	logger.Debugf("query data result query source:%s len:%d total:%d", string(b), len(result.Hits.Hits), total)
+	logger.Debugf("query_log source:%s len:%d total:%d", string(sourceBytes), len(result.Hits.Hits), total)
 
 	resultBytes, _ := json.Marshal(result)
-	logger.Debugf("query data result query source:%s result:%s", string(b), string(resultBytes))
+	logger.Debugf("query_log source:%s result:%s", string(sourceBytes), string(resultBytes))
 
 	if strings.HasPrefix(version, "6") {
 		for i := 0; i < len(result.Hits.Hits); i++ {

datasource/datasource.go

@@ -133,4 +133,5 @@ 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,52 +79,19 @@ func (d *Doris) Equal(p datasource.Datasource) bool {
 		return false
 	}
 
-	// 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
+	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
 }
 
 func (d *Doris) MakeLogQuery(ctx context.Context, query interface{}, eventTags []string, start, end int64) (interface{}, error) {
@@ -181,7 +148,7 @@ func (d *Doris) QueryData(ctx context.Context, query interface{}) ([]models.Data
 		}
 	}
 
-	items, err := d.QueryTimeseries(context.TODO(), &doris.QueryParam{
+	items, err := d.QueryTimeseries(ctx, &doris.QueryParam{
 		Database: dorisQueryParam.Database,
 		Sql:      dorisQueryParam.SQL,
 		Keys: types.Keys{

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

@@ -738,6 +738,7 @@ 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,6 +655,7 @@ 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,3 +358,6 @@ 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,6 +589,7 @@ 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,6 +104,7 @@ func getDatasourcesFromDBLoop(ctx *ctx.Context, fromAPI bool) {
 					AuthJson:       item.AuthJson,
 					Status:         item.Status,
 					IsDefault:      item.IsDefault,
+					Weight:         item.Weight,
 				}
 
 				if item.PluginType == "elasticsearch" {
@@ -236,5 +237,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,6 +39,9 @@ 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
@@ -88,13 +91,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.Password, d.User)
+	keys = append(keys, d.User, d.Password)
 	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
 	}
@@ -102,7 +105,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)
 		}
 	}()
 
@@ -121,6 +124,79 @@ 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
@@ -472,7 +548,7 @@ func (d *Doris) ExecContext(ctx context.Context, database string, sql string) er
 	timeoutCtx, cancel := d.createTimeoutContext(ctx)
 	defer cancel()
 
-	db, err := d.NewConn(timeoutCtx, database)
+	db, err := d.NewWriteConn(timeoutCtx, database)
 	if err != nil {
 		return err
 	}

dskit/doris/logs.go

@@ -10,13 +10,14 @@ 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)
+	return d.Query(ctx, query, true)
 }
 
-// 本质是查询时序数据, 取第一组, SQL由上层封装, 不再做复杂的解析和截断
+// QueryHistogram 本质是查询时序数据, 取第一组, 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,6 +15,10 @@ const (
 	TimeFieldFormatDateTime    = "datetime"
 )
 
+type noNeedCheckMaxRowKey struct{}
+
+var NoNeedCheckMaxRow = noNeedCheckMaxRowKey{}
+
 // 不再拼接SQL, 完全信赖用户的输入
 type QueryParam struct {
 	Database string     `json:"database"`
@@ -39,7 +43,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) ([]map[string]interface{}, error) {
+func (d *Doris) Query(ctx context.Context, query *QueryParam, checkMaxRow bool) ([]map[string]interface{}, error) {
 	// 校验SQL的合法性, 过滤掉 write请求
 	sqlItem := strings.Split(strings.ToUpper(query.Sql), " ")
 	for _, item := range sqlItem {
@@ -48,10 +52,12 @@ func (d *Doris) Query(ctx context.Context, query *QueryParam) ([]map[string]inte
 		}
 	}
 
-	// 检查查询结果行数
-	err := d.CheckMaxQueryRows(ctx, query.Database, query.Sql)
-	if err != nil {
-		return nil, err
+	if checkMaxRow {
+		// 检查查询结果行数
+		err := d.CheckMaxQueryRows(ctx, query.Database, query.Sql)
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	rows, err := d.ExecQuery(ctx, query.Database, query.Sql)
@@ -63,8 +69,12 @@ func (d *Doris) Query(ctx context.Context, query *QueryParam) ([]map[string]inte
 
 // 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) {
-	// 使用 Query 方法执行查询，Query方法内部已包含MaxQueryRows检查
-	rows, err := d.Query(ctx, query)
+	// 默认需要检查，除非调用方声明不需要检查
+	checkMaxRow := true
+	if noCheck, ok := ctx.Value(NoNeedCheckMaxRow).(bool); ok && noCheck {
+		checkMaxRow = false
+	}
+	rows, err := d.Query(ctx, query, checkMaxRow)
 	if err != nil {
 		return nil, err
 	}

memsto/target_cache.go

@@ -27,7 +27,8 @@ type TargetCacheType struct {
 	redis           storage.Redis
 
 	sync.RWMutex
-	targets map[string]*models.Target // key: ident
+	targets      map[string]*models.Target // key: ident
+	targetsIndex map[string][]string       // key: ip, value: ident list
 }
 
 func NewTargetCache(ctx *ctx.Context, stats *Stats, redis storage.Redis) *TargetCacheType {
@@ -38,6 +39,7 @@ 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()
@@ -51,6 +53,7 @@ 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 {
@@ -62,8 +65,17 @@ 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
@@ -78,6 +90,75 @@ 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()
@@ -105,7 +186,7 @@ func (tc *TargetCacheType) GetOffsetHost(targets []*models.Target, now, offset i
 			continue
 		}
 
-		if now-target.UpdateAt > 120 {
+		if now-target.BeatTime > 120 {
 			// means this target is not a active host, do not check offset
 			continue
 		}
@@ -147,6 +228,7 @@ 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")
@@ -170,6 +252,9 @@ 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]
 	}
@@ -186,57 +271,18 @@ 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 metaMap
+		return make(map[string]int64)
 	}
 
-	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
-	}
+	metaMap := models.FetchBeatTimesFromRedis(tc.redis, targets)
 
 	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.UpdateAt
+				metaMap[ident] = target.BeatTime
 			}
 		}
 	}

models/alert_rule.go

@@ -509,10 +509,16 @@ 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,6 +45,7 @@ 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:"-"`
 }
@@ -517,7 +518,8 @@ func (ds *Datasource) Encrypt(openRsa bool, publicKeyData []byte) error {
 // Decrypt 用于 edge 将从中心同步的数据源解密，中心不可调用
 func (ds *Datasource) Decrypt() error {
 	if rsaConfig == nil {
-		return errors.New("rsa config is nil")
+		logger.Debugf("datasource %s rsa config is nil", ds.Name)
+		return 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,6 +234,7 @@ 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,7 +212,6 @@ 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,6 +1,8 @@
 package models
 
 import (
+	"context"
+	"encoding/json"
 	"log"
 	"sort"
 	"strings"
@@ -8,6 +10,7 @@ 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"
@@ -36,6 +39,7 @@ 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:"-"`
@@ -97,12 +101,6 @@ 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
 }
@@ -182,6 +180,24 @@ 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 != "" {
@@ -203,17 +219,6 @@ 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 {
@@ -264,21 +269,6 @@ 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")
@@ -619,6 +609,66 @@ 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,6 +315,18 @@ 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 (
-	"io/ioutil"
+	"os"
 )
 
 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 = ioutil.ReadFile(file)
+		s.data, s.err = os.ReadFile(file)
 	}
 }

pkg/feishu/feishu.go

@@ -32,17 +32,18 @@ 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"`
+	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"`
 }
 
 type CallbackOutput struct {
@@ -312,6 +313,8 @@ 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,6 +106,7 @@ 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)
@@ -133,12 +134,7 @@ func (s *Set) UpdateTargets(lst []string, now int64) error {
 		return nil
 	}
 
-	if s.configs.UpdateDBTargetTimestampDisable {
-		// 如果 mysql 压力太大，关闭更新 db 的操作
-		return nil
-	}
-
-	// there are some idents not found in db, so insert them
+	// 新 target 仍需 INSERT 注册到 MySQL
 	var exists []string
 	err = s.ctx.DB.Table("target").Where("ident in ?", lst).Pluck("ident", &exists).Error
 	if err != nil {
@@ -153,35 +149,9 @@ 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")
@@ -248,7 +218,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)
+		err := storage.MSet(ctx, redis, content, 24*time.Hour)
 		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,8 +18,6 @@ type Pushgw struct {
 	UpdateTargetRetryIntervalMills int64
 	UpdateTargetTimeoutMills       int64
 	UpdateTargetBatchSize          int
-	UpdateDBTargetConcurrency      int
-	UpdateDBTargetTimestampDisable bool
 	PushConcurrency                int
 	UpdateTargetByUrlConcurrency   int
 
@@ -129,10 +127,6 @@ 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,21 +109,30 @@ func (rt *Router) debugSample(remoteAddr string, v *prompb.TimeSeries) {
 }
 
 func (rt *Router) DropSample(v *prompb.TimeSeries) bool {
-	filters := rt.Pushgw.DropSample
-	if len(filters) == 0 {
+	// 快速路径：检查仅 __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 {
 		return false
 	}
 
-	labelMap := make(map[string]string)
+	// 只有复杂过滤器存在时才创建 labelMap
+	labelMap := make(map[string]string, len(v.Labels))
 	for i := 0; i < len(v.Labels); i++ {
 		labelMap[v.Labels[i].Name] = v.Labels[i].Value
 	}
 
-	for _, filter := range filters {
-		if len(filter) == 0 {
-			continue
-		}
-
+	for _, filter := range rt.dropComplex {
 		if matchSample(filter, labelMap) {
 			return true
 		}

pushgw/router/router.go

@@ -6,6 +6,7 @@ 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"
@@ -33,6 +34,10 @@ type Router struct {
 	Ctx            *ctx.Context
 	HandleTS       HandleTSFunc
 	HeartbeatApi   string
+
+	// 预编译的 DropSample 过滤器
+	dropByNameOnly map[string]struct{} // 仅 __name__ 条件的快速匹配
+	dropComplex    []map[string]string // 多条件的复杂匹配
 }
 
 func stat() gin.HandlerFunc {
@@ -51,7 +56,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 {
-	return &Router{
+	rt := &Router{
 		HTTP:           httpConfig,
 		Pushgw:         pushgw,
 		Aconf:          aconf,
@@ -63,6 +68,38 @@ 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{}) error {
+func MSet(ctx context.Context, r Redis, m map[string]interface{}, expiration time.Duration) error {
 	pipe := r.Pipeline()
 	for k, v := range m {
-		pipe.Set(ctx, k, v, 0)
+		pipe.Set(ctx, k, v, expiration)
 	}
 	_, 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)
+	err = MSet(context.Background(), rdb, mp, 0)
 	if err != nil {
 		t.Fatalf("failed to set miniredis value: %v", err)
 	}