327 lines
15 KiB
Go
327 lines
15 KiB
Go
package ops_alert
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
"unicode/utf8"
|
|
|
|
"github.com/google/uuid"
|
|
"gorm.io/gorm"
|
|
"gorm.io/gorm/clause"
|
|
|
|
deviceModels "git.ilapage.cn/ila/yovision/Sense/server/app/sense/device/models"
|
|
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/edge_node"
|
|
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/media"
|
|
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/media_shard"
|
|
)
|
|
|
|
var (
|
|
ErrAlertNotFound = errors.New("运维告警不存在")
|
|
ErrInvalidFilter = errors.New("告警筛选条件不正确")
|
|
ErrInvalidReason = errors.New("处理说明需为 6 至 256 个字符")
|
|
ErrInvalidTransition = errors.New("当前告警状态不允许此操作")
|
|
ErrVersionConflict = errors.New("告警状态已变化，请刷新后重试")
|
|
ErrRecoveryWindowOpen = errors.New("健康恢复观察窗口尚未结束")
|
|
)
|
|
|
|
const defaultRecoveryWindow = 5 * time.Minute
|
|
|
|
type Service struct {
|
|
DB *gorm.DB
|
|
Now func() time.Time
|
|
RecoveryWindow time.Duration
|
|
}
|
|
|
|
func NewService(db *gorm.DB) *Service {
|
|
return &Service{DB: db, Now: time.Now, RecoveryWindow: defaultRecoveryWindow}
|
|
}
|
|
|
|
func (s *Service) List(request PageRequest) (PageResponse, error) {
|
|
if !validOptional(request.AlertType, alertTypes()) || !validOptional(request.State, states()) {
|
|
return PageResponse{}, ErrInvalidFilter
|
|
}
|
|
query := s.DB.Model(&Alert{})
|
|
if request.AlertType != "" {
|
|
query = query.Where("alert_type = ?", request.AlertType)
|
|
}
|
|
if request.State != "" {
|
|
query = query.Where("state = ?", request.State)
|
|
}
|
|
if keyword := strings.TrimSpace(request.Keyword); keyword != "" {
|
|
like := "%" + keyword + "%"
|
|
query = query.Where("object_name LIKE ? OR location LIKE ? OR title LIKE ? OR detail LIKE ?", like, like, like, like)
|
|
}
|
|
response := PageResponse{}
|
|
if err := query.Count(&response.Count).Error; err != nil {
|
|
return response, err
|
|
}
|
|
if err := query.Order("CASE WHEN state = 'unacknowledged' THEN 0 WHEN state = 'acknowledged' THEN 1 WHEN state = 'recovering' THEN 2 ELSE 3 END, last_seen_at DESC").Offset((request.GetPageIndex() - 1) * request.GetPageSize()).Limit(request.GetPageSize()).Find(&response.List).Error; err != nil {
|
|
return response, err
|
|
}
|
|
active := []string{StateUnacknowledged, StateAcknowledged, StateRecovering}
|
|
if err := s.DB.Model(&Alert{}).Where("state IN ?", active).Count(&response.Summary.Active).Error; err != nil {
|
|
return response, err
|
|
}
|
|
if err := s.DB.Model(&Alert{}).Where("state = ?", StateUnacknowledged).Count(&response.Summary.Unacknowledged).Error; err != nil {
|
|
return response, err
|
|
}
|
|
if err := s.DB.Model(&Alert{}).Where("state = ?", StateRecovering).Count(&response.Summary.Recovering).Error; err != nil {
|
|
return response, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (s *Service) Get(id string) (DetailResponse, error) {
|
|
response := DetailResponse{Transitions: []Transition{}}
|
|
if err := s.DB.First(&response.Alert, "id = ?", id).Error; errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return response, ErrAlertNotFound
|
|
} else if err != nil {
|
|
return response, err
|
|
}
|
|
if err := s.DB.Where("alert_id = ?", id).Order("created_at DESC").Find(&response.Transitions).Error; err != nil {
|
|
return response, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (s *Service) Acknowledge(ctx context.Context, id string, request ActionRequest, actor int) (Alert, error) {
|
|
return s.transition(ctx, id, request, actor, StateUnacknowledged, StateAcknowledged, "acknowledge")
|
|
}
|
|
|
|
func (s *Service) Recover(ctx context.Context, id string, request ActionRequest, actor int) (Alert, error) {
|
|
return s.transition(ctx, id, request, actor, StateRecovering, StateRecovered, "recover")
|
|
}
|
|
|
|
func (s *Service) transition(ctx context.Context, id string, request ActionRequest, actor int, expectedState, targetState, action string) (Alert, error) {
|
|
reason := strings.TrimSpace(request.Reason)
|
|
if utf8.RuneCountInString(reason) < 6 || utf8.RuneCountInString(reason) > 256 {
|
|
return Alert{}, ErrInvalidReason
|
|
}
|
|
if request.ExpectedVersion < 1 {
|
|
return Alert{}, ErrVersionConflict
|
|
}
|
|
now := s.Now().UTC()
|
|
result := Alert{}
|
|
err := s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
var current Alert
|
|
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(¤t, "id = ?", id).Error; errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return ErrAlertNotFound
|
|
} else if err != nil {
|
|
return err
|
|
}
|
|
if current.Version != request.ExpectedVersion {
|
|
return ErrVersionConflict
|
|
}
|
|
if current.State != expectedState {
|
|
return ErrInvalidTransition
|
|
}
|
|
if action == "recover" && (current.HealthySince == nil || now.Sub(current.HealthySince.UTC()) < s.RecoveryWindow) {
|
|
return ErrRecoveryWindowOpen
|
|
}
|
|
updates := map[string]any{"state": targetState, "version": current.Version + 1, "updated_at": now}
|
|
if action == "acknowledge" {
|
|
updates["acknowledged_at"], updates["acknowledged_by"] = now, actor
|
|
} else {
|
|
updates["recovered_at"], updates["recovered_by"] = now, actor
|
|
}
|
|
write := tx.Model(&Alert{}).Where("id = ? AND version = ? AND state = ?", id, current.Version, expectedState).Updates(updates)
|
|
if write.Error != nil {
|
|
return write.Error
|
|
}
|
|
if write.RowsAffected != 1 {
|
|
return ErrVersionConflict
|
|
}
|
|
if err := tx.Create(&Transition{ID: uuid.NewString(), AlertID: id, Cycle: current.Cycle, Action: action, FromState: expectedState, ToState: targetState, Reason: reason, ActorUserID: actor, CreatedAt: now}).Error; err != nil {
|
|
return err
|
|
}
|
|
return tx.First(&result, "id = ?", id).Error
|
|
})
|
|
return result, err
|
|
}
|
|
|
|
type observation struct {
|
|
Fingerprint, AlertType, Severity, ObjectType, ObjectID, ObjectName, Location, Title, Detail, NextAction, SourceVersion string
|
|
Abnormal bool
|
|
}
|
|
|
|
func (s *Service) EvaluateSources(ctx context.Context) (EvaluateResponse, error) {
|
|
items, err := s.sourceObservations(ctx)
|
|
if err != nil {
|
|
return EvaluateResponse{}, err
|
|
}
|
|
response := EvaluateResponse{Observed: len(items)}
|
|
for _, item := range items {
|
|
if item.Abnormal {
|
|
response.Abnormal++
|
|
}
|
|
if err = s.observe(ctx, item); err != nil {
|
|
return response, err
|
|
}
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (s *Service) observe(ctx context.Context, item observation) error {
|
|
now := s.Now().UTC()
|
|
return s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
var current Alert
|
|
err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("fingerprint = ?", item.Fingerprint).First(¤t).Error
|
|
if errors.Is(err, gorm.ErrRecordNotFound) {
|
|
if !item.Abnormal {
|
|
return nil
|
|
}
|
|
current = Alert{ID: uuid.NewString(), Fingerprint: item.Fingerprint, AlertType: item.AlertType, Severity: item.Severity, ObjectType: item.ObjectType, ObjectID: item.ObjectID, ObjectName: item.ObjectName, Location: item.Location, State: StateUnacknowledged, Title: item.Title, Detail: item.Detail, NextAction: item.NextAction, LastSourceVersion: item.SourceVersion, OccurrenceCount: 1, Cycle: 1, FirstSeenAt: now, LastSeenAt: now, OperationalOnly: true, Version: 1, CreatedAt: now, UpdatedAt: now}
|
|
if err = tx.Create(¤t).Error; err != nil {
|
|
return err
|
|
}
|
|
return tx.Create(&Transition{ID: uuid.NewString(), AlertID: current.ID, Cycle: 1, Action: "detected", ToState: StateUnacknowledged, Reason: "健康事实首次满足告警规则", CreatedAt: now}).Error
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if item.Abnormal {
|
|
updates := map[string]any{"severity": item.Severity, "object_name": item.ObjectName, "location": item.Location, "title": item.Title, "detail": item.Detail, "next_action": item.NextAction, "last_seen_at": now, "updated_at": now, "healthy_since": nil}
|
|
if item.SourceVersion != current.LastSourceVersion {
|
|
updates["last_source_version"], updates["occurrence_count"] = item.SourceVersion, current.OccurrenceCount+1
|
|
}
|
|
if current.State == StateRecovered {
|
|
updates["state"], updates["cycle"], updates["first_seen_at"], updates["occurrence_count"], updates["acknowledged_at"], updates["acknowledged_by"], updates["recovered_at"], updates["recovered_by"] = StateUnacknowledged, current.Cycle+1, now, 1, nil, 0, nil, 0
|
|
if err = tx.Create(&Transition{ID: uuid.NewString(), AlertID: current.ID, Cycle: current.Cycle + 1, Action: "reopened", FromState: StateRecovered, ToState: StateUnacknowledged, Reason: "健康事实再次异常", CreatedAt: now}).Error; err != nil {
|
|
return err
|
|
}
|
|
} else if current.State == StateRecovering {
|
|
target := StateUnacknowledged
|
|
if current.AcknowledgedAt != nil {
|
|
target = StateAcknowledged
|
|
}
|
|
updates["state"] = target
|
|
if err = tx.Create(&Transition{ID: uuid.NewString(), AlertID: current.ID, Cycle: current.Cycle, Action: "relapsed", FromState: StateRecovering, ToState: target, Reason: "恢复观察期间再次异常", CreatedAt: now}).Error; err != nil {
|
|
return err
|
|
}
|
|
}
|
|
updates["version"] = current.Version + 1
|
|
return tx.Model(&Alert{}).Where("id = ? AND version = ?", current.ID, current.Version).Updates(updates).Error
|
|
}
|
|
if current.State == StateUnacknowledged || current.State == StateAcknowledged {
|
|
if err = tx.Create(&Transition{ID: uuid.NewString(), AlertID: current.ID, Cycle: current.Cycle, Action: "health_restored", FromState: current.State, ToState: StateRecovering, Reason: "健康事实已恢复，进入稳定观察窗口", CreatedAt: now}).Error; err != nil {
|
|
return err
|
|
}
|
|
return tx.Model(&Alert{}).Where("id = ? AND version = ?", current.ID, current.Version).Updates(map[string]any{"state": StateRecovering, "healthy_since": now, "version": current.Version + 1, "updated_at": now}).Error
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
|
|
type admissionProjection struct {
|
|
DeviceID, Status, Detail string
|
|
CheckedAt time.Time
|
|
}
|
|
|
|
func (admissionProjection) TableName() string { return "sense_admission_results" }
|
|
|
|
func (s *Service) sourceObservations(ctx context.Context) ([]observation, error) {
|
|
now := s.Now().UTC()
|
|
result := []observation{}
|
|
var devices []deviceModels.Device
|
|
if err := s.DB.WithContext(ctx).Find(&devices).Error; err != nil {
|
|
return nil, err
|
|
}
|
|
var admissions []admissionProjection
|
|
if err := s.DB.WithContext(ctx).Find(&admissions).Error; err != nil && !isMissingTable(err) {
|
|
return nil, err
|
|
}
|
|
byDevice := map[string]admissionProjection{}
|
|
for _, item := range admissions {
|
|
byDevice[item.DeviceID] = item
|
|
}
|
|
for _, device := range devices {
|
|
if device.Status == deviceModels.StatusDisabled {
|
|
continue
|
|
}
|
|
state, detail, observed := device.AdapterStatus, "尚未获得设备接入结果", device.UpdatedAt
|
|
if item, ok := byDevice[device.ID]; ok {
|
|
state, detail, observed = item.Status, item.Detail, item.CheckedAt
|
|
}
|
|
text := strings.ToLower(state + " " + detail)
|
|
version := fmt.Sprintf("%d:%d", device.Version, observed.UnixNano())
|
|
base := observation{ObjectType: "device", ObjectID: device.ID, ObjectName: device.Name, Location: device.Location, SourceVersion: version}
|
|
result = append(result,
|
|
withRule(base, TypeDeviceOffline, "high", "设备离线", detail, "检查设备供电、网络和地址后重新发现", containsAny(text, "offline", "unreachable", "timeout", "离线", "无法连接", "超时")),
|
|
withRule(base, TypeAuthentication, "high", "设备认证失败", detail, "更新设备凭据后重新执行接入验证", containsAny(text, "authentication_failed", "unauthorized", "auth", "认证", "凭据")),
|
|
withRule(base, TypeClockDrift, "medium", "设备时间漂移", detail, "校准设备时间后重新执行接入验证", containsAny(text, "clock_skew", "clock_drift", "time drift", "时间漂移", "时钟")),
|
|
)
|
|
}
|
|
var routes []media.Route
|
|
if err := s.DB.WithContext(ctx).Find(&routes).Error; err != nil && !isMissingTable(err) {
|
|
return nil, err
|
|
}
|
|
for _, route := range routes {
|
|
converged := (route.Desired == media.DesiredRunning && (route.Actual == "ready" || route.Actual == "waiting")) || (route.Desired == media.DesiredStopped && route.Actual == "stopped")
|
|
base := observation{ObjectType: "media_route", ObjectID: route.ID, ObjectName: route.Path, SourceVersion: fmt.Sprintf("%d", route.Version)}
|
|
result = append(result, withRule(base, TypeReconciliation, "medium", "媒体状态未收敛", route.Detail, "在运维中心核对期望态并执行受控重试", !converged))
|
|
}
|
|
var shards []media_shard.Shard
|
|
if err := s.DB.WithContext(ctx).Find(&shards).Error; err != nil && !isMissingTable(err) {
|
|
return nil, err
|
|
}
|
|
for _, shard := range shards {
|
|
if shard.Status == media_shard.StatusDisabled {
|
|
continue
|
|
}
|
|
stale := shard.LastProbeAt == nil || now.Sub(shard.LastProbeAt.UTC()) > 30*time.Second
|
|
base := observation{ObjectType: "media_shard", ObjectID: shard.ID, ObjectName: shard.Name, SourceVersion: fmt.Sprintf("%d:%d", shard.ConfigVersion, timeValue(shard.LastProbeAt))}
|
|
result = append(result, withRule(base, TypeMediaShard, "high", "媒体分片异常", shard.Detail, "检查 MediaMTX 进程和 Control API 后重新探测", shard.Status == media_shard.StatusFailed || stale))
|
|
}
|
|
var nodes []edge_node.Node
|
|
if err := s.DB.WithContext(ctx).Find(&nodes).Error; err != nil && !isMissingTable(err) {
|
|
return nil, err
|
|
}
|
|
for _, node := range nodes {
|
|
base := observation{ObjectType: "edge_node", ObjectID: node.ID, ObjectName: node.Name, Location: node.Location, SourceVersion: fmt.Sprintf("%d:%d", node.ProjectionVersion, node.LastHeartbeatAt.UnixNano())}
|
|
result = append(result,
|
|
withRule(base, TypeDeviceOffline, "high", "边缘节点离线", "心跳超过 90 秒未更新", "检查节点进程、网络和机器身份后等待心跳恢复", now.Sub(node.LastHeartbeatAt.UTC()) > edge_node.HeartbeatTimeout),
|
|
withRule(base, TypeControlTunnel, "high", "控制隧道异常", node.ControlTunnelDetail, "检查节点到 Sense 的控制通道，不影响业务预警数据", node.ControlTunnelStatus != edge_node.ChannelReady),
|
|
)
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func withRule(base observation, typ, severity, title, detail, next string, abnormal bool) observation {
|
|
base.Fingerprint = typ + ":" + base.ObjectType + ":" + base.ObjectID
|
|
base.AlertType = typ
|
|
base.Severity = severity
|
|
base.Title = title
|
|
base.Detail = detail
|
|
base.NextAction = next
|
|
base.Abnormal = abnormal
|
|
return base
|
|
}
|
|
func containsAny(value string, needles ...string) bool {
|
|
for _, needle := range needles {
|
|
if strings.Contains(value, needle) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
func isMissingTable(err error) bool {
|
|
text := strings.ToLower(err.Error())
|
|
return strings.Contains(text, "no such table") || strings.Contains(text, "does not exist")
|
|
}
|
|
func timeValue(value *time.Time) int64 {
|
|
if value == nil {
|
|
return 0
|
|
}
|
|
return value.UnixNano()
|
|
}
|
|
func validOptional(value string, allowed map[string]bool) bool { return value == "" || allowed[value] }
|
|
func states() map[string]bool {
|
|
return map[string]bool{StateUnacknowledged: true, StateAcknowledged: true, StateRecovering: true, StateRecovered: true}
|
|
}
|
|
func alertTypes() map[string]bool {
|
|
return map[string]bool{TypeDeviceOffline: true, TypeAuthentication: true, TypeClockDrift: true, TypeReconciliation: true, TypeMediaShard: true, TypeControlTunnel: true}
|
|
}
|