amadan
status: GeoIP block on every box heartbeat; agent marks tunnel clients (geoip spec §5.5)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
4 files changed,
+82
−2
internal/host/agent.go+27 −0internal/host/agent_geoip_test.go+21 −0internal/status/snapshot.go+17 −2internal/status/snapshot_test.go+17 −0
CI passed — run details
diff --git a/internal/host/agent.go b/internal/host/agent.go| index 94dad4ed4..5b934c27e 100644 |
| --- a/internal/host/agent.go |
| +++ b/internal/host/agent.go |
| @@ -573,6 +573,11 @@ func RunAgent(args []string) { |
| // socket. It presents this box's public address, the one the edges |
| // file publishes; with none, the edge keeps the door shut and says so. |
| edgeDoorOptions(os.Getenv, readPublicIP)...) |
| + // A tunnel client takes the hub's GeoIP answer and mirrors nothing (geoip |
| + // spec §5.1); the same rule tunnelLoop arms on. |
| + if tunnelEnvComplete(os.Getenv) { |
| + edgeOpts = append(edgeOpts, edge.WithGeoTunnelClient()) |
| + } |
| edge.RunWithActivatorReady(args, act, cache, func(e *edge.Edge) { |
| go tunnelLoop(e) |
| // The edge owns the health probers, so its state table is only |
| @@ -581,6 +586,8 @@ func RunAgent(args []string) { |
| // the loop always sees the CURRENT probe set: Reload replaces probers |
| // behind this handle without the heartbeat needing to be re-wired. |
| setUpstreamSource(e.UpstreamStates) |
| + // The GeoIP block for the heartbeat (geoip spec §5.5), the same way. |
| + geoIPSource.Store(e.GeoIPState) |
| // The layout pass reloads routing through this as soon as a move |
| // commits, instead of waiting for the 2s registry poll. |
| reload := func() { _ = e.Reload() } |
| @@ -2780,6 +2787,25 @@ func shouldSkipPublisherElection(getenv func(string) string) bool { |
| // goroutine a test would have no way to stop. |
| var upstreamSource atomic.Value // func() map[string][]status.UpstreamState |
| +// geoIPSource is the edge's GeoIP state for the heartbeat, published from |
| +// the edge-ready callback like upstreamSource (geoip spec §5.5). |
| +var geoIPSource atomic.Value // func() edge.GeoIPState |
| + |
| +// geoIPState reads it, or nil before the edge is ready. |
| +func geoIPState() *status.GeoIPState { |
| + fn, _ := geoIPSource.Load().(func() edge.GeoIPState) |
| + if fn == nil { |
| + return nil |
| + } |
| + s := fn() |
| + return &status.GeoIPState{Mode: s.Mode, SHA256: s.SHA256, Type: s.Type, Error: s.Error} |
| +} |
| + |
| +// tunnelEnvComplete is tunnelLoop's own arming rule: all three set. |
| +func tunnelEnvComplete(getenv func(string) string) bool { |
| + return getenv("CARLOS_TUNNEL_URL") != "" && getenv("CARLOS_TUNNEL_TOKEN") != "" && getenv("CARLOS_TUNNEL_NAME") != "" |
| +} |
| + |
| // setUpstreamSource publishes (or, with nil, withdraws) the probe-state |
| // source. Called once, from the edge-ready callback. |
| func setUpstreamSource(fn func() map[string][]status.UpstreamState) { |
| @@ -2984,6 +3010,7 @@ func statusLoop(reg *registry.Registry) { |
| SystemChannel: boxSystemChannel(os.Getenv), |
| System: systemStateProbe(release.OptDir()), |
| Tunnel: tunnelState(), |
| + GeoIP: geoIPState(), |
| // The edge's probe table, read WHOLE every tick (never merged |
| // with the last one), so a route this box has stopped probing |
| // leaves no ghost behind in the next snapshot. nil until the |
diff --git a/internal/host/agent_geoip_test.go b/internal/host/agent_geoip_test.go| new file mode 100644 |
| index 000000000..c3ed3cc90 |
| --- /dev/null |
| +++ b/internal/host/agent_geoip_test.go |
| @@ -0,0 +1,21 @@ |
| +package host |
| + |
| +import "testing" |
| + |
| +func TestTunnelEnvComplete(t *testing.T) { |
| + env := map[string]string{"CARLOS_TUNNEL_URL": "https://t", "CARLOS_TUNNEL_TOKEN": "x", "CARLOS_TUNNEL_NAME": "n"} |
| + get := func(k string) string { return env[k] } |
| + if !tunnelEnvComplete(get) { |
| + t.Fatal("complete env") |
| + } |
| + delete(env, "CARLOS_TUNNEL_NAME") |
| + if tunnelEnvComplete(get) { |
| + t.Fatal("incomplete env must not make a tunnel client (tunnelLoop would not run)") |
| + } |
| +} |
| + |
| +func TestGeoIPStateNilBeforeTheEdgeIsReady(t *testing.T) { |
| + if geoIPState() != nil { |
| + t.Fatal("no edge published yet must mean no GeoIP block") |
| + } |
| +} |
diff --git a/internal/status/snapshot.go b/internal/status/snapshot.go| index 8f8577ff9..07e479760 100644 |
| --- a/internal/status/snapshot.go |
| +++ b/internal/status/snapshot.go |
| @@ -100,6 +100,9 @@ type Probes struct { |
| // unknown/not-applicable for both, never a guess. |
| Release string |
| Tunnel string |
| + // GeoIP is the edge's GeoIP mirror state (geoip spec §5.5); nil when |
| + // the edge has not published one (no edge yet, or an older binary). |
| + GeoIP *GeoIPState |
| // Platform is runtime.GOOS-runtime.GOARCH, so the console knows which |
| // artifacts a platform promote must mirror. SystemChannel is the box's |
| // CARLOS_SYSTEM_CHANNEL, so the console knows which boxes a promote |
| @@ -280,6 +283,10 @@ type BoxMetrics struct { |
| // naming convention, because it is not using one. |
| InstanceID string `json:"instance_id,omitempty"` |
| Tunnel string `json:"tunnel,omitempty"` // ""|"connected"|"reconnecting" |
| + // GeoIP is present on every box running a GeoIP-aware binary (geoip |
| + // spec §5.5); its absence is how the rollout verdict tells an older |
| + // binary, which does not strip forged geo headers. |
| + GeoIP *GeoIPState `json:"geoip,omitempty"` |
| // RouteVersions is host -> registry.Route.Version for every non-platform |
| // route this box carries with a Version the adopt pass has stamped |
| // (issue #78's staged_version chip) — a hostname-keyed map, which is |
| @@ -741,11 +748,11 @@ func CollectBox(routes []registry.Route, p Probes, now time.Time) Box { |
| p.InstanceID != "" || p.Platform != "" || p.SystemChannel != "" || p.System != nil || |
| len(routeVersions) > 0 || len(aliasReadings) > 0 || len(p.Upstreams) > 0 || |
| len(routeDBBytes) > 0 || p.Layout != nil || len(p.ConfigRefusals) > 0 || p.ConfigGuard != "" || |
| - p.Alerts != nil { |
| + p.Alerts != nil || p.GeoIP != nil { |
| box.Metrics = &BoxMetrics{ |
| MemTotalBytes: p.MemTotalBytes, MemAvailBytes: p.MemAvailBytes, |
| Load1: p.Load1, DiskFreeBytes: p.DiskFreeBytes, |
| - Release: p.Release, Tunnel: p.Tunnel, |
| + Release: p.Release, Tunnel: p.Tunnel, GeoIP: p.GeoIP, |
| Platform: p.Platform, SystemChannel: p.SystemChannel, System: p.System, |
| InstanceID: p.InstanceID, |
| Alerts: p.Alerts, |
| @@ -822,3 +829,11 @@ func currentEtag(ctx context.Context, s release.Store, key string) (string, erro |
| } |
| return etag, nil |
| } |
| + |
| +// GeoIPState is one box's GeoIP mirror (geoip spec §5.5). |
| +type GeoIPState struct { |
| + Mode string `json:"mode"` // mirror | tunnel | none |
| + SHA256 string `json:"sha256,omitempty"` |
| + Type string `json:"type,omitempty"` |
| + Error string `json:"error,omitempty"` |
| +} |
diff --git a/internal/status/snapshot_test.go b/internal/status/snapshot_test.go| index e72097397..85fdcb47c 100644 |
| --- a/internal/status/snapshot_test.go |
| +++ b/internal/status/snapshot_test.go |
| @@ -140,3 +140,20 @@ func TestBoxMetricsCarryAlertDelivery(t *testing.T) { |
| t.Fatalf("fleet.json leaked the alert-sink state: %s", fb) |
| } |
| } |
| + |
| +func TestCollectBoxCarriesGeoIP(t *testing.T) { |
| + box := CollectBox(nil, Probes{GeoIP: &GeoIPState{Mode: "mirror", SHA256: "abc", Type: "DBIP-City-Lite"}}, time.Now()) |
| + if box.Metrics == nil || box.Metrics.GeoIP == nil || box.Metrics.GeoIP.SHA256 != "abc" { |
| + t.Fatalf("metrics = %+v", box.Metrics) |
| + } |
| + raw, err := json.Marshal(box) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if !strings.Contains(string(raw), `"geoip":{"mode":"mirror"`) { |
| + t.Fatalf("json = %s", raw) |
| + } |
| + if b := CollectBox(nil, Probes{}, time.Now()); b.Metrics != nil { |
| + t.Fatal("no GeoIP and nothing else known must still mean no metrics block") |
| + } |
| +} |