diff --git a/client/internal/engine.go b/client/internal/engine.go index 70e7e6734..de151592d 100644 --- a/client/internal/engine.go +++ b/client/internal/engine.go @@ -901,6 +901,16 @@ func (e *Engine) handleAutoUpdateVersion(autoUpdateSettings *mgmProto.AutoUpdate e.updateManager.SetVersion(autoUpdateSettings.Version, autoUpdateSettings.AlwaysUpdate) } +// phase times a sync sub-phase: it returns a function that records the elapsed +// duration when called. Starting the timer at the call site keeps inter-phase +// glue code out of the measurement. +func (e *Engine) phase(name string) func() { + start := time.Now() + return func() { + e.clientMetrics.RecordSyncPhase(e.ctx, name, time.Since(start)) + } +} + func (e *Engine) handleSync(update *mgmProto.SyncResponse) error { started := time.Now() defer func() { @@ -920,7 +930,10 @@ func (e *Engine) handleSync(update *mgmProto.SyncResponse) error { e.handleAutoUpdateVersion(update.NetworkMap.PeerConfig.AutoUpdate) } - if err := e.updateNetbirdConfig(update.GetNetbirdConfig()); err != nil { + done := e.phase("netbird_config") + err := e.updateNetbirdConfig(update.GetNetbirdConfig()) + done() + if err != nil { return err } @@ -934,11 +947,16 @@ func (e *Engine) handleSync(update *mgmProto.SyncResponse) error { return nil } - if err := e.updateChecksIfNew(update.Checks); err != nil { + done = e.phase("checks") + err = e.updateChecksIfNew(update.Checks) + done() + if err != nil { return err } + done = e.phase("persist") e.persistSyncResponse(update) + done() // only apply new changes and ignore old ones if err := e.updateNetworkMap(nm); err != nil { @@ -1343,7 +1361,21 @@ func (e *Engine) updateNetworkMap(networkMap *mgmProto.NetworkMap) error { log.Errorf("failed to update lazy connection feature flag: %v", err) } - e.updateFirewall(networkMap) + if e.firewall != nil { + if localipfw, ok := e.firewall.(localIpUpdater); ok { + if err := localipfw.UpdateLocalIPs(); err != nil { + log.Errorf("failed to update local IPs: %v", err) + } + } + + // If we got empty rules list but management did not set the networkMap.FirewallRulesIsEmpty flag, + // then the mgmt server is older than the client, and we need to allow all traffic for routes. + // This needs to be toggled before applying routes. + isLegacy := len(networkMap.RoutesFirewallRules) == 0 && !networkMap.RoutesFirewallRulesIsEmpty + if err := e.firewall.SetLegacyManagement(isLegacy); err != nil { + log.Errorf("failed to set legacy management flag: %v", err) + } + } protoDNSConfig := networkMap.GetDNSConfig() if protoDNSConfig == nil { @@ -1352,13 +1384,16 @@ func (e *Engine) updateNetworkMap(networkMap *mgmProto.NetworkMap) error { dnsConfig := toDNSConfig(protoDNSConfig, e.wgInterface.Address()) + done := e.phase("dns_server") if err := e.dnsServer.UpdateDNSServer(serial, dnsConfig); err != nil { log.Errorf("failed to update dns server, err: %v", err) } + done() e.routeManager.SetDNSForwarderPort(dnsConfig.ForwarderPort) // apply routes first, route related actions might depend on routing being enabled + done = e.phase("routes_classify") routes := toRoutes(networkMap.GetRoutes()) serverRoutes, clientRoutes := e.routeManager.ClassifyRoutes(routes) @@ -1367,69 +1402,60 @@ func (e *Engine) updateNetworkMap(networkMap *mgmProto.NetworkMap) error { e.connMgr.UpdateRouteHAMap(clientRoutes) log.Debugf("updated lazy connection manager with %d HA groups", len(clientRoutes)) } + done() + done = e.phase("routes_apply") dnsRouteFeatureFlag := toDNSFeatureFlag(networkMap) if err := e.routeManager.UpdateRoutes(serial, serverRoutes, clientRoutes, dnsRouteFeatureFlag); err != nil { log.Errorf("failed to update routes: %v", err) } + done() + done = e.phase("filtering") if e.acl != nil { e.acl.ApplyFiltering(networkMap, dnsRouteFeatureFlag) } + done() + done = e.phase("dns_forwarder") fwdEntries := toRouteDomains(e.config.WgPrivateKey.PublicKey().String(), routes) e.updateDNSForwarder(dnsRouteFeatureFlag, fwdEntries) + done() // Ingress forward rules + done = e.phase("forward_rules") forwardingRules, err := e.updateForwardRules(networkMap.GetForwardingRules()) if err != nil { log.Errorf("failed to update forward rules, err: %v", err) } + done() - remotePeers, err := e.updateNetworkPeers(networkMap) + log.Debugf("got peers update from Management Service, total peers to connect to = %d", len(networkMap.GetRemotePeers())) + + done = e.phase("offline_peers") + e.updateOfflinePeers(networkMap.GetOfflinePeers()) + done() + + remotePeers, err := e.reconcilePeers(networkMap) if err != nil { return err } // must set the exclude list after the peers are added. Without it the manager can not figure out the peers parameters from the store + done = e.phase("lazy_exclude") excludedLazyPeers := e.toExcludedLazyPeers(forwardingRules, remotePeers) e.connMgr.SetExcludeList(e.ctx, excludedLazyPeers) + done() e.networkSerial = serial return nil } -// updateFirewall applies firewall-related state from the network map: refreshing -// local IPs and toggling legacy management mode before routes are applied. -func (e *Engine) updateFirewall(networkMap *mgmProto.NetworkMap) { - if e.firewall == nil { - return - } - - if localipfw, ok := e.firewall.(localIpUpdater); ok { - if err := localipfw.UpdateLocalIPs(); err != nil { - log.Errorf("failed to update local IPs: %v", err) - } - } - - // If we got empty rules list but management did not set the networkMap.FirewallRulesIsEmpty flag, - // then the mgmt server is older than the client, and we need to allow all traffic for routes. - // This needs to be toggled before applying routes. - isLegacy := len(networkMap.RoutesFirewallRules) == 0 && !networkMap.RoutesFirewallRulesIsEmpty - if err := e.firewall.SetLegacyManagement(isLegacy); err != nil { - log.Errorf("failed to set legacy management flag: %v", err) - } -} - -// updateNetworkPeers reconciles the remote peer set from the network map (removing, -// modifying and adding peers, and refreshing SSH state) and returns the remote peers -// with our own peer filtered out. -func (e *Engine) updateNetworkPeers(networkMap *mgmProto.NetworkMap) ([]*mgmProto.RemotePeerConfig, error) { - log.Debugf("got peers update from Management Service, total peers to connect to = %d", len(networkMap.GetRemotePeers())) - - e.updateOfflinePeers(networkMap.GetOfflinePeers()) - +// reconcilePeers applies the remote peer list from the network map (removing, +// modifying and adding peers, then updating SSH config) and returns the remote +// peers with our own peer filtered out, for use by later sync steps. +func (e *Engine) reconcilePeers(networkMap *mgmProto.NetworkMap) ([]*mgmProto.RemotePeerConfig, error) { // Filter out own peer from the remote peers list localPubKey := e.config.WgPrivateKey.PublicKey().String() remotePeers := make([]*mgmProto.RemotePeerConfig, 0, len(networkMap.GetRemotePeers())) @@ -1443,19 +1469,31 @@ func (e *Engine) updateNetworkPeers(networkMap *mgmProto.NetworkMap) ([]*mgmProt if networkMap.GetRemotePeersIsEmpty() { err := e.removeAllPeers() e.statusRecorder.FinishPeerListModifications() - return remotePeers, err + if err != nil { + return nil, err + } + return remotePeers, nil } - if err := e.removePeers(remotePeers); err != nil { - return remotePeers, err + done := e.phase("removed_peers") + err := e.removePeers(remotePeers) + done() + if err != nil { + return nil, err } - if err := e.modifyPeers(remotePeers); err != nil { - return remotePeers, err + done = e.phase("modified_peers") + err = e.modifyPeers(remotePeers) + done() + if err != nil { + return nil, err } - if err := e.addNewPeers(remotePeers); err != nil { - return remotePeers, err + done = e.phase("added_peers") + err = e.addNewPeers(remotePeers) + done() + if err != nil { + return nil, err } e.statusRecorder.FinishPeerListModifications() diff --git a/client/internal/metrics/influxdb.go b/client/internal/metrics/influxdb.go index 531f6a986..4ba14bf44 100644 --- a/client/internal/metrics/influxdb.go +++ b/client/internal/metrics/influxdb.go @@ -120,6 +120,30 @@ func (m *influxDBMetrics) RecordSyncDuration(_ context.Context, agentInfo AgentI m.trimLocked() } +func (m *influxDBMetrics) RecordSyncPhase(_ context.Context, agentInfo AgentInfo, phase string, duration time.Duration) { + tags := fmt.Sprintf("deployment_type=%s,version=%s,os=%s,arch=%s,peer_id=%s,phase=%s", + agentInfo.DeploymentType.String(), + agentInfo.Version, + agentInfo.OS, + agentInfo.Arch, + agentInfo.peerID, + phase, + ) + + m.mu.Lock() + defer m.mu.Unlock() + + m.samples = append(m.samples, influxSample{ + measurement: "netbird_sync_phase", + tags: tags, + fields: map[string]float64{ + "duration_seconds": duration.Seconds(), + }, + timestamp: time.Now(), + }) + m.trimLocked() +} + func (m *influxDBMetrics) RecordLoginDuration(_ context.Context, agentInfo AgentInfo, duration time.Duration, success bool) { result := "success" if !success { diff --git a/client/internal/metrics/infra/README.md b/client/internal/metrics/infra/README.md index 5a93dbd87..7941a30cf 100644 --- a/client/internal/metrics/infra/README.md +++ b/client/internal/metrics/infra/README.md @@ -78,6 +78,25 @@ Tags: - `os`: Operating system (linux, darwin, windows, android, ios, etc.) - `arch`: CPU architecture (amd64, arm64, etc.) +### Sync Phase Timing + +Measurement: `netbird_sync_phase` + +Breaks down where time goes inside a single sync, so the total `netbird_sync` duration can be attributed to the sub-step that dominates. + +| Field | Description | +|-------|-------------| +| `duration_seconds` | Time spent in one sub-phase of sync processing | + +Tags: +- `phase`: the sub-phase — `netbird_config`, `checks`, `persist`, `dns_server`, `routes_classify`, `routes_apply`, `filtering`, `dns_forwarder`, `forward_rules`, `offline_peers`, `removed_peers`, `modified_peers`, `added_peers`, `lazy_exclude` +- `deployment_type`: "cloud" | "selfhosted" | "unknown" +- `version`: NetBird version string +- `os`: Operating system (linux, darwin, windows, android, ios, etc.) +- `arch`: CPU architecture (amd64, arm64, etc.) + +**Note:** this is wall-time per phase — it includes both CPU work and time spent waiting on locks. A slow phase points to *where* the time goes, not *why*; pair it with lock-wait metrics to tell contention apart from real work. + ### Login Duration Measurement: `netbird_login` @@ -191,4 +210,52 @@ docker compose exec influxdb influx query \ # Check ingest server health curl http://localhost:8087/health -``` \ No newline at end of file +``` + +## Analyzing a Debug Bundle + +Metrics collection is always on, so every debug bundle ships a `metrics.txt` in InfluxDB line protocol — a timestamped time series of all recorded events (sync durations, sync phases, connection stages, login). You can replay it into the local stack and graph it, without a running client. + +The bundle's `metrics.txt` is a rolling window (capped at 5 days / ~20k samples, see [Buffer Limits](#buffer-limits)). For a connection incident the relevant window is short (connection setup is seconds), so a bundle captured during the issue is enough. + +### 1. Start the stack + +```bash +# From this directory (client/internal/metrics/infra) +INFLUXDB_ADMIN_TOKEN=admin123 INFLUXDB_ADMIN_PASSWORD=admin123 GRAFANA_ADMIN_PASSWORD=admin123 \ + docker compose up -d +``` + +(`admin123` are throwaway local credentials — fine for offline analysis.) + +### 2. Clear any previous data + +So you only see this bundle: + +```bash +docker exec influxdb influx delete --org netbird --bucket metrics --token admin123 \ + --start 1970-01-01T00:00:00Z --stop 2100-01-01T00:00:00Z +``` + +### 3. Import the bundle's metrics.txt + +InfluxDB is not exposed on the host, so import inside the container: + +```bash +docker cp /path/to/bundle/metrics.txt influxdb:/tmp/m.txt +docker exec influxdb influx write --org netbird --bucket metrics --precision ns \ + --token admin123 --file /tmp/m.txt +``` + +Re-importing the same file is idempotent (same measurement+tags+timestamp overwrites). + +### 4. View the dashboards + +Grafana on http://localhost:3001 (login `admin` / `admin123`), datasource pre-provisioned: + +- **Where sync time goes:** http://localhost:3001/d/netbird-sync-phases/netbird-sync-phases-where-time-goes +- **General client metrics:** http://localhost:3001/d/netbird-influxdb-metrics + +**Set the time range** to cover the bundle's timestamps (e.g. "Last 7 days" or an absolute range matching when the bundle was taken) — with the default short range the panels look empty. + +Bundles are distinguishable by the `version` tag; add a tag at import time (e.g. `sed 's/^netbird_\([a-z_]*\),/netbird_\1,bundle=mycase,/' metrics.txt`) if you want to compare several side by side. \ No newline at end of file diff --git a/client/internal/metrics/infra/grafana/provisioning/dashboards/json/netbird-sync-phases.json b/client/internal/metrics/infra/grafana/provisioning/dashboards/json/netbird-sync-phases.json new file mode 100644 index 000000000..69dbac0ae --- /dev/null +++ b/client/internal/metrics/infra/grafana/provisioning/dashboards/json/netbird-sync-phases.json @@ -0,0 +1,259 @@ +{ + "annotations": { + "list": [] + }, + "editable": true, + "fiscalYearStartMonth": 0, + "graphTooltip": 1, + "links": [], + "refresh": "", + "schemaVersion": 39, + "tags": [ + "netbird", + "sync" + ], + "templating": { + "list": [ + { + "current": { + "text": "All", + "value": "$__all" + }, + "datasource": { + "type": "influxdb", + "uid": "influxdb" + }, + "definition": "import \"influxdata/influxdb/schema\"\nschema.tagValues(bucket: \"metrics\", tag: \"version\")", + "includeAll": true, + "label": "version", + "multi": true, + "name": "version", + "query": "import \"influxdata/influxdb/schema\"\nschema.tagValues(bucket: \"metrics\", tag: \"version\")", + "refresh": 2, + "type": "query", + "allValue": ".*" + } + ] + }, + "time": { + "from": "now-2d", + "to": "now" + }, + "timepicker": {}, + "timezone": "", + "title": "NetBird Sync Phases (where time goes)", + "uid": "netbird-sync-phases", + "version": 1, + "panels": [ + { + "id": 1, + "title": "Time per phase over time (stacked, ms)", + "type": "timeseries", + "datasource": { + "type": "influxdb", + "uid": "influxdb" + }, + "gridPos": { + "h": 10, + "w": 24, + "x": 0, + "y": 0 + }, + "fieldConfig": { + "defaults": { + "unit": "ms", + "custom": { + "drawStyle": "bars", + "stacking": { + "mode": "normal", + "group": "A" + }, + "fillOpacity": 80, + "lineWidth": 0 + } + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "table", + "placement": "right", + "calcs": [ + "max", + "mean" + ] + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "targets": [ + { + "refId": "A", + "datasource": { + "type": "influxdb", + "uid": "influxdb" + }, + "query": "from(bucket: \"metrics\")\n |> range(start: v.timeRangeStart, stop: v.timeRangeStop)\n |> filter(fn: (r) => r._measurement == \"netbird_sync_phase\" and r._field == \"duration_seconds\")\n |> filter(fn: (r) => r.version =~ /${version:regex}/)\n |> map(fn: (r) => ({ r with _value: r._value * 1000.0 }))\n |> keep(columns: [\"_time\", \"_value\", \"phase\"])\n |> group(columns: [\"phase\"])" + } + ] + }, + { + "id": 2, + "title": "p95 per phase (ms)", + "type": "bargauge", + "datasource": { + "type": "influxdb", + "uid": "influxdb" + }, + "gridPos": { + "h": 11, + "w": 12, + "x": 0, + "y": 10 + }, + "fieldConfig": { + "defaults": { + "unit": "ms", + "color": { + "mode": "continuous-GrYlRd" + } + }, + "overrides": [] + }, + "options": { + "displayMode": "gradient", + "orientation": "horizontal", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "showUnfilled": true + }, + "targets": [ + { + "refId": "A", + "datasource": { + "type": "influxdb", + "uid": "influxdb" + }, + "query": "from(bucket: \"metrics\")\n |> range(start: v.timeRangeStart, stop: v.timeRangeStop)\n |> filter(fn: (r) => r._measurement == \"netbird_sync_phase\" and r._field == \"duration_seconds\")\n |> filter(fn: (r) => r.version =~ /${version:regex}/)\n |> map(fn: (r) => ({ r with _value: r._value * 1000.0 }))\n |> group(columns: [\"phase\"])\n |> quantile(q: 0.95)\n |> group()\n |> sort(columns: [\"_value\"], desc: true)" + } + ] + }, + { + "id": 3, + "title": "Per-phase stats (ms): mean / p95 / max", + "type": "table", + "datasource": { + "type": "influxdb", + "uid": "influxdb" + }, + "gridPos": { + "h": 11, + "w": 12, + "x": 12, + "y": 10 + }, + "fieldConfig": { + "defaults": { + "unit": "ms" + }, + "overrides": [] + }, + "options": { + "showHeader": true, + "sortBy": [ + { + "displayName": "max", + "desc": true + } + ] + }, + "transformations": [ + { + "id": "merge", + "options": {} + } + ], + "targets": [ + { + "refId": "mean", + "datasource": { + "type": "influxdb", + "uid": "influxdb" + }, + "query": "from(bucket: \"metrics\")\n |> range(start: v.timeRangeStart, stop: v.timeRangeStop)\n |> filter(fn: (r) => r._measurement == \"netbird_sync_phase\" and r._field == \"duration_seconds\")\n |> filter(fn: (r) => r.version =~ /${version:regex}/)\n |> map(fn: (r) => ({ r with _value: r._value * 1000.0 }))\n |> group(columns: [\"phase\"])\n |> mean()\n |> group()\n |> keep(columns: [\"phase\", \"_value\"])\n |> rename(columns: {_value: \"mean\"})" + }, + { + "refId": "p95", + "datasource": { + "type": "influxdb", + "uid": "influxdb" + }, + "query": "from(bucket: \"metrics\")\n |> range(start: v.timeRangeStart, stop: v.timeRangeStop)\n |> filter(fn: (r) => r._measurement == \"netbird_sync_phase\" and r._field == \"duration_seconds\")\n |> filter(fn: (r) => r.version =~ /${version:regex}/)\n |> map(fn: (r) => ({ r with _value: r._value * 1000.0 }))\n |> group(columns: [\"phase\"])\n |> quantile(q: 0.95)\n |> group()\n |> keep(columns: [\"phase\", \"_value\"])\n |> rename(columns: {_value: \"p95\"})" + }, + { + "refId": "max", + "datasource": { + "type": "influxdb", + "uid": "influxdb" + }, + "query": "from(bucket: \"metrics\")\n |> range(start: v.timeRangeStart, stop: v.timeRangeStop)\n |> filter(fn: (r) => r._measurement == \"netbird_sync_phase\" and r._field == \"duration_seconds\")\n |> filter(fn: (r) => r.version =~ /${version:regex}/)\n |> map(fn: (r) => ({ r with _value: r._value * 1000.0 }))\n |> group(columns: [\"phase\"])\n |> max()\n |> group()\n |> keep(columns: [\"phase\", \"_value\"])\n |> rename(columns: {_value: \"max\"})" + } + ] + }, + { + "id": 4, + "title": "Total sync duration (netbird_sync, ms) \u2014 reference", + "type": "timeseries", + "datasource": { + "type": "influxdb", + "uid": "influxdb" + }, + "gridPos": { + "h": 8, + "w": 24, + "x": 0, + "y": 21 + }, + "fieldConfig": { + "defaults": { + "unit": "ms", + "custom": { + "drawStyle": "points", + "pointSize": 5 + } + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "table", + "placement": "right", + "calcs": [ + "max", + "mean" + ] + }, + "tooltip": { + "mode": "single" + } + }, + "targets": [ + { + "refId": "A", + "datasource": { + "type": "influxdb", + "uid": "influxdb" + }, + "query": "from(bucket: \"metrics\")\n |> range(start: v.timeRangeStart, stop: v.timeRangeStop)\n |> filter(fn: (r) => r._measurement == \"netbird_sync\" and r._field == \"duration_seconds\")\n |> filter(fn: (r) => r.version =~ /${version:regex}/)\n |> map(fn: (r) => ({ r with _value: r._value * 1000.0 }))\n |> keep(columns: [\"_time\", \"_value\", \"version\"])\n |> group(columns: [\"version\"])" + } + ] + } + ] +} \ No newline at end of file diff --git a/client/internal/metrics/infra/ingest/main.go b/client/internal/metrics/infra/ingest/main.go index a5031a873..623a17e4d 100644 --- a/client/internal/metrics/infra/ingest/main.go +++ b/client/internal/metrics/infra/ingest/main.go @@ -59,6 +59,19 @@ var allowedMeasurements = map[string]measurementSpec{ "peer_id": true, }, }, + "netbird_sync_phase": { + allowedFields: map[string]bool{ + "duration_seconds": true, + }, + allowedTags: map[string]bool{ + "deployment_type": true, + "version": true, + "os": true, + "arch": true, + "peer_id": true, + "phase": true, + }, + }, "netbird_login": { allowedFields: map[string]bool{ "duration_seconds": true, diff --git a/client/internal/metrics/metrics.go b/client/internal/metrics/metrics.go index 4ebb43496..f18082995 100644 --- a/client/internal/metrics/metrics.go +++ b/client/internal/metrics/metrics.go @@ -56,6 +56,9 @@ type metricsImplementation interface { // RecordSyncDuration records how long it took to process a sync message RecordSyncDuration(ctx context.Context, agentInfo AgentInfo, duration time.Duration) + // RecordSyncPhase records how long a single sub-phase of sync processing took + RecordSyncPhase(ctx context.Context, agentInfo AgentInfo, phase string, duration time.Duration) + // RecordLoginDuration records how long the login to management took RecordLoginDuration(ctx context.Context, agentInfo AgentInfo, duration time.Duration, success bool) @@ -127,6 +130,18 @@ func (c *ClientMetrics) RecordSyncDuration(ctx context.Context, duration time.Du c.impl.RecordSyncDuration(ctx, agentInfo, duration) } +// RecordSyncPhase records the duration of a single sub-phase of sync processing +func (c *ClientMetrics) RecordSyncPhase(ctx context.Context, phase string, duration time.Duration) { + if c == nil { + return + } + c.mu.RLock() + agentInfo := c.agentInfo + c.mu.RUnlock() + + c.impl.RecordSyncPhase(ctx, agentInfo, phase, duration) +} + // RecordLoginDuration records how long the login to management server took func (c *ClientMetrics) RecordLoginDuration(ctx context.Context, duration time.Duration, success bool) { if c == nil { diff --git a/client/internal/metrics/push_test.go b/client/internal/metrics/push_test.go index 20a509da1..43c1b2c06 100644 --- a/client/internal/metrics/push_test.go +++ b/client/internal/metrics/push_test.go @@ -70,6 +70,9 @@ func (m *mockMetrics) RecordConnectionStages(_ context.Context, _ AgentInfo, _ s func (m *mockMetrics) RecordSyncDuration(_ context.Context, _ AgentInfo, _ time.Duration) { } +func (m *mockMetrics) RecordSyncPhase(_ context.Context, _ AgentInfo, _ string, _ time.Duration) { +} + func (m *mockMetrics) RecordLoginDuration(_ context.Context, _ AgentInfo, _ time.Duration, _ bool) { } diff --git a/shared/management/client/grpc.go b/shared/management/client/grpc.go index 016cde68a..6f5172376 100644 --- a/shared/management/client/grpc.go +++ b/shared/management/client/grpc.go @@ -55,6 +55,14 @@ type GrpcClient struct { connStateCallback ConnStateNotifier connStateCallbackLock sync.RWMutex serverURL string + + // syncStreamErr holds the last Sync stream error, or nil while the stream + // is established and healthy. GetServerKey succeeds even when the peer + // cannot sync (e.g. the server returns "settings not found"), so the + // health probe must consult this to avoid reporting a healthy management + // connection while the Sync stream keeps failing. + syncStreamMu sync.RWMutex + syncStreamErr error } type ExposeRequest struct { @@ -364,6 +372,8 @@ func (c *GrpcClient) handleSyncStream(ctx context.Context, serverPubKey wgtypes. stream, err := c.connectToSyncStream(ctx, serverPubKey, sysInfo) if err != nil { log.Debugf("failed to open Management Service stream: %s", err) + c.notifyDisconnected(err) + c.setSyncStreamDisconnected(err) if s, ok := gstatus.FromError(err); ok && s.Code() == codes.PermissionDenied { return backoff.Permanent(err) // unrecoverable error, propagate to the upper layer } @@ -372,11 +382,13 @@ func (c *GrpcClient) handleSyncStream(ctx context.Context, serverPubKey wgtypes. log.Infof("connected to the Management Service stream") c.notifyConnected() + c.setSyncStreamConnected() // blocking until error err = c.receiveUpdatesEvents(stream, serverPubKey, msgHandler) if err != nil { c.notifyDisconnected(err) + c.setSyncStreamDisconnected(err) if ctx.Err() != nil { log.Debugf("management connection context has been canceled, this usually indicates shutdown") return nil @@ -530,6 +542,13 @@ func (c *GrpcClient) IsHealthy() bool { log.Warnf("health check returned: %s", err) return false } + + if syncErr := c.syncStreamError(); syncErr != nil { + c.notifyDisconnected(syncErr) + log.Warnf("management transport is up but the Sync stream is unhealthy: %s", syncErr) + return false + } + c.notifyConnected() return true } @@ -771,6 +790,24 @@ func (c *GrpcClient) SyncMeta(sysInfo *system.Info) error { return err } +func (c *GrpcClient) setSyncStreamConnected() { + c.syncStreamMu.Lock() + defer c.syncStreamMu.Unlock() + c.syncStreamErr = nil +} + +func (c *GrpcClient) setSyncStreamDisconnected(err error) { + c.syncStreamMu.Lock() + defer c.syncStreamMu.Unlock() + c.syncStreamErr = err +} + +func (c *GrpcClient) syncStreamError() error { + c.syncStreamMu.RLock() + defer c.syncStreamMu.RUnlock() + return c.syncStreamErr +} + func (c *GrpcClient) notifyDisconnected(err error) { c.connStateCallbackLock.RLock() defer c.connStateCallbackLock.RUnlock() diff --git a/shared/signal/client/grpc.go b/shared/signal/client/grpc.go index 611ab0c45..a07867263 100644 --- a/shared/signal/client/grpc.go +++ b/shared/signal/client/grpc.go @@ -85,6 +85,7 @@ type GrpcClient struct { // receive backpressure as a dead stream: reconnecting cannot help, since the // new stream feeds the same worker, and only triggers a reconnect storm. receiveHandoffBlocked atomic.Bool + watchdogWg sync.WaitGroup } // NewClient creates a new Signal client @@ -200,10 +201,18 @@ func (c *GrpcClient) Receive(ctx context.Context, msgHandler func(msg *proto.Mes // Guard the receive direction: the transport can stay healthy while the // server stops delivering messages. The watchdog reconnects via cancelStream. c.markReceived() - go c.watchReceiveStream(streamCtx, cancelStream) + c.watchdogWg.Add(1) + go func() { + defer c.watchdogWg.Done() + c.watchReceiveStream(streamCtx, cancelStream) + }() // start receiving messages from the Signal stream (from other peers through signal) err = c.receive(stream) + + cancelStream() + c.watchdogWg.Wait() + if err != nil { // Check the parent context, not streamCtx: a watchdog-triggered // cancelStream must reconnect, only a parent cancel is shutdown. @@ -400,7 +409,12 @@ func (c *GrpcClient) encryptMessage(msg *proto.Message) (*proto.EncryptedMessage // Send sends a message to the remote Peer through the Signal Exchange. func (c *GrpcClient) Send(msg *proto.Message) error { + return c.send(c.ctx, msg) +} +// send delivers a message deriving per-attempt timeouts from parentCtx, so a +// caller can abort an in-flight send by cancelling that context. +func (c *GrpcClient) send(parentCtx context.Context, msg *proto.Message) error { if !c.Ready() { return fmt.Errorf("no connection to signal") } @@ -416,7 +430,7 @@ func (c *GrpcClient) Send(msg *proto.Message) error { if attempt > 1 { attemptTimeout = time.Duration(attempt) * 5 * time.Second } - ctx, cancel := context.WithTimeout(c.ctx, attemptTimeout) + ctx, cancel := context.WithTimeout(parentCtx, attemptTimeout) _, err = c.realClient.Send(ctx, encryptedMessage) @@ -486,7 +500,7 @@ func (c *GrpcClient) watchReceiveStream(ctx context.Context, cancelStream contex } if probeSentAt.IsZero() { - if err := c.sendReceiveProbe(); err != nil { + if err := c.sendReceiveProbe(ctx); err != nil { log.Debugf("failed to send signal receive probe: %v", err) } probeSentAt = time.Now() @@ -495,11 +509,13 @@ func (c *GrpcClient) watchReceiveStream(ctx context.Context, cancelStream contex } } -// sendReceiveProbe sends a self-addressed heartbeat. The Signal server routes it -// back to this client, exercising the exact receive path the watchdog guards. -func (c *GrpcClient) sendReceiveProbe() error { +// sendReceiveProbe sends a self-addressed heartbeat bound to ctx, so cancelStream +// aborts an in-flight probe instead of leaving the watchdog blocked on send timeouts. +// The Signal server routes it back to this client, exercising the exact receive +// path the watchdog guards. +func (c *GrpcClient) sendReceiveProbe(ctx context.Context) error { self := c.key.PublicKey().String() - return c.Send(&proto.Message{ + return c.send(ctx, &proto.Message{ Key: self, RemoteKey: self, Body: &proto.Body{Type: proto.Body_HEARTBEAT}, @@ -541,6 +557,9 @@ func (c *GrpcClient) receive(stream proto.SignalExchange_ConnectStreamClient) er if err := c.decryptionWorker.AddMsg(c.ctx, msg); err != nil { log.Errorf("failed to add message to decryption worker: %v", err) } + // Refresh liveness before clearing the flag so the window between here and + // the next Recv does not read a stale timestamp as a dead stream. + c.markReceived() c.receiveHandoffBlocked.Store(false) } } diff --git a/shared/signal/client/watchdog_test.go b/shared/signal/client/watchdog_test.go index bc6b5520b..a8bbafa29 100644 --- a/shared/signal/client/watchdog_test.go +++ b/shared/signal/client/watchdog_test.go @@ -2,6 +2,7 @@ package client import ( "context" + "io" "net" "testing" "time" @@ -74,7 +75,7 @@ func TestReceiveProbeRoundTrips(t *testing.T) { t.Fatal("signal stream did not connect within timeout") } - require.NoError(t, client.sendReceiveProbe()) + require.NoError(t, client.sendReceiveProbe(ctx)) select { case <-received: @@ -106,3 +107,72 @@ func TestReceiveAliveTreatsHandoffBlockAsLiveness(t *testing.T) { c.markReceived() require.True(t, c.receiveAlive(), "a freshly received frame must keep the stream alive") } + +// fakeRecvStream feeds the receive loop frames from a channel and reports EOF +// once the channel is closed. Only Recv is exercised by the loop. +type fakeRecvStream struct { + sigProto.SignalExchange_ConnectStreamClient + frames chan *sigProto.EncryptedMessage +} + +func (s *fakeRecvStream) Recv() (*sigProto.EncryptedMessage, error) { + msg, ok := <-s.frames + if !ok { + return nil, io.EOF + } + return msg, nil +} + +// TestReceiveLoopRefreshesLivenessAfterBlockedHandoff drives the real receive +// loop into a handoff that blocks past the inactivity threshold, then checks the +// window after the handoff drains but before the next Recv. The loop must have +// refreshed the timestamp on unblocking, otherwise that window reads the stale +// pre-handoff timestamp as a dead stream and the watchdog tears down a healthy +// connection. +func TestReceiveLoopRefreshesLivenessAfterBlockedHandoff(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + c := &GrpcClient{ctx: ctx} + + handling := make(chan struct{}, 8) + gate := make(chan struct{}) + decrypt := func(*sigProto.EncryptedMessage) (*sigProto.Message, error) { return &sigProto.Message{}, nil } + handler := func(*sigProto.Message) error { + handling <- struct{}{} + <-gate + return nil + } + c.decryptionWorker = NewWorker(decrypt, handler) + workerCtx, workerCancel := context.WithCancel(context.Background()) + go c.decryptionWorker.Work(workerCtx) + t.Cleanup(workerCancel) + + frames := make(chan *sigProto.EncryptedMessage) + t.Cleanup(func() { close(frames) }) + go func() { _ = c.receive(&fakeRecvStream{frames: frames}) }() + + // First frame: the worker drains it and parks in the blocking handler. + frames <- &sigProto.EncryptedMessage{} + <-handling + // Second frame fills the worker's single-slot pool. + frames <- &sigProto.EncryptedMessage{} + // Third frame: the pool is full, so the loop parks on the handoff. + frames <- &sigProto.EncryptedMessage{} + + require.Eventually(t, c.receiveHandoffBlocked.Load, time.Second, time.Millisecond, + "receive loop should park on the worker handoff") + + // Simulate the handoff having blocked past the inactivity threshold. + c.lastReceived.Store(time.Now().Add(-2 * receiveInactivityThreshold).UnixNano()) + require.True(t, c.receiveAlive(), "a loop parked on the handoff must stay alive") + + // Drain the worker so the handoff returns and the loop resumes reading. + close(gate) + + // Once the handoff clears, the loop is parked on the next Recv with no frame + // pending. The stream must still read as alive in that window. + require.Eventually(t, func() bool { return !c.receiveHandoffBlocked.Load() }, time.Second, time.Millisecond, + "handoff should drain once the worker is released") + require.True(t, c.receiveAlive(), + "the loop must refresh liveness when the handoff drains, before the next Recv") +}