internal/metrics/metrics.go
1
// Package metrics reports how a long-lived koment is doing: how much drift the
2
// repository carries, whether anyone is reading, and how often an agent is
3
// handed an annotation that no longer applies.
4
package metrics
6
import (
7
"context"
8
"errors"
9
"fmt"
10
"io"
11
"net"
12
"net/http"
13
"time"
15
"github.com/prometheus/client_golang/prometheus"
16
"github.com/prometheus/client_golang/prometheus/collectors"
17
"github.com/prometheus/client_golang/prometheus/promhttp"
19
"github.com/janpuc/koment/internal/anchor"
20
"github.com/janpuc/koment/internal/listen"
21
"github.com/janpuc/koment/internal/store"
22
)
24
const (
25
namespace = "koment"
26
shutdownGrace = 5 * time.Second
27
headerTimeout = 10 * time.Second
28
)
30
// Recorder is what the servers depend on, so that neither imports Prometheus
31
// and a build without metrics costs nothing (ADR 0020).
32
type Recorder interface {
33
ObserveRepository(resolved map[anchor.Status]int, files int, took time.Duration)
34
ObserveHTTP(handler string, code int, took time.Duration)
35
ObserveMCPCall(tool, outcome string, took time.Duration)
36
ObserveServed(status anchor.Status)
37
}
39
// Discard satisfies Recorder without measuring anything, which is what every
40
// server uses unless --metrics was given.
41
type Discard struct{}
43
func (Discard) ObserveRepository(map[anchor.Status]int, int, time.Duration) {}
44
func (Discard) ObserveHTTP(string, int, time.Duration) {}
45
func (Discard) ObserveMCPCall(string, string, time.Duration) {}
46
func (Discard) ObserveServed(anchor.Status) {}
48
type Metrics struct {
49
registry *prometheus.Registry
51
annotations *prometheus.GaugeVec
52
files prometheus.Gauge
53
resolveDuration prometheus.Histogram
55
httpRequests *prometheus.CounterVec
56
httpDuration *prometheus.HistogramVec
58
mcpCalls *prometheus.CounterVec
59
mcpDuration *prometheus.HistogramVec
60
mcpServed *prometheus.CounterVec
61
}
63
func New() *Metrics {
64
m := &Metrics{
65
registry: prometheus.NewRegistry(),
67
annotations: prometheus.NewGaugeVec(prometheus.GaugeOpts{
68
Namespace: namespace,
69
Name: "annotations",
70
Help: "Annotations in the served repository, by resolution status.",
71
}, []string{"status"}),
73
files: prometheus.NewGauge(prometheus.GaugeOpts{
74
Namespace: namespace,
75
Name: "files_annotated",
76
Help: "Source files carrying at least one annotation.",
77
}),
79
resolveDuration: prometheus.NewHistogram(prometheus.HistogramOpts{
80
Namespace: namespace,
81
Name: "resolve_duration_seconds",
82
Help: "Time to resolve every annotation in the repository.",
83
Buckets: prometheus.ExponentialBuckets(0.001, 4, 8),
84
}),
86
httpRequests: prometheus.NewCounterVec(prometheus.CounterOpts{
87
Namespace: namespace,
88
Name: "http_requests_total",
89
Help: "HTTP requests served.",
90
}, []string{"handler", "code"}),
92
httpDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{
93
Namespace: namespace,
94
Name: "http_request_duration_seconds",
95
Help: "Time to serve an HTTP request.",
96
Buckets: prometheus.DefBuckets,
97
}, []string{"handler"}),
99
mcpCalls: prometheus.NewCounterVec(prometheus.CounterOpts{
100
Namespace: namespace,
101
Name: "mcp_calls_total",
102
Help: "MCP tool calls, by tool and outcome.",
103
}, []string{"tool", "outcome"}),
105
mcpDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{
106
Namespace: namespace,
107
Name: "mcp_call_duration_seconds",
108
Help: "Time to answer an MCP tool call.",
109
Buckets: prometheus.DefBuckets,
110
}, []string{"tool"}),
112
mcpServed: prometheus.NewCounterVec(prometheus.CounterOpts{
113
Namespace: namespace,
114
Name: "mcp_annotations_served_total",
115
Help: "Annotations handed to an agent, by resolution status.",
116
}, []string{"status"}),
117
}
119
m.registry.MustRegister(
120
m.annotations, m.files, m.resolveDuration,
121
m.httpRequests, m.httpDuration,
122
m.mcpCalls, m.mcpDuration, m.mcpServed,
123
collectors.NewGoCollector(),
124
collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}),
125
)
126
return m
127
}
129
// ObserveRepository publishes a full sweep. Every status is set on every sweep,
130
// including the zero ones: leaving a status unset would make a drift that was
131
// fixed look like a scrape gap rather than a return to zero.
132
func (m *Metrics) ObserveRepository(resolved map[anchor.Status]int, files int, took time.Duration) {
133
for _, status := range []anchor.Status{
134
anchor.StatusOK, anchor.StatusMoved, anchor.StatusAmbiguous, anchor.StatusDrifted, anchor.StatusOrphaned,
135
} {
136
m.annotations.WithLabelValues(string(status)).Set(float64(resolved[status]))
137
}
138
m.files.Set(float64(files))
139
m.resolveDuration.Observe(took.Seconds())
140
}
142
func (m *Metrics) ObserveHTTP(handler string, code int, took time.Duration) {
143
m.httpRequests.WithLabelValues(handler, fmt.Sprint(code)).Inc()
144
m.httpDuration.WithLabelValues(handler).Observe(took.Seconds())
145
}
147
func (m *Metrics) ObserveMCPCall(tool, outcome string, took time.Duration) {
148
m.mcpCalls.WithLabelValues(tool, outcome).Inc()
149
m.mcpDuration.WithLabelValues(tool).Observe(took.Seconds())
150
}
152
func (m *Metrics) ObserveServed(status anchor.Status) {
153
m.mcpServed.WithLabelValues(string(status)).Inc()
154
}
156
func (m *Metrics) Handler() http.Handler {
157
return promhttp.HandlerFor(m.registry, promhttp.HandlerOpts{Registry: m.registry})
158
}
160
// Serve runs the metrics endpoint on its own listener. It is never mounted on
161
// the serving listener, which is unauthenticated and ingress-facing (ADR 0020).
162
func (m *Metrics) Serve(ctx context.Context, address string, stderr io.Writer) error {
163
resolved, err := listen.Address(address)
164
if err != nil {
165
return err
166
}
168
mux := http.NewServeMux()
169
mux.Handle("GET /metrics", m.Handler())
170
server := &http.Server{Handler: mux, ReadHeaderTimeout: headerTimeout}
172
listener, err := net.Listen("tcp", resolved)
173
if err != nil {
174
return fmt.Errorf("listening for metrics on %s: %w", resolved, err)
175
}
176
fmt.Fprintf(stderr, "koment: metrics on http://%s/metrics\n", listener.Addr())
178
go func() {
179
<-ctx.Done()
180
timeout, cancel := context.WithTimeout(context.Background(), shutdownGrace)
181
defer cancel()
182
if err := server.Shutdown(timeout); err != nil {
183
fmt.Fprintf(stderr, "koment: shutting down metrics: %v\n", err)
184
}
185
}()
187
if err := server.Serve(listener); !errors.Is(err, http.ErrServerClosed) {
188
return err
189
}
190
return nil
191
}
193
// Sweep resolves the whole repository so the gauges describe it. Called on a
194
// schedule rather than per request, because a sweep is O(annotations) and a
195
// scrape must not be able to drive load.
196
func Sweep(annotations *store.Store, recorder Recorder) error {
197
started := time.Now()
198
files, err := annotations.AnnotatedFiles()
199
if err != nil {
200
return err
201
}
203
counts := map[anchor.Status]int{}
204
for _, file := range files {
205
resolutions, err := anchor.ResolveStored(annotations, file)
206
if err != nil {
207
return err
208
}
209
for _, resolution := range resolutions {
210
counts[resolution.Status]++
211
}
212
}
214
recorder.ObserveRepository(counts, len(files), time.Since(started))
215
return nil
216
}