mirror of
https://github.com/netbirdio/netbird.git
synced 2026-10-27 23:09:08 +01:00
[management] Add agent-network telemetry metrics (#6561)
Surface agent-network adoption and usage in the self-hosted metrics worker: distinct accounts, providers, policies, budget rules, accounts with log collection enabled, and aggregated input/output tokens plus cost. Tokens and cost are summed from agent_network_request_usage (the always-written per-request ledger) so the figures are accurate regardless of the log-collection toggle and carry no double-counting. All values come from a handful of indexed aggregate queries run only on the worker's periodic tick. Adds store.AgentNetworkMetrics with GetAgentNetworkMetrics on the Store interface, the SqlStore implementation, and a zero-valued FileStore stub.
This commit is contained in:
@@ -55,6 +55,7 @@ type DataSource interface {
|
|||||||
GetStoreEngine() types.Engine
|
GetStoreEngine() types.Engine
|
||||||
GetCustomDomainsCounts(ctx context.Context) (total int64, validated int64, err error)
|
GetCustomDomainsCounts(ctx context.Context) (total int64, validated int64, err error)
|
||||||
GetProxyMetrics(ctx context.Context) (store.ProxyMetrics, error)
|
GetProxyMetrics(ctx context.Context) (store.ProxyMetrics, error)
|
||||||
|
GetAgentNetworkMetrics(ctx context.Context) (store.AgentNetworkMetrics, error)
|
||||||
}
|
}
|
||||||
|
|
||||||
// ConnManager peer connection manager that holds state for current active connections
|
// ConnManager peer connection manager that holds state for current active connections
|
||||||
@@ -413,6 +414,13 @@ func (w *Worker) generateProperties(ctx context.Context) properties {
|
|||||||
log.WithContext(ctx).Debugf("collect proxy metrics: %v", err)
|
log.WithContext(ctx).Debugf("collect proxy metrics: %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Agent-network adoption + usage, aggregated across all accounts in a few
|
||||||
|
// cheap queries; nil on FileStore.
|
||||||
|
agentNetworkMetrics, err := w.dataSource.GetAgentNetworkMetrics(ctx)
|
||||||
|
if err != nil {
|
||||||
|
log.WithContext(ctx).Debugf("collect agent network metrics: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
minActivePeerVersion, maxActivePeerVersion := getMinMaxVersion(peerActiveVersions)
|
minActivePeerVersion, maxActivePeerVersion := getMinMaxVersion(peerActiveVersions)
|
||||||
metricsProperties["uptime"] = uptime
|
metricsProperties["uptime"] = uptime
|
||||||
metricsProperties["accounts"] = accounts
|
metricsProperties["accounts"] = accounts
|
||||||
@@ -471,6 +479,14 @@ func (w *Worker) generateProperties(ctx context.Context) properties {
|
|||||||
metricsProperties["proxies_connected"] = proxyMetrics.ProxiesConnected
|
metricsProperties["proxies_connected"] = proxyMetrics.ProxiesConnected
|
||||||
metricsProperties["custom_domains"] = customDomains
|
metricsProperties["custom_domains"] = customDomains
|
||||||
metricsProperties["custom_domains_validated"] = customDomainsValidated
|
metricsProperties["custom_domains_validated"] = customDomainsValidated
|
||||||
|
metricsProperties["agent_network_accounts"] = agentNetworkMetrics.Accounts
|
||||||
|
metricsProperties["agent_network_providers"] = agentNetworkMetrics.Providers
|
||||||
|
metricsProperties["agent_network_policies"] = agentNetworkMetrics.Policies
|
||||||
|
metricsProperties["agent_network_budget_rules"] = agentNetworkMetrics.BudgetRules
|
||||||
|
metricsProperties["agent_network_log_collection_enabled"] = agentNetworkMetrics.LogCollectionEnabled
|
||||||
|
metricsProperties["agent_network_input_tokens"] = agentNetworkMetrics.InputTokens
|
||||||
|
metricsProperties["agent_network_output_tokens"] = agentNetworkMetrics.OutputTokens
|
||||||
|
metricsProperties["agent_network_cost_usd"] = agentNetworkMetrics.CostUSD
|
||||||
|
|
||||||
for targetType, count := range servicesTargetType {
|
for targetType, count := range servicesTargetType {
|
||||||
metricsProperties["services_target_type_"+string(targetType)] = count
|
metricsProperties["services_target_type_"+string(targetType)] = count
|
||||||
|
|||||||
@@ -277,6 +277,21 @@ func (mockDatasource) GetProxyMetrics(_ context.Context) (store.ProxyMetrics, er
|
|||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// GetAgentNetworkMetrics returns canned agent-network counts so the
|
||||||
|
// generateProperties test can assert the adoption/usage signals end-to-end.
|
||||||
|
func (mockDatasource) GetAgentNetworkMetrics(_ context.Context) (store.AgentNetworkMetrics, error) {
|
||||||
|
return store.AgentNetworkMetrics{
|
||||||
|
Accounts: 2,
|
||||||
|
Providers: 5,
|
||||||
|
Policies: 3,
|
||||||
|
BudgetRules: 1,
|
||||||
|
LogCollectionEnabled: 2,
|
||||||
|
InputTokens: 1000,
|
||||||
|
OutputTokens: 500,
|
||||||
|
CostUSD: 1.25,
|
||||||
|
}, nil
|
||||||
|
}
|
||||||
|
|
||||||
// TestGenerateProperties tests and validate the properties generation by using the mockDatasource for the Worker.generateProperties
|
// TestGenerateProperties tests and validate the properties generation by using the mockDatasource for the Worker.generateProperties
|
||||||
func TestGenerateProperties(t *testing.T) {
|
func TestGenerateProperties(t *testing.T) {
|
||||||
ds := mockDatasource{}
|
ds := mockDatasource{}
|
||||||
|
|||||||
@@ -280,3 +280,9 @@ func (s *FileStore) GetCustomDomainsCounts(_ context.Context) (int64, int64, err
|
|||||||
func (s *FileStore) GetProxyMetrics(_ context.Context) (ProxyMetrics, error) {
|
func (s *FileStore) GetProxyMetrics(_ context.Context) (ProxyMetrics, error) {
|
||||||
return ProxyMetrics{}, nil
|
return ProxyMetrics{}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// GetAgentNetworkMetrics is a no-op for FileStore — agent-network state isn't
|
||||||
|
// persisted in the JSON file format.
|
||||||
|
func (s *FileStore) GetAgentNetworkMetrics(_ context.Context) (AgentNetworkMetrics, error) {
|
||||||
|
return AgentNetworkMetrics{}, nil
|
||||||
|
}
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ package store
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
|
"fmt"
|
||||||
"math"
|
"math"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -38,6 +39,46 @@ func (s *SqlStore) GetAllAgentNetworkProviders(ctx context.Context, lockStrength
|
|||||||
return providers, nil
|
return providers, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// GetAgentNetworkMetrics returns aggregated agent-network adoption + usage
|
||||||
|
// counts for the self-hosted metrics worker. Each value is a single cheap
|
||||||
|
// aggregate; token/cost are summed over the always-collected per-request usage
|
||||||
|
// ledger (independent of the log-collection toggle) so they reflect real usage.
|
||||||
|
func (s *SqlStore) GetAgentNetworkMetrics(ctx context.Context) (AgentNetworkMetrics, error) {
|
||||||
|
var m AgentNetworkMetrics
|
||||||
|
db := s.db.WithContext(ctx)
|
||||||
|
|
||||||
|
// Providers + distinct adopting accounts in one round-trip.
|
||||||
|
provRow := db.Model(&agentNetworkTypes.Provider{}).
|
||||||
|
Select("COUNT(*) AS providers, COUNT(DISTINCT account_id) AS accounts").Row()
|
||||||
|
if err := provRow.Scan(&m.Providers, &m.Accounts); err != nil {
|
||||||
|
return AgentNetworkMetrics{}, fmt.Errorf("scan agent network provider metrics: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := db.Model(&agentNetworkTypes.Policy{}).Count(&m.Policies).Error; err != nil {
|
||||||
|
return AgentNetworkMetrics{}, fmt.Errorf("count agent network policies: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := db.Model(&agentNetworkTypes.AccountBudgetRule{}).Count(&m.BudgetRules).Error; err != nil {
|
||||||
|
return AgentNetworkMetrics{}, fmt.Errorf("count agent network budget rules: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := db.Model(&agentNetworkTypes.Settings{}).
|
||||||
|
Where("enable_log_collection = ?", true).Count(&m.LogCollectionEnabled).Error; err != nil {
|
||||||
|
return AgentNetworkMetrics{}, fmt.Errorf("count agent network log-collection accounts: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
// COALESCE so an empty ledger scans as 0 instead of NULL.
|
||||||
|
usageRow := db.Model(&agentNetworkTypes.AgentNetworkUsage{}).
|
||||||
|
Select("COALESCE(SUM(input_tokens), 0) AS input_tokens, " +
|
||||||
|
"COALESCE(SUM(output_tokens), 0) AS output_tokens, " +
|
||||||
|
"COALESCE(SUM(cost_usd), 0) AS cost_usd").Row()
|
||||||
|
if err := usageRow.Scan(&m.InputTokens, &m.OutputTokens, &m.CostUSD); err != nil {
|
||||||
|
return AgentNetworkMetrics{}, fmt.Errorf("scan agent network usage metrics: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return m, nil
|
||||||
|
}
|
||||||
|
|
||||||
func (s *SqlStore) GetAccountAgentNetworkProviders(ctx context.Context, lockStrength LockingStrength, accountID string) ([]*agentNetworkTypes.Provider, error) {
|
func (s *SqlStore) GetAccountAgentNetworkProviders(ctx context.Context, lockStrength LockingStrength, accountID string) ([]*agentNetworkTypes.Provider, error) {
|
||||||
tx := s.db
|
tx := s.db
|
||||||
if lockStrength != LockingStrengthNone {
|
if lockStrength != LockingStrengthNone {
|
||||||
|
|||||||
@@ -334,6 +334,11 @@ type Store interface {
|
|||||||
// return a zero-valued struct.
|
// return a zero-valued struct.
|
||||||
GetProxyMetrics(ctx context.Context) (ProxyMetrics, error)
|
GetProxyMetrics(ctx context.Context) (ProxyMetrics, error)
|
||||||
|
|
||||||
|
// GetAgentNetworkMetrics returns aggregated agent-network adoption + usage
|
||||||
|
// counts for the self-hosted metrics worker. Self-hosted only — file-based
|
||||||
|
// stores return a zero-valued struct.
|
||||||
|
GetAgentNetworkMetrics(ctx context.Context) (AgentNetworkMetrics, error)
|
||||||
|
|
||||||
GetRoutingPeerNetworks(ctx context.Context, accountID, peerID string) ([]string, error)
|
GetRoutingPeerNetworks(ctx context.Context, accountID, peerID string) ([]string, error)
|
||||||
|
|
||||||
// Agent Network persistence (providers, policies, guardrails, settings).
|
// Agent Network persistence (providers, policies, guardrails, settings).
|
||||||
@@ -389,6 +394,32 @@ type ProxyMetrics struct {
|
|||||||
ProxiesConnected int64
|
ProxiesConnected int64
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// AgentNetworkMetrics aggregates self-hosted agent-network adoption + usage
|
||||||
|
// signals surfaced to the telemetry payload. Each field is best-effort: when a
|
||||||
|
// store cannot answer (e.g. FileStore) all fields are zero.
|
||||||
|
type AgentNetworkMetrics struct {
|
||||||
|
// Accounts is the number of distinct accounts with at least one provider
|
||||||
|
// configured (agent-network adoption).
|
||||||
|
Accounts int64
|
||||||
|
// Providers is the total number of configured providers across all accounts.
|
||||||
|
Providers int64
|
||||||
|
// Policies is the total number of agent-network policies across all accounts.
|
||||||
|
Policies int64
|
||||||
|
// BudgetRules is the total number of account-level budget rules ("budget
|
||||||
|
// limits") across all accounts.
|
||||||
|
BudgetRules int64
|
||||||
|
// LogCollectionEnabled is the number of accounts that have agent-network
|
||||||
|
// log collection turned on.
|
||||||
|
LogCollectionEnabled int64
|
||||||
|
// InputTokens / OutputTokens / CostUSD are summed over the always-collected
|
||||||
|
// per-request usage ledger (agent_network_request_usage), independent of the
|
||||||
|
// log-collection toggle. They reflect total metered LLM usage served through
|
||||||
|
// agent networks.
|
||||||
|
InputTokens int64
|
||||||
|
OutputTokens int64
|
||||||
|
CostUSD float64
|
||||||
|
}
|
||||||
|
|
||||||
const (
|
const (
|
||||||
postgresDsnEnv = "NB_STORE_ENGINE_POSTGRES_DSN"
|
postgresDsnEnv = "NB_STORE_ENGINE_POSTGRES_DSN"
|
||||||
postgresDsnEnvLegacy = "NETBIRD_STORE_ENGINE_POSTGRES_DSN"
|
postgresDsnEnvLegacy = "NETBIRD_STORE_ENGINE_POSTGRES_DSN"
|
||||||
|
|||||||
@@ -25,6 +25,21 @@ func (mr *MockStoreMockRecorder) GetAllAgentNetworkProviders(ctx, lockStrength i
|
|||||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetAllAgentNetworkProviders", reflect.TypeOf((*MockStore)(nil).GetAllAgentNetworkProviders), ctx, lockStrength)
|
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetAllAgentNetworkProviders", reflect.TypeOf((*MockStore)(nil).GetAllAgentNetworkProviders), ctx, lockStrength)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// GetAgentNetworkMetrics mocks base method.
|
||||||
|
func (m *MockStore) GetAgentNetworkMetrics(ctx context.Context) (AgentNetworkMetrics, error) {
|
||||||
|
m.ctrl.T.Helper()
|
||||||
|
ret := m.ctrl.Call(m, "GetAgentNetworkMetrics", ctx)
|
||||||
|
ret0, _ := ret[0].(AgentNetworkMetrics)
|
||||||
|
ret1, _ := ret[1].(error)
|
||||||
|
return ret0, ret1
|
||||||
|
}
|
||||||
|
|
||||||
|
// GetAgentNetworkMetrics indicates an expected call of GetAgentNetworkMetrics.
|
||||||
|
func (mr *MockStoreMockRecorder) GetAgentNetworkMetrics(ctx interface{}) *gomock.Call {
|
||||||
|
mr.mock.ctrl.T.Helper()
|
||||||
|
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetAgentNetworkMetrics", reflect.TypeOf((*MockStore)(nil).GetAgentNetworkMetrics), ctx)
|
||||||
|
}
|
||||||
|
|
||||||
// GetAccountAgentNetworkProviders mocks base method.
|
// GetAccountAgentNetworkProviders mocks base method.
|
||||||
func (m *MockStore) GetAccountAgentNetworkProviders(ctx context.Context, lockStrength LockingStrength, accountID string) ([]*agentNetworkTypes.Provider, error) {
|
func (m *MockStore) GetAccountAgentNetworkProviders(ctx context.Context, lockStrength LockingStrength, accountID string) ([]*agentNetworkTypes.Provider, error) {
|
||||||
m.ctrl.T.Helper()
|
m.ctrl.T.Helper()
|
||||||
|
|||||||
Reference in New Issue
Block a user