mirror of
https://github.com/netbirdio/netbird.git
synced 2026-07-17 20:19:55 +00:00
[management] Move agent-network access-log ingest into the agentnetwork module (#6568)
The agent-network access-log ingest path (metaKey wire contract, flatten, usage derivation, and the dual-write of the usage ledger + settings-gated full row) lived in the reverseproxy accesslogs manager, even though the agentnetwork module already owns the rest of that domain — types, read (ListAccessLogs / GetUsageOverview), the budget-counter writes, and retention cleanup. Move it next to the rest: a stateless agentnetwork.IngestAccessLog(ctx, store, entry) that the reverseproxy SaveAccessLog delegates to when the entry is agent-network. Removes the agentNetworkTypes import from the reverseproxy manager. No behavior change; the write/read table separation is unchanged. Adds real-store coverage for the disable->enable log-collection toggle (usage ledger always written, full row gated) plus the metadata parse and group-dedup helpers, which previously had no dedicated tests.
This commit is contained in:
Maycon Santos
215
management/internals/modules/agentnetwork/accesslog_ingest.go
Normal file
215
management/internals/modules/agentnetwork/accesslog_ingest.go
Normal file
@@ -0,0 +1,215 @@
|
||||
package agentnetwork
|
||||
|
||||
import (
|
||||
"context"
|
||||
"math"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
|
||||
"github.com/netbirdio/netbird/management/internals/modules/agentnetwork/types"
|
||||
"github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs"
|
||||
"github.com/netbirdio/netbird/management/server/store"
|
||||
)
|
||||
|
||||
// Metadata keys the proxy stamps on agent-network access-log entries. These
|
||||
// mirror the constants in proxy/internal/middleware/keys.go and form the wire
|
||||
// contract between the proxy and management; management flattens them into
|
||||
// queryable columns. Keep in sync with the proxy side.
|
||||
const (
|
||||
metaKeyProvider = "llm.provider"
|
||||
metaKeyModel = "llm.model"
|
||||
metaKeyResolvedProviderID = "llm.resolved_provider_id"
|
||||
metaKeySelectedPolicyID = "llm.selected_policy_id"
|
||||
metaKeyPolicyDecision = "llm_policy.decision"
|
||||
metaKeyPolicyReason = "llm_policy.reason"
|
||||
metaKeyInputTokens = "llm.input_tokens" //nolint:gosec // metadata key name, not a credential
|
||||
metaKeyOutputTokens = "llm.output_tokens" //nolint:gosec // metadata key name, not a credential
|
||||
metaKeyTotalTokens = "llm.total_tokens" //nolint:gosec // metadata key name, not a credential
|
||||
metaKeyCostUSDTotal = "cost.usd_total"
|
||||
metaKeyStream = "llm.stream"
|
||||
metaKeySessionID = "llm.session_id"
|
||||
metaKeyAuthorisingGroups = "llm.authorising_groups"
|
||||
metaKeyRequestPrompt = "llm.request_prompt"
|
||||
metaKeyResponseCompletion = "llm.response_completion"
|
||||
)
|
||||
|
||||
// IngestAccessLog flattens the metadata-bearing reverse-proxy access-log entry
|
||||
// and persists it in the dedicated agent-network tables (instead of the shared
|
||||
// reverse-proxy table), in two parts:
|
||||
//
|
||||
// - The stripped usage record is written unconditionally — usage/cost is
|
||||
// collected on every request regardless of the account's log-collection
|
||||
// toggle (the proxy ships a usage-only entry when logging is disabled).
|
||||
// - The full access-log row (with request detail + prompt) is written only
|
||||
// when the account's EnableLogCollection setting is on. This setting read
|
||||
// is the authoritative gate; the proxy-side strip is defense in depth.
|
||||
func IngestAccessLog(ctx context.Context, s store.Store, logEntry *accesslogs.AccessLogEntry) error {
|
||||
entry, groups := flattenAccessLog(logEntry)
|
||||
|
||||
usage, usageGroups := usageFromFlattenedLog(entry, groups)
|
||||
if err := s.CreateAgentNetworkUsage(ctx, usage, usageGroups); err != nil {
|
||||
log.WithContext(ctx).WithFields(log.Fields{
|
||||
"account_id": entry.AccountID,
|
||||
"model": entry.Model,
|
||||
}).Errorf("failed to save agent-network usage: %v", err)
|
||||
return err
|
||||
}
|
||||
|
||||
settings, err := s.GetAgentNetworkSettings(ctx, store.LockingStrengthNone, entry.AccountID)
|
||||
if err != nil {
|
||||
// No settings row (or a transient read error) means we can't confirm
|
||||
// log collection is enabled — usage is already saved, so skip the full
|
||||
// row rather than fail the whole ingest.
|
||||
log.WithContext(ctx).Debugf("skipping full agent-network access-log row for account %s: %v", entry.AccountID, err)
|
||||
return nil
|
||||
}
|
||||
if !settings.EnableLogCollection {
|
||||
return nil
|
||||
}
|
||||
|
||||
if err := s.CreateAgentNetworkAccessLog(ctx, entry, groups); err != nil {
|
||||
log.WithContext(ctx).WithFields(log.Fields{
|
||||
"account_id": entry.AccountID,
|
||||
"service_id": entry.ServiceID,
|
||||
"model": entry.Model,
|
||||
"status": entry.StatusCode,
|
||||
}).Errorf("failed to save agent-network access log: %v", err)
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// flattenAccessLog converts a reverse-proxy AccessLogEntry (whose LLM
|
||||
// dimensions live in the opaque Metadata map) into the flattened
|
||||
// agent-network row + authorising-group child rows.
|
||||
func flattenAccessLog(e *accesslogs.AccessLogEntry) (*types.AgentNetworkAccessLog, []types.AgentNetworkAccessLogGroup) {
|
||||
meta := e.Metadata
|
||||
|
||||
var sourceIP string
|
||||
if e.GeoLocation.ConnectionIP != nil {
|
||||
sourceIP = e.GeoLocation.ConnectionIP.String()
|
||||
}
|
||||
|
||||
entry := &types.AgentNetworkAccessLog{
|
||||
ID: e.ID,
|
||||
AccountID: e.AccountID,
|
||||
ServiceID: e.ServiceID,
|
||||
Timestamp: e.Timestamp,
|
||||
UserID: e.UserId,
|
||||
SourceIP: sourceIP,
|
||||
Method: e.Method,
|
||||
Host: e.Host,
|
||||
Path: e.Path,
|
||||
Duration: e.Duration,
|
||||
StatusCode: e.StatusCode,
|
||||
AuthMethod: e.AuthMethodUsed,
|
||||
BytesUpload: e.BytesUpload,
|
||||
BytesDownload: e.BytesDownload,
|
||||
|
||||
Provider: meta[metaKeyProvider],
|
||||
Model: meta[metaKeyModel],
|
||||
SessionID: meta[metaKeySessionID],
|
||||
ResolvedProviderID: meta[metaKeyResolvedProviderID],
|
||||
SelectedPolicyID: meta[metaKeySelectedPolicyID],
|
||||
Decision: meta[metaKeyPolicyDecision],
|
||||
DenyReason: meta[metaKeyPolicyReason],
|
||||
InputTokens: parseMetaInt(meta, metaKeyInputTokens),
|
||||
OutputTokens: parseMetaInt(meta, metaKeyOutputTokens),
|
||||
TotalTokens: parseMetaInt(meta, metaKeyTotalTokens),
|
||||
CostUSD: parseMetaFloat(meta, metaKeyCostUSDTotal),
|
||||
Stream: parseMetaBool(meta, metaKeyStream),
|
||||
RequestPrompt: meta[metaKeyRequestPrompt],
|
||||
ResponseCompletion: meta[metaKeyResponseCompletion],
|
||||
}
|
||||
|
||||
var groups []types.AgentNetworkAccessLogGroup
|
||||
for _, gid := range parseGroupCSV(meta[metaKeyAuthorisingGroups]) {
|
||||
groups = append(groups, types.AgentNetworkAccessLogGroup{
|
||||
LogID: entry.ID,
|
||||
GroupID: gid,
|
||||
AccountID: entry.AccountID,
|
||||
})
|
||||
}
|
||||
return entry, groups
|
||||
}
|
||||
|
||||
// usageFromFlattenedLog derives the stripped usage record (and its group child
|
||||
// rows) from an already-flattened access-log entry. The usage row shares the
|
||||
// log's ID so the two correlate.
|
||||
func usageFromFlattenedLog(e *types.AgentNetworkAccessLog, groups []types.AgentNetworkAccessLogGroup) (*types.AgentNetworkUsage, []types.AgentNetworkUsageGroup) {
|
||||
usage := &types.AgentNetworkUsage{
|
||||
ID: e.ID,
|
||||
AccountID: e.AccountID,
|
||||
Timestamp: e.Timestamp,
|
||||
UserID: e.UserID,
|
||||
ResolvedProviderID: e.ResolvedProviderID,
|
||||
Provider: e.Provider,
|
||||
Model: e.Model,
|
||||
SessionID: e.SessionID,
|
||||
InputTokens: e.InputTokens,
|
||||
OutputTokens: e.OutputTokens,
|
||||
TotalTokens: e.TotalTokens,
|
||||
CostUSD: e.CostUSD,
|
||||
}
|
||||
|
||||
usageGroups := make([]types.AgentNetworkUsageGroup, 0, len(groups))
|
||||
for _, g := range groups {
|
||||
usageGroups = append(usageGroups, types.AgentNetworkUsageGroup{
|
||||
UsageID: usage.ID,
|
||||
GroupID: g.GroupID,
|
||||
AccountID: g.AccountID,
|
||||
})
|
||||
}
|
||||
return usage, usageGroups
|
||||
}
|
||||
|
||||
// parseMetaInt parses a non-negative token count. Negative or unparseable
|
||||
// values are clamped to 0 so a malformed metric can't persist a negative
|
||||
// counter.
|
||||
func parseMetaInt(meta map[string]string, key string) int64 {
|
||||
if v, err := strconv.ParseInt(strings.TrimSpace(meta[key]), 10, 64); err == nil && v >= 0 {
|
||||
return v
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
// parseMetaFloat parses a non-negative, finite cost. Negative, NaN, Inf, or
|
||||
// unparseable values are clamped to 0 so a malformed metric can't poison the
|
||||
// stored cost.
|
||||
func parseMetaFloat(meta map[string]string, key string) float64 {
|
||||
if v, err := strconv.ParseFloat(strings.TrimSpace(meta[key]), 64); err == nil && v >= 0 && !math.IsInf(v, 0) {
|
||||
return v
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func parseMetaBool(meta map[string]string, key string) bool {
|
||||
v, _ := strconv.ParseBool(strings.TrimSpace(meta[key]))
|
||||
return v
|
||||
}
|
||||
|
||||
// parseGroupCSV splits the comma-separated authorising-group id list the proxy
|
||||
// emits, trimming blanks and de-duplicating. Dedup matters because the group
|
||||
// rows are keyed by (log_id, group_id) / (usage_id, group_id): a repeated id
|
||||
// in the CSV would otherwise produce a duplicate primary key and fail the
|
||||
// insert transaction.
|
||||
func parseGroupCSV(raw string) []string {
|
||||
if raw == "" {
|
||||
return nil
|
||||
}
|
||||
parts := strings.Split(raw, ",")
|
||||
out := make([]string, 0, len(parts))
|
||||
seen := make(map[string]struct{}, len(parts))
|
||||
for _, p := range parts {
|
||||
if p = strings.TrimSpace(p); p != "" {
|
||||
if _, dup := seen[p]; dup {
|
||||
continue
|
||||
}
|
||||
seen[p] = struct{}{}
|
||||
out = append(out, p)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,124 @@
|
||||
package agentnetwork
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/netbirdio/netbird/management/internals/modules/agentnetwork/types"
|
||||
"github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs"
|
||||
"github.com/netbirdio/netbird/management/server/store"
|
||||
)
|
||||
|
||||
// newIngestTestEntry builds an agent-network reverse-proxy access-log entry whose
|
||||
// LLM dimensions live in the opaque Metadata map, as the proxy ships it.
|
||||
func newIngestTestEntry() *accesslogs.AccessLogEntry {
|
||||
return &accesslogs.AccessLogEntry{
|
||||
ID: "log-1",
|
||||
AccountID: testAccountID,
|
||||
ServiceID: "svc-1",
|
||||
Timestamp: time.Now().UTC(),
|
||||
Method: "POST",
|
||||
Host: testEndpoint,
|
||||
Path: "/v1/chat/completions",
|
||||
StatusCode: 200,
|
||||
UserId: "user-1",
|
||||
AgentNetwork: true,
|
||||
Metadata: map[string]string{
|
||||
metaKeyProvider: "openai",
|
||||
metaKeyModel: "gpt-5.4",
|
||||
metaKeyResolvedProviderID: "prov-1",
|
||||
metaKeySessionID: "sess-1",
|
||||
metaKeyInputTokens: "100",
|
||||
metaKeyOutputTokens: "50",
|
||||
metaKeyTotalTokens: "150",
|
||||
metaKeyCostUSDTotal: "0.0123",
|
||||
metaKeyStream: "true",
|
||||
metaKeyRequestPrompt: "hello",
|
||||
metaKeyResponseCompletion: "world",
|
||||
// repeated id must be de-duplicated before the group rows insert.
|
||||
metaKeyAuthorisingGroups: "grp-eng,grp-eng,grp-ops",
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// TestIngestAccessLog_RealStore_LogCollectionOff persists the usage ledger
|
||||
// unconditionally but skips the full access-log row when the account hasn't
|
||||
// opted into log collection.
|
||||
func TestIngestAccessLog_RealStore_LogCollectionOff(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
s, cleanup, err := store.NewTestStoreFromSQL(ctx, "", t.TempDir())
|
||||
require.NoError(t, err, "real sqlite test store must come up")
|
||||
defer cleanup()
|
||||
|
||||
settings := newSynthTestSettings()
|
||||
settings.EnableLogCollection = false
|
||||
require.NoError(t, s.SaveAgentNetworkSettings(ctx, settings))
|
||||
|
||||
require.NoError(t, IngestAccessLog(ctx, s, newIngestTestEntry()))
|
||||
|
||||
usage, err := s.GetAgentNetworkUsageRows(ctx, store.LockingStrengthNone, testAccountID, types.AgentNetworkAccessLogFilter{})
|
||||
require.NoError(t, err)
|
||||
require.Len(t, usage, 1, "usage row must be written even with log collection off")
|
||||
assert.Equal(t, int64(100), usage[0].InputTokens, "input tokens must round-trip from metadata")
|
||||
assert.Equal(t, int64(50), usage[0].OutputTokens, "output tokens must round-trip from metadata")
|
||||
assert.InDelta(t, 0.0123, usage[0].CostUSD, 1e-9, "cost must round-trip from metadata")
|
||||
|
||||
logs, _, err := s.GetAgentNetworkAccessLogs(ctx, store.LockingStrengthNone, testAccountID, types.AgentNetworkAccessLogFilter{})
|
||||
require.NoError(t, err)
|
||||
assert.Empty(t, logs, "full access-log row must be skipped while log collection is off")
|
||||
}
|
||||
|
||||
// TestIngestAccessLog_RealStore_LogCollectionOn writes both the usage ledger and
|
||||
// the full access-log row once the account opts in, carrying the request detail
|
||||
// and prompt through.
|
||||
func TestIngestAccessLog_RealStore_LogCollectionOn(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
s, cleanup, err := store.NewTestStoreFromSQL(ctx, "", t.TempDir())
|
||||
require.NoError(t, err, "real sqlite test store must come up")
|
||||
defer cleanup()
|
||||
|
||||
settings := newSynthTestSettings()
|
||||
settings.EnableLogCollection = true
|
||||
require.NoError(t, s.SaveAgentNetworkSettings(ctx, settings))
|
||||
|
||||
require.NoError(t, IngestAccessLog(ctx, s, newIngestTestEntry()))
|
||||
|
||||
usage, err := s.GetAgentNetworkUsageRows(ctx, store.LockingStrengthNone, testAccountID, types.AgentNetworkAccessLogFilter{})
|
||||
require.NoError(t, err)
|
||||
require.Len(t, usage, 1, "usage row must be written when log collection is on")
|
||||
|
||||
logs, total, err := s.GetAgentNetworkAccessLogs(ctx, store.LockingStrengthNone, testAccountID, types.AgentNetworkAccessLogFilter{})
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, int64(1), total, "exactly one access-log row expected")
|
||||
require.Len(t, logs, 1, "full access-log row must be written when log collection is on")
|
||||
assert.Equal(t, "gpt-5.4", logs[0].Model, "model must flatten from metadata")
|
||||
assert.Equal(t, "hello", logs[0].RequestPrompt, "prompt must be retained when log collection is on")
|
||||
assert.Equal(t, "world", logs[0].ResponseCompletion, "completion must be retained when log collection is on")
|
||||
assert.True(t, logs[0].Stream, "stream flag must flatten from metadata")
|
||||
}
|
||||
|
||||
func TestParseGroupCSV_DedupAndTrim(t *testing.T) {
|
||||
assert.Nil(t, parseGroupCSV(""), "empty CSV yields no groups")
|
||||
assert.Equal(t, []string{"a", "b"}, parseGroupCSV(" a , b , a ,"),
|
||||
"group CSV must trim, drop blanks, and de-duplicate preserving first-seen order")
|
||||
}
|
||||
|
||||
func TestParseMetaInt_ClampsNegativeAndJunk(t *testing.T) {
|
||||
meta := map[string]string{"ok": " 42 ", "neg": "-5", "junk": "abc"}
|
||||
assert.Equal(t, int64(42), parseMetaInt(meta, "ok"), "valid count parses with surrounding space trimmed")
|
||||
assert.Equal(t, int64(0), parseMetaInt(meta, "neg"), "negative count clamps to 0")
|
||||
assert.Equal(t, int64(0), parseMetaInt(meta, "junk"), "unparseable count clamps to 0")
|
||||
assert.Equal(t, int64(0), parseMetaInt(meta, "missing"), "missing key clamps to 0")
|
||||
}
|
||||
|
||||
func TestParseMetaFloat_ClampsNegativeInfAndJunk(t *testing.T) {
|
||||
meta := map[string]string{"ok": "1.5", "neg": "-1", "inf": "Inf", "junk": "x"}
|
||||
assert.InDelta(t, 1.5, parseMetaFloat(meta, "ok"), 1e-9, "valid cost parses")
|
||||
assert.Equal(t, float64(0), parseMetaFloat(meta, "neg"), "negative cost clamps to 0")
|
||||
assert.Equal(t, float64(0), parseMetaFloat(meta, "inf"), "non-finite cost clamps to 0")
|
||||
assert.Equal(t, float64(0), parseMetaFloat(meta, "junk"), "unparseable cost clamps to 0")
|
||||
}
|
||||
@@ -2,15 +2,13 @@ package manager
|
||||
|
||||
import (
|
||||
"context"
|
||||
"math"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
|
||||
"github.com/netbirdio/netbird/management/internals/modules/agentnetwork"
|
||||
"github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs"
|
||||
agentNetworkTypes "github.com/netbirdio/netbird/management/internals/modules/agentnetwork/types"
|
||||
"github.com/netbirdio/netbird/management/server/geolocation"
|
||||
"github.com/netbirdio/netbird/management/server/permissions"
|
||||
"github.com/netbirdio/netbird/management/server/permissions/modules"
|
||||
@@ -19,28 +17,6 @@ import (
|
||||
"github.com/netbirdio/netbird/shared/management/status"
|
||||
)
|
||||
|
||||
// Metadata keys the proxy stamps on agent-network access-log entries. These
|
||||
// mirror the constants in proxy/internal/middleware/keys.go and form the wire
|
||||
// contract between the proxy and management; management flattens them into
|
||||
// queryable columns. Keep in sync with the proxy side.
|
||||
const (
|
||||
metaKeyProvider = "llm.provider"
|
||||
metaKeyModel = "llm.model"
|
||||
metaKeyResolvedProviderID = "llm.resolved_provider_id"
|
||||
metaKeySelectedPolicyID = "llm.selected_policy_id"
|
||||
metaKeyPolicyDecision = "llm_policy.decision"
|
||||
metaKeyPolicyReason = "llm_policy.reason"
|
||||
metaKeyInputTokens = "llm.input_tokens" //nolint:gosec // metadata key name, not a credential
|
||||
metaKeyOutputTokens = "llm.output_tokens" //nolint:gosec // metadata key name, not a credential
|
||||
metaKeyTotalTokens = "llm.total_tokens" //nolint:gosec // metadata key name, not a credential
|
||||
metaKeyCostUSDTotal = "cost.usd_total"
|
||||
metaKeyStream = "llm.stream"
|
||||
metaKeySessionID = "llm.session_id"
|
||||
metaKeyAuthorisingGroups = "llm.authorising_groups"
|
||||
metaKeyRequestPrompt = "llm.request_prompt"
|
||||
metaKeyResponseCompletion = "llm.response_completion"
|
||||
)
|
||||
|
||||
type managerImpl struct {
|
||||
store store.Store
|
||||
permissionsManager permissions.Manager
|
||||
@@ -61,7 +37,7 @@ func NewManager(store store.Store, permissionsManager permissions.Manager, geo g
|
||||
// LLM columns + group child rows) instead of the shared reverse-proxy table.
|
||||
func (m *managerImpl) SaveAccessLog(ctx context.Context, logEntry *accesslogs.AccessLogEntry) error {
|
||||
if logEntry.AgentNetwork {
|
||||
return m.saveAgentNetworkAccessLog(ctx, logEntry)
|
||||
return agentnetwork.IngestAccessLog(ctx, m.store, logEntry)
|
||||
}
|
||||
|
||||
if m.geo != nil && logEntry.GeoLocation.ConnectionIP != nil {
|
||||
@@ -92,184 +68,6 @@ func (m *managerImpl) SaveAccessLog(ctx context.Context, logEntry *accesslogs.Ac
|
||||
return nil
|
||||
}
|
||||
|
||||
// saveAgentNetworkAccessLog flattens the metadata-bearing access-log entry and
|
||||
// persists it in two parts:
|
||||
//
|
||||
// - The stripped usage record is written unconditionally — usage/cost is
|
||||
// collected on every request regardless of the account's log-collection
|
||||
// toggle (the proxy ships a usage-only entry when logging is disabled).
|
||||
// - The full access-log row (with request detail + prompt) is written only
|
||||
// when the account's EnableLogCollection setting is on. This setting read
|
||||
// is the authoritative gate; the proxy-side strip is defense in depth.
|
||||
func (m *managerImpl) saveAgentNetworkAccessLog(ctx context.Context, logEntry *accesslogs.AccessLogEntry) error {
|
||||
entry, groups := flattenAgentNetworkLog(logEntry)
|
||||
|
||||
usage, usageGroups := usageFromFlattenedLog(entry, groups)
|
||||
if err := m.store.CreateAgentNetworkUsage(ctx, usage, usageGroups); err != nil {
|
||||
log.WithContext(ctx).WithFields(log.Fields{
|
||||
"account_id": entry.AccountID,
|
||||
"model": entry.Model,
|
||||
}).Errorf("failed to save agent-network usage: %v", err)
|
||||
return err
|
||||
}
|
||||
|
||||
settings, err := m.store.GetAgentNetworkSettings(ctx, store.LockingStrengthNone, entry.AccountID)
|
||||
if err != nil {
|
||||
// No settings row (or a transient read error) means we can't confirm
|
||||
// log collection is enabled — usage is already saved, so skip the full
|
||||
// row rather than fail the whole ingest.
|
||||
log.WithContext(ctx).Debugf("skipping full agent-network access-log row for account %s: %v", entry.AccountID, err)
|
||||
return nil
|
||||
}
|
||||
if !settings.EnableLogCollection {
|
||||
return nil
|
||||
}
|
||||
|
||||
if err := m.store.CreateAgentNetworkAccessLog(ctx, entry, groups); err != nil {
|
||||
log.WithContext(ctx).WithFields(log.Fields{
|
||||
"account_id": entry.AccountID,
|
||||
"service_id": entry.ServiceID,
|
||||
"model": entry.Model,
|
||||
"status": entry.StatusCode,
|
||||
}).Errorf("failed to save agent-network access log: %v", err)
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// flattenAgentNetworkLog converts a reverse-proxy AccessLogEntry (whose LLM
|
||||
// dimensions live in the opaque Metadata map) into the flattened
|
||||
// agent-network row + authorising-group child rows.
|
||||
func flattenAgentNetworkLog(e *accesslogs.AccessLogEntry) (*agentNetworkTypes.AgentNetworkAccessLog, []agentNetworkTypes.AgentNetworkAccessLogGroup) {
|
||||
meta := e.Metadata
|
||||
|
||||
var sourceIP string
|
||||
if e.GeoLocation.ConnectionIP != nil {
|
||||
sourceIP = e.GeoLocation.ConnectionIP.String()
|
||||
}
|
||||
|
||||
entry := &agentNetworkTypes.AgentNetworkAccessLog{
|
||||
ID: e.ID,
|
||||
AccountID: e.AccountID,
|
||||
ServiceID: e.ServiceID,
|
||||
Timestamp: e.Timestamp,
|
||||
UserID: e.UserId,
|
||||
SourceIP: sourceIP,
|
||||
Method: e.Method,
|
||||
Host: e.Host,
|
||||
Path: e.Path,
|
||||
Duration: e.Duration,
|
||||
StatusCode: e.StatusCode,
|
||||
AuthMethod: e.AuthMethodUsed,
|
||||
BytesUpload: e.BytesUpload,
|
||||
BytesDownload: e.BytesDownload,
|
||||
|
||||
Provider: meta[metaKeyProvider],
|
||||
Model: meta[metaKeyModel],
|
||||
SessionID: meta[metaKeySessionID],
|
||||
ResolvedProviderID: meta[metaKeyResolvedProviderID],
|
||||
SelectedPolicyID: meta[metaKeySelectedPolicyID],
|
||||
Decision: meta[metaKeyPolicyDecision],
|
||||
DenyReason: meta[metaKeyPolicyReason],
|
||||
InputTokens: parseMetaInt(meta, metaKeyInputTokens),
|
||||
OutputTokens: parseMetaInt(meta, metaKeyOutputTokens),
|
||||
TotalTokens: parseMetaInt(meta, metaKeyTotalTokens),
|
||||
CostUSD: parseMetaFloat(meta, metaKeyCostUSDTotal),
|
||||
Stream: parseMetaBool(meta, metaKeyStream),
|
||||
RequestPrompt: meta[metaKeyRequestPrompt],
|
||||
ResponseCompletion: meta[metaKeyResponseCompletion],
|
||||
}
|
||||
|
||||
var groups []agentNetworkTypes.AgentNetworkAccessLogGroup
|
||||
for _, gid := range parseGroupCSV(meta[metaKeyAuthorisingGroups]) {
|
||||
groups = append(groups, agentNetworkTypes.AgentNetworkAccessLogGroup{
|
||||
LogID: entry.ID,
|
||||
GroupID: gid,
|
||||
AccountID: entry.AccountID,
|
||||
})
|
||||
}
|
||||
return entry, groups
|
||||
}
|
||||
|
||||
// usageFromFlattenedLog derives the stripped usage record (and its group child
|
||||
// rows) from an already-flattened access-log entry. The usage row shares the
|
||||
// log's ID so the two correlate.
|
||||
func usageFromFlattenedLog(e *agentNetworkTypes.AgentNetworkAccessLog, groups []agentNetworkTypes.AgentNetworkAccessLogGroup) (*agentNetworkTypes.AgentNetworkUsage, []agentNetworkTypes.AgentNetworkUsageGroup) {
|
||||
usage := &agentNetworkTypes.AgentNetworkUsage{
|
||||
ID: e.ID,
|
||||
AccountID: e.AccountID,
|
||||
Timestamp: e.Timestamp,
|
||||
UserID: e.UserID,
|
||||
ResolvedProviderID: e.ResolvedProviderID,
|
||||
Provider: e.Provider,
|
||||
Model: e.Model,
|
||||
SessionID: e.SessionID,
|
||||
InputTokens: e.InputTokens,
|
||||
OutputTokens: e.OutputTokens,
|
||||
TotalTokens: e.TotalTokens,
|
||||
CostUSD: e.CostUSD,
|
||||
}
|
||||
|
||||
usageGroups := make([]agentNetworkTypes.AgentNetworkUsageGroup, 0, len(groups))
|
||||
for _, g := range groups {
|
||||
usageGroups = append(usageGroups, agentNetworkTypes.AgentNetworkUsageGroup{
|
||||
UsageID: usage.ID,
|
||||
GroupID: g.GroupID,
|
||||
AccountID: g.AccountID,
|
||||
})
|
||||
}
|
||||
return usage, usageGroups
|
||||
}
|
||||
|
||||
// parseMetaInt parses a non-negative token count. Negative or unparseable
|
||||
// values are clamped to 0 so a malformed metric can't persist a negative
|
||||
// counter.
|
||||
func parseMetaInt(meta map[string]string, key string) int64 {
|
||||
if v, err := strconv.ParseInt(strings.TrimSpace(meta[key]), 10, 64); err == nil && v >= 0 {
|
||||
return v
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
// parseMetaFloat parses a non-negative, finite cost. Negative, NaN, Inf, or
|
||||
// unparseable values are clamped to 0 so a malformed metric can't poison the
|
||||
// stored cost.
|
||||
func parseMetaFloat(meta map[string]string, key string) float64 {
|
||||
if v, err := strconv.ParseFloat(strings.TrimSpace(meta[key]), 64); err == nil && v >= 0 && !math.IsInf(v, 0) {
|
||||
return v
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func parseMetaBool(meta map[string]string, key string) bool {
|
||||
v, _ := strconv.ParseBool(strings.TrimSpace(meta[key]))
|
||||
return v
|
||||
}
|
||||
|
||||
// parseGroupCSV splits the comma-separated authorising-group id list the proxy
|
||||
// emits, trimming blanks and de-duplicating. Dedup matters because the group
|
||||
// rows are keyed by (log_id, group_id) / (usage_id, group_id): a repeated id
|
||||
// in the CSV would otherwise produce a duplicate primary key and fail the
|
||||
// insert transaction.
|
||||
func parseGroupCSV(raw string) []string {
|
||||
if raw == "" {
|
||||
return nil
|
||||
}
|
||||
parts := strings.Split(raw, ",")
|
||||
out := make([]string, 0, len(parts))
|
||||
seen := make(map[string]struct{}, len(parts))
|
||||
for _, p := range parts {
|
||||
if p = strings.TrimSpace(p); p != "" {
|
||||
if _, dup := seen[p]; dup {
|
||||
continue
|
||||
}
|
||||
seen[p] = struct{}{}
|
||||
out = append(out, p)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// GetAllAccessLogs retrieves access logs for an account with pagination and filtering
|
||||
func (m *managerImpl) GetAllAccessLogs(ctx context.Context, accountID, userID string, filter *accesslogs.AccessLogFilter) ([]*accesslogs.AccessLogEntry, int64, error) {
|
||||
ok, ctx, err := m.permissionsManager.ValidateUserPermissions(ctx, accountID, userID, modules.Services, operations.Read)
|
||||
|
||||
Reference in New Issue
Block a user