mirror of
https://github.com/netbirdio/netbird.git
synced 2026-10-14 09:29:07 +02:00
[management, client] Add management-controlled client metrics push (#5886)
* [management, client] Add management-controlled client metrics push Allow enabling/disabling client metrics push from the dashboard via account settings instead of requiring env vars on every client. - Add MetricsConfig proto message to NetbirdConfig - Add MetricsPushEnabled to account Settings (DB-persisted) - Expose metrics_push_enabled in OpenAPI and dashboard API handler - Populate MetricsConfig in sync and login responses - Client dynamically starts/stops push based on management config - NB_METRICS_PUSH_ENABLED env var overrides management when explicitly set - Add activity events for metrics push enable/disable * Remove log line * [management] Fix peer update test for MetricsConfig in NetbirdConfig Update TestUpdateAccountPeers assertions: NetbirdConfig is no longer nil in peer update responses since it now carries MetricsConfig even when STUN/TURN config is absent. * Regenerate proto files with protoc v7.34.1 * [management] Read metrics push setting in Postgres account query getAccountPgx omitted settings_metrics_push_enabled from its hand-written SELECT and Scan, so the toggle was always read back as false on Postgres and never reached clients. * [client] Fix metrics push getting stuck off after engine restart Engine restarts (backoff retries within the same login session) cancel e.ctx, which the push goroutine's lifetime was tied to. The goroutine died silently but ClientMetrics.push stayed non-nil since only an explicit stop clears it, so the next UpdatePushFromMgm call saw a "push already running" state and never restarted it. Give the Engine its own metricsCtx sourced from ConnectClient.ctx, which outlives engine restarts, so handleMetricsUpdate stops tying the push to the wrong-scoped context. Additionally make ClientMetrics.push an atomic.Pointer that the push goroutine clears via CompareAndSwap on exit, so the tracked state can never drift from the goroutine's actual lifetime regardless of which context a future caller passes in. * [management] Regenerate OpenAPI types with oapi-codegen v2.7.1 types.gen.go was regenerated with a stale local v2.6.0 binary, causing the CI git-diff check against generate.sh's pinned v2.7.1 to fail.
This commit is contained in:
@@ -314,6 +314,10 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
|
||||
c.clientMetrics.RecordLoginDuration(engineCtx, time.Since(loginStarted), true)
|
||||
c.statusRecorder.MarkManagementConnected()
|
||||
|
||||
if metricsConfig := loginResp.GetNetbirdConfig().GetMetrics(); metricsConfig != nil {
|
||||
c.clientMetrics.UpdatePushFromMgm(c.ctx, metricsConfig.GetEnabled())
|
||||
}
|
||||
|
||||
localPeerState := peer.LocalPeerState{
|
||||
IP: loginResp.GetPeerConfig().GetAddress(),
|
||||
PubKey: myPrivateKey.PublicKey().String(),
|
||||
@@ -399,6 +403,7 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
|
||||
StateManager: stateManager,
|
||||
UpdateManager: c.updateManager,
|
||||
ClientMetrics: c.clientMetrics,
|
||||
MetricsCtx: c.ctx,
|
||||
}, mobileDependency)
|
||||
engine.SetSyncResponsePersistence(c.persistSyncResponse)
|
||||
c.engine = engine
|
||||
|
||||
@@ -172,6 +172,7 @@ type EngineServices struct {
|
||||
StateManager *statemanager.Manager
|
||||
UpdateManager *updater.Manager
|
||||
ClientMetrics *metrics.ClientMetrics
|
||||
MetricsCtx context.Context
|
||||
}
|
||||
|
||||
// Engine is a mechanism responsible for reacting on Signal and Management stream events and managing connections to the remote peers.
|
||||
@@ -264,6 +265,7 @@ type Engine struct {
|
||||
|
||||
// clientMetrics collects and pushes metrics
|
||||
clientMetrics *metrics.ClientMetrics
|
||||
metricsCtx context.Context
|
||||
|
||||
jobExecutor *jobexec.Executor
|
||||
jobExecutorWG sync.WaitGroup
|
||||
@@ -316,6 +318,7 @@ func NewEngine(
|
||||
probeStunTurn: relay.NewStunTurnProbe(relay.DefaultCacheTTL),
|
||||
jobExecutor: jobexec.NewExecutor(),
|
||||
clientMetrics: services.ClientMetrics,
|
||||
metricsCtx: services.MetricsCtx,
|
||||
updateManager: services.UpdateManager,
|
||||
syncStoreDir: config.StateDir,
|
||||
}
|
||||
@@ -997,6 +1000,8 @@ func (e *Engine) updateNetbirdConfig(wCfg *mgmProto.NetbirdConfig) error {
|
||||
return fmt.Errorf("handle the flow configuration: %w", err)
|
||||
}
|
||||
|
||||
e.handleMetricsUpdate(wCfg.GetMetrics())
|
||||
|
||||
if err := e.PopulateNetbirdConfig(wCfg, nil); err != nil {
|
||||
log.Warnf("Failed to update DNS server config: %v", err)
|
||||
}
|
||||
@@ -1066,6 +1071,14 @@ func (e *Engine) handleFlowUpdate(config *mgmProto.FlowConfig) error {
|
||||
return e.flowManager.Update(flowConfig)
|
||||
}
|
||||
|
||||
func (e *Engine) handleMetricsUpdate(config *mgmProto.MetricsConfig) {
|
||||
if config == nil {
|
||||
return
|
||||
}
|
||||
log.Infof("received metrics configuration from management: enabled=%v", config.GetEnabled())
|
||||
e.clientMetrics.UpdatePushFromMgm(e.metricsCtx, config.GetEnabled())
|
||||
}
|
||||
|
||||
func toFlowLoggerConfig(config *mgmProto.FlowConfig) (*nftypes.FlowConfig, error) {
|
||||
if config.GetInterval() == nil {
|
||||
return nil, errors.New("flow interval is nil")
|
||||
|
||||
@@ -60,6 +60,13 @@ func getMetricsInterval() time.Duration {
|
||||
return interval
|
||||
}
|
||||
|
||||
// isMetricsPushEnvSet returns true if NB_METRICS_PUSH_ENABLED is explicitly set (to any value).
|
||||
// When set, the env var takes full precedence over management server configuration.
|
||||
func isMetricsPushEnvSet() bool {
|
||||
_, set := os.LookupEnv(EnvMetricsPushEnabled)
|
||||
return set
|
||||
}
|
||||
|
||||
func isForceSending() bool {
|
||||
force, _ := strconv.ParseBool(os.Getenv(EnvMetricsForceSending))
|
||||
return force
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"fmt"
|
||||
"io"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
@@ -75,7 +76,7 @@ type ClientMetrics struct {
|
||||
agentInfo AgentInfo
|
||||
mu sync.RWMutex
|
||||
|
||||
push *Push
|
||||
push atomic.Pointer[Push]
|
||||
pushMu sync.Mutex
|
||||
wg sync.WaitGroup
|
||||
pushCancel context.CancelFunc
|
||||
@@ -167,10 +168,7 @@ func (c *ClientMetrics) UpdateAgentInfo(agentInfo AgentInfo, publicKey string) {
|
||||
c.agentInfo = agentInfo
|
||||
c.mu.Unlock()
|
||||
|
||||
c.pushMu.Lock()
|
||||
push := c.push
|
||||
c.pushMu.Unlock()
|
||||
if push != nil {
|
||||
if push := c.push.Load(); push != nil {
|
||||
push.SetPeerID(agentInfo.peerID)
|
||||
}
|
||||
}
|
||||
@@ -184,7 +182,7 @@ func (c *ClientMetrics) Export(w io.Writer) error {
|
||||
return c.impl.Export(w)
|
||||
}
|
||||
|
||||
// StartPush starts periodic pushing of metrics with the given configuration
|
||||
// StartPush starts periodic pushing of metrics with the given configuration.
|
||||
// Precedence: PushConfig.ServerAddress > remote config server_url
|
||||
func (c *ClientMetrics) StartPush(ctx context.Context, config PushConfig) {
|
||||
if c == nil {
|
||||
@@ -194,11 +192,58 @@ func (c *ClientMetrics) StartPush(ctx context.Context, config PushConfig) {
|
||||
c.pushMu.Lock()
|
||||
defer c.pushMu.Unlock()
|
||||
|
||||
if c.push != nil {
|
||||
if c.push.Load() != nil {
|
||||
log.Warnf("metrics push already running")
|
||||
return
|
||||
}
|
||||
|
||||
c.startPushLocked(ctx, config)
|
||||
}
|
||||
|
||||
// StopPush stops the periodic metrics push.
|
||||
func (c *ClientMetrics) StopPush() {
|
||||
if c == nil {
|
||||
return
|
||||
}
|
||||
c.pushMu.Lock()
|
||||
defer c.pushMu.Unlock()
|
||||
|
||||
c.stopPushLocked()
|
||||
}
|
||||
|
||||
// UpdatePushFromMgm updates metrics push based on management server configuration.
|
||||
// If NB_METRICS_PUSH_ENABLED is explicitly set (true or false), management config is ignored.
|
||||
// When unset, management controls whether push is enabled.
|
||||
func (c *ClientMetrics) UpdatePushFromMgm(ctx context.Context, enabled bool) {
|
||||
if c == nil {
|
||||
return
|
||||
}
|
||||
|
||||
if isMetricsPushEnvSet() {
|
||||
log.Debugf("ignoring management config, env var is explicitly set: %s", EnvMetricsPushEnabled)
|
||||
return
|
||||
}
|
||||
|
||||
c.pushMu.Lock()
|
||||
defer c.pushMu.Unlock()
|
||||
|
||||
if enabled {
|
||||
if c.push.Load() != nil {
|
||||
return
|
||||
}
|
||||
log.Infof("enabled metrics push by management")
|
||||
c.startPushLocked(ctx, PushConfigFromEnv())
|
||||
} else {
|
||||
if c.push.Load() == nil {
|
||||
return
|
||||
}
|
||||
log.Infof("disabled metrics push by management")
|
||||
c.stopPushLocked()
|
||||
}
|
||||
}
|
||||
|
||||
// startPushLocked starts push. Caller must hold pushMu.
|
||||
func (c *ClientMetrics) startPushLocked(ctx context.Context, config PushConfig) {
|
||||
c.mu.RLock()
|
||||
agentVersion := c.agentInfo.Version
|
||||
peerID := c.agentInfo.peerID
|
||||
@@ -214,26 +259,23 @@ func (c *ClientMetrics) StartPush(ctx context.Context, config PushConfig) {
|
||||
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
c.pushCancel = cancel
|
||||
c.push.Store(push)
|
||||
|
||||
c.wg.Add(1)
|
||||
go func() {
|
||||
defer c.wg.Done()
|
||||
push.Start(ctx)
|
||||
c.push.CompareAndSwap(push, nil)
|
||||
}()
|
||||
c.push = push
|
||||
}
|
||||
|
||||
func (c *ClientMetrics) StopPush() {
|
||||
if c == nil {
|
||||
return
|
||||
}
|
||||
c.pushMu.Lock()
|
||||
defer c.pushMu.Unlock()
|
||||
if c.push == nil {
|
||||
// stopPushLocked stops push. Caller must hold pushMu.
|
||||
func (c *ClientMetrics) stopPushLocked() {
|
||||
if c.push.Load() == nil {
|
||||
return
|
||||
}
|
||||
|
||||
c.pushCancel()
|
||||
c.wg.Wait()
|
||||
c.push = nil
|
||||
c.push.Store(nil)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user