diff --git a/pkg/hold/pds/scan_broadcaster.go b/pkg/hold/pds/scan_broadcaster.go index 52e94bb..25d7012 100644 --- a/pkg/hold/pds/scan_broadcaster.go +++ b/pkg/hold/pds/scan_broadcaster.go @@ -21,6 +21,25 @@ import ( "github.com/gorilla/websocket" ) +const ( + // pendingReclaimAfter is how long a job may sit in 'pending' before the + // re-dispatch loop offers it to a scanner again. Enqueue and + // drainPendingJobs are both one-shot, so without a periodic re-offer a row + // whose only dispatch attempt failed waits for the next scanner connect. + pendingReclaimAfter = 1 * time.Minute + + // pendingStaleAfter is how long a pending job counts as active for dispatch + // throttling. Assigned and processing jobs need no such bound — the ack and + // processing timeouts always resolve them — but 'pending' is otherwise + // unbounded, and one row stuck there froze proactive scanning for nine days. + pendingStaleAfter = 15 * time.Minute + + // capacityStallWarnAfter is how long the dispatch loop waits with no + // capacity before it says so. A silent stall is what made this class of + // failure invisible until users noticed missing scans. + capacityStallWarnAfter = 10 * time.Minute +) + // ScanBroadcaster manages scanner WebSocket connections and dispatches scan jobs // using a competing-consumer pattern. Jobs are persisted in SQLite and dispatched // round-robin to connected scanners. @@ -396,7 +415,7 @@ func (sb *ScanBroadcaster) dispatchJob(job *ScanJobEvent) { sb.nextIdx++ // Mark as assigned in database - _, err := sb.db.Exec(` + res, err := sb.db.Exec(` UPDATE scan_jobs SET status = 'assigned', assigned_to = ?, assigned_at = ? WHERE seq = ? AND status = 'pending' `, sub.id, time.Now(), job.Seq) @@ -404,6 +423,13 @@ func (sb *ScanBroadcaster) dispatchJob(job *ScanJobEvent) { slog.Error("Failed to assign scan job", "seq", job.Seq, "error", err) return } + // Two paths hand out pending rows now (drainPendingJobs on connect, and the + // re-dispatch loop), so a row that is no longer pending was claimed by the + // other one. Sending it anyway would scan it twice. + if n, err := res.RowsAffected(); err == nil && n == 0 { + slog.Debug("Scan job no longer pending, skipping dispatch", "seq", job.Seq) + return + } // Send to subscriber select { @@ -790,9 +816,16 @@ func (sb *ScanBroadcaster) reDispatchLoop() { } // reDispatchTimedOut finds jobs that were assigned but not acked/completed within timeout, +// re-offers jobs that have been sitting in 'pending' with nobody to hand them to, // and also marks stuck processing jobs as failed. // Collects timed-out rows first, closes cursor, then resets and re-dispatches // to avoid holding a SELECT cursor open during UPDATEs (prevents SQLite BUSY). +// +// Pending rows matter as much as assigned ones: Enqueue and drainPendingJobs are +// both one-shot dispatchers, so before this loop covered 'pending' a row whose +// single dispatch attempt failed (or that a disconnect unassigned while no +// scanner reconnected afterwards) stayed pending forever — and hasActiveJobs() +// counted it, which stopped proactive scanning deployment-wide. func (sb *ScanBroadcaster) reDispatchTimedOut() { timeout := time.Now().Add(-sb.ackTimeout) @@ -809,25 +842,31 @@ func (sb *ScanBroadcaster) reDispatchTimedOut() { } rows, err := sb.db.Query(` - SELECT seq, manifest_digest, repository, tag, user_did, user_handle, hold_did, hold_endpoint, tier, config_json, layers_json + SELECT seq, manifest_digest, repository, tag, user_did, user_handle, hold_did, hold_endpoint, tier, config_json, layers_json, status FROM scan_jobs - WHERE status = 'assigned' AND assigned_at < ? + WHERE (status = 'assigned' AND assigned_at < ?) + OR (status = 'pending' AND datetime(created_at) < datetime('now', ?)) ORDER BY seq ASC - `, timeout) + `, timeout, sqliteAgoModifier(pendingReclaimAfter)) if err != nil { slog.Error("Failed to query timed-out scan jobs", "error", err) return } - var jobs []*ScanJobEvent + type reclaimable struct { + job *ScanJobEvent + status string + } + + var jobs []reclaimable for rows.Next() { job := &ScanJobEvent{Type: "job"} - var configJSON, layersJSON string + var configJSON, layersJSON, status string err := rows.Scan( &job.Seq, &job.ManifestDigest, &job.Repository, &job.Tag, &job.UserDID, &job.UserHandle, &job.HoldDID, &job.HoldEndpoint, - &job.Tier, &configJSON, &layersJSON, + &job.Tier, &configJSON, &layersJSON, &status, ) if err != nil { continue @@ -835,27 +874,41 @@ func (sb *ScanBroadcaster) reDispatchTimedOut() { job.Config = json.RawMessage(configJSON) job.Layers = json.RawMessage(layersJSON) - jobs = append(jobs, job) + jobs = append(jobs, reclaimable{job: job, status: status}) } rows.Close() - for _, job := range jobs { - _, err = sb.db.Exec(` - UPDATE scan_jobs SET status = 'pending', assigned_to = NULL, assigned_at = NULL - WHERE seq = ? - `, job.Seq) - if err != nil { - continue + for _, r := range jobs { + job := r.job + + // Already-pending rows need no reset — and skipping the UPDATE keeps a + // row another dispatcher claimed between the SELECT and here assigned, + // so dispatchJob's status guard can drop it instead of double-sending. + if r.status != "pending" { + _, err = sb.db.Exec(` + UPDATE scan_jobs SET status = 'pending', assigned_to = NULL, assigned_at = NULL + WHERE seq = ? + `, job.Seq) + if err != nil { + continue + } } - slog.Info("Re-dispatching timed-out scan job", + slog.Info("Re-dispatching scan job", "seq", job.Seq, - "repository", job.Repository) + "repository", job.Repository, + "previousStatus", r.status) sb.dispatchJob(job) } } +// sqliteAgoModifier renders a duration as a SQLite datetime() modifier that +// walks backwards from 'now', e.g. 15m becomes "-900 seconds". +func sqliteAgoModifier(d time.Duration) string { + return fmt.Sprintf("-%d seconds", int64(d.Seconds())) +} + // Close stops background goroutines and closes the scan broadcaster's database connection func (sb *ScanBroadcaster) Close() error { if sb.stopCh != nil { @@ -1273,10 +1326,20 @@ func (sb *ScanBroadcaster) dispatchLoop() { // waitForCapacity blocks until there are no active proactive scan jobs. // Returns false if stopCh is closed. func (sb *ScanBroadcaster) waitForCapacity() bool { + blockedSince := time.Now() + var lastWarn time.Time + for { if !sb.hasActiveJobs() { return true } + + if blocked := time.Since(blockedSince); blocked >= capacityStallWarnAfter && + time.Since(lastWarn) >= capacityStallWarnAfter { + sb.logStalledCapacity(blocked) + lastWarn = time.Now() + } + select { case <-sb.stopCh: return false @@ -1488,13 +1551,23 @@ func (sb *ScanBroadcaster) hasConnectedScanners() bool { return len(sb.subscribers) > 0 } -// hasActiveJobs returns true if there are any pending, assigned, or processing scan jobs. +// hasActiveJobs returns true if there are any assigned or processing scan jobs, +// or any recently-created pending ones. +// +// Pending rows older than pendingStaleAfter are deliberately not counted. A job +// that has been pending that long is one no scanner can be given (dispatch +// failed, or no scanner is connected), and counting it blocks the proactive +// dispatch loop for as long as the row exists — which is how a single +// undispatchable job stopped scanning deployment-wide. The re-dispatch loop +// keeps re-offering such rows, so ignoring them here costs nothing when a +// scanner is available. func (sb *ScanBroadcaster) hasActiveJobs() bool { var count int err := sb.db.QueryRow(` SELECT COUNT(*) FROM scan_jobs - WHERE status IN ('pending', 'assigned', 'processing') - `).Scan(&count) + WHERE status IN ('assigned', 'processing') + OR (status = 'pending' AND datetime(created_at) > datetime('now', ?)) + `, sqliteAgoModifier(pendingStaleAfter)).Scan(&count) if err != nil { slog.Error("Failed to check active scan jobs", "error", err) return true // Assume busy on error @@ -1502,6 +1575,32 @@ func (sb *ScanBroadcaster) hasActiveJobs() bool { return count > 0 } +// logStalledCapacity reports what is holding the dispatch loop back, so a stall +// shows up in the hold's logs instead of only as an absence of scan results. +func (sb *ScanBroadcaster) logStalledCapacity(blockedFor time.Duration) { + var ( + count int + oldestSeq sql.NullInt64 + statuses sql.NullString + ) + err := sb.db.QueryRow(` + SELECT COUNT(*), MIN(seq), GROUP_CONCAT(DISTINCT status) + FROM scan_jobs + WHERE status IN ('pending', 'assigned', 'processing') + `).Scan(&count, &oldestSeq, &statuses) + if err != nil { + slog.Warn("Proactive scan dispatch stalled; could not inspect active jobs", + "blockedFor", blockedFor.Truncate(time.Minute), "error", err) + return + } + + slog.Warn("Proactive scan dispatch stalled waiting on active jobs", + "blockedFor", blockedFor.Truncate(time.Minute), + "activeJobs", count, + "oldestSeq", oldestSeq.Int64, + "statuses", statuses.String) +} + func generateSubscriberID() string { b := make([]byte, 8) _, _ = rand.Read(b) diff --git a/pkg/hold/pds/scan_broadcaster_stall_test.go b/pkg/hold/pds/scan_broadcaster_stall_test.go new file mode 100644 index 0000000..d590b41 --- /dev/null +++ b/pkg/hold/pds/scan_broadcaster_stall_test.go @@ -0,0 +1,167 @@ +package pds + +import ( + "testing" + "time" +) + +// backdateJob ages a job's created_at so the staleness windows in +// hasActiveJobs and reDispatchTimedOut can be exercised without sleeping. +func backdateJob(t *testing.T, sb *ScanBroadcaster, seq int64, minutes int) { + t.Helper() + + _, err := sb.db.Exec( + `UPDATE scan_jobs SET created_at = datetime('now', ?) WHERE seq = ?`, + sqliteAgoModifier(time.Duration(minutes)*time.Minute), seq) + if err != nil { + t.Fatalf("backdate job %d: %v", seq, err) + } +} + +func jobStatus(t *testing.T, sb *ScanBroadcaster, seq int64) string { + t.Helper() + + var status string + if err := sb.db.QueryRow(`SELECT status FROM scan_jobs WHERE seq = ?`, seq).Scan(&status); err != nil { + t.Fatalf("query status for %d: %v", seq, err) + } + return status +} + +// TestScanHasActiveJobs_IgnoresLongPendingJob is the regression test for the +// nine-day deployment-wide scanning outage. A single job sat in 'pending' with +// nothing left to dispatch it, hasActiveJobs() counted it forever, and +// waitForCapacity() therefore never let the proactive dispatch loop enqueue +// another job — so discovery kept finding unscanned images and creating none. +func TestScanHasActiveJobs_IgnoresLongPendingJob(t *testing.T) { + sb := newTestScanBroadcaster(t) + seedPendingJobs(t, sb, 1) + + if !sb.hasActiveJobs() { + t.Fatal("a freshly enqueued pending job must count as active") + } + + backdateJob(t, sb, 1, 60) + + if sb.hasActiveJobs() { + t.Error("a job pending for an hour must not block dispatch capacity") + } +} + +// TestScanHasActiveJobs_CountsAssignedAndProcessing guards the other half: +// assigned and processing jobs have their own reclaim timeouts, so they must +// still hold capacity no matter how old the row is. +func TestScanHasActiveJobs_CountsAssignedAndProcessing(t *testing.T) { + for _, status := range []string{"assigned", "processing"} { + t.Run(status, func(t *testing.T) { + sb := newTestScanBroadcaster(t) + seedPendingJobs(t, sb, 1) + backdateJob(t, sb, 1, 60) + + if _, err := sb.db.Exec(`UPDATE scan_jobs SET status = ? WHERE seq = 1`, status); err != nil { + t.Fatalf("set status: %v", err) + } + + if !sb.hasActiveJobs() { + t.Errorf("%s job must hold dispatch capacity", status) + } + }) + } +} + +// TestScanHasActiveJobs_IgnoresTerminalJobs keeps completed and failed rows out +// of the capacity check — the table holds tens of thousands of them. +func TestScanHasActiveJobs_IgnoresTerminalJobs(t *testing.T) { + sb := newTestScanBroadcaster(t) + seedPendingJobs(t, sb, 2) + + if _, err := sb.db.Exec(`UPDATE scan_jobs SET status = 'completed' WHERE seq = 1`); err != nil { + t.Fatalf("complete: %v", err) + } + if _, err := sb.db.Exec(`UPDATE scan_jobs SET status = 'failed' WHERE seq = 2`); err != nil { + t.Fatalf("fail: %v", err) + } + + if sb.hasActiveJobs() { + t.Error("terminal jobs must not hold dispatch capacity") + } +} + +// TestScanReDispatch_ReoffersStalePendingJob covers the missing lease reclaim. +// Enqueue and drainPendingJobs are both one-shot, so a pending row used to wait +// for the next scanner connect — nine days, in production. The re-dispatch loop +// now hands it to a connected scanner on its own. +func TestScanReDispatch_ReoffersStalePendingJob(t *testing.T) { + sb := newTestScanBroadcaster(t) + sub := newTestScanSubscriber(t, sb, 4) + seedPendingJobs(t, sb, 1) + backdateJob(t, sb, 1, 30) + + sb.reDispatchTimedOut() + + if got := jobStatus(t, sb, 1); got != "assigned" { + t.Errorf("stale pending job not re-dispatched, status=%q", got) + } + if len(sub.send) != 1 { + t.Errorf("scanner received %d jobs, want 1", len(sub.send)) + } +} + +// TestScanReDispatch_LeavesFreshPendingJobAlone stops the loop from racing the +// dispatch Enqueue just did, which would scan the same manifest twice. +func TestScanReDispatch_LeavesFreshPendingJobAlone(t *testing.T) { + sb := newTestScanBroadcaster(t) + sub := newTestScanSubscriber(t, sb, 4) + seedPendingJobs(t, sb, 1) + + sb.reDispatchTimedOut() + + if got := jobStatus(t, sb, 1); got != "pending" { + t.Errorf("just-enqueued job was re-dispatched, status=%q", got) + } + if len(sub.send) != 0 { + t.Errorf("scanner received %d jobs, want 0", len(sub.send)) + } +} + +// TestScanReDispatch_PendingSurvivesWithNoScanner verifies the reclaim is a +// no-op rather than a corruption when no scanner is connected: the row stays +// pending and gets offered again on the next tick. +func TestScanReDispatch_PendingSurvivesWithNoScanner(t *testing.T) { + sb := newTestScanBroadcaster(t) + seedPendingJobs(t, sb, 1) + backdateJob(t, sb, 1, 30) + + sb.reDispatchTimedOut() + + if got := jobStatus(t, sb, 1); got != "pending" { + t.Errorf("job status changed with no scanner connected, status=%q", got) + } +} + +// TestScanDispatchJob_SkipsClaimedJob covers the double-send window opened by +// having two dispatchers for pending rows: whichever loses the UPDATE race must +// not also push the job onto a scanner's queue. +func TestScanDispatchJob_SkipsClaimedJob(t *testing.T) { + sb := newTestScanBroadcaster(t) + sub := newTestScanSubscriber(t, sb, 4) + seedPendingJobs(t, sb, 1) + + // Another dispatcher got there first. + if _, err := sb.db.Exec(`UPDATE scan_jobs SET status='assigned', assigned_to='other' WHERE seq = 1`); err != nil { + t.Fatalf("claim: %v", err) + } + + sb.dispatchJob(&ScanJobEvent{Type: "job", Seq: 1, Repository: "repo"}) + + if len(sub.send) != 0 { + t.Errorf("already-claimed job was dispatched again (%d sends)", len(sub.send)) + } + var assignedTo string + if err := sb.db.QueryRow(`SELECT assigned_to FROM scan_jobs WHERE seq = 1`).Scan(&assignedTo); err != nil { + t.Fatalf("query assigned_to: %v", err) + } + if assignedTo != "other" { + t.Errorf("assignment stolen from the first dispatcher, assigned_to=%q", assignedTo) + } +} diff --git a/scanner/cmd/scanner/main.go b/scanner/cmd/scanner/main.go index 0491392..b2d8c01 100644 --- a/scanner/cmd/scanner/main.go +++ b/scanner/cmd/scanner/main.go @@ -12,6 +12,7 @@ import ( "github.com/spf13/cobra" + "atcr.io/pkg/logging" "atcr.io/scanner/internal/client" "atcr.io/scanner/internal/config" "atcr.io/scanner/internal/queue" @@ -50,6 +51,20 @@ Environment variables always override file values (SCANNER_ prefix).`, return fmt.Errorf("failed to load config: %w", err) } + // Without this the scanner runs on slog's default handler: log_level and + // log_shipper are silently ignored, nothing is shipped anywhere, and an + // idle scanner is indistinguishable from a wedged one. + logging.InitLoggerWithShipper(cfg.LogLevel, logging.ShipperConfig{ + Backend: cfg.LogShipper.Backend, + URL: cfg.LogShipper.URL, + BatchSize: cfg.LogShipper.BatchSize, + FlushInterval: cfg.LogShipper.FlushInterval, + Service: "scanner", + Username: cfg.LogShipper.Username, + Password: cfg.LogShipper.Password, + }) + defer logging.Shutdown() + slog.Info("Starting ATCR scanner", "hold_url", cfg.Hold.URL, "workers", cfg.Scanner.Workers, diff --git a/scanner/internal/scan/grype.go b/scanner/internal/scan/grype.go index 8df8d14..956b60f 100644 --- a/scanner/internal/scan/grype.go +++ b/scanner/internal/scan/grype.go @@ -125,11 +125,11 @@ func scanVulnerabilities(ctx context.Context, s *sbom.SBOM, vulnDBPath string) ( "low", summary.Low, "total", summary.Total) - report := map[string]interface{}{ + report := map[string]any{ "matches": allMatches.Sorted(), "source": s.Source, "distro": s.Artifacts.LinuxDistribution, - "descriptor": map[string]interface{}{ + "descriptor": map[string]any{ "name": "grype", "version": "v0.107.1", }, diff --git a/scanner/internal/scan/worker.go b/scanner/internal/scan/worker.go index b5f35d1..4b140e4 100644 --- a/scanner/internal/scan/worker.go +++ b/scanner/internal/scan/worker.go @@ -9,6 +9,7 @@ import ( "log/slog" "os" "runtime" + "strings" "sync" "time" @@ -151,6 +152,37 @@ var unscannableConfigTypes = map[string]bool{ "application/vnd.dsse.envelope.v1+json": true, // DSSE envelopes (SLSA) } +// skipReason reports why a job cannot be scanned, or "" when it can be. +func skipReason(job *scanner.ScanJob) string { + if unscannableConfigTypes[job.Config.MediaType] { + return fmt.Sprintf("unscannable artifact type %s", job.Config.MediaType) + } + + // A buildx attestation manifest carries an ordinary image config with a + // single in-toto or DSSE payload as its layer, so the config media type + // alone does not identify it. buildOCILayout drops every non-tar layer, + // which would hand Syft an image with nothing in it. + if len(job.Layers) > 0 && !hasScannableLayer(job.Layers) { + return fmt.Sprintf("no scannable layers (%s)", job.Layers[0].MediaType) + } + + return "" +} + +// hasScannableLayer mirrors the layer filter in buildOCILayout: anything that +// is not a tar of some flavour is not something Syft can read. +func hasScannableLayer(layers []scanner.BlobDescriptor) bool { + for _, layer := range layers { + if layer.Digest == "" { + continue + } + if layer.MediaType == "" || strings.Contains(layer.MediaType, "tar") { + return true + } + } + return false +} + func (wp *WorkerPool) processJob(ctx context.Context, job *scanner.ScanJob) (*scanner.ScanResult, error) { startTime := time.Now() @@ -158,8 +190,8 @@ func (wp *WorkerPool) processJob(ctx context.Context, job *scanner.ScanJob) (*sc // Returning *SkipError tells the worker dispatch loop to send a "skipped" // message rather than an "error" — the hold marks these records as // permanently skipped and won't retry them on the rescan interval. - if unscannableConfigTypes[job.Config.MediaType] { - return nil, &SkipError{Reason: fmt.Sprintf("unscannable artifact type %s", job.Config.MediaType)} + if reason := skipReason(job); reason != "" { + return nil, &SkipError{Reason: reason} } // Ensure tmp dir exists diff --git a/scanner/internal/scan/worker_skip_test.go b/scanner/internal/scan/worker_skip_test.go new file mode 100644 index 0000000..b6e3e14 --- /dev/null +++ b/scanner/internal/scan/worker_skip_test.go @@ -0,0 +1,83 @@ +package scan + +import ( + "testing" + + scanner "atcr.io/scanner" +) + +// TestSkipReason covers the artifact shapes the scanner must refuse before it +// spends a download on them. The attestation case is the one that reached +// production: an in-toto SLSA provenance manifest carries an ordinary image +// config, so the config media type check alone waved it through and Syft was +// handed an OCI layout with no layers in it. +func TestSkipReason(t *testing.T) { + tests := []struct { + name string + configType string + layerTypes []string + wantSkip bool + }{ + { + name: "container image", + configType: "application/vnd.oci.image.config.v1+json", + layerTypes: []string{"application/vnd.oci.image.layer.v1.tar+gzip"}, + }, + { + name: "docker image", + configType: "application/vnd.docker.container.image.v1+json", + layerTypes: []string{"application/vnd.docker.image.rootfs.diff.tar.gzip"}, + }, + { + name: "layer media type absent", + configType: "application/vnd.oci.image.config.v1+json", + layerTypes: []string{""}, + }, + { + name: "helm chart", + configType: "application/vnd.cncf.helm.config.v1+json", + layerTypes: []string{"application/vnd.cncf.helm.chart.content.v1.tar+gzip"}, + wantSkip: true, + }, + { + name: "in-toto attestation with an image config", + configType: "application/vnd.oci.image.config.v1+json", + layerTypes: []string{"application/vnd.in-toto+json"}, + wantSkip: true, + }, + { + name: "dsse envelope layer", + configType: "application/vnd.oci.image.config.v1+json", + layerTypes: []string{"application/vnd.dsse.envelope.v1+json"}, + wantSkip: true, + }, + { + name: "mixed layers keep the scannable one", + configType: "application/vnd.oci.image.config.v1+json", + layerTypes: []string{"application/vnd.in-toto+json", "application/vnd.oci.image.layer.v1.tar"}, + }, + { + name: "no layers at all is left to the pipeline", + configType: "application/vnd.oci.image.config.v1+json", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + job := &scanner.ScanJob{ + Config: scanner.BlobDescriptor{MediaType: tt.configType, Digest: "sha256:config"}, + } + for i, mt := range tt.layerTypes { + job.Layers = append(job.Layers, scanner.BlobDescriptor{ + MediaType: mt, + Digest: "sha256:layer" + string(rune('a'+i)), + }) + } + + reason := skipReason(job) + if got := reason != ""; got != tt.wantSkip { + t.Errorf("skipReason = %q, wantSkip=%v", reason, tt.wantSkip) + } + }) + } +} diff --git a/scanner/types.go b/scanner/types.go index 03e4fd2..4bf9000 100644 --- a/scanner/types.go +++ b/scanner/types.go @@ -6,16 +6,16 @@ import "encoding/json" // ScanJob represents a vulnerability scanning job received from the hold service type ScanJob struct { - Seq int64 `json:"seq"` - ManifestDigest string `json:"manifestDigest"` - Repository string `json:"repository"` - Tag string `json:"tag"` - UserDID string `json:"userDid"` - UserHandle string `json:"userHandle"` - HoldDID string `json:"holdDid"` - HoldEndpoint string `json:"holdEndpoint"` - Tier string `json:"tier"` - Config BlobDescriptor `json:"config"` + Seq int64 `json:"seq"` + ManifestDigest string `json:"manifestDigest"` + Repository string `json:"repository"` + Tag string `json:"tag"` + UserDID string `json:"userDid"` + UserHandle string `json:"userHandle"` + HoldDID string `json:"holdDid"` + HoldEndpoint string `json:"holdEndpoint"` + Tier string `json:"tier"` + Config BlobDescriptor `json:"config"` Layers []BlobDescriptor `json:"layers"` }