From c48cc4e2d4f99e38e29dd2c2907b79e16ddf2378 Mon Sep 17 00:00:00 2001 From: Denozordec Date: Mon, 6 Apr 2026 17:10:29 +0700 Subject: [PATCH] feat: enhance job processing with detailed logging and UI improvements. Introduce revision log entries in the Worker to capture job metadata, including source and community details. Update the Svelte UI to display job logs with expandable details, improving user experience and job tracking capabilities. --- internal/jobs/worker.go | 174 ++++++++++++++++++++- web/src/routes/operations/+page.svelte | 207 ++++++++++++++++++++++++- 2 files changed, 368 insertions(+), 13 deletions(-) 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 @@