Persist and expose aggregate usage metrics
This commit is contained in:
Ollie Agent
parent
a945445c1f
commit
741ac689f8
|
|
@ -9,6 +9,7 @@ import (
|
|||
"strings"
|
||||
|
||||
"ollie/cmd/olliesrv/internal/backend"
|
||||
"ollie/cmd/olliesrv/internal/metrics"
|
||||
"ollie/toolsrv"
|
||||
)
|
||||
|
||||
|
|
@ -173,6 +174,9 @@ func (ag *Agent) executeTurn(ctx context.Context, input string) string {
|
|||
CostUSD: costUSD,
|
||||
}
|
||||
ag.history.addUsage(usage, est != 0)
|
||||
if err := metrics.RecordUsage(ag.sessionID, ag.id, ag.BackendName(), ag.ModelName(), usage, est != 0); err != nil {
|
||||
ag.log.Error("record metrics: %v", err)
|
||||
}
|
||||
ag.log.Debug("usage: model=%s input=%d cached=%d cache_create=%d output=%d hit_ratio=%.3f cost=$%.6f", ag.runtime.Backend.Model(), in, cached, creation, out, ag.history.cacheHitRatio(), costUSD)
|
||||
ag.notifyChange()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -15,6 +15,7 @@ import (
|
|||
|
||||
"ollie/cmd/olliesrv/internal/agent"
|
||||
"ollie/cmd/olliesrv/internal/backend"
|
||||
"ollie/cmd/olliesrv/internal/metrics"
|
||||
"ollie/cmd/olliesrv/internal/session"
|
||||
"ollie/paths"
|
||||
"ollie/virtfs"
|
||||
|
|
@ -152,6 +153,33 @@ func buildTreeSpec(cfg *Config) virtfs.FsNodeDecl {
|
|||
return []byte(sb.String()), nil
|
||||
}),
|
||||
),
|
||||
virtfs.FileNode("stats", 0444,
|
||||
virtfs.Read(func() ([]byte, error) {
|
||||
a, err := metrics.AggregateAll()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return []byte(metrics.Format(a)), nil
|
||||
}),
|
||||
),
|
||||
virtfs.FileNode("metrics", 0444,
|
||||
virtfs.Read(func() ([]byte, error) {
|
||||
a, err := metrics.AggregateAll()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return []byte(metrics.Format(a)), nil
|
||||
}),
|
||||
),
|
||||
virtfs.FileNode("metrics.by-backend-model", 0444,
|
||||
virtfs.Read(func() ([]byte, error) {
|
||||
groups, err := metrics.GroupedAggregate("", "")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return []byte(metrics.FormatGroups(groups)), nil
|
||||
}),
|
||||
),
|
||||
|
||||
// session/
|
||||
virtfs.DirNode("session",
|
||||
|
|
@ -245,6 +273,33 @@ func buildSessionChildren(
|
|||
return []byte("false\n"), nil
|
||||
}),
|
||||
),
|
||||
virtfs.FileNode("stats", 0444,
|
||||
virtfs.Read(func() ([]byte, error) {
|
||||
a, err := metrics.AggregateSession(s.ID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return []byte(metrics.Format(a)), nil
|
||||
}),
|
||||
),
|
||||
virtfs.FileNode("metrics", 0444,
|
||||
virtfs.Read(func() ([]byte, error) {
|
||||
a, err := metrics.AggregateSession(s.ID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return []byte(metrics.Format(a)), nil
|
||||
}),
|
||||
),
|
||||
virtfs.FileNode("metrics.by-backend-model", 0444,
|
||||
virtfs.Read(func() ([]byte, error) {
|
||||
groups, err := metrics.GroupedAggregate(s.ID, "")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return []byte(metrics.FormatGroups(groups)), nil
|
||||
}),
|
||||
),
|
||||
virtfs.FileNode("ctl", 0666,
|
||||
virtfs.Rdwr(func(_ context.Context, data []byte) ([]byte, error) {
|
||||
return dispatch(map[string]func([]string) ([]byte, error){
|
||||
|
|
@ -778,7 +833,29 @@ func buildAgentChildren(a *agent.Agent, s *session.Session) []virtfs.FsNodeDecl
|
|||
),
|
||||
virtfs.FileNode("stats", 0444,
|
||||
virtfs.Read(func() ([]byte, error) {
|
||||
return stats(nil)
|
||||
a, err := metrics.AggregateAgent(s.ID, a.ID())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return []byte(metrics.Format(a)), nil
|
||||
}),
|
||||
),
|
||||
virtfs.FileNode("metrics", 0444,
|
||||
virtfs.Read(func() ([]byte, error) {
|
||||
a, err := metrics.AggregateAgent(s.ID, a.ID())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return []byte(metrics.Format(a)), nil
|
||||
}),
|
||||
),
|
||||
virtfs.FileNode("metrics.by-backend-model", 0444,
|
||||
virtfs.Read(func() ([]byte, error) {
|
||||
groups, err := metrics.GroupedAggregate(s.ID, a.ID())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return []byte(metrics.FormatGroups(groups)), nil
|
||||
}),
|
||||
),
|
||||
virtfs.FileNode("name", 0666,
|
||||
|
|
|
|||
|
|
@ -0,0 +1,172 @@
|
|||
package metrics
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"ollie/cmd/olliesrv/internal/backend"
|
||||
"ollie/paths"
|
||||
)
|
||||
|
||||
// Record is one backend usage event. Records are append-only and independently
|
||||
// queryable, so aggregates can be rebuilt without session state.
|
||||
type Record struct {
|
||||
Time time.Time `json:"time"`
|
||||
SessionID string `json:"session"`
|
||||
AgentID string `json:"agent"`
|
||||
Backend string `json:"backend,omitempty"`
|
||||
Model string `json:"model,omitempty"`
|
||||
InputTokens int `json:"inputTokens"`
|
||||
CachedInputTokens int `json:"cachedInputTokens"`
|
||||
CacheCreationTokens int `json:"cacheCreationTokens"`
|
||||
OutputTokens int `json:"outputTokens"`
|
||||
CostUSD float64 `json:"costUSD"`
|
||||
Estimated bool `json:"estimated,omitempty"`
|
||||
}
|
||||
|
||||
// Aggregate contains cumulative token, cache, request, and cost metrics.
|
||||
type Aggregate struct {
|
||||
InputTokens int
|
||||
CachedInputTokens int
|
||||
CacheCreationTokens int
|
||||
OutputTokens int
|
||||
Requests int
|
||||
CostUSD float64
|
||||
Estimated bool
|
||||
}
|
||||
|
||||
func (a Aggregate) CacheHitRatio() float64 {
|
||||
total := a.InputTokens + a.CachedInputTokens
|
||||
if total == 0 {
|
||||
return 0
|
||||
}
|
||||
return float64(a.CachedInputTokens) / float64(total)
|
||||
}
|
||||
|
||||
func (a *Aggregate) add(r Record) {
|
||||
a.InputTokens += r.InputTokens
|
||||
a.CachedInputTokens += r.CachedInputTokens
|
||||
a.CacheCreationTokens += r.CacheCreationTokens
|
||||
a.OutputTokens += r.OutputTokens
|
||||
a.Requests++
|
||||
a.CostUSD += r.CostUSD
|
||||
a.Estimated = a.Estimated || r.Estimated
|
||||
}
|
||||
|
||||
var appendMu sync.Mutex
|
||||
|
||||
// EventsPath returns the append-only metrics event log path.
|
||||
func EventsPath() string { return filepath.Join(paths.DataDir(), "metrics", "events.jsonl") }
|
||||
|
||||
// RecordUsage appends one usage event to the global metrics log.
|
||||
func RecordUsage(sessionID, agentID, backendName, model string, u backend.Usage, estimated bool) error {
|
||||
r := Record{Time: time.Now().UTC(), SessionID: sessionID, AgentID: agentID, Backend: backendName, Model: model, InputTokens: u.InputTokens, CachedInputTokens: u.CachedInputTokens, CacheCreationTokens: u.CacheCreationTokens, OutputTokens: u.OutputTokens, CostUSD: u.CostUSD, Estimated: estimated}
|
||||
data, err := json.Marshal(r)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
appendMu.Lock()
|
||||
defer appendMu.Unlock()
|
||||
path := EventsPath()
|
||||
if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil {
|
||||
return err
|
||||
}
|
||||
f, err := os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0600)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer f.Close()
|
||||
_, err = fmt.Fprintf(f, "%s\n", data)
|
||||
return err
|
||||
}
|
||||
|
||||
func AggregateAll() (Aggregate, error) { return aggregate(func(Record) bool { return true }) }
|
||||
func AggregateSession(sessionID string) (Aggregate, error) {
|
||||
return aggregate(func(r Record) bool { return r.SessionID == sessionID })
|
||||
}
|
||||
func AggregateAgent(sessionID, agentID string) (Aggregate, error) {
|
||||
return aggregate(func(r Record) bool { return r.SessionID == sessionID && r.AgentID == agentID })
|
||||
}
|
||||
|
||||
func aggregate(match func(Record) bool) (Aggregate, error) {
|
||||
f, err := os.Open(EventsPath())
|
||||
if os.IsNotExist(err) {
|
||||
return Aggregate{}, nil
|
||||
}
|
||||
if err != nil {
|
||||
return Aggregate{}, err
|
||||
}
|
||||
defer f.Close()
|
||||
var out Aggregate
|
||||
scanner := bufio.NewScanner(f)
|
||||
for scanner.Scan() {
|
||||
var r Record
|
||||
if json.Unmarshal(scanner.Bytes(), &r) == nil && match(r) {
|
||||
out.add(r)
|
||||
}
|
||||
}
|
||||
if err := scanner.Err(); err != nil {
|
||||
return Aggregate{}, err
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// GroupedAggregate returns aggregates grouped by backend and model.
|
||||
func GroupedAggregate(sessionID, agentID string) (map[string]Aggregate, error) {
|
||||
f, err := os.Open(EventsPath())
|
||||
if os.IsNotExist(err) {
|
||||
return map[string]Aggregate{}, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer f.Close()
|
||||
groups := make(map[string]Aggregate)
|
||||
scanner := bufio.NewScanner(f)
|
||||
for scanner.Scan() {
|
||||
var r Record
|
||||
if json.Unmarshal(scanner.Bytes(), &r) != nil {
|
||||
continue
|
||||
}
|
||||
if (sessionID != "" && r.SessionID != sessionID) || (agentID != "" && r.AgentID != agentID) {
|
||||
continue
|
||||
}
|
||||
key := r.Backend + "\t" + r.Model
|
||||
a := groups[key]
|
||||
a.add(r)
|
||||
groups[key] = a
|
||||
}
|
||||
if err := scanner.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return groups, nil
|
||||
}
|
||||
|
||||
// Format renders an aggregate as stable key=value lines for the 9P API.
|
||||
func Format(a Aggregate) string {
|
||||
return fmt.Sprintf("inputTokens=%d\ncachedInputTokens=%d\ncacheCreationTokens=%d\noutputTokens=%d\nrequests=%d\ncostUSD=%.6f\ncacheHitRatio=%g\nestimated=%t\n", a.InputTokens, a.CachedInputTokens, a.CacheCreationTokens, a.OutputTokens, a.Requests, a.CostUSD, a.CacheHitRatio(), a.Estimated)
|
||||
}
|
||||
|
||||
// FormatGroups renders backend/model aggregates as tab-separated records.
|
||||
func FormatGroups(groups map[string]Aggregate) string {
|
||||
keys := make([]string, 0, len(groups))
|
||||
for key := range groups {
|
||||
keys = append(keys, key)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
var sb strings.Builder
|
||||
sb.WriteString("backend\tmodel\tinputTokens\tcachedInputTokens\tcacheCreationTokens\toutputTokens\trequests\tcostUSD\tcacheHitRatio\testimated\n")
|
||||
for _, key := range keys {
|
||||
parts := strings.SplitN(key, "\t", 2)
|
||||
a := groups[key]
|
||||
fmt.Fprintf(&sb, "%s\t%s\t%d\t%d\t%d\t%d\t%d\t%.6f\t%g\t%t\n", parts[0], parts[1], a.InputTokens, a.CachedInputTokens, a.CacheCreationTokens, a.OutputTokens, a.Requests, a.CostUSD, a.CacheHitRatio(), a.Estimated)
|
||||
}
|
||||
return sb.String()
|
||||
}
|
||||
Loading…
Reference in New Issue