diff --git a/cmd/ctcloud/main.go b/cmd/ctcloud/main.go index 9f405ac..9c856dc 100644 --- a/cmd/ctcloud/main.go +++ b/cmd/ctcloud/main.go @@ -135,7 +135,13 @@ func run() error { // Scanner prober: a separate, never-started Prober used synchronously by the // on-demand scan endpoint (Probe is stateless). scanProber := probe.New(probe.Config{Workers: 1, Timeout: cfg.ProbeTimeout, Deep: cfg.ProbeDeep, Metrics: m}) - scanner := &scan.Scanner{Prober: scanProber, Indexer: indexer, Enricher: enricher, Store: st} + // Backfill does one DB write per target, so keep its concurrency near the + // pool size to avoid starving the live pipeline and the UI. + backfillWorkers := cfg.ProbeWorkers + if backfillWorkers > 16 { + backfillWorkers = 16 + } + scanner := &scan.Scanner{Prober: scanProber, Indexer: indexer, Enricher: enricher, Store: st, BackfillWorkers: backfillWorkers} var prober *probe.Prober if cfg.ActiveProbing { diff --git a/internal/pipeline/dnsqueue.go b/internal/pipeline/dnsqueue.go new file mode 100644 index 0000000..c6e00da --- /dev/null +++ b/internal/pipeline/dnsqueue.go @@ -0,0 +1,71 @@ +package pipeline + +import ( + mathrand "math/rand/v2" + "sync" +) + +// stringSample is a memory-bounded queue that keeps a random sample of recent +// hostnames instead of a FIFO. A FIFO fills with the first N hostnames seen at +// startup and drops everything after, so DNS enrichment only ever sees stale +// hostnames and CNAME discovery stalls. Random replacement makes coverage +// representative at bounded memory. Not on the certificate hot path. +type stringSample struct { + mu sync.Mutex + items []string + cap int + notify chan struct{} + stop chan struct{} + once sync.Once +} + +func newStringSample(capacity int) *stringSample { + return &stringSample{ + cap: capacity, + notify: make(chan struct{}, 1), + stop: make(chan struct{}), + } +} + +func (q *stringSample) push(s string) { + q.mu.Lock() + if len(q.items) < q.cap { + q.items = append(q.items, s) + } else { + q.items[mathrand.IntN(len(q.items))] = s + } + q.mu.Unlock() + select { + case q.notify <- struct{}{}: + default: + } +} + +func (q *stringSample) pop() (string, bool) { + q.mu.Lock() + n := len(q.items) + if n == 0 { + q.mu.Unlock() + return "", false + } + i := mathrand.IntN(n) + s := q.items[i] + q.items[i] = q.items[n-1] + q.items = q.items[:n-1] + q.mu.Unlock() + return s, true +} + +func (q *stringSample) len() int { + q.mu.Lock() + defer q.mu.Unlock() + return len(q.items) +} + +func (q *stringSample) close() { + q.once.Do(func() { close(q.stop) }) + select { + case q.notify <- struct{}{}: + default: + } +} diff --git a/internal/pipeline/pipeline.go b/internal/pipeline/pipeline.go index 6fcae68..ac91dfd 100644 --- a/internal/pipeline/pipeline.go +++ b/internal/pipeline/pipeline.go @@ -50,7 +50,7 @@ type Pipeline struct { cfg Config directQ chan model.Candidate - dnsQ chan string + dnsQ *stringSample dbQ chan Finding dedup *dns.TTLCache[bool] // hostnames already queued for DNS @@ -90,14 +90,14 @@ func New(cfg Config) *Pipeline { p := &Pipeline{ cfg: cfg, directQ: make(chan model.Candidate, cfg.DirectQueueSize), - dnsQ: make(chan string, cfg.DNSQueueSize), + dnsQ: newStringSample(cfg.DNSQueueSize), dbQ: make(chan Finding, cfg.DBQueueSize), dedup: dns.NewTTLCache[bool](cfg.DedupSize, 64), seen: dns.NewTTLCache[bool](cfg.DedupSize, 64), } if p.cfg.Metrics != nil { p.cfg.Metrics.SetQueueDepth("direct", func() float64 { return float64(len(p.directQ)) }) - p.cfg.Metrics.SetQueueDepth("dns", func() float64 { return float64(len(p.dnsQ)) }) + p.cfg.Metrics.SetQueueDepth("dns", func() float64 { return float64(p.dnsQ.len()) }) p.cfg.Metrics.SetQueueDepth("db", func() float64 { return float64(len(p.dbQ)) }) } return p @@ -119,14 +119,14 @@ func (p *Pipeline) Start(ctx context.Context) { // Depths reports the current length of the direct, DNS and DB queues. func (p *Pipeline) Depths() (direct, dns, db int) { - return len(p.directQ), len(p.dnsQ), len(p.dbQ) + return len(p.directQ), p.dnsQ.len(), len(p.dbQ) } // Close stops intake and drains all queues. It must only be called after the // CertSpotter monitor has stopped calling HandleIdentifiers. func (p *Pipeline) Close() { close(p.directQ) - close(p.dnsQ) + p.dnsQ.close() p.producer.Wait() close(p.dbQ) p.workers.Wait() @@ -175,11 +175,7 @@ func (p *Pipeline) HandleHost(host string) { return } p.dedup.Set(host, true, p.cfg.DedupTTL) - select { - case p.dnsQ <- host: - default: - p.drop() - } + p.dnsQ.push(host) } func (p *Pipeline) drop() { @@ -239,12 +235,17 @@ func (p *Pipeline) dnsWorker(ctx context.Context) { select { case <-ctx.Done(): return - case host, ok := <-p.dnsQ: + case <-p.dnsQ.stop: + return + case <-p.dnsQ.notify: + } + for { + host, ok := p.dnsQ.pop() if !ok { - return + break } if len(p.dbQ) >= cap(p.dbQ)/2 { - // Load shedding: optional CNAME enrichment is dropped before + // Load shedding: optional CNAME enrichment is skipped before // direct matches when the database writer is backed up. p.drop() continue diff --git a/internal/pipeline/pipeline_test.go b/internal/pipeline/pipeline_test.go index acfbd1f..93e05c7 100644 --- a/internal/pipeline/pipeline_test.go +++ b/internal/pipeline/pipeline_test.go @@ -269,7 +269,7 @@ func TestDNSDedup(t *testing.T) { p := New(Config{DNSQueueSize: 8, Metrics: m}) p.HandleHost("files.example.com") p.HandleHost("files.example.com") - if got := len(p.dnsQ); got != 1 { + if got := p.dnsQ.len(); got != 1 { t.Fatalf("dns queue len = %d, want 1", got) } } diff --git a/internal/scan/scanner.go b/internal/scan/scanner.go index 4de6cd4..33b57a7 100644 --- a/internal/scan/scanner.go +++ b/internal/scan/scanner.go @@ -8,6 +8,8 @@ import ( "context" "net" "strings" + "sync" + "sync/atomic" "tea.karlcom.de/karl/bucketeer/internal/classify" "tea.karlcom.de/karl/bucketeer/internal/dns" @@ -37,10 +39,11 @@ type Enricher interface { } type Scanner struct { - Prober Prober - Indexer Lister - Enricher Enricher - Store Store + Prober Prober + Indexer Lister + Enricher Enricher + Store Store + BackfillWorkers int } // Result is the outcome of scanning one resolved provider resource. @@ -107,47 +110,84 @@ type BackfillReport struct { // Backfill re-probes assets already in the database (optionally filtered by // exposure_state) with the current checks and indexes any that are readable. -// This recovers findings discovered before a check was improved. +// This is how already-discovered "reachable" endpoints actually get verified. +// It runs a bounded worker pool so it stays interactive. func (s *Scanner) Backfill(ctx context.Context, exposure string, limit int) (BackfillReport, error) { targets, err := s.Store.ListProbeTargets(ctx, exposure, limit) if err != nil { return BackfillReport{}, err } - var rep BackfillReport - for _, t := range targets { - if err := ctx.Err(); err != nil { - return rep, err - } - c := model.Candidate{ - Hostname: t.Hostname, Original: t.Hostname, Resource: t.Resource, Region: t.Region, - Provider: model.ParseProvider(t.Provider), Service: model.ParseService(t.Service), - Source: model.SourceManual, - } - pr := s.Prober.ProbeDetailed(ctx, c) - rep.Probed++ - if pr.Exposure == model.ExposurePublicRead { - rep.Readable++ - } - if pr.Listable { - rep.Listable++ - } - if err := s.Store.UpdateExposure(ctx, t.Provider, t.Service, t.Hostname, - pr.Exposure, pr.Detail, pr.Listable, pr.ObjectReadable); err != nil { - rep.Errors++ - continue - } - if pr.Listable && s.Indexer != nil { - entries, err := s.Indexer.List(ctx, c) - if err == nil && len(entries) > 0 { - if n, werr := s.Store.ReplaceEntries(ctx, t.Provider, t.Service, t.Hostname, entries); werr == nil { - rep.Indexed += n - } else { - rep.Errors++ + workers := s.BackfillWorkers + if workers <= 0 { + workers = 32 + } + var ( + probed, readable, listable, indexed, errs atomic.Int64 + wg sync.WaitGroup + jobs = make(chan model.ProbeTarget) + ) + for i := 0; i < workers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for t := range jobs { + if ctx.Err() != nil { + return } + s.probeTarget(ctx, t, &probed, &readable, &listable, &indexed, &errs) + } + }() + } + for _, t := range targets { + select { + case jobs <- t: + case <-ctx.Done(): + close(jobs) + wg.Wait() + return s.report(&probed, &readable, &listable, &indexed, &errs), ctx.Err() + } + } + close(jobs) + wg.Wait() + return s.report(&probed, &readable, &listable, &indexed, &errs), nil +} + +func (s *Scanner) probeTarget(ctx context.Context, t model.ProbeTarget, probed, readable, listable, indexed, errs *atomic.Int64) { + c := model.Candidate{ + Hostname: t.Hostname, Original: t.Hostname, Resource: t.Resource, Region: t.Region, + Provider: model.ParseProvider(t.Provider), Service: model.ParseService(t.Service), + Source: model.SourceManual, + } + pr := s.Prober.ProbeDetailed(ctx, c) + probed.Add(1) + if pr.Exposure == model.ExposurePublicRead { + readable.Add(1) + } + if pr.Listable { + listable.Add(1) + } + if err := s.Store.UpdateExposure(ctx, t.Provider, t.Service, t.Hostname, + pr.Exposure, pr.Detail, pr.Listable, pr.ObjectReadable); err != nil { + errs.Add(1) + return + } + if pr.Listable && s.Indexer != nil { + entries, err := s.Indexer.List(ctx, c) + if err == nil && len(entries) > 0 { + if n, werr := s.Store.ReplaceEntries(ctx, t.Provider, t.Service, t.Hostname, entries); werr == nil { + indexed.Add(int64(n)) + } else { + errs.Add(1) } } } - return rep, nil +} + +func (s *Scanner) report(probed, readable, listable, indexed, errs *atomic.Int64) BackfillReport { + return BackfillReport{ + Probed: int(probed.Load()), Readable: int(readable.Load()), + Listable: int(listable.Load()), Indexed: int(indexed.Load()), Errors: int(errs.Load()), + } } func (s *Scanner) run(ctx context.Context, c model.Candidate, chain []string) Result { @@ -161,6 +201,10 @@ func (s *Scanner) run(ctx context.Context, c model.Candidate, chain []string) Re pr := s.Prober.ProbeDetailed(ctx, c) r.Exposure, r.Detail = pr.Exposure.String(), pr.Detail r.Listable, r.ObjectReadable = pr.Listable, pr.ObjectReadable + if err := s.Store.UpdateExposure(ctx, c.Provider.String(), c.Service.String(), c.Hostname, + pr.Exposure, pr.Detail, pr.Listable, pr.ObjectReadable); err != nil { + r.Error = "store: " + err.Error() + } if pr.Listable && s.Indexer != nil { entries, err := s.Indexer.List(ctx, c) if err == nil && len(entries) > 0 { diff --git a/internal/webui/webui.go b/internal/webui/webui.go index d79282f..b15ac96 100644 --- a/internal/webui/webui.go +++ b/internal/webui/webui.go @@ -8,9 +8,12 @@ import ( "embed" "encoding/json" "io/fs" + "log/slog" "net/http" "strconv" "strings" + "sync/atomic" + "time" "tea.karlcom.de/karl/bucketeer/internal/scan" "tea.karlcom.de/karl/bucketeer/internal/store" @@ -111,19 +114,33 @@ func New(s Store, cfg Config) http.Handler { } writeJSON(w, cfg.Scanner.Scan(r.Context(), target)) }) + var backfillRunning atomic.Bool mux.HandleFunc("/api/reprobe", func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { http.Error(w, "POST only", http.StatusMethodNotAllowed) return } - q := r.URL.Query() - limit, _ := strconv.Atoi(q.Get("limit")) - rep, err := cfg.Scanner.Backfill(r.Context(), strings.TrimSpace(q.Get("exposure")), limit) - if err != nil { - http.Error(w, err.Error(), http.StatusInternalServerError) + if !backfillRunning.CompareAndSwap(false, true) { + http.Error(w, "backfill already running", http.StatusConflict) return } - writeJSON(w, rep) + q := r.URL.Query() + limit, _ := strconv.Atoi(q.Get("limit")) + exposure := strings.TrimSpace(q.Get("exposure")) + go func() { + defer backfillRunning.Store(false) + start := time.Now() + rep, err := cfg.Scanner.Backfill(context.Background(), exposure, limit) + if err != nil { + slog.Warn("backfill failed", "event", "backfill", "error", err) + return + } + slog.Info("backfill complete", "event", "backfill", "exposure", exposure, + "probed", rep.Probed, "readable", rep.Readable, "listable", rep.Listable, + "indexed", rep.Indexed, "errors", rep.Errors, "took", time.Since(start).String()) + }() + w.WriteHeader(http.StatusAccepted) + writeJSON(w, map[string]any{"started": true, "exposure": exposure, "limit": limit}) }) } mux.HandleFunc("/api/stats", func(w http.ResponseWriter, r *http.Request) {