From f53d394cbe6b5205fa4449754132778a1652f192 Mon Sep 17 00:00:00 2001 From: Andy Aylward Date: Wed, 2 Sep 2026 09:36:00 -0400 Subject: [PATCH 1/2] stats: where scraper, bot, and scanner traffic comes from (#1467) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Each request's client address is placed in a country from DB-IP's free country CSV, loaded at boot from the stats bucket into a sorted range table — no library, a binary search is the whole lookup — and geo_stats keys day, host, agent class, and country with request, 403, and probe counts. Served at /stats/v1/countries. GEO_DB_KEY names the object; unset, every row reads "--", and a key that fails to load fails boot. --- deploy/consolidated/README.md | 4 +- deploy/consolidated/compose.yaml | 3 + domains/platform/apis/stats/BUILD.bazel | 2 + domains/platform/apis/stats/README.md | 21 ++- domains/platform/apis/stats/aggregate.go | 39 +++-- domains/platform/apis/stats/aggregate_test.go | 59 +++++++ domains/platform/apis/stats/api.go | 12 ++ domains/platform/apis/stats/api_test.go | 35 ++++ domains/platform/apis/stats/geo.go | 159 ++++++++++++++++++ domains/platform/apis/stats/geo_test.go | 115 +++++++++++++ domains/platform/apis/stats/loop.go | 5 + domains/platform/apis/stats/loop_test.go | 19 +++ domains/platform/apis/stats/main/main.go | 43 +++-- domains/platform/apis/stats/store.go | 58 ++++++- domains/platform/apis/stats/store_test.go | 27 +++ 15 files changed, 573 insertions(+), 28 deletions(-) create mode 100644 domains/platform/apis/stats/geo.go create mode 100644 domains/platform/apis/stats/geo_test.go diff --git a/deploy/consolidated/README.md b/deploy/consolidated/README.md index a619e5fe..b81dce9c 100644 --- a/deploy/consolidated/README.md +++ b/deploy/consolidated/README.md @@ -220,7 +220,9 @@ COMPOSE_PROFILES=stats deploy.sh runs compose in `~`, where compose reads that file, so every normal deploy includes the trio. Alongside it live `STATS_AWS_ACCESS_KEY_ID`, `STATS_AWS_SECRET_ACCESS_KEY`, -`STATS_S3_BUCKET`, `STATS_S3_REGION`, and `STATS_DB_PASSWORD` (same URL-safe rules as the +`STATS_S3_BUCKET`, `STATS_S3_REGION`, optionally `STATS_GEO_DB_KEY` (the key of +DB-IP's country CSV in that bucket; unset, the geo rows all read `--`), and +`STATS_DB_PASSWORD` (same URL-safe rules as the other database passwords: it rides in a libpq URL and a single-quoted SQL literal). Without the `COMPOSE_PROFILES` line the containers keep running after a deploy but silently stop being updated — compose ignores profile-gated services on an unflagged `up -d`. diff --git a/deploy/consolidated/compose.yaml b/deploy/consolidated/compose.yaml index d0af252b..15290b93 100644 --- a/deploy/consolidated/compose.yaml +++ b/deploy/consolidated/compose.yaml @@ -662,6 +662,9 @@ services: - AWS_SECRET_ACCESS_KEY=${STATS_AWS_SECRET_ACCESS_KEY} - S3_BUCKET=${STATS_S3_BUCKET} - S3_REGION=${STATS_S3_REGION:-us-east-1} + # DB-IP's country CSV in the same bucket (#1467); empty means every + # geo row reads "--". A key that is set has to load, or boot fails. + - GEO_DB_KEY=${STATS_GEO_DB_KEY:-} - PORT=8092 networks: - app_network diff --git a/domains/platform/apis/stats/BUILD.bazel b/domains/platform/apis/stats/BUILD.bazel index 0b833936..c1d3af27 100644 --- a/domains/platform/apis/stats/BUILD.bazel +++ b/domains/platform/apis/stats/BUILD.bazel @@ -10,6 +10,7 @@ go_library( "aggregate.go", "api.go", "classify.go", + "geo.go", "loop.go", "queries.go", "store.go", @@ -33,6 +34,7 @@ go_test( "aggregate_test.go", "api_test.go", "classify_test.go", + "geo_test.go", "loop_test.go", "queries_test.go", "store_test.go", diff --git a/domains/platform/apis/stats/README.md b/domains/platform/apis/stats/README.md index 755838eb..a63a0985 100644 --- a/domains/platform/apis/stats/README.md +++ b/domains/platform/apis/stats/README.md @@ -56,9 +56,20 @@ The pass runs while the API serves, so for its length the counts climb back up from zero — minutes at this scale, and the log says when it is done. -What stays ad hoc: IP-range clusters (a /24 key is caller-shaped and -unbounded) and geography (nothing in the repo maps addresses to -countries). Both are one query over the raw partitions in S3, which keep +The geo rollup (#1467) places each request's `client_ip` in a country +and keys `geo_stats` on day, host, agent class, and the two-letter code, +with request, 403, and probe counts — where the scrapers, bots, and +scanners come from. The database is DB-IP's free country CSV +(`dbip-country-lite-YYYY-MM.csv.gz`, CC BY 4.0, attribution on the +dashboard), uploaded by the operator to the stats bucket under the key +`GEO_DB_KEY` names; the service loads it at boot into a sorted range +table and binary-searches it, no library. An address outside every range, +or no database at all, files under `--`. A new monthly file is a restart. +Rows aggregated before the database was uploaded stay `--` until a +re-aggregation (bump `RollupVersion`). + +What stays ad hoc: IP-range clusters — a /24 key is caller-shaped and +unbounded — which is one query over the raw partitions in S3, keeping `request.remote_ip` and `request.client_ip` per line. ## The API @@ -77,6 +88,8 @@ countries). Both are one query over the raw partitions in S3, which keep Rows from before #1468 landed on `api.muchq.com` and `gpt.muchq.com` overcount served, and the summary's error count there rose with the change, because scanner traffic now gets the 404 it always deserved. +- `GET /stats/v1/countries?days=30&limit=2000` — per host/class/country + request, 403, and probe counts, busiest first - `GET /stats/v1/one_d4/queries?days=30` — one_d4 queries per day/entry/source/outcome/cache - `GET /stats/v1/one_d4/terms?days=30&limit=200` — which fields, motifs, @@ -89,7 +102,7 @@ stay in the log, not on the wire. ## Configuration `STATS_DB_URL` (postgres), `S3_BUCKET`, `S3_REGION`, `AWS_ACCESS_KEY_ID`, -`AWS_SECRET_ACCESS_KEY` — the same stats IAM user the shipper writes with, +`AWS_SECRET_ACCESS_KEY`, and optionally `GEO_DB_KEY` — the same stats IAM user the shipper writes with, which therefore needs `s3:GetObject` and `s3:ListBucket` on the `logs/*` prefix as well as `s3:PutObject`. `AGGREGATE_INTERVAL` and `PORT` (default 8092) are optional. diff --git a/domains/platform/apis/stats/aggregate.go b/domains/platform/apis/stats/aggregate.go index 4808a7f5..f9217b0d 100644 --- a/domains/platform/apis/stats/aggregate.go +++ b/domains/platform/apis/stats/aggregate.go @@ -59,11 +59,15 @@ const ( // so a crash between the two reprocesses the object rather than losing or // double-counting it. type Rollup struct { - Requests map[RequestKey]int64 - Slugs map[SlugKey]int64 - Probes map[ProbeKey]int64 - Queries map[QueryKey]int64 - Terms map[TermKey]int64 + Requests map[RequestKey]int64 + Slugs map[SlugKey]int64 + Probes map[ProbeKey]int64 + Queries map[QueryKey]int64 + Terms map[TermKey]int64 + Countries map[GeoKey]GeoStat + + // Geo places client addresses; NoLocator when no database is loaded. + Geo Locator tailAgents map[string]bool } @@ -75,6 +79,8 @@ func NewRollup() *Rollup { Probes: map[ProbeKey]int64{}, Queries: map[QueryKey]int64{}, Terms: map[TermKey]int64{}, + Countries: map[GeoKey]GeoStat{}, + Geo: NoLocator{}, tailAgents: map[string]bool{}, } } @@ -100,13 +106,24 @@ func (r *Rollup) boundedAgent(class, agent string) string { type caddyLine struct { Status int `json:"status"` Request struct { - Host string `json:"host"` - Method string `json:"method"` - URI string `json:"uri"` - Headers map[string][]string `json:"headers"` + Host string `json:"host"` + Method string `json:"method"` + URI string `json:"uri"` + ClientIP string `json:"client_ip"` + RemoteIP string `json:"remote_ip"` + Headers map[string][]string `json:"headers"` } `json:"request"` } +// Caddy is the edge, so client_ip and remote_ip agree; older lines carry +// only remote_ip. +func (l *caddyLine) clientIP() string { + if l.Request.ClientIP != "" { + return l.Request.ClientIP + } + return l.Request.RemoteIP +} + func (l *caddyLine) userAgent() string { values := l.Request.Headers["User-Agent"] if len(values) == 0 { @@ -158,9 +175,11 @@ func (r *Rollup) Consume(reader io.Reader, date string) (skipped int, err error) AgentClass: agentClass, Agent: r.boundedAgent(agentClass, agent), }]++ - if probe := ProbeOf(parsed.Request.URI); probe != "" { + probe := ProbeOf(parsed.Request.URI) + if probe != "" { r.Probes[ProbeKey{Date: date, Host: parsed.Request.Host, Probe: probe, Status: parsed.Status}]++ } + r.addGeo(date, parsed.Request.Host, agentClass, parsed.clientIP(), parsed.Status, probe != "") if slug := SlugOf(parsed.Request.Host, parsed.Request.Method, parsed.Request.URI); slug != "" { r.Slugs[SlugKey{Date: date, Slug: slug, Status: parsed.Status}]++ } diff --git a/domains/platform/apis/stats/aggregate_test.go b/domains/platform/apis/stats/aggregate_test.go index 18b5562f..1ef92b20 100644 --- a/domains/platform/apis/stats/aggregate_test.go +++ b/domains/platform/apis/stats/aggregate_test.go @@ -112,6 +112,65 @@ func TestConsumeCapsTheAnonymousAgentTailPerObject(t *testing.T) { } } +type fakeLocator map[string]string + +func (f fakeLocator) Country(ip string) string { return f[ip] } + +func TestConsumeRollsUpWhereEachClassCameFromWithBlocksAndProbes(t *testing.T) { + lines := strings.Join([]string{ + `{"status":200,"request":{"host":"h","method":"GET","uri":"/","client_ip":"1.0.0.7","headers":{"User-Agent":["Mozilla/5.0 (Macintosh) Chrome/126.0"]}}}`, + `{"status":403,"request":{"host":"h","method":"GET","uri":"/x","client_ip":"57.141.3.4","headers":{"User-Agent":["meta-externalagent/1.1"]}}}`, + `{"status":404,"request":{"host":"h","method":"GET","uri":"/.env","client_ip":"195.178.110.199","headers":{"User-Agent":["TLM-Audit-Scanner/1.0"]}}}`, + `{"status":404,"request":{"host":"h","method":"GET","uri":"/wp-login.php","client_ip":"195.178.110.199","headers":{"User-Agent":["TLM-Audit-Scanner/1.0"]}}}`, + // Older lines carry remote_ip only; an address the locator does not know is "--". + `{"status":200,"request":{"host":"h","method":"GET","uri":"/","remote_ip":"10.1.2.3","headers":{"User-Agent":["curl/8.6.0"]}}}`, + }, "\n") + rollup := NewRollup() + rollup.Geo = fakeLocator{"1.0.0.7": "AU", "57.141.3.4": "US", "195.178.110.199": "GB"} + + if _, err := rollup.Consume(strings.NewReader(lines), "2026-08-30"); err != nil { + t.Fatal(err) + } + + want := map[GeoKey]GeoStat{ + {"2026-08-30", "h", AgentBrowser, "AU"}: {1, 0, 0}, + {"2026-08-30", "h", AgentAIScraper, "US"}: {1, 1, 0}, + {"2026-08-30", "h", AgentBot, "GB"}: {2, 0, 2}, + {"2026-08-30", "h", AgentBot, "--"}: {1, 0, 0}, + } + if len(rollup.Countries) != len(want) { + t.Errorf("geo rows = %v, want %v", rollup.Countries, want) + } + for key, stat := range want { + if rollup.Countries[key] != stat { + t.Errorf("geo %+v = %+v, want %+v", key, rollup.Countries[key], stat) + } + } +} + +// Without a database every row is "--"; the class split still holds, so the +// table is honest rather than empty. +func TestConsumeWithoutAGeoDatabaseFilesEverythingUnderUnknown(t *testing.T) { + rollup := NewRollup() + if _, err := rollup.Consume(strings.NewReader(sampleLines), "2026-08-30"); err != nil { + t.Fatal(err) + } + var total int64 + for key, stat := range rollup.Countries { + if key.Country != UnknownCountry { + t.Errorf("row %+v placed without a database", key) + } + total += stat.Requests + } + var requests int64 + for _, count := range rollup.Requests { + requests += count + } + if total != requests { + t.Errorf("geo rows count %d requests, the request rollup %d; every request is somewhere", total, requests) + } +} + func TestConsumeSurvivesOversizedLines(t *testing.T) { huge := `{"status":200,"request":{"host":"x","method":"GET","uri":"/` + strings.Repeat("a", 2*1024*1024) + `","headers":{}}}` diff --git a/domains/platform/apis/stats/api.go b/domains/platform/apis/stats/api.go index 5ec25cfa..5cec34c1 100644 --- a/domains/platform/apis/stats/api.go +++ b/domains/platform/apis/stats/api.go @@ -17,6 +17,7 @@ type Reader interface { Probes(ctx context.Context, days int) ([]ProbeRow, error) Queries(ctx context.Context, days int) ([]QueryRow, error) QueryTerms(ctx context.Context, days, limit int) ([]TermRow, error) + Countries(ctx context.Context, days, limit int) ([]CountryRow, error) } type Handlers struct { @@ -113,6 +114,17 @@ func (h *Handlers) GetQueryTerms(w http.ResponseWriter, r *http.Request) { writeJSON(w, map[string]any{"days": days, "rows": emptyIfNil(rows)}) } +func (h *Handlers) GetCountries(w http.ResponseWriter, r *http.Request) { + days := queryInt(r, "days", 30, 365) + limit := queryInt(r, "limit", 2000, 5000) + rows, err := h.reader.Countries(r.Context(), days, limit) + if err != nil { + h.serverError(w, "countries", err) + return + } + writeJSON(w, map[string]any{"days": days, "rows": emptyIfNil(rows)}) +} + func (h *Handlers) serverError(w http.ResponseWriter, what string, err error) { h.logger.Error("stats query failed", "query", what, "error", err) // The reason goes to the log, not the wire: these endpoints are public. diff --git a/domains/platform/apis/stats/api_test.go b/domains/platform/apis/stats/api_test.go index 59e0fd6c..38b3ecd1 100644 --- a/domains/platform/apis/stats/api_test.go +++ b/domains/platform/apis/stats/api_test.go @@ -19,6 +19,7 @@ type fakeReader struct { probes []ProbeRow queries []QueryRow terms []TermRow + countries []CountryRow lastDays int lastLimit int fail bool @@ -72,6 +73,14 @@ func (f *fakeReader) QueryTerms(_ context.Context, days, limit int) ([]TermRow, return f.terms, nil } +func (f *fakeReader) Countries(_ context.Context, days, limit int) ([]CountryRow, error) { + if f.fail { + return nil, errors.New("db is having a day") + } + f.lastDays, f.lastLimit = days, limit + return f.countries, nil +} + func handlersWith(reader *fakeReader) *Handlers { return NewHandlers(reader, slog.New(slog.NewTextHandler(io.Discard, nil))) } @@ -247,3 +256,29 @@ func TestOneD4QueryEndpointsShareTheWindowRules(t *testing.T) { } } } + +func TestCountriesShareTheWindowRules(t *testing.T) { + reader := &fakeReader{countries: []CountryRow{ + {Host: "git.muchq.com", AgentClass: AgentAIScraper, Country: "US", Requests: 9, Blocked: 9, Probes: 0}, + }} + handlers := handlersWith(reader) + + _, body := get(t, handlers.GetCountries, "/stats/v1/countries") + if reader.lastDays != 30 || reader.lastLimit != 2000 { + t.Errorf("default countries (days, limit) = (%d, %d), want (30, 2000)", reader.lastDays, reader.lastLimit) + } + row := body["rows"].([]any)[0].(map[string]any) + if row["country"] != "US" || row["blocked"] != float64(9) || row["probes"] != float64(0) { + t.Errorf("country row = %v", row) + } + get(t, handlers.GetCountries, "/stats/v1/countries?days=99999&limit=99999") + if reader.lastDays != 365 || reader.lastLimit != 5000 { + t.Errorf("clamped countries (days, limit) = (%d, %d), want (365, 5000)", reader.lastDays, reader.lastLimit) + } + if _, body := get(t, handlersWith(&fakeReader{}).GetCountries, "/x"); body["rows"] == nil { + t.Error("countries rows serialized as null; want []") + } + if recorder, _ := get(t, handlersWith(&fakeReader{fail: true}).GetCountries, "/x"); recorder.Code != http.StatusInternalServerError { + t.Errorf("countries on a failing store = %d, want 500", recorder.Code) + } +} diff --git a/domains/platform/apis/stats/geo.go b/domains/platform/apis/stats/geo.go new file mode 100644 index 00000000..2c16bd84 --- /dev/null +++ b/domains/platform/apis/stats/geo.go @@ -0,0 +1,159 @@ +package stats + +import ( + "bufio" + "compress/gzip" + "encoding/csv" + "fmt" + "io" + "net/netip" + "sort" + "strings" +) + +// Locator answers which country an address is in, as an ISO 3166-1 alpha-2 +// code, or "" when it does not know. The rollup keys geo rows on the +// answer, so the vocabulary is the ~250 codes plus UnknownCountry. +type Locator interface { + Country(ip string) string +} + +// UnknownCountry is the geo row for an address no locator placed: a private +// range, a malformed field, or no database loaded at all. +const UnknownCountry = "--" + +// NoLocator places nothing; the geo rows all read UnknownCountry. +type NoLocator struct{} + +func (NoLocator) Country(string) string { return "" } + +// Geo is an in-memory range table built from DB-IP's country CSV +// (dbip-country-lite: start,end,country per line, IPv4 and IPv6 mixed). +// Some hundreds of thousands of ranges, binary-searched; no database +// library, because a sorted slice is the whole data structure. +type Geo struct { + ranges []ipRange // sorted by start, non-overlapping +} + +type ipRange struct { + start, end netip.Addr + country string +} + +// ParseDBIP reads the CSV. Lines that do not parse are counted and +// skipped rather than failing the load — one bad row in a vendor file +// must not switch geo off — but a file with no usable rows is an error, +// since that is a wrong object, not a bad line. +func ParseDBIP(r io.Reader) (*Geo, int, error) { + reader := csv.NewReader(bufio.NewReader(r)) + reader.FieldsPerRecord = -1 + reader.ReuseRecord = true + var ranges []ipRange + skipped := 0 + for { + record, err := reader.Read() + if err == io.EOF { + break + } + if err != nil { + return nil, skipped, fmt.Errorf("reading geo csv: %w", err) + } + if len(record) < 3 { + skipped++ + continue + } + start, err1 := netip.ParseAddr(strings.TrimSpace(record[0])) + end, err2 := netip.ParseAddr(strings.TrimSpace(record[1])) + country := strings.ToUpper(strings.TrimSpace(record[2])) + if err1 != nil || err2 != nil || len(country) != 2 || start.Is4() != end.Is4() || end.Less(start) { + skipped++ + continue + } + ranges = append(ranges, ipRange{start: start.Unmap(), end: end.Unmap(), country: country}) + } + if len(ranges) == 0 { + return nil, skipped, fmt.Errorf("geo csv holds no usable ranges (%d lines skipped)", skipped) + } + sort.Slice(ranges, func(i, j int) bool { return ranges[i].start.Less(ranges[j].start) }) + return &Geo{ranges: ranges}, skipped, nil +} + +// Country finds the range holding ip. Addresses come from Caddy's log as +// text; anything that does not parse is "", like an address in no range. +func (g *Geo) Country(ip string) string { + addr, err := netip.ParseAddr(ip) + if err != nil { + return "" + } + addr = addr.Unmap() + // The first range starting after addr; the candidate is the one before it. + // netip orders every IPv4 address before every IPv6 one, so a v6 address + // whose predecessor is a v4 range fails the end check like any other gap. + i := sort.Search(len(g.ranges), func(i int) bool { return addr.Less(g.ranges[i].start) }) + if i == 0 { + return "" + } + candidate := g.ranges[i-1] + if candidate.end.Less(addr) { + return "" + } + return candidate.country +} + +// LoadGeo reads the CSV object (gzipped when the key says so) from the +// bucket the logs live in. The key is the operator's: DB-IP publishes a +// new file monthly under CC BY 4.0, and the dashboard carries the +// attribution. +func LoadGeo(objects ObjectStore, key string) (*Geo, int, error) { + body, err := objects.Get(key) + if err != nil { + return nil, 0, fmt.Errorf("fetching geo database %s: %w", key, err) + } + defer body.Close() + var reader io.Reader = body + if strings.HasSuffix(key, ".gz") { + gz, err := gzip.NewReader(body) + if err != nil { + return nil, 0, fmt.Errorf("geo database %s is not gzip: %w", key, err) + } + defer gz.Close() + reader = gz + } + return ParseDBIP(reader) +} + +// GeoKey is one row of the per-day geo rollup: where a host's traffic of +// each class came from. Bounded by the country vocabulary times the four +// classes. +type GeoKey struct { + Date string + Host string + AgentClass string + Country string +} + +// GeoStat is what a GeoKey accumulates: requests, how many were refused +// (403), and how many were scanner probes — the three columns the "where +// does junk traffic come from" question reads. +type GeoStat struct { + Requests int64 + Blocked int64 + Probes int64 +} + +func (r *Rollup) addGeo(date, host, agentClass, ip string, status int, probed bool) { + country := r.Geo.Country(ip) + if country == "" { + country = UnknownCountry + } + key := GeoKey{Date: date, Host: host, AgentClass: agentClass, Country: country} + stat := r.Countries[key] + stat.Requests++ + if status == 403 { + stat.Blocked++ + } + if probed { + stat.Probes++ + } + r.Countries[key] = stat +} diff --git a/domains/platform/apis/stats/geo_test.go b/domains/platform/apis/stats/geo_test.go new file mode 100644 index 00000000..4d321b92 --- /dev/null +++ b/domains/platform/apis/stats/geo_test.go @@ -0,0 +1,115 @@ +package stats + +import ( + "bytes" + "compress/gzip" + "errors" + "strings" + "testing" +) + +const dbipSample = `1.0.0.0,1.0.0.255,AU +1.0.1.0,1.0.3.255,CN +57.141.0.0,57.141.255.255,US +195.178.110.0,195.178.110.255,GB +not,a,row +2001:db8::,2001:db8:ffff:ffff:ffff:ffff:ffff:ffff,DE +9.9.9.9,9.9.9.1,ZZ +10.0.0.0,10.255.255.255,xx +` + +func TestGeoPlacesAddressesByRangeAndSaysNothingOutsideThem(t *testing.T) { + geo, skipped, err := ParseDBIP(strings.NewReader(dbipSample)) + if err != nil { + t.Fatal(err) + } + // The unparseable row and the inverted range are skipped; the lowercase + // country is a real row, uppercased. + if skipped != 2 { + t.Errorf("skipped = %d, want 2", skipped) + } + cases := map[string]string{ + "1.0.0.7": "AU", + "1.0.0.255": "AU", // range end is inclusive + "1.0.1.0": "CN", // and so is the next start + "1.0.4.0": "", // the gap after CN + "57.141.3.4": "US", + "195.178.110.199": "GB", + "195.178.111.1": "", + "2001:db8::1": "DE", + "2001:db9::1": "", + "::ffff:1.0.0.7": "AU", // v4-mapped v6, as some stacks log it + "10.1.2.3": "XX", + "0.0.0.1": "", // before the first range + "255.255.255.255": "", + "not an address": "", + "": "", + "1.0.0.7 extra": "", + "192.168.1.1": "", + "2001:db8::": "DE", + "9.9.9.5": "", // the inverted range never loaded + "1.0.0.7\n": "", + "1.0.3.255": "CN", + "1.0.0.0": "AU", + "1.0.256.1": "", + "57.140.255.255": "", + "57.142.0.0": "", + "195.178.110.0": "GB", + "195.178.109.255": "", + "ffff::1": "", + "::1": "", + "1.0.0.7%eth0": "", + "[1.0.0.7]": "", + "01.0.0.7": "", + "1.0.0.7/32": "", + "1.0.0.7:8080": "", + } + for ip, want := range cases { + if got := geo.Country(ip); got != want { + t.Errorf("Country(%q) = %q, want %q", ip, got, want) + } + } + if NoLocator.Country(NoLocator{}, "1.0.0.7") != "" { + t.Error("NoLocator placed an address") + } +} + +func TestAGeoFileWithNoUsableRowsIsAnError(t *testing.T) { + if _, _, err := ParseDBIP(strings.NewReader("not,a,row\n")); err == nil { + t.Error("a database that places nothing loaded without complaint") + } + if _, _, err := ParseDBIP(strings.NewReader("")); err == nil { + t.Error("an empty database loaded without complaint") + } +} + +func TestLoadGeoReadsAGzippedObjectAndReportsAMissingOne(t *testing.T) { + var buf bytes.Buffer + w := gzip.NewWriter(&buf) + if _, err := w.Write([]byte(dbipSample)); err != nil { + t.Fatal(err) + } + if err := w.Close(); err != nil { + t.Fatal(err) + } + objects := &fakeObjects{objects: map[string][]byte{ + "geo/dbip-country-lite.csv.gz": buf.Bytes(), + "geo/plain.csv": []byte(dbipSample), + }, fail: map[string]error{"geo/missing.csv": errors.New("404")}} + + for _, key := range []string{"geo/dbip-country-lite.csv.gz", "geo/plain.csv"} { + geo, _, err := LoadGeo(objects, key) + if err != nil { + t.Fatalf("%s: %v", key, err) + } + if geo.Country("57.141.3.4") != "US" { + t.Errorf("%s loaded but places nothing", key) + } + } + if _, _, err := LoadGeo(objects, "geo/missing.csv"); err == nil { + t.Error("a missing database loaded without complaint") + } + if _, _, err := LoadGeo(&fakeObjects{objects: map[string][]byte{"geo/x.gz": []byte("not gzip")}}, "geo/x.gz"); err == nil { + t.Error("a .gz that is not gzip loaded without complaint") + } +} diff --git a/domains/platform/apis/stats/loop.go b/domains/platform/apis/stats/loop.go index 6db06683..18de664f 100644 --- a/domains/platform/apis/stats/loop.go +++ b/domains/platform/apis/stats/loop.go @@ -51,6 +51,8 @@ type Aggregator struct { Objects ObjectStore Store Applier Logger *slog.Logger + // Geo places client addresses for the geo rollup; nil means none loaded. + Geo Locator } // RunOnce aggregates every not-yet-processed object under the source @@ -112,6 +114,9 @@ func (a *Aggregator) processObject(ctx context.Context, key string) error { return fmt.Errorf("no parser for source %q", source) } rollup := NewRollup() + if a.Geo != nil { + rollup.Geo = a.Geo + } skipped, err := consume(rollup, reader, date) if err != nil { return err diff --git a/domains/platform/apis/stats/loop_test.go b/domains/platform/apis/stats/loop_test.go index 75496ddb..0c448e6b 100644 --- a/domains/platform/apis/stats/loop_test.go +++ b/domains/platform/apis/stats/loop_test.go @@ -106,6 +106,25 @@ func TestRunOnceAggregatesNewObjectsAndSkipsProcessedAndForeignKeys(t *testing.T } } +func TestRunOnceHandsTheAggregatorsLocatorToEveryRollup(t *testing.T) { + objects := &fakeObjects{objects: map[string][]byte{ + "logs/source=caddy/dt=2026-08-30/a.log.gz": gzipped(t, + `{"status":200,"request":{"host":"h","method":"GET","uri":"/","client_ip":"1.0.0.7","headers":{}}}`), + }} + store := newFakeApplier() + agg := testAggregator(objects, store) + agg.Geo = fakeLocator{"1.0.0.7": "AU"} + + if _, err := agg.RunOnce(context.Background()); err != nil { + t.Fatal(err) + } + + rollup := store.applied["logs/source=caddy/dt=2026-08-30/a.log.gz"] + if rollup.Countries[GeoKey{"2026-08-30", "h", AgentOther, "AU"}].Requests != 1 { + t.Errorf("geo rows = %v; the locator did not reach the rollup", rollup.Countries) + } +} + func TestRunOnceReadsOneD4ObjectsAsQueryEvents(t *testing.T) { objects := &fakeObjects{objects: map[string][]byte{ "logs/source=one_d4/dt=2026-09-01/query_events-2026-09-01T14.log.gz": gzipped(t, diff --git a/domains/platform/apis/stats/main/main.go b/domains/platform/apis/stats/main/main.go index 92142973..aa49b610 100644 --- a/domains/platform/apis/stats/main/main.go +++ b/domains/platform/apis/stats/main/main.go @@ -49,19 +49,37 @@ func main() { } defer store.Close() - aggregator := &stats.Aggregator{ - Objects: &s3lite.S3{ - Bucket: requireEnv(logger, "S3_BUCKET"), - Region: requireEnv(logger, "S3_REGION"), - Creds: s3lite.Credentials{ - AccessKeyID: requireEnv(logger, "AWS_ACCESS_KEY_ID"), - SecretAccessKey: requireEnv(logger, "AWS_SECRET_ACCESS_KEY"), - }, - Client: &http.Client{Timeout: 5 * time.Minute}, - Now: time.Now, + objects := &s3lite.S3{ + Bucket: requireEnv(logger, "S3_BUCKET"), + Region: requireEnv(logger, "S3_REGION"), + Creds: s3lite.Credentials{ + AccessKeyID: requireEnv(logger, "AWS_ACCESS_KEY_ID"), + SecretAccessKey: requireEnv(logger, "AWS_SECRET_ACCESS_KEY"), }, - Store: store, - Logger: logger, + Client: &http.Client{Timeout: 5 * time.Minute}, + Now: time.Now, + } + // The geo database is optional and, when named, required to load: a + // configured key that fails is a deployment fault, not a reason to + // silently file every address under "--". + var geo stats.Locator = stats.NoLocator{} + if key := os.Getenv("GEO_DB_KEY"); key != "" { + loaded, skipped, err := stats.LoadGeo(objects, key) + if err != nil { + logger.Error("cannot load the geo database", "key", key, "error", err) + os.Exit(1) + } + logger.Info("geo database loaded", "key", key, "skipped_lines", skipped) + geo = loaded + } else { + logger.Warn("GEO_DB_KEY is not set; geo rows will all read " + stats.UnknownCountry) + } + + aggregator := &stats.Aggregator{ + Objects: objects, + Store: store, + Logger: logger, + Geo: geo, } go func() { ticker := time.NewTicker(interval) @@ -86,6 +104,7 @@ func main() { router.HandleFunc("GET /stats/v1/probes", handlers.GetProbes) router.HandleFunc("GET /stats/v1/one_d4/queries", handlers.GetQueries) router.HandleFunc("GET /stats/v1/one_d4/terms", handlers.GetQueryTerms) + router.HandleFunc("GET /stats/v1/countries", handlers.GetCountries) logger.Info("stats started", "port", port, "interval", interval.String()) if err := http.ListenAndServe(":"+port, router); err != nil { diff --git a/domains/platform/apis/stats/store.go b/domains/platform/apis/stats/store.go index e74f982f..5fd044d3 100644 --- a/domains/platform/apis/stats/store.go +++ b/domains/platform/apis/stats/store.go @@ -52,6 +52,16 @@ var schema = []string{ requests bigint NOT NULL, PRIMARY KEY (dt, entry, source, outcome, cache) )`, + `CREATE TABLE IF NOT EXISTS geo_stats ( + dt date NOT NULL, + host text NOT NULL, + agent_class text NOT NULL, + country text NOT NULL, + requests bigint NOT NULL, + blocked bigint NOT NULL, + probes bigint NOT NULL, + PRIMARY KEY (dt, host, agent_class, country) + )`, `CREATE TABLE IF NOT EXISTS query_term_stats ( dt date NOT NULL, entry text NOT NULL, @@ -85,7 +95,7 @@ func currentRollupVersion() string { return rollupVersionFor(RollupVersion, sche // belongs here, or a version bump leaves it double-counted. var rollupTables = []string{ "processed_log_objects", "request_stats", "iili_slug_stats", "probe_stats", - "query_stats", "query_term_stats", + "query_stats", "query_term_stats", "geo_stats", } const metaSchema = `CREATE TABLE IF NOT EXISTS stats_meta ( @@ -241,6 +251,18 @@ func (s *Store) ApplyRollup(ctx context.Context, key string, rollup *Rollup) err return err } } + for k, stat := range rollup.Countries { + if _, err := tx.Exec(ctx, + `INSERT INTO geo_stats (dt, host, agent_class, country, requests, blocked, probes) + VALUES ($1, $2, $3, $4, $5, $6, $7) + ON CONFLICT (dt, host, agent_class, country) + DO UPDATE SET requests = geo_stats.requests + EXCLUDED.requests, + blocked = geo_stats.blocked + EXCLUDED.blocked, + probes = geo_stats.probes + EXCLUDED.probes`, + k.Date, k.Host, k.AgentClass, k.Country, stat.Requests, stat.Blocked, stat.Probes); err != nil { + return err + } + } for k, count := range rollup.Terms { if _, err := tx.Exec(ctx, `INSERT INTO query_term_stats (dt, entry, kind, term, requests) @@ -439,3 +461,37 @@ func (s *Store) QueryTerms(ctx context.Context, days, limit int) ([]TermRow, err }) return out, err } + +// CountryRow is one host's traffic of one class from one country over the +// window: requests, how many were refused, and how many were probes. +type CountryRow struct { + Host string `json:"host"` + AgentClass string `json:"agent_class"` + Country string `json:"country"` + Requests int64 `json:"requests"` + Blocked int64 `json:"blocked"` + Probes int64 `json:"probes"` +} + +func (s *Store) Countries(ctx context.Context, days, limit int) ([]CountryRow, error) { + rows, err := s.pool.Query(ctx, + `SELECT host, agent_class, country, + SUM(requests) AS requests, SUM(blocked) AS blocked, SUM(probes) AS probes + FROM geo_stats + WHERE dt >= current_date - $1::int + GROUP BY host, agent_class, country + ORDER BY requests DESC, host, agent_class, country + LIMIT $2`, days, limit) + if err != nil { + return nil, err + } + var out []CountryRow + var row CountryRow + _, err = pgx.ForEachRow(rows, + []any{&row.Host, &row.AgentClass, &row.Country, &row.Requests, &row.Blocked, &row.Probes}, + func() error { + out = append(out, row) + return nil + }) + return out, err +} diff --git a/domains/platform/apis/stats/store_test.go b/domains/platform/apis/stats/store_test.go index da711383..d6aaa7eb 100644 --- a/domains/platform/apis/stats/store_test.go +++ b/domains/platform/apis/stats/store_test.go @@ -55,6 +55,8 @@ func TestApplyRollupIsTransactionalIdempotentAndReadable(t *testing.T) { // unique host serves as one and keeps the rows tellable and the counts exact. rollup.Queries[QueryKey{date, host, "ui", "ok", "live"}] = 2 rollup.Terms[TermKey{date, host, KindField, "white.elo"}] = 2 + rollup.Countries[GeoKey{date, host, AgentBot, "GB"}] = GeoStat{5, 1, 4} + rollup.Countries[GeoKey{date, host, AgentBot, "--"}] = GeoStat{2, 0, 0} if pending, err := store.Unprocessed(ctx, []string{key}); err != nil || len(pending) != 1 { t.Fatalf("Unprocessed = (%v, %v), want the fresh key pending", pending, err) @@ -146,6 +148,9 @@ func TestApplyRollupIsTransactionalIdempotentAndReadable(t *testing.T) { if row := queryRowFor(t, store, host); row == nil || row.Requests != 2 || row.Cache != "live" { t.Errorf("query row = %+v, want exactly the 2 applied once", row) } + if row := countryRowFor(t, store, host, "GB"); row == nil || *row != (CountryRow{host, AgentBot, "GB", 5, 1, 4}) { + t.Errorf("country row = %+v, want exactly the rollup applied once", row) + } if row := termRowFor(t, store, host); row == nil || row.Requests != 2 || row.Kind != KindField { t.Errorf("term row = %+v, want exactly the 2 applied once", row) } @@ -155,9 +160,13 @@ func TestApplyRollupIsTransactionalIdempotentAndReadable(t *testing.T) { second := NewRollup() second.Queries[QueryKey{date, host, "ui", "ok", "live"}] = 3 second.Terms[TermKey{date, host, KindField, "white.elo"}] = 1 + second.Countries[GeoKey{date, host, AgentBot, "GB"}] = GeoStat{1, 1, 0} if err := store.ApplyRollup(ctx, key+".second", second); err != nil { t.Fatal(err) } + if row := countryRowFor(t, store, host, "GB"); row == nil || *row != (CountryRow{host, AgentBot, "GB", 6, 2, 4}) { + t.Errorf("country row after a second object = %+v, want the three columns summed", row) + } if row := queryRowFor(t, store, host); row == nil || row.Requests != 5 { t.Errorf("query row after a second object = %+v, want 5", row) } @@ -166,6 +175,20 @@ func TestApplyRollupIsTransactionalIdempotentAndReadable(t *testing.T) { } } +func countryRowFor(t *testing.T, store *Store, host, country string) *CountryRow { + t.Helper() + rows, err := store.Countries(context.Background(), 2, 5000) + if err != nil { + t.Fatal(err) + } + for i := range rows { + if rows[i].Host == host && rows[i].Country == country { + return &rows[i] + } + } + return nil +} + func queryRowFor(t *testing.T, store *Store, entry string) *QueryRow { t.Helper() rows, err := store.Queries(context.Background(), 2) @@ -230,6 +253,7 @@ func TestAVersionBumpDropsAggregatesAndMarkersForReaggregation(t *testing.T) { rollup.Probes[ProbeKey{date, host, ProbeGit, 404}] = 1 rollup.Queries[QueryKey{date, host, "ui", "ok", "live"}] = 1 rollup.Terms[TermKey{date, host, KindField, "eco"}] = 1 + rollup.Countries[GeoKey{date, host, AgentBot, "GB"}] = GeoStat{1, 0, 0} if err := store.ApplyRollup(ctx, key, rollup); err != nil { t.Fatal(err) } @@ -292,6 +316,9 @@ func TestAVersionBumpDropsAggregatesAndMarkersForReaggregation(t *testing.T) { if row := termRowFor(t, reopened, host); row != nil { t.Errorf("term aggregates survived the version bump: %+v", row) } + if row := countryRowFor(t, reopened, host, "GB"); row != nil { + t.Errorf("geo aggregates survived the version bump: %+v", row) + } var recorded string if err := reopened.pool.QueryRow(ctx, `SELECT value FROM stats_meta WHERE key = 'rollup_version'`).Scan(&recorded); err != nil || recorded != currentRollupVersion() { From 8faee61fcce991c0653a7a190e9216364e85636a Mon Sep 17 00:00:00 2001 From: Andy Aylward Date: Wed, 2 Sep 2026 09:48:45 -0400 Subject: [PATCH 2/2] stats: review-panel fixes for the geo rollup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A geo key that will not load is retried while the bucket wakes up, then logged and run without — an all-"--" table, never a boot failure that takes the other endpoints with it. Ranges are validated after unmapping, overlaps lose to the range they sit in, and country codes are interned so the table does not pin its CSV lines. Tests cover the refused probe, the unplaced row, and the boot path; the deploy test pins GEO_DB_KEY; the READMEs say which prefixes the IAM user needs and where the file goes. --- deploy/consolidated/README.md | 5 +- deploy/consolidated/compose.yaml | 2 +- deploy/consolidated/deploy_config_test.go | 5 ++ domains/platform/apis/stats/README.md | 24 ++++--- domains/platform/apis/stats/aggregate_test.go | 4 +- domains/platform/apis/stats/api_test.go | 4 +- domains/platform/apis/stats/geo.go | 71 ++++++++++++++++--- domains/platform/apis/stats/geo_test.go | 67 ++++++++++++++--- domains/platform/apis/stats/main/main.go | 25 ++++--- domains/platform/apis/stats/store_test.go | 3 + domains/platform/apps/log_shipper/README.md | 2 +- 11 files changed, 165 insertions(+), 47 deletions(-) diff --git a/deploy/consolidated/README.md b/deploy/consolidated/README.md index b81dce9c..f24e743c 100644 --- a/deploy/consolidated/README.md +++ b/deploy/consolidated/README.md @@ -221,7 +221,10 @@ COMPOSE_PROFILES=stats deploy.sh runs compose in `~`, where compose reads that file, so every normal deploy includes the trio. Alongside it live `STATS_AWS_ACCESS_KEY_ID`, `STATS_AWS_SECRET_ACCESS_KEY`, `STATS_S3_BUCKET`, `STATS_S3_REGION`, optionally `STATS_GEO_DB_KEY` (the key of -DB-IP's country CSV in that bucket; unset, the geo rows all read `--`), and +DB-IP's country CSV in that bucket — upload `dbip-country-lite-YYYY-MM.csv.gz` +under `geo/` and restart `stats` for a new month; unset, the geo rows all read +`--`, and a key that will not load is an error in the log, not a boot +failure), and `STATS_DB_PASSWORD` (same URL-safe rules as the other database passwords: it rides in a libpq URL and a single-quoted SQL literal). Without the `COMPOSE_PROFILES` line the containers keep running after a deploy but silently stop diff --git a/deploy/consolidated/compose.yaml b/deploy/consolidated/compose.yaml index 15290b93..b6b5a934 100644 --- a/deploy/consolidated/compose.yaml +++ b/deploy/consolidated/compose.yaml @@ -663,7 +663,7 @@ services: - S3_BUCKET=${STATS_S3_BUCKET} - S3_REGION=${STATS_S3_REGION:-us-east-1} # DB-IP's country CSV in the same bucket (#1467); empty means every - # geo row reads "--". A key that is set has to load, or boot fails. + # geo row reads "--", and so does a key that will not load. - GEO_DB_KEY=${STATS_GEO_DB_KEY:-} - PORT=8092 networks: diff --git a/deploy/consolidated/deploy_config_test.go b/deploy/consolidated/deploy_config_test.go index 269118e0..8270730f 100644 --- a/deploy/consolidated/deploy_config_test.go +++ b/deploy/consolidated/deploy_config_test.go @@ -2116,6 +2116,11 @@ func TestTheStatsPairIsProfileGatedTogether(t *testing.T) { if !strings.Contains(serviceBlock(t, "compose.yaml", "stats"), "postgresql://stats:") { t.Errorf("stats names no stats database URL; the aggregates have nowhere to land") } + // The geo database key rides the stats block (#1467); dropping it is an + // all-"--" table with nothing else to say so. + if !strings.Contains(serviceBlock(t, "compose.yaml", "stats"), "GEO_DB_KEY=") { + t.Errorf("the stats service does not pass GEO_DB_KEY; the geo rollup has no database to load") + } } // catchAllIsLastHandle reports whether the site block ends its handle diff --git a/domains/platform/apis/stats/README.md b/domains/platform/apis/stats/README.md index a63a0985..a0cc2898 100644 --- a/domains/platform/apis/stats/README.md +++ b/domains/platform/apis/stats/README.md @@ -56,15 +56,18 @@ The pass runs while the API serves, so for its length the counts climb back up from zero — minutes at this scale, and the log says when it is done. -The geo rollup (#1467) places each request's `client_ip` in a country -and keys `geo_stats` on day, host, agent class, and the two-letter code, -with request, 403, and probe counts — where the scrapers, bots, and -scanners come from. The database is DB-IP's free country CSV -(`dbip-country-lite-YYYY-MM.csv.gz`, CC BY 4.0, attribution on the -dashboard), uploaded by the operator to the stats bucket under the key -`GEO_DB_KEY` names; the service loads it at boot into a sorted range -table and binary-searches it, no library. An address outside every range, -or no database at all, files under `--`. A new monthly file is a restart. +The geo rollup (#1467) places each request's `client_ip` (`remote_ip` on +older lines) in a country and keys `geo_stats` on day, host, agent class, +and the two-letter code, with request, 403, and probe counts — where the +scrapers, bots, and scanners come from. The database is DB-IP's free +country CSV (`dbip-country-lite-YYYY-MM.csv.gz`, CC BY 4.0; muchq.com's +stats page carries the attribution), uploaded by the operator to the stats +bucket under the key `GEO_DB_KEY` names; the service loads it at boot +into a sorted range table and binary-searches it, no library. Overlapping +rows lose to the range they sit in. An address outside every range, or no +database at all, files under `--`. A key that will not load, after a few +retries for a bucket still waking up, is an error in the log and an +all-`--` table, never a boot failure. A new monthly file is a restart. Rows aggregated before the database was uploaded stay `--` until a re-aggregation (bump `RollupVersion`). @@ -104,7 +107,8 @@ stay in the log, not on the wire. `STATS_DB_URL` (postgres), `S3_BUCKET`, `S3_REGION`, `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, and optionally `GEO_DB_KEY` — the same stats IAM user the shipper writes with, which therefore needs `s3:GetObject` and `s3:ListBucket` on the `logs/*` -prefix as well as `s3:PutObject`. `AGGREGATE_INTERVAL` and `PORT` +prefix as well as `s3:PutObject`, and `s3:GetObject` on the geo key's +prefix (`geo/*` in the deployment). `AGGREGATE_INTERVAL` and `PORT` (default 8092) are optional. The store integration test needs `STATS_TEST_DB_URL` and skips without it, diff --git a/domains/platform/apis/stats/aggregate_test.go b/domains/platform/apis/stats/aggregate_test.go index 1ef92b20..d8313bb9 100644 --- a/domains/platform/apis/stats/aggregate_test.go +++ b/domains/platform/apis/stats/aggregate_test.go @@ -122,6 +122,8 @@ func TestConsumeRollsUpWhereEachClassCameFromWithBlocksAndProbes(t *testing.T) { `{"status":403,"request":{"host":"h","method":"GET","uri":"/x","client_ip":"57.141.3.4","headers":{"User-Agent":["meta-externalagent/1.1"]}}}`, `{"status":404,"request":{"host":"h","method":"GET","uri":"/.env","client_ip":"195.178.110.199","headers":{"User-Agent":["TLM-Audit-Scanner/1.0"]}}}`, `{"status":404,"request":{"host":"h","method":"GET","uri":"/wp-login.php","client_ip":"195.178.110.199","headers":{"User-Agent":["TLM-Audit-Scanner/1.0"]}}}`, + // Refused and a probe at once: the production shape once refuse_bots answers a scanner. + `{"status":403,"request":{"host":"h","method":"GET","uri":"/.git/config","client_ip":"195.178.110.199","headers":{"User-Agent":["TLM-Audit-Scanner/1.0"]}}}`, // Older lines carry remote_ip only; an address the locator does not know is "--". `{"status":200,"request":{"host":"h","method":"GET","uri":"/","remote_ip":"10.1.2.3","headers":{"User-Agent":["curl/8.6.0"]}}}`, }, "\n") @@ -135,7 +137,7 @@ func TestConsumeRollsUpWhereEachClassCameFromWithBlocksAndProbes(t *testing.T) { want := map[GeoKey]GeoStat{ {"2026-08-30", "h", AgentBrowser, "AU"}: {1, 0, 0}, {"2026-08-30", "h", AgentAIScraper, "US"}: {1, 1, 0}, - {"2026-08-30", "h", AgentBot, "GB"}: {2, 0, 2}, + {"2026-08-30", "h", AgentBot, "GB"}: {3, 1, 3}, {"2026-08-30", "h", AgentBot, "--"}: {1, 0, 0}, } if len(rollup.Countries) != len(want) { diff --git a/domains/platform/apis/stats/api_test.go b/domains/platform/apis/stats/api_test.go index 38b3ecd1..f6f21683 100644 --- a/domains/platform/apis/stats/api_test.go +++ b/domains/platform/apis/stats/api_test.go @@ -259,7 +259,7 @@ func TestOneD4QueryEndpointsShareTheWindowRules(t *testing.T) { func TestCountriesShareTheWindowRules(t *testing.T) { reader := &fakeReader{countries: []CountryRow{ - {Host: "git.muchq.com", AgentClass: AgentAIScraper, Country: "US", Requests: 9, Blocked: 9, Probes: 0}, + {Host: "git.muchq.com", AgentClass: AgentAIScraper, Country: "US", Requests: 9, Blocked: 4, Probes: 2}, }} handlers := handlersWith(reader) @@ -268,7 +268,7 @@ func TestCountriesShareTheWindowRules(t *testing.T) { t.Errorf("default countries (days, limit) = (%d, %d), want (30, 2000)", reader.lastDays, reader.lastLimit) } row := body["rows"].([]any)[0].(map[string]any) - if row["country"] != "US" || row["blocked"] != float64(9) || row["probes"] != float64(0) { + if row["country"] != "US" || row["requests"] != float64(9) || row["blocked"] != float64(4) || row["probes"] != float64(2) { t.Errorf("country row = %v", row) } get(t, handlers.GetCountries, "/stats/v1/countries?days=99999&limit=99999") diff --git a/domains/platform/apis/stats/geo.go b/domains/platform/apis/stats/geo.go index 2c16bd84..9caac681 100644 --- a/domains/platform/apis/stats/geo.go +++ b/domains/platform/apis/stats/geo.go @@ -40,14 +40,18 @@ type ipRange struct { country string } -// ParseDBIP reads the CSV. Lines that do not parse are counted and -// skipped rather than failing the load — one bad row in a vendor file -// must not switch geo off — but a file with no usable rows is an error, -// since that is a wrong object, not a bad line. +// ParseDBIP reads the CSV. Rows that do not parse, and rows that overlap +// an earlier one, are counted and skipped rather than failing the load — +// one bad row in a vendor file must not switch geo off — but a file with +// no usable rows is an error, since that is a wrong object, not a bad +// line. Country codes are interned: the reader allocates a string per +// record, and a two-byte code must not keep its whole line alive. func ParseDBIP(r io.Reader) (*Geo, int, error) { reader := csv.NewReader(bufio.NewReader(r)) reader.FieldsPerRecord = -1 reader.ReuseRecord = true + reader.LazyQuotes = true + codes := map[string]string{} var ranges []ipRange skipped := 0 for { @@ -65,17 +69,39 @@ func ParseDBIP(r io.Reader) (*Geo, int, error) { start, err1 := netip.ParseAddr(strings.TrimSpace(record[0])) end, err2 := netip.ParseAddr(strings.TrimSpace(record[1])) country := strings.ToUpper(strings.TrimSpace(record[2])) - if err1 != nil || err2 != nil || len(country) != 2 || start.Is4() != end.Is4() || end.Less(start) { + if err1 != nil || err2 != nil || len(country) != 2 { skipped++ continue } - ranges = append(ranges, ipRange{start: start.Unmap(), end: end.Unmap(), country: country}) + // Compared as stored: a v4-mapped v6 end against a v6 start is an + // inverted range once both are unmapped, whatever they looked like. + start, end = start.Unmap(), end.Unmap() + if start.Is4() != end.Is4() || end.Less(start) { + skipped++ + continue + } + interned, ok := codes[country] + if !ok { + interned = strings.Clone(country) + codes[country] = interned + } + ranges = append(ranges, ipRange{start: start, end: end, country: interned}) + } + sort.SliceStable(ranges, func(i, j int) bool { return ranges[i].start.Less(ranges[j].start) }) + // The lookup reads one predecessor, so ranges must not overlap; a row + // that starts inside the one before it is dropped and counted. + kept := ranges[:0] + for _, candidate := range ranges { + if len(kept) > 0 && !kept[len(kept)-1].end.Less(candidate.start) { + skipped++ + continue + } + kept = append(kept, candidate) } - if len(ranges) == 0 { + if len(kept) == 0 { return nil, skipped, fmt.Errorf("geo csv holds no usable ranges (%d lines skipped)", skipped) } - sort.Slice(ranges, func(i, j int) bool { return ranges[i].start.Less(ranges[j].start) }) - return &Geo{ranges: ranges}, skipped, nil + return &Geo{ranges: kept}, skipped, nil } // Country finds the range holding ip. Addresses come from Caddy's log as @@ -102,7 +128,7 @@ func (g *Geo) Country(ip string) string { // LoadGeo reads the CSV object (gzipped when the key says so) from the // bucket the logs live in. The key is the operator's: DB-IP publishes a -// new file monthly under CC BY 4.0, and the dashboard carries the +// new file monthly under CC BY 4.0, and muchq.com/stats carries the // attribution. func LoadGeo(objects ObjectStore, key string) (*Geo, int, error) { body, err := objects.Get(key) @@ -122,6 +148,31 @@ func LoadGeo(objects ObjectStore, key string) (*Geo, int, error) { return ParseDBIP(reader) } +// Locate is the boot path: no key means no database, and a key that +// fails is retried — a bucket that is briefly unreachable while the +// container starts must not take every endpoint down with it — then +// reported for the caller to carry on without. It never fails a boot: +// an all-"--" table with an error in the log is the visible outcome, and +// a crash loop would hide every other table behind it. +func Locate(objects ObjectStore, key string, attempts int, wait func()) (Locator, int, error) { + if key == "" { + return NoLocator{}, 0, nil + } + var err error + for attempt := 0; attempt < attempts; attempt++ { + if attempt > 0 { + wait() + } + var geo *Geo + var skipped int + geo, skipped, err = LoadGeo(objects, key) + if err == nil { + return geo, skipped, nil + } + } + return NoLocator{}, 0, err +} + // GeoKey is one row of the per-day geo rollup: where a host's traffic of // each class came from. Bounded by the country vocabulary times the four // classes. diff --git a/domains/platform/apis/stats/geo_test.go b/domains/platform/apis/stats/geo_test.go index 4d321b92..6d769a4c 100644 --- a/domains/platform/apis/stats/geo_test.go +++ b/domains/platform/apis/stats/geo_test.go @@ -8,13 +8,23 @@ import ( "testing" ) +// Rows in the shape of dbip-country-lite-YYYY-MM.csv (start,end,country, +// no header, v4 and v6 mixed; the sample addresses are made up), plus the +// malformed rows a vendor file could carry: short, unparseable, a name +// rather than a code, a v4/v6 pair, inverted, overlapping, and one with +// a trailing field, which loads. const dbipSample = `1.0.0.0,1.0.0.255,AU 1.0.1.0,1.0.3.255,CN 57.141.0.0,57.141.255.255,US 195.178.110.0,195.178.110.255,GB not,a,row +1.2.3.0,1.2.3.255 +2.0.0.0,2.0.0.255,United States +3.0.0.0,2001:db8::,FR 2001:db8::,2001:db8:ffff:ffff:ffff:ffff:ffff:ffff,DE 9.9.9.9,9.9.9.1,ZZ +1.0.2.0,1.0.2.255,NZ +4.0.0.0,4.0.0.255,FR,extra 10.0.0.0,10.255.255.255,xx ` @@ -23,10 +33,11 @@ func TestGeoPlacesAddressesByRangeAndSaysNothingOutsideThem(t *testing.T) { if err != nil { t.Fatal(err) } - // The unparseable row and the inverted range are skipped; the lowercase - // country is a real row, uppercased. - if skipped != 2 { - t.Errorf("skipped = %d, want 2", skipped) + // Six rows skipped: unparseable, short, a country name, a v4/v6 pair, an + // inverted range, and the NZ row nested inside CN's. The lowercase + // country and the row with a trailing field are real rows. + if skipped != 6 { + t.Errorf("skipped = %d, want 6", skipped) } cases := map[string]string{ "1.0.0.7": "AU", @@ -40,7 +51,12 @@ func TestGeoPlacesAddressesByRangeAndSaysNothingOutsideThem(t *testing.T) { "2001:db9::1": "", "::ffff:1.0.0.7": "AU", // v4-mapped v6, as some stacks log it "10.1.2.3": "XX", - "0.0.0.1": "", // before the first range + "4.0.0.7": "FR", // the trailing field is ignored + "2.0.0.7": "", // a name is not a code + "3.0.0.7": "", // a v4/v6 pair never loaded + "1.0.2.7": "CN", // the overlapping NZ row lost to the range it sits in + "1.0.3.0": "CN", // and the rest of CN's range is not lost with it + "0.0.0.1": "", // before the first range "255.255.255.255": "", "not an address": "", "": "", @@ -98,12 +114,12 @@ func TestLoadGeoReadsAGzippedObjectAndReportsAMissingOne(t *testing.T) { }, fail: map[string]error{"geo/missing.csv": errors.New("404")}} for _, key := range []string{"geo/dbip-country-lite.csv.gz", "geo/plain.csv"} { - geo, _, err := LoadGeo(objects, key) + geo, skipped, err := LoadGeo(objects, key) if err != nil { t.Fatalf("%s: %v", key, err) } - if geo.Country("57.141.3.4") != "US" { - t.Errorf("%s loaded but places nothing", key) + if geo.Country("57.141.3.4") != "US" || skipped != 6 { + t.Errorf("%s loaded but places nothing, or lost the skip count (%d)", key, skipped) } } if _, _, err := LoadGeo(objects, "geo/missing.csv"); err == nil { @@ -113,3 +129,38 @@ func TestLoadGeoReadsAGzippedObjectAndReportsAMissingOne(t *testing.T) { t.Error("a .gz that is not gzip loaded without complaint") } } + +// A quoted field with a stray quote is a bad row, not a dead database. +func TestAStrayQuoteIsOneSkippedRowNotAFailedLoad(t *testing.T) { + geo, skipped, err := ParseDBIP(strings.NewReader("1.0.0.0,1.0.0.255,AU\n\"1.0.1.0,1.0.1.255,\"C\"N\n")) + if err != nil || skipped != 1 || geo.Country("1.0.0.7") != "AU" { + t.Errorf("ParseDBIP = (%v, %d, %v); want the good row kept and the bad one counted", geo != nil, skipped, err) + } +} + +func TestLocateNeverFailsABootAndRetriesABucketThatIsNotThereYet(t *testing.T) { + geo, _, err := Locate(&fakeObjects{}, "", 3, func() {}) + if _, none := geo.(NoLocator); !none || err != nil { + t.Errorf("Locate with no key = (%T, %v), want NoLocator and no error", geo, err) + } + + objects := &fakeObjects{objects: map[string][]byte{"geo/plain.csv": []byte(dbipSample)}, + fail: map[string]error{"geo/plain.csv": errors.New("503 slow down")}} + waits := 0 + // The bucket answers on the third try. + geo, _, err = Locate(objects, "geo/plain.csv", 5, func() { + waits++ + if waits == 2 { + delete(objects.fail, "geo/plain.csv") + } + }) + if err != nil || geo.Country("57.141.3.4") != "US" || waits != 2 { + t.Errorf("Locate after a flaky bucket = (%v, %v, %d waits); want the database on the third try", geo, err, waits) + } + + geo, _, err = Locate(&fakeObjects{fail: map[string]error{"geo/missing.csv": errors.New("404")}}, + "geo/missing.csv", 3, func() {}) + if _, none := geo.(NoLocator); !none || err == nil { + t.Errorf("Locate with a key that never loads = (%T, %v); want NoLocator and the error to log", geo, err) + } +} diff --git a/domains/platform/apis/stats/main/main.go b/domains/platform/apis/stats/main/main.go index aa49b610..d7d3fc49 100644 --- a/domains/platform/apis/stats/main/main.go +++ b/domains/platform/apis/stats/main/main.go @@ -59,20 +59,19 @@ func main() { Client: &http.Client{Timeout: 5 * time.Minute}, Now: time.Now, } - // The geo database is optional and, when named, required to load: a - // configured key that fails is a deployment fault, not a reason to - // silently file every address under "--". - var geo stats.Locator = stats.NoLocator{} - if key := os.Getenv("GEO_DB_KEY"); key != "" { - loaded, skipped, err := stats.LoadGeo(objects, key) - if err != nil { - logger.Error("cannot load the geo database", "key", key, "error", err) - os.Exit(1) - } - logger.Info("geo database loaded", "key", key, "skipped_lines", skipped) - geo = loaded - } else { + // The geo database is optional. A key that will not load is logged and + // the service runs without it — every geo row reads "--" — rather than + // crash-looping the other endpoints behind a bucket hiccup or a typo. + geoKey := os.Getenv("GEO_DB_KEY") + geo, skipped, err := stats.Locate(objects, geoKey, 5, func() { time.Sleep(10 * time.Second) }) + switch { + case err != nil: + logger.Error("cannot load the geo database; geo rows will all read "+stats.UnknownCountry, + "key", geoKey, "error", err) + case geoKey == "": logger.Warn("GEO_DB_KEY is not set; geo rows will all read " + stats.UnknownCountry) + default: + logger.Info("geo database loaded", "key", geoKey, "skipped_lines", skipped) } aggregator := &stats.Aggregator{ diff --git a/domains/platform/apis/stats/store_test.go b/domains/platform/apis/stats/store_test.go index d6aaa7eb..9c181f93 100644 --- a/domains/platform/apis/stats/store_test.go +++ b/domains/platform/apis/stats/store_test.go @@ -151,6 +151,9 @@ func TestApplyRollupIsTransactionalIdempotentAndReadable(t *testing.T) { if row := countryRowFor(t, store, host, "GB"); row == nil || *row != (CountryRow{host, AgentBot, "GB", 5, 1, 4}) { t.Errorf("country row = %+v, want exactly the rollup applied once", row) } + if row := countryRowFor(t, store, host, UnknownCountry); row == nil || *row != (CountryRow{host, AgentBot, "--", 2, 0, 0}) { + t.Errorf("unplaced row = %+v, want it stored and read back like any country", row) + } if row := termRowFor(t, store, host); row == nil || row.Requests != 2 || row.Kind != KindField { t.Errorf("term row = %+v, want exactly the 2 applied once", row) } diff --git a/domains/platform/apps/log_shipper/README.md b/domains/platform/apps/log_shipper/README.md index 7f7655bf..866f2fc4 100644 --- a/domains/platform/apps/log_shipper/README.md +++ b/domains/platform/apps/log_shipper/README.md @@ -41,7 +41,7 @@ and anything that is not a rolled log are never touched. | --- | --- | | `S3_BUCKET` | Destination bucket (required) | | `S3_REGION` | Bucket's region (required) | -| `AWS_ACCESS_KEY_ID` / `AWS_SECRET_ACCESS_KEY` | The stats IAM user: `s3:PutObject` for this shipper, plus `s3:GetObject`/`s3:ListBucket` for the aggregator, all scoped to the bucket's `logs/*` prefix (required) | +| `AWS_ACCESS_KEY_ID` / `AWS_SECRET_ACCESS_KEY` | The stats IAM user: `s3:PutObject` for this shipper, plus `s3:GetObject`/`s3:ListBucket` for the aggregator, all scoped to the bucket's `logs/*` prefix, plus `s3:GetObject` on the geo database's prefix (required) | | `LOG_DIRS` | `label=dir` pairs, comma-separated: each directory ships under `logs/source=