Add metrics logging and an Activity page to show requests

config.example.yaml

@@ -15,6 +15,18 @@ healthCheckTimeout: 500
 # - Valid log levels: debug, info, warn, error
 logLevel: info
 
+# metricsMaxInMemory: maximum number of metrics to keep in memory
+# - optional, default: 1000
+# - controls how many metrics are stored in memory before older ones are discarded
+# - useful for limiting memory usage when processing large volumes of metrics
+metricsMaxInMemory: 1000
+
+# metricsLogPath: sets the path to the metrics log file
+# - optional, default: empty string
+# - if set, metrics will be logged to the specified file
+# - useful for persistent logging of metrics
+metricsLogPath: /path/to/metrics.log
+
 # startPort: sets the starting port number for the automatic ${PORT} macro.
 # - optional, default: 5800
 # - the ${PORT} macro can be used in model.cmd and model.proxy settings
@@ -200,4 +212,4 @@ groups:
     members:
       - "forever-modelA"
       - "forever-modelB"
-      - "forever-modelc"
+      - "forever-modelc"

docker/config.example.yaml

@@ -1,5 +1,6 @@
 healthCheckTimeout: 300
 logRequests: true
+metricsMaxInMemory: 1000
 
 models:
   "qwen2.5":

proxy/config.go

@@ -142,6 +142,8 @@ type Config struct {
 	HealthCheckTimeout int                    `yaml:"healthCheckTimeout"`
 	LogRequests        bool                   `yaml:"logRequests"`
 	LogLevel           string                 `yaml:"logLevel"`
+	MetricsLogPath     string                 `yaml:"metricsLogPath"`
+	MetricsMaxInMemory int                    `yaml:"metricsMaxInMemory"`
 	Models             map[string]ModelConfig `yaml:"models"` /* key is model ID */
 	Profiles           map[string][]string    `yaml:"profiles"`
 	Groups             map[string]GroupConfig `yaml:"groups"` /* key is group ID */
@@ -194,6 +196,8 @@ func LoadConfigFromReader(r io.Reader) (Config, error) {
 		HealthCheckTimeout: 120,
 		StartPort:          5800,
 		LogLevel:           "info",
+		MetricsLogPath:     "",
+		MetricsMaxInMemory: 1000,
 	}
 	err = yaml.Unmarshal(data, &config)
 	if err != nil {

proxy/config_posix_test.go

@@ -196,6 +196,7 @@ groups:
 			},
 		},
 		HealthCheckTimeout: 15,
+		MetricsMaxInMemory: 1000,
 		Profiles: map[string][]string{
 			"test": {"model1", "model2"},
 		},

proxy/metrics_parser.go

@@ -0,0 +1,195 @@
+package proxy
+
+import (
+	"bufio"
+	"encoding/json"
+	"os"
+	"regexp"
+	"strconv"
+	"sync"
+	"time"
+)
+
+// TokenMetrics represents parsed token statistics from llama-server logs
+type TokenMetrics struct {
+	Timestamp       time.Time `json:"timestamp"`
+	Model           string    `json:"model"`
+	InputTokens     int       `json:"input_tokens"`
+	OutputTokens    int       `json:"output_tokens"`
+	TokensPerSecond float64   `json:"tokens_per_second"`
+	DurationMs      int       `json:"duration_ms"`
+}
+
+// MetricsParser parses llama-server output for token statistics
+type MetricsParser struct {
+	mu              sync.RWMutex
+	metrics         []TokenMetrics
+	maxMetrics      int
+	logPath         string
+	promptEvalRegex *regexp.Regexp
+	evalRegex       *regexp.Regexp
+	proxyLogger     *LogMonitor
+}
+
+// NewMetricsParser creates a new metrics parser
+func NewMetricsParser(config *Config, proxyLogger *LogMonitor) *MetricsParser {
+	maxMetrics := config.MetricsMaxInMemory
+	if maxMetrics <= 0 {
+		maxMetrics = 1000 // Default fallback
+	}
+
+	mp := &MetricsParser{
+		maxMetrics:      maxMetrics,
+		logPath:         config.MetricsLogPath,
+		promptEvalRegex: regexp.MustCompile(`prompt eval time\s*=\s*(\d+(?:\.\d+)?)\s*ms\s*/\s*(\d+)\s*tokens\s*\(\s*(\d+(?:\.\d+)?)\s*ms per token,\s*(\d+(?:\.\d+)?)\s*tokens per second\s*\)`),
+		evalRegex:       regexp.MustCompile(`eval time\s*=\s*(\d+(?:\.\d+)?)\s*ms\s*/\s*(\d+)\s*tokens\s*\(\s*(\d+(?:\.\d+)?)\s*ms per token,\s*(\d+(?:\.\d+)?)\s*tokens per second\s*\)`),
+		proxyLogger:     proxyLogger,
+	}
+
+	// Load existing metrics from file if path is provided
+	if config.MetricsLogPath != "" {
+		_ = mp.LoadMetrics() // Only warn, don't error as requested
+	}
+
+	return mp
+}
+
+// LoadMetrics loads metrics from the JSONL file
+func (mp *MetricsParser) LoadMetrics() error {
+	if mp.logPath == "" {
+		return nil
+	}
+
+	file, err := os.Open(mp.logPath)
+	if err != nil {
+		if os.IsNotExist(err) {
+			// File doesn't exist yet, which is fine
+			return nil
+		}
+		if mp.proxyLogger != nil {
+			mp.proxyLogger.Warnf("Failed to open metrics log file for reading: %v", err)
+		}
+		return err
+	}
+	defer file.Close()
+
+	mp.mu.Lock()
+	defer mp.mu.Unlock()
+
+	// Use bufio.Scanner to read line by line
+	scanner := bufio.NewScanner(file)
+	lineNum := 0
+	for scanner.Scan() {
+		lineNum++
+		line := scanner.Text()
+		if len(line) == 0 {
+			continue
+		}
+
+		var metric TokenMetrics
+		if err := json.Unmarshal([]byte(line), &metric); err != nil {
+			if mp.proxyLogger != nil {
+				mp.proxyLogger.Warnf("Skipping malformed metrics line %d: %v", lineNum, err)
+			}
+			continue
+		}
+		mp.metrics = append(mp.metrics, metric)
+	}
+
+	// Keep only the most recent metrics if we loaded too many
+	if len(mp.metrics) > mp.maxMetrics {
+		mp.metrics = mp.metrics[len(mp.metrics)-mp.maxMetrics:]
+	}
+
+	if err := scanner.Err(); err != nil && mp.proxyLogger != nil {
+		mp.proxyLogger.Warnf("Error reading metrics log file: %v", err)
+	}
+
+	return scanner.Err()
+}
+
+// addMetrics adds a new metric to the collection and appends to file if configured
+func (mp *MetricsParser) addMetrics(metric TokenMetrics) {
+	mp.mu.Lock()
+	defer mp.mu.Unlock()
+
+	mp.metrics = append(mp.metrics, metric)
+	if len(mp.metrics) > mp.maxMetrics {
+		mp.metrics = mp.metrics[len(mp.metrics)-mp.maxMetrics:]
+	}
+
+	// Append to JSONL file if path is configured
+	if mp.logPath != "" {
+		mp.appendToFile(metric)
+	}
+}
+
+// appendToFile appends a single metric to the JSONL file
+func (mp *MetricsParser) appendToFile(metric TokenMetrics) {
+	file, err := os.OpenFile(mp.logPath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644)
+	if err != nil {
+		if mp.proxyLogger != nil {
+			mp.proxyLogger.Warnf("Failed to open metrics log file for appending: %v", err)
+		}
+		return
+	}
+	defer file.Close()
+
+	jsonData, err := json.Marshal(metric)
+	if err != nil {
+		if mp.proxyLogger != nil {
+			mp.proxyLogger.Warnf("Failed to marshal metrics data: %v", err)
+		}
+		return
+	}
+
+	// Append newline and write
+	jsonData = append(jsonData, '\n')
+	if _, err := file.Write(jsonData); err != nil {
+		if mp.proxyLogger != nil {
+			mp.proxyLogger.Warnf("Failed to write metrics to log file: %v", err)
+		}
+	}
+}
+
+// ParseLogLine parses a single log line for token metrics
+func (mp *MetricsParser) ParseLogLine(line string, modelName string) {
+	if matches := mp.promptEvalRegex.FindStringSubmatch(line); matches != nil {
+		// Check for prompt evaluation metrics (input tokens)
+		durationMs, _ := strconv.ParseFloat(matches[1], 64)
+		tokens, _ := strconv.Atoi(matches[2])
+		tokensPerSecond, _ := strconv.ParseFloat(matches[4], 64)
+
+		metrics := TokenMetrics{
+			Timestamp:       time.Now(),
+			Model:           modelName,
+			InputTokens:     tokens,
+			OutputTokens:    0,
+			TokensPerSecond: tokensPerSecond,
+			DurationMs:      int(durationMs),
+		}
+		mp.addMetrics(metrics)
+	} else if matches := mp.evalRegex.FindStringSubmatch(line); matches != nil {
+		// Check for evaluation metrics (output tokens)
+		durationMs, _ := strconv.ParseFloat(matches[1], 64)
+		tokens, _ := strconv.Atoi(matches[2])
+		tokensPerSecond, _ := strconv.ParseFloat(matches[4], 64)
+
+		metrics := TokenMetrics{
+			Timestamp:       time.Now(),
+			Model:           modelName,
+			InputTokens:     0,
+			OutputTokens:    tokens,
+			TokensPerSecond: tokensPerSecond,
+			DurationMs:      int(durationMs),
+		}
+		mp.addMetrics(metrics)
+	}
+}
+
+// GetMetricsJSON returns metrics as JSON
+func (mp *MetricsParser) GetMetricsJSON() ([]byte, error) {
+	mp.mu.RLock()
+	defer mp.mu.RUnlock()
+	return json.Marshal(mp.metrics)
+}

proxy/process.go

@@ -1,6 +1,7 @@
 package proxy
 
 import (
+	"bufio"
 	"context"
 	"errors"
 	"fmt"
@@ -49,6 +50,7 @@ type Process struct {
 
 	processLogger *LogMonitor
 	proxyLogger   *LogMonitor
+	metricsParser *MetricsParser
 
 	healthCheckTimeout      int
 	healthCheckLoopInterval time.Duration
@@ -73,7 +75,7 @@ type Process struct {
 	failedStartCount int
 }
 
-func NewProcess(ID string, healthCheckTimeout int, config ModelConfig, processLogger *LogMonitor, proxyLogger *LogMonitor) *Process {
+func NewProcess(ID string, healthCheckTimeout int, config ModelConfig, processLogger *LogMonitor, proxyLogger *LogMonitor, metricsParser *MetricsParser) *Process {
 	concurrentLimit := 10
 	if config.ConcurrencyLimit > 0 {
 		concurrentLimit = config.ConcurrencyLimit
@@ -86,6 +88,7 @@ func NewProcess(ID string, healthCheckTimeout int, config ModelConfig, processLo
 		cancelUpstream:          nil,
 		processLogger:           processLogger,
 		proxyLogger:             proxyLogger,
+		metricsParser:           metricsParser,
 		healthCheckTimeout:      healthCheckTimeout,
 		healthCheckLoopInterval: 5 * time.Second, /* default, can not be set by user - used for testing */
 		state:                   StateStopped,
@@ -193,15 +196,23 @@ func (p *Process) start() error {
 	defer p.waitStarting.Done()
 	cmdContext, ctxCancelUpstream := context.WithCancel(context.Background())
 
+	// Create a pipe to capture stdout and feed both LogMonitor and MetricsParser
+	stdoutReader, stdoutWriter := io.Pipe()
+	stderrReader, stderrWriter := io.Pipe()
+
 	p.cmd = exec.CommandContext(cmdContext, args[0], args[1:]...)
-	p.cmd.Stdout = p.processLogger
-	p.cmd.Stderr = p.processLogger
+	p.cmd.Stdout = stdoutWriter
+	p.cmd.Stderr = stderrWriter
 	p.cmd.Env = append(p.cmd.Environ(), p.config.Env...)
 	p.cmd.Cancel = p.cmdStopUpstreamProcess
 	p.cmd.WaitDelay = p.gracefulStopTimeout
 	p.cancelUpstream = ctxCancelUpstream
 	p.cmdWaitChan = make(chan struct{})
 
+	// Start goroutines to process stdout and stderr
+	go p.processOutput(stdoutReader, "stdout")
+	go p.processOutput(stderrReader, "stderr")
+
 	p.failedStartCount++ // this will be reset to zero when the process has successfully started
 
 	p.proxyLogger.Debugf("<%s> Executing start command: %s, env: %s", p.ID, strings.Join(args, " "), strings.Join(p.config.Env, ", "))
@@ -512,6 +523,31 @@ func (p *Process) waitForCmd() {
 	close(p.cmdWaitChan)
 }
 
+// processOutput reads from a pipe and sends data to both LogMonitor and MetricsParser
+func (p *Process) processOutput(reader *io.PipeReader, streamType string) {
+	defer reader.Close()
+
+	scanner := bufio.NewScanner(reader)
+	for scanner.Scan() {
+		line := scanner.Text()
+		if line == "" {
+			continue
+		}
+
+		// Send to LogMonitor for logging
+		p.processLogger.Write([]byte(line + "\n"))
+
+		// Send to MetricsParser for parsing
+		if p.metricsParser != nil {
+			p.metricsParser.ParseLogLine(line, p.ID)
+		}
+	}
+
+	if err := scanner.Err(); err != nil && err != io.EOF {
+		p.proxyLogger.Errorf("<%s> Error reading %s: %v", p.ID, streamType, err)
+	}
+}
+
 // cmdStopUpstreamProcess attemps to stop the upstream process gracefully
 func (p *Process) cmdStopUpstreamProcess() error {
 	p.processLogger.Debugf("<%s> cmdStopUpstreamProcess() initiating graceful stop of upstream process", p.ID)