diff --git a/Bell/README.md b/Bell/README.md index f4317e0..7d71e20 100644 --- a/Bell/README.md +++ b/Bell/README.md @@ -66,6 +66,10 @@ When `BELL_ENV=development` or `test`, administrators can open “合成事件 In `BELL_ENV=production` those routes are not registered and return `404`; hiding the menu is not the security boundary. +### Rules and alerts + +Administrators configure and enable rules from “规则配置”. Each accepted Event is evaluated in the same database transaction that creates its Event/Receipt. A matching Event creates or joins an open Alert correlated by rule and location; one Event may match several rules and one Alert may collect several related Events. Every evaluation records its rule version and explanation, while every match also stores the actual rule snapshot. Unmatched Events remain visible as Events and are never labelled as notified or handled. + ## Independent smoke check Run PostgreSQL with an empty Bell-only database, set `BELL_DATABASE_URL`, start the backend and then the web client. Keep Sense and Brain stopped. Verify `/healthz`, `/readyz`, the workbench shell, and a direct request to an unregistered framework demo route returns `404`. diff --git a/Bell/server/app/alert/model/types.go b/Bell/server/app/alert/model/types.go new file mode 100644 index 0000000..a90bfdc --- /dev/null +++ b/Bell/server/app/alert/model/types.go @@ -0,0 +1,36 @@ +package model + +import "encoding/json" + +type Alert struct { + ID string `json:"id"` + Status string `json:"status"` + Severity string `json:"severity"` + Summary string `json:"summary"` + Location string `json:"location"` + PrimaryRuleID string `json:"primary_rule_id"` + RuleName string `json:"rule_name"` + EventCount int `json:"event_count"` + CreatedAt string `json:"created_at"` + UpdatedAt string `json:"updated_at"` +} +type LinkedEvent struct { + ID string `json:"id"` + EventType string `json:"event_type"` + OccurredAt string `json:"occurred_at"` + Location string `json:"location"` + Severity string `json:"severity"` +} +type Match struct { + EventID string `json:"event_id"` + RuleID string `json:"rule_id"` + RuleVersion int `json:"rule_version"` + RuleSnapshot json.RawMessage `json:"rule_snapshot"` + Explanation string `json:"explanation"` + MatchedAt string `json:"matched_at"` +} +type Detail struct { + Alert Alert `json:"alert"` + Events []LinkedEvent `json:"events"` + Matches []Match `json:"matches"` +} diff --git a/Bell/server/app/alert/query/http.go b/Bell/server/app/alert/query/http.go new file mode 100644 index 0000000..ed3fe0b --- /dev/null +++ b/Bell/server/app/alert/query/http.go @@ -0,0 +1,61 @@ +package query + +import ( + "encoding/json" + "errors" + "net/http" + "strconv" + + "git.ilapage.cn/ila/yovision/Bell/server/app/auth" + "git.ilapage.cn/ila/yovision/Bell/server/app/rbac" + "github.com/jackc/pgx/v5" +) + +type HTTP struct { + Service Service + Auth auth.HTTP +} + +func (h HTTP) Register(mux *http.ServeMux) { + mux.Handle("GET /api/v1/alerts", h.Auth.Require(rbac.AlertsRead, http.HandlerFunc(h.list))) + mux.Handle("GET /api/v1/alerts/{id}", h.Auth.Require(rbac.AlertsRead, http.HandlerFunc(h.get))) + mux.Handle("GET /api/v1/events/{id}/alerts", h.Auth.Require(rbac.EventsRead, http.HandlerFunc(h.forEvent))) +} +func (h HTTP) list(w http.ResponseWriter, r *http.Request) { + limit, _ := strconv.Atoi(r.URL.Query().Get("limit")) + items, err := h.Service.List(r.Context(), limit, r.URL.Query().Get("before")) + if err != nil { + writeJSON(w, 500, map[string]string{"error": "读取预警失败"}) + return + } + next := "" + if len(items) > 0 { + next = items[len(items)-1].ID + } + writeJSON(w, 200, map[string]any{"items": items, "next": next}) +} +func (h HTTP) get(w http.ResponseWriter, r *http.Request) { + item, err := h.Service.Get(r.Context(), r.PathValue("id")) + if err != nil { + status := 500 + if errors.Is(err, pgx.ErrNoRows) { + status = 404 + } + writeJSON(w, status, map[string]string{"error": "预警不存在"}) + return + } + writeJSON(w, 200, item) +} +func (h HTTP) forEvent(w http.ResponseWriter, r *http.Request) { + items, err := h.Service.ForEvent(r.Context(), r.PathValue("id")) + if err != nil { + writeJSON(w, 500, map[string]string{"error": "读取关联预警失败"}) + return + } + writeJSON(w, 200, map[string]any{"items": items}) +} +func writeJSON(w http.ResponseWriter, status int, value any) { + w.Header().Set("Content-Type", "application/json; charset=utf-8") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(value) +} diff --git a/Bell/server/app/alert/query/service.go b/Bell/server/app/alert/query/service.go new file mode 100644 index 0000000..0b7209c --- /dev/null +++ b/Bell/server/app/alert/query/service.go @@ -0,0 +1,86 @@ +package query + +import ( + "context" + + "git.ilapage.cn/ila/yovision/Bell/server/app/alert/model" + "github.com/jackc/pgx/v5/pgxpool" +) + +type Service struct{ DB *pgxpool.Pool } + +const base = `SELECT a.id::text,a.status,a.severity,a.summary,a.location,a.primary_rule_id::text,r.name,(SELECT count(*) FROM bell_alert_events ae WHERE ae.alert_id=a.id),a.created_at::text,a.updated_at::text FROM bell_alerts a JOIN bell_rules r ON r.id=a.primary_rule_id` + +func scanAlert(row interface{ Scan(...any) error }) (model.Alert, error) { + var item model.Alert + err := row.Scan(&item.ID, &item.Status, &item.Severity, &item.Summary, &item.Location, &item.PrimaryRuleID, &item.RuleName, &item.EventCount, &item.CreatedAt, &item.UpdatedAt) + return item, err +} +func (s Service) List(ctx context.Context, limit int, before string) ([]model.Alert, error) { + if limit < 1 || limit > 100 { + limit = 25 + } + rows, err := s.DB.Query(ctx, base+` WHERE ($2='' OR (a.created_at,a.id)<((SELECT created_at FROM bell_alerts WHERE id=nullif($2,'')::uuid),nullif($2,'')::uuid)) ORDER BY a.created_at DESC,a.id DESC LIMIT $1`, limit, before) + if err != nil { + return nil, err + } + defer rows.Close() + items := []model.Alert{} + for rows.Next() { + item, err := scanAlert(rows) + if err != nil { + return nil, err + } + items = append(items, item) + } + return items, rows.Err() +} +func (s Service) Get(ctx context.Context, id string) (model.Detail, error) { + item, err := scanAlert(s.DB.QueryRow(ctx, base+` WHERE a.id=$1`, id)) + if err != nil { + return model.Detail{}, err + } + detail := model.Detail{Alert: item, Events: []model.LinkedEvent{}, Matches: []model.Match{}} + rows, err := s.DB.Query(ctx, `SELECT e.id::text,e.event_type,e.occurred_at::text,e.location,e.severity FROM bell_alert_events ae JOIN bell_events e ON e.id=ae.event_id WHERE ae.alert_id=$1 ORDER BY e.occurred_at,e.id`, id) + if err != nil { + return model.Detail{}, err + } + for rows.Next() { + var linked model.LinkedEvent + if err := rows.Scan(&linked.ID, &linked.EventType, &linked.OccurredAt, &linked.Location, &linked.Severity); err != nil { + rows.Close() + return model.Detail{}, err + } + detail.Events = append(detail.Events, linked) + } + rows.Close() + rows, err = s.DB.Query(ctx, `SELECT event_id::text,rule_id::text,rule_version,rule_snapshot,explanation,matched_at::text FROM bell_rule_matches WHERE alert_id=$1 ORDER BY matched_at,event_id`, id) + if err != nil { + return model.Detail{}, err + } + defer rows.Close() + for rows.Next() { + var match model.Match + if err := rows.Scan(&match.EventID, &match.RuleID, &match.RuleVersion, &match.RuleSnapshot, &match.Explanation, &match.MatchedAt); err != nil { + return model.Detail{}, err + } + detail.Matches = append(detail.Matches, match) + } + return detail, rows.Err() +} +func (s Service) ForEvent(ctx context.Context, eventID string) ([]model.Alert, error) { + rows, err := s.DB.Query(ctx, base+` JOIN bell_alert_events link ON link.alert_id=a.id WHERE link.event_id=$1 ORDER BY a.created_at,a.id`, eventID) + if err != nil { + return nil, err + } + defer rows.Close() + items := []model.Alert{} + for rows.Next() { + item, err := scanAlert(rows) + if err != nil { + return nil, err + } + items = append(items, item) + } + return items, rows.Err() +} diff --git a/Bell/server/app/auth/http.go b/Bell/server/app/auth/http.go index 17016bb..2990b64 100644 --- a/Bell/server/app/auth/http.go +++ b/Bell/server/app/auth/http.go @@ -165,6 +165,7 @@ func (h HTTP) setCookie(w http.ResponseWriter, value string, expires time.Time) http.SetCookie(w, &http.Cookie{Name: CookieName, Value: value, Path: "/", HttpOnly: true, Secure: h.SecureCookie, SameSite: http.SameSiteStrictMode, Expires: expires, MaxAge: int(time.Until(expires).Seconds())}) } func principal(ctx context.Context) User { user, _ := ctx.Value(principalKey{}).(User); return user } +func Principal(ctx context.Context) User { return principal(ctx) } func decodeJSON(w http.ResponseWriter, r *http.Request, value any) error { r.Body = http.MaxBytesReader(w, r.Body, 1<<20) decoder := json.NewDecoder(r.Body) diff --git a/Bell/server/app/event/service.go b/Bell/server/app/event/service.go index 75abe21..479865c 100644 --- a/Bell/server/app/event/service.go +++ b/Bell/server/app/event/service.go @@ -15,7 +15,11 @@ import ( var ErrConflict = errors.New("幂等键已用于不同事件载荷") var ErrInvalid = errors.New("事件字段无效") -type Service struct{ DB *pgxpool.Pool } +type PersistHook func(context.Context, pgx.Tx, Event) error +type Service struct { + DB *pgxpool.Pool + AfterPersist PersistHook +} func normalize(command Command) ([]byte, [32]byte, error) { command.ProducerID = strings.TrimSpace(command.ProducerID) @@ -80,6 +84,11 @@ func (s Service) Ingest(ctx context.Context, command Command) (Result, error) { return Result{}, ErrConflict } result.Duplicate = !created + if created && s.AfterPersist != nil { + if err := s.AfterPersist(ctx, tx, result.Event); err != nil { + return Result{}, fmt.Errorf("process accepted Event: %w", err) + } + } if err = tx.Commit(ctx); err != nil { return Result{}, err } diff --git a/Bell/server/app/event/service_test.go b/Bell/server/app/event/service_test.go index e0c6fb0..ff8362d 100644 --- a/Bell/server/app/event/service_test.go +++ b/Bell/server/app/event/service_test.go @@ -11,6 +11,7 @@ import ( "git.ilapage.cn/ila/yovision/Bell/server/internal/platform" "git.ilapage.cn/ila/yovision/Bell/server/migrations" + "github.com/jackc/pgx/v5" ) func TestNormalizeRejectsUnsafeEvidence(t *testing.T) { @@ -96,4 +97,17 @@ func TestPostgresIdempotencyAndImmutability(t *testing.T) { if afterRestart.Event.ID != eventID || afterRestart.Receipt.ID != receiptID || !afterRestart.Duplicate { t.Fatalf("receipt changed after restart: %#v", afterRestart) } + failing := command + failing.SourceEventID = key + "-rollback" + failedService := Service{DB: reopened, AfterPersist: func(context.Context, pgx.Tx, Event) error { return errors.New("forced matching failure") }} + if _, err := failedService.Ingest(ctx, failing); err == nil { + t.Fatal("expected hook failure") + } + var persisted int + if err := reopened.QueryRow(ctx, `SELECT count(*) FROM bell_events WHERE producer_id=$1 AND source_event_id=$2`, failing.ProducerID, failing.SourceEventID).Scan(&persisted); err != nil { + t.Fatal(err) + } + if persisted != 0 { + t.Fatal("Event committed without atomic match") + } } diff --git a/Bell/server/app/rule/http.go b/Bell/server/app/rule/http.go new file mode 100644 index 0000000..d2988d1 --- /dev/null +++ b/Bell/server/app/rule/http.go @@ -0,0 +1,72 @@ +package rule + +import ( + "encoding/json" + "net/http" + + "git.ilapage.cn/ila/yovision/Bell/server/app/audit" + "git.ilapage.cn/ila/yovision/Bell/server/app/auth" + "git.ilapage.cn/ila/yovision/Bell/server/app/rbac" +) + +type HTTP struct { + Service Service + Auth auth.HTTP + Audit audit.Store +} + +func (h HTTP) Register(mux *http.ServeMux) { + mux.Handle("GET /api/v1/rules", h.Auth.Require(rbac.RulesRead, http.HandlerFunc(h.list))) + mux.Handle("POST /api/v1/rules", h.Auth.Require(rbac.RulesWrite, http.HandlerFunc(h.create))) + mux.Handle("PUT /api/v1/rules/{id}/enabled", h.Auth.Require(rbac.RulesWrite, http.HandlerFunc(h.setEnabled))) +} +func (h HTTP) list(w http.ResponseWriter, r *http.Request) { + items, err := h.Service.List(r.Context()) + if err != nil { + writeJSON(w, 500, map[string]string{"error": "读取规则失败"}) + return + } + writeJSON(w, 200, map[string]any{"items": items}) +} +func (h HTTP) create(w http.ResponseWriter, r *http.Request) { + var input CreateInput + if err := decode(w, r, &input); err != nil { + writeJSON(w, 400, map[string]string{"error": "请求格式无效"}) + return + } + item, err := h.Service.Create(r.Context(), input) + if err != nil { + writeJSON(w, 400, map[string]string{"error": err.Error()}) + return + } + actor := auth.Principal(r.Context()) + _ = h.Audit.Record(r.Context(), &actor.ID, "rule.created", "rule", &item.ID, "success", map[string]any{"version": item.Version}) + writeJSON(w, 201, item) +} +func (h HTTP) setEnabled(w http.ResponseWriter, r *http.Request) { + var input struct { + Enabled bool `json:"enabled"` + } + if err := decode(w, r, &input); err != nil { + writeJSON(w, 400, map[string]string{"error": "请求格式无效"}) + return + } + item, err := h.Service.SetEnabled(r.Context(), r.PathValue("id"), input.Enabled) + if err != nil { + writeJSON(w, 404, map[string]string{"error": "规则不存在"}) + return + } + actor := auth.Principal(r.Context()) + _ = h.Audit.Record(r.Context(), &actor.ID, "rule.enabled_changed", "rule", &item.ID, "success", map[string]any{"enabled": item.Enabled, "version": item.Version}) + writeJSON(w, 200, item) +} +func decode(w http.ResponseWriter, r *http.Request, value any) error { + decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)) + decoder.DisallowUnknownFields() + return decoder.Decode(value) +} +func writeJSON(w http.ResponseWriter, status int, value any) { + w.Header().Set("Content-Type", "application/json; charset=utf-8") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(value) +} diff --git a/Bell/server/app/rule/service.go b/Bell/server/app/rule/service.go new file mode 100644 index 0000000..f70fefc --- /dev/null +++ b/Bell/server/app/rule/service.go @@ -0,0 +1,132 @@ +package rule + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + + "git.ilapage.cn/ila/yovision/Bell/server/app/event" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +var ErrInvalid = errors.New("规则字段无效") + +type Service struct{ DB *pgxpool.Pool } + +func validSeverity(value string) bool { + return value == "low" || value == "medium" || value == "high" || value == "critical" +} +func severityRank(value string) int { + return map[string]int{"low": 1, "medium": 2, "high": 3, "critical": 4}[value] +} +func (s Service) Create(ctx context.Context, input CreateInput) (Rule, error) { + input.Code = strings.ToLower(strings.TrimSpace(input.Code)) + input.Name = strings.TrimSpace(input.Name) + input.MinimumSeverity = strings.ToLower(strings.TrimSpace(input.MinimumSeverity)) + if input.Code == "" || input.Name == "" || !validSeverity(input.MinimumSeverity) { + return Rule{}, ErrInvalid + } + if input.EventType != nil { + v := strings.TrimSpace(*input.EventType) + if v == "" { + input.EventType = nil + } else { + input.EventType = &v + } + } + if input.LocationContains != nil { + v := strings.TrimSpace(*input.LocationContains) + if v == "" { + input.LocationContains = nil + } else { + input.LocationContains = &v + } + } + var result Rule + err := s.DB.QueryRow(ctx, `INSERT INTO bell_rules(code,name,event_type,minimum_severity,location_contains) VALUES($1,$2,$3,$4,$5) RETURNING id::text,code,name,enabled,event_type,minimum_severity,location_contains,version,created_at::text,updated_at::text`, input.Code, input.Name, input.EventType, input.MinimumSeverity, input.LocationContains).Scan(&result.ID, &result.Code, &result.Name, &result.Enabled, &result.EventType, &result.MinimumSeverity, &result.LocationContains, &result.Version, &result.CreatedAt, &result.UpdatedAt) + return result, err +} +func (s Service) SetEnabled(ctx context.Context, id string, enabled bool) (Rule, error) { + var result Rule + err := s.DB.QueryRow(ctx, `UPDATE bell_rules SET enabled=$2,version=version+1,updated_at=now() WHERE id=$1 RETURNING id::text,code,name,enabled,event_type,minimum_severity,location_contains,version,created_at::text,updated_at::text`, id, enabled).Scan(&result.ID, &result.Code, &result.Name, &result.Enabled, &result.EventType, &result.MinimumSeverity, &result.LocationContains, &result.Version, &result.CreatedAt, &result.UpdatedAt) + return result, err +} +func (s Service) List(ctx context.Context) ([]Rule, error) { + rows, err := s.DB.Query(ctx, `SELECT id::text,code,name,enabled,event_type,minimum_severity,location_contains,version,created_at::text,updated_at::text FROM bell_rules ORDER BY created_at,id LIMIT 200`) + if err != nil { + return nil, err + } + defer rows.Close() + items := []Rule{} + for rows.Next() { + var item Rule + if err := rows.Scan(&item.ID, &item.Code, &item.Name, &item.Enabled, &item.EventType, &item.MinimumSeverity, &item.LocationContains, &item.Version, &item.CreatedAt, &item.UpdatedAt); err != nil { + return nil, err + } + items = append(items, item) + } + return items, rows.Err() +} + +func (s Service) Match(ctx context.Context, tx pgx.Tx, evt event.Event) error { + rows, err := tx.Query(ctx, `SELECT id::text,code,name,enabled,event_type,minimum_severity,location_contains,version,created_at::text,updated_at::text FROM bell_rules WHERE enabled=true ORDER BY id`) + if err != nil { + return err + } + rules := []Rule{} + for rows.Next() { + var item Rule + if err := rows.Scan(&item.ID, &item.Code, &item.Name, &item.Enabled, &item.EventType, &item.MinimumSeverity, &item.LocationContains, &item.Version, &item.CreatedAt, &item.UpdatedAt); err != nil { + rows.Close() + return err + } + rules = append(rules, item) + } + rows.Close() + if err := rows.Err(); err != nil { + return err + } + for _, current := range rules { + matched := true + reasons := []string{} + if current.EventType != nil && *current.EventType != evt.EventType { + matched = false + reasons = append(reasons, "事件类型不匹配") + } + if severityRank(evt.Severity) < severityRank(current.MinimumSeverity) { + matched = false + reasons = append(reasons, "风险等级低于阈值") + } + if current.LocationContains != nil && !strings.Contains(strings.ToLower(evt.Location), strings.ToLower(*current.LocationContains)) { + matched = false + reasons = append(reasons, "地点条件不匹配") + } + explanation := "全部条件命中" + if !matched { + explanation = strings.Join(reasons, "；") + } + if _, err := tx.Exec(ctx, `INSERT INTO bell_rule_evaluations(event_id,rule_id,rule_version,matched,explanation) VALUES($1,$2,$3,$4,$5) ON CONFLICT DO NOTHING`, evt.ID, current.ID, current.Version, matched, explanation); err != nil { + return err + } + if !matched { + continue + } + snapshot, _ := json.Marshal(current) + summary := fmt.Sprintf("%s：%s", current.Name, evt.EventType) + var alertID string + err := tx.QueryRow(ctx, `INSERT INTO bell_alerts(primary_rule_id,correlation_key,severity,summary,location) VALUES($1,$2,$3,$4,$5) ON CONFLICT(primary_rule_id,correlation_key) WHERE status='open' DO UPDATE SET updated_at=now(),severity=CASE WHEN array_position(ARRAY['low','medium','high','critical'],EXCLUDED.severity)>array_position(ARRAY['low','medium','high','critical'],bell_alerts.severity) THEN EXCLUDED.severity ELSE bell_alerts.severity END RETURNING id::text`, current.ID, strings.ToLower(evt.Location), evt.Severity, summary, evt.Location).Scan(&alertID) + if err != nil { + return err + } + if _, err := tx.Exec(ctx, `INSERT INTO bell_alert_events(alert_id,event_id) VALUES($1,$2) ON CONFLICT DO NOTHING`, alertID, evt.ID); err != nil { + return err + } + if _, err := tx.Exec(ctx, `INSERT INTO bell_rule_matches(alert_id,event_id,rule_id,rule_version,rule_snapshot,explanation) VALUES($1,$2,$3,$4,$5,$6) ON CONFLICT DO NOTHING`, alertID, evt.ID, current.ID, current.Version, snapshot, explanation); err != nil { + return err + } + } + return nil +} diff --git a/Bell/server/app/rule/service_test.go b/Bell/server/app/rule/service_test.go new file mode 100644 index 0000000..4f5b71e --- /dev/null +++ b/Bell/server/app/rule/service_test.go @@ -0,0 +1,87 @@ +package rule + +import ( + "context" + "fmt" + "os" + "testing" + "time" + + "git.ilapage.cn/ila/yovision/Bell/server/app/event" + "git.ilapage.cn/ila/yovision/Bell/server/internal/platform" + "git.ilapage.cn/ila/yovision/Bell/server/migrations" +) + +func TestPostgresMatchingAndNavigationShape(t *testing.T) { + url := os.Getenv("BELL_TEST_DATABASE_URL") + if url == "" { + t.Skip("BELL_TEST_DATABASE_URL is not set") + } + ctx := context.Background() + db, err := platform.OpenDatabase(ctx, url) + if err != nil { + t.Fatal(err) + } + defer db.Close() + if err := migrations.Apply(ctx, db); err != nil { + t.Fatal(err) + } + service := Service{DB: db} + prefix := fmt.Sprintf("rule-test-%d", time.Now().UnixNano()) + eventType := "rule-test-danger" + location := "north" + first, err := service.Create(ctx, CreateInput{Code: prefix + "-one", Name: "规则一", EventType: &eventType, MinimumSeverity: "medium", LocationContains: &location}) + if err != nil { + t.Fatal(err) + } + second, err := service.Create(ctx, CreateInput{Code: prefix + "-two", Name: "规则二", EventType: &eventType, MinimumSeverity: "low"}) + if err != nil { + t.Fatal(err) + } + eventService := event.Service{DB: db, AfterPersist: service.Match} + base := event.Command{ProducerID: prefix, SourceEventID: "1", EventType: eventType, OccurredAt: time.Now().UTC(), Location: "north gate", Severity: "high", Attributes: map[string]any{}} + one, err := eventService.Ingest(ctx, base) + if err != nil { + t.Fatal(err) + } + base.SourceEventID = "2" + two, err := eventService.Ingest(ctx, base) + if err != nil { + t.Fatal(err) + } + var alertCount, linkCount int + if err := db.QueryRow(ctx, `SELECT count(DISTINCT m.alert_id),count(*) FROM bell_rule_matches m WHERE m.rule_id=ANY($1::uuid[])`, []string{first.ID, second.ID}).Scan(&alertCount, &linkCount); err != nil { + t.Fatal(err) + } + if alertCount != 2 || linkCount != 4 { + t.Fatalf("alerts=%d links=%d", alertCount, linkCount) + } + var shared int + if err := db.QueryRow(ctx, `SELECT count(*) FROM bell_alert_events a JOIN bell_alert_events b ON b.alert_id=a.alert_id JOIN bell_alerts current ON current.id=a.alert_id WHERE a.event_id=$1 AND b.event_id=$2 AND current.primary_rule_id=ANY($3::uuid[])`, one.Event.ID, two.Event.ID, []string{first.ID, second.ID}).Scan(&shared); err != nil { + t.Fatal(err) + } + if shared != 2 { + t.Fatalf("expected two correlated alerts, got %d", shared) + } + base.SourceEventID = "3" + base.EventType = "unmatched" + unmatched, err := eventService.Ingest(ctx, base) + if err != nil { + t.Fatal(err) + } + var evaluations, matches int + if err := db.QueryRow(ctx, `SELECT count(*),count(*) FILTER(WHERE matched) FROM bell_rule_evaluations WHERE event_id=$1`, unmatched.Event.ID).Scan(&evaluations, &matches); err != nil { + t.Fatal(err) + } + if evaluations < 2 || matches != 0 { + t.Fatalf("evaluations=%d matches=%d", evaluations, matches) + } + disabled, err := service.SetEnabled(ctx, first.ID, false) + if err != nil || disabled.Enabled || disabled.Version != 2 { + t.Fatalf("disable: %#v %v", disabled, err) + } + enabled, err := service.SetEnabled(ctx, first.ID, true) + if err != nil || !enabled.Enabled || enabled.Version != 3 { + t.Fatalf("enable: %#v %v", enabled, err) + } +} diff --git a/Bell/server/app/rule/types.go b/Bell/server/app/rule/types.go new file mode 100644 index 0000000..01801f6 --- /dev/null +++ b/Bell/server/app/rule/types.go @@ -0,0 +1,21 @@ +package rule + +type Rule struct { + ID string `json:"id"` + Code string `json:"code"` + Name string `json:"name"` + Enabled bool `json:"enabled"` + EventType *string `json:"event_type,omitempty"` + MinimumSeverity string `json:"minimum_severity"` + LocationContains *string `json:"location_contains,omitempty"` + Version int `json:"version"` + CreatedAt string `json:"created_at"` + UpdatedAt string `json:"updated_at"` +} +type CreateInput struct { + Code string `json:"code"` + Name string `json:"name"` + EventType *string `json:"event_type,omitempty"` + MinimumSeverity string `json:"minimum_severity"` + LocationContains *string `json:"location_contains,omitempty"` +} diff --git a/Bell/server/cmd/bell/root.go b/Bell/server/cmd/bell/root.go index aa51d04..8e043e5 100644 --- a/Bell/server/cmd/bell/root.go +++ b/Bell/server/cmd/bell/root.go @@ -12,9 +12,11 @@ import ( "syscall" "time" + alertQuery "git.ilapage.cn/ila/yovision/Bell/server/app/alert/query" "git.ilapage.cn/ila/yovision/Bell/server/app/audit" "git.ilapage.cn/ila/yovision/Bell/server/app/auth" "git.ilapage.cn/ila/yovision/Bell/server/app/event" + "git.ilapage.cn/ila/yovision/Bell/server/app/rule" "git.ilapage.cn/ila/yovision/Bell/server/app/synthetic" "git.ilapage.cn/ila/yovision/Bell/server/config" "git.ilapage.cn/ila/yovision/Bell/server/internal/platform" @@ -67,9 +69,13 @@ func Run(ctx context.Context, args []string) error { app := platform.NewHTTPApp(db) authHTTP := auth.HTTP{Service: authService, Store: authStore, Audit: auditStore, SecureCookie: cfg.CookieSecure} authHTTP.Register(app.Router()) - event.HTTP{Service: event.Service{DB: db}, Auth: authHTTP}.Register(app.Router()) + ruleService := rule.Service{DB: db} + eventService := event.Service{DB: db, AfterPersist: ruleService.Match} + event.HTTP{Service: eventService, Auth: authHTTP}.Register(app.Router()) + rule.HTTP{Service: ruleService, Auth: authHTTP, Audit: auditStore}.Register(app.Router()) + alertQuery.HTTP{Service: alertQuery.Service{DB: db}, Auth: authHTTP}.Register(app.Router()) if cfg.Environment == "development" || cfg.Environment == "test" { - synthetic.HTTP{Service: synthetic.Service{Events: event.Service{DB: db}}, Auth: authHTTP}.Register(app.Router()) + synthetic.HTTP{Service: synthetic.Service{Events: eventService}, Auth: authHTTP}.Register(app.Router()) } server := &http.Server{Addr: cfg.HTTPAddress, Handler: app.Handler()} serverCtx, stop := signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM) diff --git a/Bell/server/migrations/2026081204_rule_alert_base.sql b/Bell/server/migrations/2026081204_rule_alert_base.sql new file mode 100644 index 0000000..6dd01ec --- /dev/null +++ b/Bell/server/migrations/2026081204_rule_alert_base.sql @@ -0,0 +1,51 @@ +CREATE TABLE bell_rules ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + code text NOT NULL UNIQUE, + name text NOT NULL, + enabled boolean NOT NULL DEFAULT true, + event_type text, + minimum_severity text NOT NULL CHECK (minimum_severity IN ('low','medium','high','critical')), + location_contains text, + version integer NOT NULL DEFAULT 1 CHECK (version > 0), + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now() +); +CREATE TABLE bell_alerts ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + primary_rule_id uuid NOT NULL REFERENCES bell_rules(id), + correlation_key text NOT NULL, + status text NOT NULL DEFAULT 'open' CHECK (status IN ('open','acknowledged','closed')), + severity text NOT NULL CHECK (severity IN ('low','medium','high','critical')), + summary text NOT NULL, + location text NOT NULL, + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now() +); +CREATE UNIQUE INDEX bell_alert_open_correlation_idx ON bell_alerts(primary_rule_id,correlation_key) WHERE status='open'; +CREATE INDEX bell_alerts_page_idx ON bell_alerts(created_at DESC,id DESC); +CREATE TABLE bell_alert_events ( + alert_id uuid NOT NULL REFERENCES bell_alerts(id), + event_id uuid NOT NULL REFERENCES bell_events(id), + linked_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY(alert_id,event_id) +); +CREATE INDEX bell_alert_events_event_idx ON bell_alert_events(event_id,alert_id); +CREATE TABLE bell_rule_matches ( + alert_id uuid NOT NULL REFERENCES bell_alerts(id), + event_id uuid NOT NULL REFERENCES bell_events(id), + rule_id uuid NOT NULL REFERENCES bell_rules(id), + rule_version integer NOT NULL, + rule_snapshot jsonb NOT NULL, + explanation text NOT NULL, + matched_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY(event_id,rule_id) +); +CREATE TABLE bell_rule_evaluations ( + event_id uuid NOT NULL REFERENCES bell_events(id), + rule_id uuid NOT NULL REFERENCES bell_rules(id), + rule_version integer NOT NULL, + matched boolean NOT NULL, + explanation text NOT NULL, + evaluated_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY(event_id,rule_id) +); diff --git a/Bell/server/migrations/runner.go b/Bell/server/migrations/runner.go index d97c04f..b2c836a 100644 --- a/Bell/server/migrations/runner.go +++ b/Bell/server/migrations/runner.go @@ -14,7 +14,18 @@ import ( var files embed.FS func Apply(ctx context.Context, pool *pgxpool.Pool) error { - if _, err := pool.Exec(ctx, `CREATE TABLE IF NOT EXISTS bell_schema_migrations (name text PRIMARY KEY, applied_at timestamptz NOT NULL DEFAULT now())`); err != nil { + conn, err := pool.Acquire(ctx) + if err != nil { + return fmt.Errorf("acquire migration connection: %w", err) + } + defer conn.Release() + if _, err = conn.Exec(ctx, `SELECT pg_advisory_lock(hashtext('yovision-bell-migrations'))`); err != nil { + return fmt.Errorf("lock migrations: %w", err) + } + defer func() { + _, _ = conn.Exec(context.Background(), `SELECT pg_advisory_unlock(hashtext('yovision-bell-migrations'))`) + }() + if _, err := conn.Exec(ctx, `CREATE TABLE IF NOT EXISTS bell_schema_migrations (name text PRIMARY KEY, applied_at timestamptz NOT NULL DEFAULT now())`); err != nil { return fmt.Errorf("create migration ledger: %w", err) } entries, err := fs.ReadDir(files, ".") @@ -27,7 +38,7 @@ func Apply(ctx context.Context, pool *pgxpool.Pool) error { continue } var applied bool - if err := pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM bell_schema_migrations WHERE name=$1)`, entry.Name()).Scan(&applied); err != nil { + if err := conn.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM bell_schema_migrations WHERE name=$1)`, entry.Name()).Scan(&applied); err != nil { return err } if applied { @@ -37,7 +48,7 @@ func Apply(ctx context.Context, pool *pgxpool.Pool) error { if err != nil { return err } - tx, err := pool.Begin(ctx) + tx, err := conn.Begin(ctx) if err != nil { return err } diff --git a/Bell/web/src/api/alert/index.js b/Bell/web/src/api/alert/index.js new file mode 100644 index 0000000..1e3e400 --- /dev/null +++ b/Bell/web/src/api/alert/index.js @@ -0,0 +1,4 @@ +import request from '../../bootstrap/request' +export const listAlerts = params => request.get('/api/v1/alerts', { params }) +export const getAlert = id => request.get(`/api/v1/alerts/${id}`) +export const alertsForEvent = id => request.get(`/api/v1/events/${id}/alerts`) diff --git a/Bell/web/src/api/rule/index.js b/Bell/web/src/api/rule/index.js new file mode 100644 index 0000000..f713c15 --- /dev/null +++ b/Bell/web/src/api/rule/index.js @@ -0,0 +1,4 @@ +import request from '../../bootstrap/request' +export const listRules = () => request.get('/api/v1/rules') +export const createRule = data => request.post('/api/v1/rules', data) +export const setRuleEnabled = (id, enabled) => request.put(`/api/v1/rules/${id}/enabled`, { enabled }) diff --git a/Bell/web/src/bootstrap/router.js b/Bell/web/src/bootstrap/router.js index ed7cde0..b07112c 100644 --- a/Bell/web/src/bootstrap/router.js +++ b/Bell/web/src/bootstrap/router.js @@ -5,6 +5,8 @@ import Users from '../views/system/Users.vue' import Audit from '../views/system/Audit.vue' import Events from '../views/event/Events.vue' import SyntheticEvent from '../views/synthetic-event/SyntheticEvent.vue' +import Rules from '../views/rule/Rules.vue' +import Alerts from '../views/alert/list/Alerts.vue' import { listFixtures } from '../api/synthetic' import store from './store' @@ -14,6 +16,8 @@ const router = createRouter({ { path: '/login', name: 'login', component: Login, meta: { title: '登录', public: true } }, { path: '/', name: 'dashboard', component: Dashboard, meta: { title: '工作台', permission: 'dashboard:read' } }, { path: '/events', name: 'events', component: Events, meta: { title: '事件查询', permission: 'events:read' } }, + { path: '/alerts', name: 'alerts', component: Alerts, meta: { title: '预警管理', permission: 'alerts:read' } }, + { path: '/rules', name: 'rules', component: Rules, meta: { title: '规则配置', permission: 'rules:read' } }, { path: '/development/synthetic-events', name: 'synthetic-events', component: SyntheticEvent, meta: { title: '合成事件测试', permission: 'synthetic:write', developmentOnly: true } }, { path: '/system/users', name: 'users', component: Users, meta: { title: '用户与角色', permission: 'users:read' } }, { path: '/system/audit', name: 'audit', component: Audit, meta: { title: '认证审计', permission: 'audit:read' } } diff --git a/Bell/web/src/layout/AppLayout.vue b/Bell/web/src/layout/AppLayout.vue index 9f74186..6b5554d 100644 --- a/Bell/web/src/layout/AppLayout.vue +++ b/Bell/web/src/layout/AppLayout.vue @@ -5,6 +5,8 @@