diff --git a/deploy/consolidated/README.md b/deploy/consolidated/README.md index a619e5fe..f24e743c 100644 --- a/deploy/consolidated/README.md +++ b/deploy/consolidated/README.md @@ -220,7 +220,12 @@ 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 — 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 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..b6b5a934 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 "--", and so does a key that will not load. + - GEO_DB_KEY=${STATS_GEO_DB_KEY:-} - PORT=8092 networks: - app_network 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/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..a0cc2898 100644 --- a/domains/platform/apis/stats/README.md +++ b/domains/platform/apis/stats/README.md @@ -56,9 +56,23 @@ 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` (`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`). + +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 +91,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,9 +105,10 @@ 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` +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.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..d8313bb9 100644 --- a/domains/platform/apis/stats/aggregate_test.go +++ b/domains/platform/apis/stats/aggregate_test.go @@ -112,6 +112,67 @@ 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"]}}}`, + // 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") + 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"}: {3, 1, 3}, + {"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..f6f21683 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: 4, Probes: 2}, + }} + 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["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") + 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..9caac681 --- /dev/null +++ b/domains/platform/apis/stats/geo.go @@ -0,0 +1,210 @@ +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. 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 { + 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 { + skipped++ + continue + } + // 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(kept) == 0 { + return nil, skipped, fmt.Errorf("geo csv holds no usable ranges (%d lines skipped)", skipped) + } + return &Geo{ranges: kept}, 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 muchq.com/stats 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) +} + +// 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. +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..6d769a4c --- /dev/null +++ b/domains/platform/apis/stats/geo_test.go @@ -0,0 +1,166 @@ +package stats + +import ( + "bytes" + "compress/gzip" + "errors" + "strings" + "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 +` + +func TestGeoPlacesAddressesByRangeAndSaysNothingOutsideThem(t *testing.T) { + geo, skipped, err := ParseDBIP(strings.NewReader(dbipSample)) + if err != nil { + t.Fatal(err) + } + // 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", + "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", + "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": "", + "": "", + "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, skipped, err := LoadGeo(objects, key) + if err != nil { + t.Fatalf("%s: %v", key, err) + } + 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 { + 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") + } +} + +// 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/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..d7d3fc49 100644 --- a/domains/platform/apis/stats/main/main.go +++ b/domains/platform/apis/stats/main/main.go @@ -49,19 +49,36 @@ 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. 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{ + Objects: objects, + Store: store, + Logger: logger, + Geo: geo, } go func() { ticker := time.NewTicker(interval) @@ -86,6 +103,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..9c181f93 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,12 @@ 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 := 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) } @@ -155,9 +163,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 +178,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 +256,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 +319,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() { 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=