diff --git a/.env.example b/.env.example index b92fab1..933ac5a 100644 --- a/.env.example +++ b/.env.example @@ -12,5 +12,7 @@ DATABASE_URL=postgres://ctcloud:change-me-strong@postgres:5432/ctcloud?sslmode=d DNS_SERVERS=1.1.1.1:53,8.8.8.8:53,1.0.0.1:53 DNS_WORKERS=16 +DNS_CACHE_SIZE=262144 +DEDUP_SIZE=262144 LOG_LEVEL=info ACTIVE_PROBING=false diff --git a/README.md b/README.md index ac2e014..4a2d0a3 100644 --- a/README.md +++ b/README.md @@ -111,7 +111,8 @@ Environment variables only; no config files, no runtime pattern files. | `DNS_SERVERS` | `1.1.1.1:53,8.8.8.8:53,1.0.0.1:53` | comma-separated resolvers (round-robin + failover); `system` for the host resolver | | `DNS_WORKERS` | `16` | DNS enrichment workers | | `DNS_TIMEOUT` | `5s` | per-lookup timeout | -| `DNS_CACHE_SIZE` | `1048576` | max DNS cache entries | +| `DNS_CACHE_SIZE` | `262144` | max DNS cache entries | +| `DEDUP_SIZE` | `262144` | max entries in each dedup cache | | `DNS_MAX_CNAME_DEPTH` | `8` | max CNAME hops | | `DNS_POSITIVE_TTL` / `DNS_NEGATIVE_TTL` | `10m` / `1m` | DNS cache TTLs | | `DIRECT_QUEUE_SIZE` | `8192` | direct-match queue capacity | diff --git a/cmd/ctcloud/main.go b/cmd/ctcloud/main.go index 3b723e7..18dcd17 100644 --- a/cmd/ctcloud/main.go +++ b/cmd/ctcloud/main.go @@ -114,6 +114,7 @@ func run() error { BatchSize: cfg.DBBatchSize, FlushInterval: cfg.DBFlushInterval, DedupTTL: cfg.DedupeTTL, + DedupSize: cfg.DedupSize, Enricher: enricher, Write: write, Metrics: m, diff --git a/internal/config/config.go b/internal/config/config.go index 07285a5..9cdbcc4 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -22,6 +22,8 @@ type Config struct { DNSPositiveTTL time.Duration DNSNegativeTTL time.Duration + DedupSize int + DirectQueueSize int DNSQueueSize int DBQueueSize int @@ -57,7 +59,8 @@ func Load() (*Config, error) { DNSServers: envList("DNS_SERVERS", DefaultDNSServers), DNSWorkers: envInt("DNS_WORKERS", 16), DNSTimeout: envDuration("DNS_TIMEOUT", 5*time.Second), - DNSCacheSize: envInt("DNS_CACHE_SIZE", 1<<20), + DNSCacheSize: envInt("DNS_CACHE_SIZE", 1<<18), + DedupSize: envInt("DEDUP_SIZE", 1<<18), DNSMaxCNAMEdepth: envInt("DNS_MAX_CNAME_DEPTH", 8), DNSPositiveTTL: envDuration("DNS_POSITIVE_TTL", 10*time.Minute), DNSNegativeTTL: envDuration("DNS_NEGATIVE_TTL", 1*time.Minute), diff --git a/internal/dns/cache.go b/internal/dns/cache.go index 67e4c4b..ab9dc75 100644 --- a/internal/dns/cache.go +++ b/internal/dns/cache.go @@ -41,7 +41,11 @@ func NewTTLCache[V any](maxEntries, shards int) *TTLCache[V] { c := &TTLCache[V]{mask: uint64(n - 1), maxShard: maxShard} c.shards = make([]shard[V], n) for i := range c.shards { - c.shards[i].m = make(map[string]cacheEntry[V], maxShard) + // Deliberately no capacity hint: pre-sizing every shard to maxShard + // allocates the full theoretical capacity (hundreds of MB across the + // caches) up front. Let the maps grow on demand; maxShard still bounds + // them via eviction. + c.shards[i].m = make(map[string]cacheEntry[V]) } return c } diff --git a/internal/pipeline/pipeline.go b/internal/pipeline/pipeline.go index bb61a39..f2cfebf 100644 --- a/internal/pipeline/pipeline.go +++ b/internal/pipeline/pipeline.go @@ -208,20 +208,6 @@ func appendIPs(f *Finding, ips []net.IP) { } } -func (p *Pipeline) enrich(ctx context.Context, c model.Candidate) Finding { - f := Finding{Candidate: c} - if p.cfg.Enricher == nil { - return f - } - res, ok := p.cfg.Enricher.Resolve(ctx, c.Hostname) - if !ok { - return f - } - f.CNAMEChain = res.Chain - appendIPs(&f, res.IPs) - return f -} - func (p *Pipeline) directWorker(ctx context.Context) { defer p.producer.Done() for { @@ -232,7 +218,12 @@ func (p *Pipeline) directWorker(ctx context.Context) { if !ok { return } - p.pushDB(ctx, p.enrich(ctx, c)) + // Direct provider matches are written immediately and are never + // queued behind optional DNS work, so they cannot be shed in favour + // of speculative CNAME enrichment. The matched provider hostname is + // the finding; A/AAAA enrichment is a DNS-bound luxury that the + // shared resolver cannot keep up with at these volumes. + p.pushDB(ctx, Finding{Candidate: c}) } } } diff --git a/internal/pipeline/pipeline_test.go b/internal/pipeline/pipeline_test.go index af6ab5a..a019921 100644 --- a/internal/pipeline/pipeline_test.go +++ b/internal/pipeline/pipeline_test.go @@ -195,6 +195,61 @@ func TestCNAMEMatchIPv6(t *testing.T) { } } +type blockingResolver struct{ release chan struct{} } + +func (b *blockingResolver) LookupCNAME(ctx context.Context, host string) (string, error) { + select { + case <-b.release: + case <-ctx.Done(): + } + return host, nil +} + +func (b *blockingResolver) LookupIP(ctx context.Context, host string) ([]net.IP, error) { + return nil, nil +} + +// TestDirectNotStarvedByDNSBacklog verifies direct provider matches are written +// even while the DNS queue is saturated with optional CNAME work. +func TestDirectNotStarvedByDNSBacklog(t *testing.T) { + col := &collector{} + release := make(chan struct{}) + p := New(Config{ + DNSQueueSize: 1, + DNSWorkers: 1, + Enricher: newEnricher(&blockingResolver{release: release}), + Write: col.write, + BatchSize: 10, + FlushInterval: 10 * time.Millisecond, + }) + ctx, cancel := context.WithCancel(context.Background()) + p.Start(ctx) + for i := 0; i < 100; i++ { + p.HandleHost("api" + strconv.Itoa(i) + ".example.com") // fills/blocks DNS queue + } + p.HandleHost("bucket.s3.eu-central-1.amazonaws.com") + + found := false + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + for _, f := range col.all() { + if f.Hostname == "bucket.s3.eu-central-1.amazonaws.com" { + found = true + } + } + if found { + break + } + time.Sleep(5 * time.Millisecond) + } + close(release) + cancel() + p.Close() + if !found { + t.Fatal("direct match was not written while the DNS queue was saturated") + } +} + func TestQueueSaturationDrops(t *testing.T) { m := metrics.New() p := New(Config{DirectQueueSize: 1, DNSQueueSize: 1, Metrics: m})