diff --git a/internal/jobs/worker.go b/internal/jobs/worker.go index 5eb910a..babe217 100644 --- a/internal/jobs/worker.go +++ b/internal/jobs/worker.go @@ -2,8 +2,11 @@ package jobs import ( "context" + "fmt" "net/http" + "net/netip" "os" + "sort" "strings" "time" @@ -37,20 +40,29 @@ func mergeBirdPostApplyMeta(j *Job) { } const ( - KindModuleRefresh = "module_refresh" - KindDeployApply = "deploy_apply" + KindModuleRefresh = "module_refresh" + KindDeployApply = "deploy_apply" KindRevisionRollback = "revision_rollback" - KindBirdReload = "bird_reload" + KindBirdReload = "bird_reload" ) // Worker executes queued jobs against store.Backend (memory or SQL). type Worker struct { - Store store.Backend - HTTPClient *http.Client // optional; CDN refresh uses this (default 45s timeout). + Store store.Backend + HTTPClient *http.Client // optional; CDN refresh uses this (default 45s timeout). // Registry is set after BootstrapWorkers creates the job queue; used to chain deploy_apply after refresh/rollback. Registry *Registry } +type revisionLogEntry struct { + Kind string `json:"kind"` + Source string `json:"source"` + Community string `json:"community"` + PrefixCount int `json:"prefix_count"` + Sample []string `json:"sample,omitempty"` + Message string `json:"message"` +} + var defaultWorkerHTTP = &http.Client{Timeout: 45 * time.Second} func (w *Worker) httpClient() *http.Client { @@ -90,6 +102,15 @@ func (w *Worker) Process(j *Job) { return } j.mergeMeta(map[string]any{"revision_id": rev}) + if entries, total, err := w.buildRevisionLogEntries(j.TenantID, rev); err == nil { + j.mergeMeta(map[string]any{ + "log_entries": entries, + "log_total": total, + "log_generated": time.Now().UTC().Format(time.RFC3339Nano), + }) + } else { + j.mergeMeta(map[string]any{"log_build_error": err.Error()}) + } w.enqueueDeployAllSpeakers(j, j.TenantID, rev) j.Succeed() case KindDeployApply: @@ -169,6 +190,7 @@ func (w *Worker) runDeployApply(j *Job) { return } } + applied := make([]string, 0, 8) applyOne := func(speakerID string) error { if err := w.Store.SetLastAppliedRevision(j.TenantID, speakerID, revID); err != nil { return err @@ -177,6 +199,7 @@ func (w *Worker) runDeployApply(j *Job) { if err := w.Store.PublishRevisionForSpeaker(speakerID, revID); err != nil { return err } + applied = append(applied, speakerID) return nil } if hasSpeaker && spk != "" { @@ -194,6 +217,14 @@ func (w *Worker) runDeployApply(j *Job) { return } } + j.mergeMeta(map[string]any{ + "apply_summary": map[string]any{ + "revision_id": revID, + "speakers_count": len(applied), + "speaker_ids": applied, + "message": fmt.Sprintf("Ревизия %s применена на %d спикерах", shortID(revID), len(applied)), + }, + }) mergeBirdPostApplyMeta(j) j.Succeed() } @@ -210,6 +241,139 @@ func (w *Worker) runRollback(j *Job) { return } j.mergeMeta(map[string]any{"new_revision_id": newID}) + j.mergeMeta(map[string]any{ + "rollback_summary": map[string]any{ + "source_revision_id": src, + "new_revision_id": newID, + "message": fmt.Sprintf("Rollback %s -> %s", shortID(src), shortID(newID)), + }, + }) w.enqueueDeployAllSpeakers(j, j.TenantID, newID) j.Succeed() } + +func (w *Worker) buildRevisionLogEntries(tenantID, revID string) ([]map[string]any, int, error) { + if w == nil || w.Store == nil { + return nil, 0, fmt.Errorf("store not configured") + } + var all []store.PrefixRow + cursor := "" + for { + rows, next, more := w.Store.ListRevisionPrefixes(tenantID, revID, cursor, 1000) + all = append(all, rows...) + if !more { + break + } + cursor = next + if strings.TrimSpace(cursor) == "" { + break + } + } + type agg struct { + kind string + source string + community string + count int + sample []string + } + groups := map[string]*agg{} + for _, p := range all { + src := strings.TrimSpace(p.Source) + comm := "none" + if p.CommunityID != nil && strings.TrimSpace(*p.CommunityID) != "" { + comm = strings.TrimSpace(*p.CommunityID) + } + kind, sourceName := classifySource(src) + k := kind + "|" + sourceName + "|" + comm + g, ok := groups[k] + if !ok { + g = &agg{kind: kind, source: sourceName, community: comm} + groups[k] = g + } + g.count++ + if len(g.sample) < 5 { + g.sample = append(g.sample, p.Prefix) + } + } + keys := make([]string, 0, len(groups)) + for k := range groups { + keys = append(keys, k) + } + sort.Strings(keys) + out := make([]map[string]any, 0, len(keys)) + for _, k := range keys { + g := groups[k] + msg := humanLogMessage(g.kind, g.source, g.count, g.community, g.sample) + out = append(out, map[string]any{ + "kind": g.kind, + "source": g.source, + "community": g.community, + "prefix_count": g.count, + "sample": g.sample, + "message": msg, + }) + } + return out, len(all), nil +} + +func classifySource(src string) (kind, name string) { + switch { + case strings.HasPrefix(src, "as:"): + return "asn", strings.TrimPrefix(src, "as:") + case strings.HasPrefix(src, "domain:"): + return "domain", strings.TrimPrefix(src, "domain:") + case strings.HasPrefix(src, "cdn:"): + return "cdn", strings.TrimPrefix(src, "cdn:") + case src == "ip_range": + return "ip_range", "manual_ranges" + default: + if src == "" { + return "unknown", "unknown" + } + return "source", src + } +} + +func humanLogMessage(kind, source string, count int, community string, sample []string) string { + switch kind { + case "asn": + return fmt.Sprintf("AS%s -> %d префиксов добавлены в community %s", source, count, community) + case "domain": + ips := strings.Join(prettyDomainSample(sample), " ") + if ips == "" { + ips = "-" + } + return fmt.Sprintf("%s -> ip (%s) -> добавлены в community %s", source, ips, community) + case "cdn": + return fmt.Sprintf("CDN source %s -> %d префиксов добавлены в community %s", source, count, community) + case "ip_range": + return fmt.Sprintf("IP ranges -> %d префиксов добавлены в community %s", count, community) + default: + return fmt.Sprintf("%s -> %d префиксов добавлены в community %s", source, count, community) + } +} + +func prettyDomainSample(sample []string) []string { + out := make([]string, 0, len(sample)) + for _, s := range sample { + p, err := netip.ParsePrefix(strings.TrimSpace(s)) + if err != nil { + out = append(out, s) + continue + } + if (p.Addr().Is4() && p.Bits() == 32) || (p.Addr().Is6() && p.Bits() == 128) { + out = append(out, p.Addr().String()) + continue + } + out = append(out, s) + } + return out +} + +func shortID(id string) string { + s := strings.TrimSpace(id) + if len(s) <= 8 { + return s + } + return s[:8] +} diff --git a/web/src/routes/operations/+page.svelte b/web/src/routes/operations/+page.svelte index 0615c95..6b9caa9 100644 --- a/web/src/routes/operations/+page.svelte +++ b/web/src/routes/operations/+page.svelte @@ -1,5 +1,6 @@