340 lines
17 KiB
Go
340 lines
17 KiB
Go
package alerts
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"log/slog"
|
|
"strings"
|
|
"time"
|
|
|
|
"data-terminal-cloud/internal/outbox"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
type Signal struct {
|
|
TenantID, RuleCode, Category, Severity, Fingerprint, Title, Summary string
|
|
ResourceType, ResourcePublicID, RequestID string
|
|
}
|
|
|
|
type Scanner struct {
|
|
pool *pgxpool.Pool
|
|
logger *slog.Logger
|
|
interval time.Duration
|
|
now func() time.Time
|
|
outbox *outbox.Store
|
|
}
|
|
|
|
func NewScanner(pool *pgxpool.Pool, logger *slog.Logger, interval time.Duration) (*Scanner, error) {
|
|
if pool == nil || logger == nil || interval <= 0 {
|
|
return nil, errors.New("alert scanner dependencies are required")
|
|
}
|
|
return &Scanner{pool: pool, logger: logger, interval: interval, now: time.Now, outbox: outbox.NewStore(pool)}, nil
|
|
}
|
|
|
|
func (s *Scanner) Run(ctx context.Context) {
|
|
ticker := time.NewTicker(s.interval)
|
|
defer ticker.Stop()
|
|
for {
|
|
if count, err := s.ScanOnce(ctx); err != nil && !errors.Is(err, context.Canceled) {
|
|
s.logger.Error("alert scan failed", "error", err)
|
|
} else if count > 0 {
|
|
s.logger.Info("alert signals processed", "count", count)
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *Scanner) ScanOnce(ctx context.Context) (int, error) {
|
|
now := s.now().UTC()
|
|
signals := []Signal{}
|
|
queries := []struct {
|
|
sql string
|
|
build func(row pgx.Row) (Signal, error)
|
|
}{
|
|
{`SELECT tenant_id::text,count(*) FROM outbox_event WHERE status='dead' GROUP BY tenant_id`, func(r pgx.Row) (Signal, error) {
|
|
var tenant string
|
|
var count int64
|
|
e := r.Scan(&tenant, &count)
|
|
return Signal{TenantID: tenant, RuleCode: "OUTBOX_DEAD_LETTER", Category: "TECHNICAL", Severity: "HIGH", Fingerprint: "outbox-dead", Title: "可靠任务出现死信", Summary: fmt.Sprintf("当前客户组织存在%d条死信待办，请按错误码修复并受控重放。", count), ResourceType: "outbox_queue"}, e
|
|
}},
|
|
{`SELECT tenant_id::text,public_id::text,status FROM privacy_request WHERE status='PARTIAL_FAILED' OR (request_type='DELETE' AND status<>'COMPLETED' AND created_at<now()-interval '1 hour')`, func(r pgx.Row) (Signal, error) {
|
|
var tenant, id, status string
|
|
e := r.Scan(&tenant, &id, &status)
|
|
return Signal{TenantID: tenant, RuleCode: "PRIVACY_DELETE_OVERDUE", Category: "BUSINESS", Severity: "CRITICAL", Fingerprint: "privacy-delete:" + id, Title: "隐私删除需要人工处置", Summary: "主体删除未在预期时间内完成全部子系统对账，必须立即核查失败子任务。", ResourceType: "privacy_request", ResourcePublicID: id}, e
|
|
}},
|
|
{`SELECT tenant_id::text,public_id::text,COALESCE(last_error_code,'UNKNOWN') FROM upload_batch WHERE status='FAILED'`, func(r pgx.Row) (Signal, error) {
|
|
var tenant, id, code string
|
|
e := r.Scan(&tenant, &id, &code)
|
|
return Signal{TenantID: tenant, RuleCode: "MEDIA_UPLOAD_FAILED", Category: "BUSINESS", Severity: "MEDIUM", Fingerprint: "media-upload:" + id, Title: "录音上传批次校验失败", Summary: "录音上传批次进入失败终态，错误码：" + code + "。", ResourceType: "upload_batch", ResourcePublicID: id}, e
|
|
}},
|
|
{`SELECT fa.tenant_id::text,fr.public_id::text,count(*) FROM firmware_assignment fa JOIN firmware_rollout fr ON fr.id=fa.rollout_id WHERE fa.status='FAILED' GROUP BY fa.tenant_id,fr.public_id`, func(r pgx.Row) (Signal, error) {
|
|
var tenant, id string
|
|
var count int64
|
|
e := r.Scan(&tenant, &id, &count)
|
|
return Signal{TenantID: tenant, RuleCode: "FIRMWARE_ROLLOUT_FAILURE", Category: "TECHNICAL", Severity: "HIGH", Fingerprint: "firmware-rollout:" + id, Title: "固件灰度出现失败设备", Summary: fmt.Sprintf("固件灰度批次已有%d台设备失败，应暂停后续分配并核查回滚条件。", count), ResourceType: "firmware_rollout", ResourcePublicID: id}, e
|
|
}},
|
|
{`WITH spend AS (SELECT qp.tenant_id,qp.id,qp.currency,qp.daily_limit_micros,COALESCE(SUM(ccl.cost_micros) FILTER(WHERE cue.occurred_at>=date_trunc('day',now() AT TIME ZONE qp.billing_timezone) AT TIME ZONE qp.billing_timezone),0)::bigint used FROM usage_quota_policy qp LEFT JOIN conversation_cost_line ccl ON ccl.tenant_id=qp.tenant_id AND ccl.currency=qp.currency LEFT JOIN conversation_usage_event cue ON cue.id=ccl.usage_event_id WHERE qp.scope_type='TENANT' AND qp.status='ACTIVE' GROUP BY qp.tenant_id,qp.id) SELECT tenant_id::text,currency,used,daily_limit_micros FROM spend WHERE used*100>=daily_limit_micros*80`, func(r pgx.Row) (Signal, error) {
|
|
var tenant, currency string
|
|
var used, limit int64
|
|
e := r.Scan(&tenant, ¤cy, &used, &limit)
|
|
return Signal{TenantID: tenant, RuleCode: "COST_DAILY_QUOTA_NEAR", Category: "COST", Severity: "MEDIUM", Fingerprint: "cost-daily-quota", Title: "日费用接近硬额度", Summary: fmt.Sprintf("本账期已使用%s %.6f，达到日额度的80%%以上。", currency, float64(used)/1e6), ResourceType: "usage_quota_policy"}, e
|
|
}},
|
|
{`SELECT tenant_id::text,public_id::text,status FROM conversation WHERE (status='STARTING_AGENT' AND updated_at<now()-interval '5 minutes') OR (status='STOPPING' AND updated_at<now()-interval '5 minutes')`, func(r pgx.Row) (Signal, error) {
|
|
var tenant, id, status string
|
|
e := r.Scan(&tenant, &id, &status)
|
|
return Signal{TenantID: tenant, RuleCode: "VOICE_TASK_STUCK", Category: "TECHNICAL", Severity: "HIGH", Fingerprint: "voice-task-stuck:" + id, Title: "语音任务长时间停留在中间状态", Summary: "通话持续停留在" + status + "，自动恢复未及时收敛，需要检查供应商任务和恢复记录。", ResourceType: "conversation", ResourcePublicID: id}, e
|
|
}},
|
|
{`SELECT tenant_id::text,provider,COALESCE(last_error_code,'PROVIDER_TEMPORARY_ERROR'),consecutive_failures FROM provider_dependency_state WHERE status='UNAVAILABLE'`, func(r pgx.Row) (Signal, error) {
|
|
var tenant, provider, code string
|
|
var failures int
|
|
e := r.Scan(&tenant, &provider, &code, &failures)
|
|
return Signal{TenantID: tenant, RuleCode: "PROVIDER_UNAVAILABLE", Category: "TECHNICAL", Severity: "CRITICAL", Fingerprint: "provider-unavailable:" + provider, Title: "语音供应商暂时不可用", Summary: fmt.Sprintf("供应商%s连续临时失败%d次，错误码%s；新启动已暂停，停止与回收仍继续重试。", provider, failures, code), ResourceType: "provider_dependency"}, e
|
|
}},
|
|
{`SELECT tenant_id::text,provider,count(*),min(received_at) FROM provider_event WHERE processing_status IN ('pending','failed') AND received_at<now()-interval '1 minute' GROUP BY tenant_id,provider`, func(r pgx.Row) (Signal, error) {
|
|
var tenant, provider string
|
|
var count int64
|
|
var oldest time.Time
|
|
e := r.Scan(&tenant, &provider, &count, &oldest)
|
|
return Signal{TenantID: tenant, RuleCode: "PROVIDER_NOTIFICATION_BACKLOG", Category: "TECHNICAL", Severity: "HIGH", Fingerprint: "provider-notification-backlog:" + provider, Title: "供应商状态通知处理积压", Summary: fmt.Sprintf("供应商%s有%d条通知超过1分钟仍未完成处理，最早接收时间为%s。", provider, count, oldest.UTC().Format(time.RFC3339)), ResourceType: "provider_event_queue"}, e
|
|
}},
|
|
{`SELECT tenant_id::text,public_id::text,display_name,expires_at FROM ops_resource_reference WHERE resource_type='CERTIFICATE' AND status='ACTIVE' AND expires_at<=now()+interval '30 days'`, func(r pgx.Row) (Signal, error) {
|
|
var tenant, id, name string
|
|
var expires time.Time
|
|
e := r.Scan(&tenant, &id, &name, &expires)
|
|
severity := "HIGH"
|
|
if !expires.After(now.Add(7 * 24 * time.Hour)) {
|
|
severity = "CRITICAL"
|
|
}
|
|
return Signal{TenantID: tenant, RuleCode: "CERTIFICATE_EXPIRING", Category: "SECURITY", Severity: severity, Fingerprint: "certificate-expiring:" + id, Title: "受控证书即将到期", Summary: fmt.Sprintf("证书%s将在%s到期，请按轮换手册更新引用并验证回退路径。", name, expires.UTC().Format(time.RFC3339)), ResourceType: "ops_resource_reference", ResourcePublicID: id}, e
|
|
}},
|
|
}
|
|
for _, q := range queries {
|
|
rows, err := s.pool.Query(ctx, q.sql)
|
|
if err != nil {
|
|
return len(signals), err
|
|
}
|
|
for rows.Next() {
|
|
signal, scanErr := q.build(rows)
|
|
if scanErr != nil {
|
|
rows.Close()
|
|
return len(signals), scanErr
|
|
}
|
|
signals = append(signals, signal)
|
|
}
|
|
if err = rows.Err(); err != nil {
|
|
rows.Close()
|
|
return len(signals), err
|
|
}
|
|
rows.Close()
|
|
}
|
|
processed := 0
|
|
for _, signal := range signals {
|
|
if err := s.detect(ctx, signal, now); err != nil {
|
|
return processed, err
|
|
}
|
|
processed++
|
|
}
|
|
if err := s.escalate(ctx, now); err != nil {
|
|
return processed, err
|
|
}
|
|
return processed, nil
|
|
}
|
|
|
|
func (s *Scanner) detect(ctx context.Context, signal Signal, now time.Time) error {
|
|
signal.RuleCode = strings.ToUpper(strings.TrimSpace(signal.RuleCode))
|
|
if !rulePattern.MatchString(signal.RuleCode) || !validCategory(signal.Category) || !validSeverity(signal.Severity) || signal.Fingerprint == "" || signal.Title == "" || signal.Summary == "" {
|
|
return ErrInvalid
|
|
}
|
|
tx, err := s.pool.BeginTx(ctx, pgx.TxOptions{})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback(context.Background())
|
|
if _, err = tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,0))`, signal.TenantID+":"+signal.Fingerprint); err != nil {
|
|
return err
|
|
}
|
|
var tenantPublicID string
|
|
if err = tx.QueryRow(ctx, `SELECT public_id::text FROM tenant WHERE id=$1::uuid`, signal.TenantID).Scan(&tenantPublicID); err != nil {
|
|
return err
|
|
}
|
|
var routeID, receiverGroup, runbookRef *string
|
|
_ = tx.QueryRow(ctx, `SELECT id::text,receiver_group,runbook_ref FROM alert_routing_policy WHERE tenant_id=$1::uuid AND category=$2 AND severity=$3 AND status='ACTIVE'`, signal.TenantID, signal.Category, signal.Severity).Scan(&routeID, &receiverGroup, &runbookRef)
|
|
var silenceUntil *time.Time
|
|
err = tx.QueryRow(ctx, `SELECT max(ends_at) FROM alert_silence WHERE tenant_id=$1::uuid AND revoked_at IS NULL AND starts_at<=$2 AND ends_at>$2 AND (rule_code IS NULL OR rule_code=$3) AND (category IS NULL OR category=$4) AND (resource_public_id IS NULL OR resource_public_id=$5::uuid)`, signal.TenantID, now, signal.RuleCode, signal.Category, nullUUID(signal.ResourcePublicID)).Scan(&silenceUntil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
status := "OPEN"
|
|
if silenceUntil != nil {
|
|
status = "SILENCED"
|
|
}
|
|
var incidentID, previousStatus string
|
|
err = tx.QueryRow(ctx, `SELECT id::text,status FROM alert_incident WHERE tenant_id=$1::uuid AND fingerprint=$2 AND status<>'RESOLVED' FOR UPDATE`, signal.TenantID, signal.Fingerprint).Scan(&incidentID, &previousStatus)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
incidentPublicID := uuid.NewString()
|
|
err = tx.QueryRow(ctx, `INSERT INTO alert_incident(public_id,tenant_id,rule_code,category,severity,status,fingerprint,title,summary,resource_type,resource_public_id,correlation_request_id,routing_policy_id,first_detected_at,last_detected_at,silenced_until) VALUES($1::uuid,$2::uuid,$3,$4,$5,$6,$7,$8,$9,NULLIF($10,''),$11::uuid,NULLIF($12,''),$13::uuid,$14,$14,$15) RETURNING id::text`, incidentPublicID, signal.TenantID, signal.RuleCode, signal.Category, signal.Severity, status, signal.Fingerprint, signal.Title, signal.Summary, signal.ResourceType, nullUUID(signal.ResourcePublicID), signal.RequestID, routeID, now, silenceUntil).Scan(&incidentID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = tx.Exec(ctx, `INSERT INTO alert_action(tenant_id,incident_id,action_type,to_status,actor_type,note) VALUES($1::uuid,$2::uuid,'DETECTED',$3,'system',$4)`, signal.TenantID, incidentID, status, "监测规则 "+signal.RuleCode+" 首次命中")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if status == "SILENCED" {
|
|
if _, err = tx.Exec(ctx, `INSERT INTO alert_action(tenant_id,incident_id,action_type,to_status,actor_type,note) VALUES($1::uuid,$2::uuid,'SILENCED','SILENCED','system','首次命中时已匹配有效静默策略')`, signal.TenantID, incidentID); err != nil {
|
|
return err
|
|
}
|
|
} else if err = s.enqueueNotification(ctx, tx, alertNotification{
|
|
Action: "DETECTED", IncidentID: incidentPublicID, TenantID: tenantPublicID,
|
|
RuleCode: signal.RuleCode, Category: signal.Category, Severity: signal.Severity,
|
|
Status: status, Title: signal.Title, Summary: signal.Summary,
|
|
ResourceType: signal.ResourceType, ResourceID: signal.ResourcePublicID,
|
|
ReceiverGroup: valueOrEmpty(receiverGroup), RunbookRef: valueOrEmpty(runbookRef),
|
|
OccurredAt: now,
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
return tx.Commit(ctx)
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
nextStatus := previousStatus
|
|
if silenceUntil != nil {
|
|
nextStatus = "SILENCED"
|
|
} else if previousStatus == "SILENCED" {
|
|
nextStatus = "OPEN"
|
|
}
|
|
_, err = tx.Exec(ctx, `UPDATE alert_incident SET status=$2,last_detected_at=$3,occurrence_count=occurrence_count+1,summary=$4,routing_policy_id=COALESCE($5::uuid,routing_policy_id),silenced_until=$6,row_version=row_version+1,updated_at=$3 WHERE id=$1::uuid`, incidentID, nextStatus, now, signal.Summary, routeID, silenceUntil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if nextStatus != previousStatus {
|
|
action := "REOPENED"
|
|
note := "告警静默窗口已结束，重新进入待处置队列"
|
|
if nextStatus == "SILENCED" {
|
|
action = "SILENCED"
|
|
note = "告警命中有效静默策略"
|
|
}
|
|
if _, err = tx.Exec(ctx, `INSERT INTO alert_action(tenant_id,incident_id,action_type,from_status,to_status,actor_type,note) VALUES($1::uuid,$2::uuid,$3,$4,$5,'system',$6)`, signal.TenantID, incidentID, action, previousStatus, nextStatus, note); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return tx.Commit(ctx)
|
|
}
|
|
|
|
func (s *Scanner) escalate(ctx context.Context, now time.Time) error {
|
|
tx, err := s.pool.BeginTx(ctx, pgx.TxOptions{})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback(context.Background())
|
|
rows, err := tx.Query(ctx, `
|
|
WITH due AS (
|
|
SELECT ai.id,ai.tenant_id,ai.public_id,ai.rule_code,ai.category,ai.severity,ai.status,
|
|
ai.title,ai.summary,ai.resource_type,ai.resource_public_id,
|
|
t.public_id AS tenant_public_id,arp.escalation_receiver_group,arp.runbook_ref
|
|
FROM alert_incident ai
|
|
JOIN alert_routing_policy arp ON arp.id=ai.routing_policy_id
|
|
JOIN tenant t ON t.id=ai.tenant_id
|
|
WHERE ai.status IN('OPEN','ACKNOWLEDGED') AND ai.escalated_at IS NULL
|
|
AND ai.first_detected_at+(arp.escalation_after_minutes||' minutes')::interval<=$1
|
|
FOR UPDATE OF ai
|
|
), updated AS (
|
|
UPDATE alert_incident ai SET escalated_at=$1,row_version=row_version+1,updated_at=$1
|
|
FROM due WHERE ai.id=due.id
|
|
RETURNING ai.id,ai.tenant_id,ai.public_id,ai.rule_code,ai.category,ai.severity,ai.status,
|
|
ai.title,ai.summary,ai.resource_type,ai.resource_public_id
|
|
), actions AS (
|
|
INSERT INTO alert_action(tenant_id,incident_id,action_type,from_status,to_status,actor_type,note)
|
|
SELECT tenant_id,id,'ESCALATED',status,status,'system','超过升级时限，已进入升级接收组' FROM updated
|
|
RETURNING incident_id
|
|
)
|
|
SELECT u.public_id::text,d.tenant_public_id::text,u.rule_code,u.category,u.severity,u.status,
|
|
u.title,u.summary,COALESCE(u.resource_type,''),COALESCE(u.resource_public_id::text,''),
|
|
d.escalation_receiver_group,d.runbook_ref
|
|
FROM updated u JOIN due d ON d.id=u.id JOIN actions a ON a.incident_id=u.id`, now)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
notifications := []alertNotification{}
|
|
for rows.Next() {
|
|
var item alertNotification
|
|
item.Action = "ESCALATED"
|
|
item.OccurredAt = now
|
|
if err = rows.Scan(&item.IncidentID, &item.TenantID, &item.RuleCode, &item.Category, &item.Severity, &item.Status, &item.Title, &item.Summary, &item.ResourceType, &item.ResourceID, &item.ReceiverGroup, &item.RunbookRef); err != nil {
|
|
rows.Close()
|
|
return err
|
|
}
|
|
notifications = append(notifications, item)
|
|
}
|
|
if err = rows.Err(); err != nil {
|
|
rows.Close()
|
|
return err
|
|
}
|
|
rows.Close()
|
|
for _, item := range notifications {
|
|
if err = s.enqueueNotification(ctx, tx, item); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return tx.Commit(ctx)
|
|
}
|
|
|
|
type alertNotification struct {
|
|
Action string `json:"action"`
|
|
IncidentID string `json:"incident_id"`
|
|
TenantID string `json:"tenant_id"`
|
|
RuleCode string `json:"rule_code"`
|
|
Category string `json:"category"`
|
|
Severity string `json:"severity"`
|
|
Status string `json:"status"`
|
|
Title string `json:"title"`
|
|
Summary string `json:"summary"`
|
|
ResourceType string `json:"resource_type,omitempty"`
|
|
ResourceID string `json:"resource_id,omitempty"`
|
|
ReceiverGroup string `json:"receiver_group,omitempty"`
|
|
RunbookRef string `json:"runbook_ref,omitempty"`
|
|
OccurredAt time.Time `json:"occurred_at"`
|
|
}
|
|
|
|
func (s *Scanner) enqueueNotification(ctx context.Context, tx pgx.Tx, item alertNotification) error {
|
|
payload, err := json.Marshal(item)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
tenantInternalID := item.TenantID
|
|
if err = tx.QueryRow(ctx, `SELECT id::text FROM tenant WHERE public_id=$1::uuid`, item.TenantID).Scan(&tenantInternalID); err != nil {
|
|
return err
|
|
}
|
|
deduplicationKey := "alert-webhook:" + strings.ToLower(item.Action) + ":" + item.IncidentID
|
|
_, err = s.outbox.Enqueue(ctx, tx, outbox.NewEvent{
|
|
TenantID: &tenantInternalID, AggregateType: "alert_incident", AggregateID: item.IncidentID,
|
|
EventType: "alert.webhook_requested", DeduplicationKey: &deduplicationKey, Payload: payload,
|
|
})
|
|
return err
|
|
}
|
|
|
|
func valueOrEmpty(value *string) string {
|
|
if value == nil {
|
|
return ""
|
|
}
|
|
return *value
|
|
}
|
|
|
|
func nullUUID(value string) any {
|
|
if strings.TrimSpace(value) == "" {
|
|
return nil
|
|
}
|
|
return value
|
|
}
|