Skip to content

Commit ff28b5b

Browse files
authored
fix(alert): fix eval goroutine leak and nil-pointer panics (#262)
- Submit(alert/recording/consumer): 覆盖 ContextMap 前先 cancel 旧的 context, 避免同 ruleId/faultCenter 旧评估协程泄漏 + 新老规则并行重复评估、重复告警。 - processSingleDatasource: 数据源启用判断由 !*instance.Enabled 改为 instance.GetEnabled()(nil 安全), 避免 DB 中 enabled 为 NULL 时 panic; 否则会被 Eval 的 recover→Restart 拖进无限重启循环。 - rule.Update: 对 First(&oldRule) 增加错误检查(规则不存在时直接返回); oldRule.Enabled / r.Enabled 判 nil 后再解引用, 避免更新已删除规则时 panic。 - oidc.CallBack: 畸形 access_token 时返回真实错误, 而非 (nil, nil) 把认证 失败当成功, 掩盖错误。
1 parent 435f3a3 commit ff28b5b

5 files changed

Lines changed: 27 additions & 8 deletions

File tree

‎alert/consumer/consumer.go‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -123,6 +123,11 @@ func (c *Consume) Submit(faultCenter models.FaultCenter) {
123123
c.ctx.Mux.Lock()
124124
defer c.ctx.Mux.Unlock()
125125

126+
// 先停掉同 faultCenter 的旧消费协程, 避免泄漏
127+
if cancel, exists := c.ctx.ContextMap[faultCenter.ID]; exists {
128+
cancel()
129+
}
130+
126131
withCtx, cancel := context.WithCancel(context.Background())
127132
c.ctx.ContextMap[faultCenter.ID] = cancel
128133
go c.Watch(withCtx, faultCenter)

‎alert/eval/alert_rule.go‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -76,6 +76,11 @@ func (t *AlertRule) Submit(rule models.AlertRule) {
7676
t.ctx.Mux.Lock()
7777
defer t.ctx.Mux.Unlock()
7878

79+
// 先停掉同 ruleId 的旧评估协程, 避免泄漏与重复评估
80+
if cancel, exists := t.ctx.ContextMap[rule.RuleId]; exists {
81+
cancel()
82+
}
83+
7984
c, cancel := context.WithCancel(context.Background())
8085
t.ctx.ContextMap[rule.RuleId] = cancel
8186
go t.Eval(c, rule)
@@ -199,7 +204,7 @@ func (t *AlertRule) processSingleDatasource(dsId string, rule models.AlertRule)
199204
}
200205

201206
// 检查数据源是否启用
202-
if !*instance.Enabled {
207+
if !instance.GetEnabled() {
203208
logc.Errorf(t.ctx.Ctx, "Datasource %s is disabled", dsId)
204209
return nil
205210
}

‎alert/eval/recording_rule.go‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,11 @@ func (t *RecordingRule) Submit(rule models.RecordingRule) {
4040
t.ctx.Mux.Lock()
4141
defer t.ctx.Mux.Unlock()
4242

43+
// 先停掉同 ruleId 的旧评估协程, 避免泄漏与重复评估
44+
if cancel, exists := t.ctx.ContextMap[rule.RuleId]; exists {
45+
cancel()
46+
}
47+
4348
c, cancel := context.WithCancel(context.Background())
4449
t.ctx.ContextMap[rule.RuleId] = cancel
4550
go t.Eval(c, rule)
@@ -122,7 +127,7 @@ func (t *RecordingRule) processSingleDatasource(rule models.RecordingRule) {
122127
}
123128

124129
// 检查数据源是否启用
125-
if !*instance.Enabled {
130+
if !instance.GetEnabled() {
126131
logc.Errorf(t.ctx.Ctx, "Datasource %s is disabled", rule.DatasourceId)
127132
return
128133
}

‎internal/services/oidc.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -68,7 +68,7 @@ func (os oidcService) CallBack(ctx *gin.Context, req interface{}) (interface{},
6868

6969
parts := strings.Split(data.AccessToken, ".")
7070
if len(parts) < 2 {
71-
return nil, err
71+
return nil, fmt.Errorf("OIDC 返回的 access_token 不是合法 JWT")
7272
}
7373

7474
payloadBytes, err := base64.RawURLEncoding.DecodeString(parts[1])

‎internal/services/rule.go‎

Lines changed: 9 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -98,9 +98,11 @@ func (rs ruleService) Create(req interface{}) (interface{}, interface{}) {
9898
func (rs ruleService) Update(req interface{}) (interface{}, interface{}) {
9999
r := req.(*types.RequestRuleUpdate)
100100
oldRule := models.AlertRule{}
101-
rs.ctx.DB.DB().Model(&models.AlertRule{}).
101+
if err := rs.ctx.DB.DB().Model(&models.AlertRule{}).
102102
Where("tenant_id = ? AND rule_id = ?", r.TenantId, r.RuleId).
103-
First(&oldRule)
103+
First(&oldRule).Error; err != nil {
104+
return nil, fmt.Errorf("规则不存在, rule_id: %s", r.RuleId)
105+
}
104106

105107
if oldRule.FaultCenterId != r.FaultCenterId {
106108
fingerprints := rs.ctx.Redis.Alert().GetFingerprintsByRuleId(oldRule.TenantId, oldRule.FaultCenterId, oldRule.RuleId)
@@ -114,11 +116,13 @@ func (rs ruleService) Update(req interface{}) (interface{}, interface{}) {
114116
判断当前状态是否是false 并且 历史状态是否为true
115117
*/
116118
var action string
117-
if *oldRule.Enabled == true && *r.Enabled == false {
119+
oldEnabled := oldRule.Enabled != nil && *oldRule.Enabled
120+
newEnabled := r.Enabled != nil && *r.Enabled
121+
if oldEnabled && !newEnabled {
118122
action = tools.ActionDisable
119-
} else if *oldRule.Enabled == false && *r.Enabled == true {
123+
} else if !oldEnabled && newEnabled {
120124
action = tools.ActionEnable
121-
} else if *oldRule.Enabled == true && *r.Enabled == true {
125+
} else if oldEnabled && newEnabled {
122126
action = tools.ActionUpdate
123127
}
124128

0 commit comments

Comments
 (0)