p2.7: prometheus /metrics
Default GET /metrics is Prometheus text; JSON stays at /metrics.json and ?format=json. :8443 needs the metrics caller bearer; loopback stays open. Restricted callers have no per-caller series and no tavily:restricted gauge.
This commit is contained in:
+149
-1
@@ -22,6 +22,7 @@ import (
|
||||
"git.wntrmute.dev/kyle/sift/internal/fetch"
|
||||
"git.wntrmute.dev/kyle/sift/internal/fleetctx"
|
||||
"git.wntrmute.dev/kyle/sift/internal/guard"
|
||||
"git.wntrmute.dev/kyle/sift/internal/metrics"
|
||||
"git.wntrmute.dev/kyle/sift/internal/recover"
|
||||
"git.wntrmute.dev/kyle/sift/internal/search"
|
||||
"git.wntrmute.dev/kyle/sift/internal/selfip"
|
||||
@@ -48,6 +49,7 @@ type Server struct {
|
||||
// PermitRestricted is context.permit_restricted: those callers never
|
||||
// have a fetch host written to the log (host=-). Independent of Honcho.
|
||||
PermitRestricted []string
|
||||
Prom *metrics.Set
|
||||
|
||||
global chan struct{}
|
||||
perMu sync.Mutex
|
||||
@@ -71,9 +73,11 @@ func New(version string, g *guard.Guard, f *fetch.Fetcher, keys *auth.Store, log
|
||||
global: make(chan struct{}, 8),
|
||||
per: map[string]chan struct{}{},
|
||||
}
|
||||
s.Prom = metrics.New(s)
|
||||
s.Mux.HandleFunc("GET /healthz", s.healthz)
|
||||
s.Mux.HandleFunc("GET /healthz/", s.healthz)
|
||||
s.Mux.HandleFunc("GET /metrics", s.metrics)
|
||||
s.Mux.HandleFunc("GET /metrics.json", s.metrics)
|
||||
s.Mux.HandleFunc("POST /v2/scrape", s.scrape)
|
||||
s.Mux.HandleFunc("POST /v1/fetch", s.v1fetch)
|
||||
s.Mux.HandleFunc("POST /v1/search", s.v1search)
|
||||
@@ -100,10 +104,38 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
ctx, cancel := context.WithTimeout(r.Context(), requestDeadline)
|
||||
defer cancel()
|
||||
r = r.WithContext(ctx)
|
||||
cw := &codeWriter{ResponseWriter: w}
|
||||
start := time.Now()
|
||||
s.Mux.ServeHTTP(cw, r)
|
||||
code := cw.code
|
||||
if code == 0 {
|
||||
code = http.StatusOK
|
||||
}
|
||||
s.Prom.Observe(r.URL.Path, time.Since(start), code)
|
||||
return
|
||||
}
|
||||
s.Mux.ServeHTTP(w, r)
|
||||
}
|
||||
|
||||
type codeWriter struct {
|
||||
http.ResponseWriter
|
||||
code int
|
||||
}
|
||||
|
||||
func (c *codeWriter) WriteHeader(status int) {
|
||||
if c.code == 0 {
|
||||
c.code = status
|
||||
}
|
||||
c.ResponseWriter.WriteHeader(status)
|
||||
}
|
||||
|
||||
func (c *codeWriter) Write(b []byte) (int, error) {
|
||||
if c.code == 0 {
|
||||
c.code = http.StatusOK
|
||||
}
|
||||
return c.ResponseWriter.Write(b)
|
||||
}
|
||||
|
||||
func (s *Server) healthz(w http.ResponseWriter, _ *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
out := map[string]any{"status": "ok", "version": s.Version}
|
||||
@@ -119,7 +151,44 @@ func (s *Server) healthz(w http.ResponseWriter, _ *http.Request) {
|
||||
_ = json.NewEncoder(w).Encode(out)
|
||||
}
|
||||
|
||||
func (s *Server) metrics(w http.ResponseWriter, _ *http.Request) {
|
||||
func (s *Server) metrics(w http.ResponseWriter, r *http.Request) {
|
||||
if !s.allowMetrics(w, r) {
|
||||
return
|
||||
}
|
||||
if r.URL.Path == "/metrics.json" || r.URL.Query().Get("format") == "json" {
|
||||
s.writeMetricsJSON(w)
|
||||
return
|
||||
}
|
||||
s.Prom.Handler().ServeHTTP(w, r)
|
||||
}
|
||||
|
||||
func (s *Server) allowMetrics(w http.ResponseWriter, r *http.Request) bool {
|
||||
if metrics.LoopbackOpen(r) {
|
||||
return true
|
||||
}
|
||||
h := r.Header.Get("Authorization")
|
||||
tok := strings.TrimSpace(strings.TrimPrefix(h, "Bearer"))
|
||||
tok = strings.TrimSpace(tok)
|
||||
if tok == "" || s.Auth == nil {
|
||||
s.Log.Warn("unauthorized", "caller", "-", "host", "-", "reason", "missing metrics key")
|
||||
writeErr(w, 401, "unauthorized", "")
|
||||
return false
|
||||
}
|
||||
c := s.Auth.LookupBearer(tok)
|
||||
if c == nil {
|
||||
s.Log.Warn("unauthorized", "caller", "-", "host", "-", "reason", "unknown metrics key")
|
||||
writeErr(w, 401, "unauthorized", "")
|
||||
return false
|
||||
}
|
||||
if !c.MetricsOnly() {
|
||||
s.Log.Warn("unauthorized", "caller", c.Name, "host", "-", "reason", "not metrics caller")
|
||||
writeErr(w, 403, "unauthorized", "")
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func (s *Server) writeMetricsJSON(w http.ResponseWriter) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
fails := uint64(0)
|
||||
if s.SelfIP != nil {
|
||||
@@ -136,9 +205,80 @@ func (s *Server) metrics(w http.ResponseWriter, _ *http.Request) {
|
||||
m[k] = v
|
||||
}
|
||||
}
|
||||
calls, fb := uint64(0), map[string]uint64{}
|
||||
var inflight, waiters int
|
||||
if s.Digest != nil {
|
||||
calls, fb = s.Digest.Stats()
|
||||
inflight, waiters = s.Digest.Gate.Snapshot()
|
||||
}
|
||||
m["digest_calls"] = calls
|
||||
m["digest_fallbacks"] = fb
|
||||
m["dixie_inflight"] = inflight
|
||||
m["dixie_queue_waiters"] = waiters
|
||||
var dH, dM, eH, eM uint64
|
||||
if cache := s.cache(); cache != nil {
|
||||
dH, dM, eH, eM, _, _, _ = cache.Stats()
|
||||
}
|
||||
m["dcache_digest_hit"] = dH
|
||||
m["dcache_digest_miss"] = dM
|
||||
m["dcache_embed_hit"] = eH
|
||||
m["dcache_embed_miss"] = eM
|
||||
_ = json.NewEncoder(w).Encode(m)
|
||||
}
|
||||
|
||||
func (s *Server) cache() *dcache.DB {
|
||||
if s.Digest != nil && s.Digest.Cache != nil {
|
||||
return s.Digest.Cache
|
||||
}
|
||||
if s.Search != nil {
|
||||
return s.Search.Cache
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Server) Snapshot() metrics.Snapshot {
|
||||
out := metrics.Snapshot{
|
||||
Blocks: s.blockN.Load(),
|
||||
Scrapes: s.scrapeN.Load(),
|
||||
Restricted: map[string]struct{}{},
|
||||
}
|
||||
for _, n := range s.PermitRestricted {
|
||||
out.Restricted[n] = struct{}{}
|
||||
}
|
||||
if s.SelfIP != nil {
|
||||
out.SelfIPFailures = s.SelfIP.Failures.Load()
|
||||
}
|
||||
if s.Digest != nil {
|
||||
out.DigestCalls, out.DigestFallbacks = s.Digest.Stats()
|
||||
out.DixieInflight, out.DixieWaiters = s.Digest.Gate.Snapshot()
|
||||
}
|
||||
if cache := s.cache(); cache != nil {
|
||||
out.DigestHit, out.DigestMiss, out.EmbedHit, out.EmbedMiss, _, _, _ = cache.Stats()
|
||||
}
|
||||
if s.Search != nil {
|
||||
m := s.Search.Metrics()
|
||||
if v, ok := m["search_queries"].(map[string]map[string]uint64); ok {
|
||||
out.SearchQueries = v
|
||||
}
|
||||
if v, ok := m["search_credits"].(map[string]map[string]uint64); ok {
|
||||
out.SearchCredits = v
|
||||
}
|
||||
if v, ok := m["search_daily_credits"].(map[string]uint64); ok {
|
||||
out.SearchDaily = v
|
||||
}
|
||||
if v, ok := m["search_capped"].(map[string]uint64); ok {
|
||||
out.SearchCapped = v
|
||||
}
|
||||
if v, ok := m["search_cache_hit"].(uint64); ok {
|
||||
out.SearchHit = v
|
||||
}
|
||||
if v, ok := m["search_cache_miss"].(uint64); ok {
|
||||
out.SearchMiss = v
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
type scrapeReq struct {
|
||||
URL string `json:"url"`
|
||||
Formats []string `json:"formats"`
|
||||
@@ -177,6 +317,11 @@ func (s *Server) caller(w http.ResponseWriter, r *http.Request) *auth.Caller {
|
||||
writeErr(w, 401, "unauthorized", "")
|
||||
return nil
|
||||
}
|
||||
if c.MetricsOnly() {
|
||||
s.Log.Warn("unauthorized", "caller", c.Name, "host", "-", "reason", "metrics caller")
|
||||
writeErr(w, 403, "unauthorized", "")
|
||||
return nil
|
||||
}
|
||||
return c
|
||||
}
|
||||
|
||||
@@ -327,6 +472,9 @@ func (s *Server) v1fetch(w http.ResponseWriter, r *http.Request) {
|
||||
} else {
|
||||
dr := s.Digest.Run(r.Context(), content, q, res.FinalURL, s.callerClass(c))
|
||||
digestMS = dr.DigestMS
|
||||
if !dr.Cached {
|
||||
s.Prom.ObserveDigest(time.Duration(dr.DigestMS) * time.Millisecond)
|
||||
}
|
||||
if dr.Fallback != "" {
|
||||
warn = append(warn, "digest fell back to clean: "+dr.Fallback)
|
||||
mode = "clean"
|
||||
|
||||
@@ -0,0 +1,256 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.wntrmute.dev/kyle/sift/internal/auth"
|
||||
"git.wntrmute.dev/kyle/sift/internal/digest"
|
||||
dto "github.com/prometheus/client_model/go"
|
||||
"github.com/prometheus/common/expfmt"
|
||||
)
|
||||
|
||||
func loopbackGET(path string) *http.Request {
|
||||
req := httptest.NewRequest(http.MethodGet, path, nil)
|
||||
req.Host = "127.0.0.1:8890"
|
||||
req.RemoteAddr = "127.0.0.1:9"
|
||||
return req
|
||||
}
|
||||
|
||||
func serveMetrics(t *testing.T, s *Server, req *http.Request) *httptest.ResponseRecorder {
|
||||
t.Helper()
|
||||
rec := httptest.NewRecorder()
|
||||
s.ServeHTTP(rec, req)
|
||||
return rec
|
||||
}
|
||||
|
||||
func parseProm(t *testing.T, body string) map[string]*dto.MetricFamily {
|
||||
t.Helper()
|
||||
parser := expfmt.NewDecoder(strings.NewReader(body), expfmt.NewFormat(expfmt.TypeTextPlain))
|
||||
out := map[string]*dto.MetricFamily{}
|
||||
for {
|
||||
var mf dto.MetricFamily
|
||||
if err := parser.Decode(&mf); err != nil {
|
||||
if err == io.EOF {
|
||||
break
|
||||
}
|
||||
t.Fatalf("expfmt: %v\n%s", err, body)
|
||||
}
|
||||
out[mf.GetName()] = &mf
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func TestMetricsJSONSidePath(t *testing.T) {
|
||||
s, key, _, _, _ := searchServer(t)
|
||||
postJSON(t, s, "/v1/search", key, map[string]any{"query": "same q", "k": 5, "rerank": false})
|
||||
postJSON(t, s, "/v1/search", key, map[string]any{"query": "same q", "k": 5, "rerank": false})
|
||||
for _, path := range []string{"/metrics.json", "/metrics?format=json"} {
|
||||
rec := serveMetrics(t, s, loopbackGET(path))
|
||||
if rec.Code != 200 {
|
||||
t.Fatalf("%s %d %s", path, rec.Code, rec.Body.String())
|
||||
}
|
||||
if !strings.Contains(rec.Header().Get("Content-Type"), "json") {
|
||||
t.Fatalf("%s ctype %s", path, rec.Header().Get("Content-Type"))
|
||||
}
|
||||
var m map[string]any
|
||||
if err := json.Unmarshal(rec.Body.Bytes(), &m); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if m["search_cache_hit"] == 0 && m["search_cache_miss"] == 0 {
|
||||
t.Fatalf("%s %v", path, m)
|
||||
}
|
||||
if _, ok := m["requests"]; !ok {
|
||||
t.Fatalf("missing requests: %v", m)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestMetricsPrometheusLoopback(t *testing.T) {
|
||||
s, key, _, _, _ := searchServer(t)
|
||||
postJSON(t, s, "/v1/search", key, map[string]any{"query": "fleet q", "k": 5, "rerank": false})
|
||||
rec := serveMetrics(t, s, loopbackGET("/metrics"))
|
||||
if rec.Code != 200 {
|
||||
t.Fatalf("%d %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
ct := rec.Header().Get("Content-Type")
|
||||
if !strings.Contains(ct, "text/plain") {
|
||||
t.Fatalf("ctype %s", ct)
|
||||
}
|
||||
fams := parseProm(t, rec.Body.String())
|
||||
if fams["sift_search_queries_total"] == nil {
|
||||
t.Fatalf("missing queries total: %v", rec.Body.String())
|
||||
}
|
||||
if fams["sift_search_queries_caller_total"] == nil {
|
||||
t.Fatalf("missing caller total")
|
||||
}
|
||||
if !strings.Contains(rec.Body.String(), `caller="test"`) {
|
||||
t.Fatal("fleet caller label")
|
||||
}
|
||||
if strings.Contains(rec.Body.String(), "fleet q") {
|
||||
t.Fatal("query leaked")
|
||||
}
|
||||
}
|
||||
|
||||
func TestMetricsAuth(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
keys := filepath.Join(dir, "keys.toml")
|
||||
fleetOut := filepath.Join(dir, "fleet")
|
||||
metOut := filepath.Join(dir, "met")
|
||||
if err := auth.Add(keys, "test", "hermes-fleet", fleetOut); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := auth.Add(keys, auth.NameMetrics, "should-be-cleared", metOut); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
store, err := auth.Load(keys)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
mc := store.LookupBearer(strings.TrimSpace(mustRead(t, metOut)))
|
||||
if mc == nil || mc.Workspace != "" || !mc.MetricsOnly() {
|
||||
t.Fatalf("metrics caller %+v", mc)
|
||||
}
|
||||
s, fleetKey := setupAPI(t, nil)
|
||||
s.Auth = store
|
||||
fleetKey = strings.TrimSpace(mustRead(t, fleetOut))
|
||||
metKey := strings.TrimSpace(mustRead(t, metOut))
|
||||
|
||||
req := httptest.NewRequest(http.MethodGet, "/metrics", nil) // 192.0.2.1 + example.com
|
||||
rec := serveMetrics(t, s, req)
|
||||
if rec.Code != 401 {
|
||||
t.Fatalf("no bearer: %d", rec.Code)
|
||||
}
|
||||
req = httptest.NewRequest(http.MethodGet, "/metrics", nil)
|
||||
req.Header.Set("Authorization", "Bearer "+fleetKey)
|
||||
rec = serveMetrics(t, s, req)
|
||||
if rec.Code != 403 {
|
||||
t.Fatalf("fleet bearer: %d %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
req = httptest.NewRequest(http.MethodGet, "/metrics", nil)
|
||||
req.Header.Set("Authorization", "Bearer "+metKey)
|
||||
rec = serveMetrics(t, s, req)
|
||||
if rec.Code != 200 {
|
||||
t.Fatalf("metrics bearer: %d %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
|
||||
fwd := loopbackGET("/metrics")
|
||||
fwd.Header.Set("X-Forwarded-For", "100.1.2.3")
|
||||
rec = serveMetrics(t, s, fwd)
|
||||
if rec.Code != 401 {
|
||||
t.Fatalf("forwarded loopback: %d", rec.Code)
|
||||
}
|
||||
|
||||
rec = postJSON(t, s, "/v1/fetch", metKey, map[string]string{"url": "https://example.com/"})
|
||||
if rec.Code != 403 {
|
||||
t.Fatalf("metrics fetch: %d", rec.Code)
|
||||
}
|
||||
rec = postJSON(t, s, "/v1/search", metKey, map[string]any{"query": "x"})
|
||||
if rec.Code != 403 {
|
||||
t.Fatalf("metrics search: %d", rec.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMetricsRestrictedNoCallerLabel(t *testing.T) {
|
||||
s, fleetKey, gsKey, _, _ := searchServer(t)
|
||||
q := "secret-restricted-query-42"
|
||||
postJSON(t, s, "/v1/search", fleetKey, map[string]any{"query": "fleet-only-q", "k": 5, "rerank": false})
|
||||
postJSON(t, s, "/v1/search", gsKey, map[string]any{"query": q, "k": 5, "rerank": false})
|
||||
rec := serveMetrics(t, s, loopbackGET("/metrics"))
|
||||
body := rec.Body.String()
|
||||
if rec.Code != 200 {
|
||||
t.Fatal(body)
|
||||
}
|
||||
if strings.Contains(body, q) || strings.Contains(body, "fleet-only-q") {
|
||||
t.Fatal("query in prometheus")
|
||||
}
|
||||
if strings.Contains(body, "example.com") {
|
||||
t.Fatal("host in prometheus")
|
||||
}
|
||||
if strings.Contains(body, `caller="straylight-gs"`) {
|
||||
t.Fatal("restricted caller series")
|
||||
}
|
||||
if strings.Contains(body, "tavily:restricted") {
|
||||
t.Fatal("restricted cap partition")
|
||||
}
|
||||
if !strings.Contains(body, `caller="test"`) {
|
||||
t.Fatal("fleet caller missing")
|
||||
}
|
||||
js := serveMetrics(t, s, loopbackGET("/metrics.json"))
|
||||
var m map[string]any
|
||||
_ = json.Unmarshal(js.Body.Bytes(), &m)
|
||||
queries, _ := m["search_queries"].(map[string]any)
|
||||
if queries["straylight-gs"] == nil {
|
||||
t.Fatalf("json still has restricted caller: %v", m["search_queries"])
|
||||
}
|
||||
}
|
||||
|
||||
func TestMetricsDigestFallbackAndGauges(t *testing.T) {
|
||||
s, _ := setupAPI(t, nil)
|
||||
g := digest.NewGate(1, 0, time.Second)
|
||||
s.Digest = &digest.Engine{Gate: g}
|
||||
ok, _ := g.Acquire(context.Background())
|
||||
if !ok {
|
||||
t.Fatal("hold permit")
|
||||
}
|
||||
s.Digest.Run(context.Background(), "doc", "q", "https://ex.test/p", "fleet")
|
||||
rec := serveMetrics(t, s, loopbackGET("/metrics"))
|
||||
body := rec.Body.String()
|
||||
if !strings.Contains(body, `sift_digest_fallbacks_total{reason="queue_full"}`) &&
|
||||
!strings.Contains(body, `reason="queue_full"`) {
|
||||
t.Fatal(body)
|
||||
}
|
||||
if !strings.Contains(body, "sift_dixie_inflight") || !strings.Contains(body, "sift_dixie_queue_waiters") {
|
||||
t.Fatal(body)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMetricsDoesNotCountScrapes(t *testing.T) {
|
||||
s, key := setupAPI(t, nil)
|
||||
before := serveMetrics(t, s, loopbackGET("/metrics"))
|
||||
serveMetrics(t, s, loopbackGET("/metrics"))
|
||||
after := serveMetrics(t, s, loopbackGET("/metrics"))
|
||||
if before.Body.String() != after.Body.String() && strings.Contains(after.Body.String(), `endpoint="/metrics"`) {
|
||||
t.Fatal(after.Body.String())
|
||||
}
|
||||
postJSON(t, s, "/v2/scrape", key, map[string]string{"url": "http://127.0.0.1/"})
|
||||
rec := serveMetrics(t, s, loopbackGET("/metrics"))
|
||||
if !strings.Contains(rec.Body.String(), `sift_requests_total{code="403",endpoint="/v2/scrape"}`) &&
|
||||
!strings.Contains(rec.Body.String(), `endpoint="/v2/scrape"`) {
|
||||
t.Fatal(rec.Body.String())
|
||||
}
|
||||
if !strings.Contains(rec.Body.String(), "sift_request_duration_seconds") {
|
||||
t.Fatal("histogram missing")
|
||||
}
|
||||
}
|
||||
|
||||
func TestMetricsCacheHitMiss(t *testing.T) {
|
||||
s, key, _, _, _ := searchServer(t)
|
||||
postJSON(t, s, "/v1/search", key, map[string]any{"query": "same q", "k": 5, "rerank": false})
|
||||
postJSON(t, s, "/v1/search", key, map[string]any{"query": "same q", "k": 5, "rerank": false})
|
||||
req := loopbackGET("/metrics.json")
|
||||
rec := httptest.NewRecorder()
|
||||
s.ServeHTTP(rec, req)
|
||||
var m map[string]any
|
||||
_ = json.Unmarshal(rec.Body.Bytes(), &m)
|
||||
if m["search_cache_hit"] == 0 && m["search_cache_miss"] == 0 {
|
||||
t.Fatalf("%v", m)
|
||||
}
|
||||
}
|
||||
|
||||
func mustRead(t *testing.T, path string) string {
|
||||
t.Helper()
|
||||
b, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return string(b)
|
||||
}
|
||||
@@ -133,20 +133,6 @@ func TestHealthzMissingKey(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestMetricsCacheHitMiss(t *testing.T) {
|
||||
s, key, _, _, _ := searchServer(t)
|
||||
postJSON(t, s, "/v1/search", key, map[string]any{"query": "same q", "k": 5, "rerank": false})
|
||||
postJSON(t, s, "/v1/search", key, map[string]any{"query": "same q", "k": 5, "rerank": false})
|
||||
req := httptest.NewRequest(http.MethodGet, "/metrics", nil)
|
||||
rec := httptest.NewRecorder()
|
||||
s.ServeHTTP(rec, req)
|
||||
var m map[string]any
|
||||
_ = json.Unmarshal(rec.Body.Bytes(), &m)
|
||||
if m["search_cache_hit"] == 0 && m["search_cache_miss"] == 0 {
|
||||
t.Fatalf("%v", m)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNoSearchResults422(t *testing.T) {
|
||||
s, key := setupAPI(t, nil)
|
||||
s.Search = &search.Engine{DefaultK: 8}
|
||||
|
||||
@@ -16,12 +16,19 @@ import (
|
||||
"github.com/pelletier/go-toml/v2"
|
||||
)
|
||||
|
||||
const NameMetrics = "metrics"
|
||||
|
||||
type Caller struct {
|
||||
Name string `toml:"name"`
|
||||
Workspace string `toml:"workspace"`
|
||||
KeySHA256 string `toml:"key_sha256"`
|
||||
}
|
||||
|
||||
// MetricsOnly is the scrape-token caller: empty workspace, no fetch/search.
|
||||
func (c *Caller) MetricsOnly() bool {
|
||||
return c != nil && c.Name == NameMetrics
|
||||
}
|
||||
|
||||
type file struct {
|
||||
Caller []Caller `toml:"caller"`
|
||||
}
|
||||
@@ -107,6 +114,9 @@ func GenerateKey() (string, error) {
|
||||
}
|
||||
|
||||
func Add(path, name, workspace, keyOut string) error {
|
||||
if name == NameMetrics {
|
||||
workspace = ""
|
||||
}
|
||||
if _, err := os.Stat(keyOut); err == nil {
|
||||
return fmt.Errorf("refusing to overwrite %s", keyOut)
|
||||
}
|
||||
|
||||
@@ -3,6 +3,7 @@ package auth
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
@@ -69,3 +70,22 @@ func TestAddPreservesKeysMode(t *testing.T) {
|
||||
t.Fatalf("mode %o", st.Mode().Perm())
|
||||
}
|
||||
}
|
||||
|
||||
func TestAddMetricsClearsWorkspace(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
keys := filepath.Join(dir, "keys.toml")
|
||||
out := filepath.Join(dir, "k")
|
||||
if err := Add(keys, NameMetrics, "hermes-fleet", out); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cs, err := List(keys)
|
||||
if err != nil || len(cs) != 1 || cs[0].Name != NameMetrics || cs[0].Workspace != "" {
|
||||
t.Fatalf("%v %v", cs, err)
|
||||
}
|
||||
b, _ := os.ReadFile(out)
|
||||
s, _ := Load(keys)
|
||||
c := s.LookupBearer(strings.TrimSpace(string(b)))
|
||||
if c == nil || !c.MetricsOnly() || c.Workspace != "" {
|
||||
t.Fatalf("%+v", c)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -217,6 +217,15 @@ func (d *DB) PutEmbed(key string, vec []float64) {
|
||||
d.evict()
|
||||
}
|
||||
|
||||
func (d *DB) Stats() (digestH, digestM, embedH, embedM, searchH, searchM, evictions uint64) {
|
||||
if d == nil {
|
||||
return 0, 0, 0, 0, 0, 0, 0
|
||||
}
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
return d.DigestH, d.DigestM, d.EmbedH, d.EmbedM, d.SearchH, d.SearchM, d.Evictions
|
||||
}
|
||||
|
||||
func (d *DB) Size() int64 {
|
||||
if d == nil {
|
||||
return 0
|
||||
|
||||
@@ -154,6 +154,31 @@ func (g *Gate) Release() {
|
||||
g.wakeLocked()
|
||||
}
|
||||
|
||||
func (g *Gate) Snapshot() (inflight, waiters int) {
|
||||
if g == nil {
|
||||
return 0, 0
|
||||
}
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
return g.inflight, len(g.waiters)
|
||||
}
|
||||
|
||||
func (e *Engine) Stats() (calls uint64, fallback map[string]uint64) {
|
||||
if e == nil {
|
||||
return 0, nil
|
||||
}
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
if e.Fallback == nil {
|
||||
return e.Calls, map[string]uint64{}
|
||||
}
|
||||
out := make(map[string]uint64, len(e.Fallback))
|
||||
for k, v := range e.Fallback {
|
||||
out[k] = v
|
||||
}
|
||||
return e.Calls, out
|
||||
}
|
||||
|
||||
type Engine struct {
|
||||
ChatURL string
|
||||
EmbedURL string
|
||||
|
||||
@@ -17,6 +17,8 @@ import (
|
||||
"sync/atomic"
|
||||
"time"
|
||||
"unicode/utf8"
|
||||
|
||||
"git.wntrmute.dev/kyle/sift/internal/auth"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -202,6 +204,9 @@ func (c *Client) CheckMappings(callers []CallerWS) {
|
||||
have := make(map[string]struct{}, len(callers))
|
||||
for _, a := range callers {
|
||||
have[a.Name] = struct{}{}
|
||||
if a.Name == auth.NameMetrics {
|
||||
continue
|
||||
}
|
||||
if permitted(a.Name, c.Permit) && !restricted(a.Workspace) {
|
||||
if c.Log != nil {
|
||||
c.Log.Error("restricted caller mapped off guardian-star", "caller", a.Name, "workspace", a.Workspace)
|
||||
|
||||
@@ -202,4 +202,9 @@ func TestCheckMappingsWarnsMissingKey(t *testing.T) {
|
||||
if strings.Contains(buf.String(), "no key") || strings.Contains(buf.String(), "mapped off") {
|
||||
t.Fatalf("mapped ok still logged: %s", buf.String())
|
||||
}
|
||||
buf.Reset()
|
||||
c.CheckMappings([]CallerWS{{Name: "metrics", Workspace: ""}})
|
||||
if strings.Contains(buf.String(), "mapped off") {
|
||||
t.Fatalf("metrics caller treated as restricted map: %s", buf.String())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,256 @@
|
||||
// Package metrics is sift's Prometheus registry and the loopback-scrape predicate.
|
||||
package metrics
|
||||
|
||||
import (
|
||||
"net"
|
||||
"net/http"
|
||||
"net/netip"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"github.com/prometheus/client_golang/prometheus/promhttp"
|
||||
)
|
||||
|
||||
var durationBuckets = []float64{0.1, 0.5, 1, 2, 5, 10, 15, 20, 30, 45}
|
||||
|
||||
var fallbackOK = map[string]struct{}{
|
||||
"queue_full": {}, "queue_wait": {}, "canceled": {},
|
||||
"embed": {}, "dixie": {}, "timeout": {}, "empty": {}, "unset": {},
|
||||
}
|
||||
|
||||
// Snapshot is process state gathered on each Prometheus scrape.
|
||||
type Snapshot struct {
|
||||
Blocks, Scrapes, SelfIPFailures uint64
|
||||
DigestCalls uint64
|
||||
DigestFallbacks map[string]uint64
|
||||
DixieInflight, DixieWaiters int
|
||||
DigestHit, DigestMiss uint64
|
||||
EmbedHit, EmbedMiss uint64
|
||||
SearchHit, SearchMiss uint64
|
||||
SearchQueries map[string]map[string]uint64 // caller → source
|
||||
SearchCredits map[string]map[string]uint64
|
||||
SearchDaily map[string]uint64
|
||||
SearchCapped map[string]uint64
|
||||
Restricted map[string]struct{}
|
||||
}
|
||||
|
||||
// Source is implemented by api.Server.
|
||||
type Source interface {
|
||||
Snapshot() Snapshot
|
||||
}
|
||||
|
||||
// Set is a private registry of sift_* series.
|
||||
type Set struct {
|
||||
Reg *prometheus.Registry
|
||||
Requests *prometheus.CounterVec
|
||||
Duration *prometheus.HistogramVec
|
||||
handler http.Handler
|
||||
}
|
||||
|
||||
func New(src Source) *Set {
|
||||
reg := prometheus.NewRegistry()
|
||||
requests := prometheus.NewCounterVec(prometheus.CounterOpts{
|
||||
Name: "sift_requests_total",
|
||||
Help: "HTTP API requests by endpoint and status.",
|
||||
}, []string{"endpoint", "code"})
|
||||
duration := prometheus.NewHistogramVec(prometheus.HistogramOpts{
|
||||
Name: "sift_request_duration_seconds",
|
||||
Help: "API wall time (fetch/search) or digest time (digest).",
|
||||
Buckets: durationBuckets,
|
||||
}, []string{"op"})
|
||||
reg.MustRegister(requests, duration, &collector{src: src})
|
||||
return &Set{
|
||||
Reg: reg,
|
||||
Requests: requests,
|
||||
Duration: duration,
|
||||
handler: promhttp.HandlerFor(reg, promhttp.HandlerOpts{EnableOpenMetrics: false}),
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Set) Observe(endpoint string, dur time.Duration, code int) {
|
||||
if s == nil {
|
||||
return
|
||||
}
|
||||
s.Requests.WithLabelValues(endpoint, strconv.Itoa(code)).Inc()
|
||||
switch endpoint {
|
||||
case "/v1/fetch", "/v2/scrape":
|
||||
s.Duration.WithLabelValues("fetch").Observe(dur.Seconds())
|
||||
case "/v1/search", "/v2/search":
|
||||
s.Duration.WithLabelValues("search").Observe(dur.Seconds())
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Set) ObserveDigest(d time.Duration) {
|
||||
if s == nil {
|
||||
return
|
||||
}
|
||||
s.Duration.WithLabelValues("digest").Observe(d.Seconds())
|
||||
}
|
||||
|
||||
func (s *Set) Handler() http.Handler {
|
||||
if s == nil {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
http.Error(w, "metrics unavailable", http.StatusServiceUnavailable)
|
||||
})
|
||||
}
|
||||
return s.handler
|
||||
}
|
||||
|
||||
// LoopbackOpen is curl http://127.0.0.1:8890/metrics. tailscale serve on :8443
|
||||
// is not open: it sets forwarded/Tailscale headers and a non-loopback Host.
|
||||
func LoopbackOpen(r *http.Request) bool {
|
||||
if r == nil {
|
||||
return false
|
||||
}
|
||||
host, _, err := net.SplitHostPort(r.RemoteAddr)
|
||||
if err != nil {
|
||||
host = r.RemoteAddr
|
||||
}
|
||||
addr, err := netip.ParseAddr(host)
|
||||
if err != nil || !addr.IsLoopback() {
|
||||
return false
|
||||
}
|
||||
for _, h := range []string{
|
||||
"X-Forwarded-For", "X-Forwarded-Proto", "X-Forwarded-Host",
|
||||
"Tailscale-User-Login", "Tailscale-Headers-Info",
|
||||
} {
|
||||
if r.Header.Get(h) != "" {
|
||||
return false
|
||||
}
|
||||
}
|
||||
h := r.Host
|
||||
if hh, _, err := net.SplitHostPort(h); err == nil {
|
||||
h = hh
|
||||
}
|
||||
if strings.EqualFold(h, "localhost") {
|
||||
return true
|
||||
}
|
||||
ip, err := netip.ParseAddr(h)
|
||||
return err == nil && ip.IsLoopback()
|
||||
}
|
||||
|
||||
func CoerceFallback(reason string) string {
|
||||
if _, ok := fallbackOK[reason]; ok {
|
||||
return reason
|
||||
}
|
||||
return "other"
|
||||
}
|
||||
|
||||
func skipRestrictedSource(source string) bool {
|
||||
return strings.HasSuffix(source, ":restricted")
|
||||
}
|
||||
|
||||
type collector struct {
|
||||
src Source
|
||||
}
|
||||
|
||||
func (c *collector) Describe(ch chan<- *prometheus.Desc) {
|
||||
ch <- dBlocks
|
||||
ch <- dSelfIP
|
||||
ch <- dDigestCalls
|
||||
ch <- dDigestFB
|
||||
ch <- dDixieIn
|
||||
ch <- dDixieWait
|
||||
ch <- dCache
|
||||
ch <- dQTotal
|
||||
ch <- dQCaller
|
||||
ch <- dCTotal
|
||||
ch <- dCCaller
|
||||
ch <- dDaily
|
||||
ch <- dCapped
|
||||
}
|
||||
|
||||
func (c *collector) Collect(ch chan<- prometheus.Metric) {
|
||||
if c.src == nil {
|
||||
return
|
||||
}
|
||||
s := c.src.Snapshot()
|
||||
ch <- prometheus.MustNewConstMetric(dBlocks, prometheus.CounterValue, float64(s.Blocks))
|
||||
ch <- prometheus.MustNewConstMetric(dSelfIP, prometheus.CounterValue, float64(s.SelfIPFailures))
|
||||
ch <- prometheus.MustNewConstMetric(dDigestCalls, prometheus.CounterValue, float64(s.DigestCalls))
|
||||
fb := map[string]uint64{}
|
||||
for reason, n := range s.DigestFallbacks {
|
||||
fb[CoerceFallback(reason)] += n
|
||||
}
|
||||
for reason, n := range fb {
|
||||
ch <- prometheus.MustNewConstMetric(dDigestFB, prometheus.CounterValue, float64(n), reason)
|
||||
}
|
||||
ch <- prometheus.MustNewConstMetric(dDixieIn, prometheus.GaugeValue, float64(s.DixieInflight))
|
||||
ch <- prometheus.MustNewConstMetric(dDixieWait, prometheus.GaugeValue, float64(s.DixieWaiters))
|
||||
emitCache := func(cache string, hit, miss uint64) {
|
||||
ch <- prometheus.MustNewConstMetric(dCache, prometheus.CounterValue, float64(hit), cache, "hit")
|
||||
ch <- prometheus.MustNewConstMetric(dCache, prometheus.CounterValue, float64(miss), cache, "miss")
|
||||
}
|
||||
emitCache("digest", s.DigestHit, s.DigestMiss)
|
||||
emitCache("embed", s.EmbedHit, s.EmbedMiss)
|
||||
emitCache("search", s.SearchHit, s.SearchMiss)
|
||||
|
||||
sumQ := map[string]uint64{}
|
||||
sumC := map[string]uint64{}
|
||||
for caller, bySrc := range s.SearchQueries {
|
||||
restricted := false
|
||||
if s.Restricted != nil {
|
||||
_, restricted = s.Restricted[caller]
|
||||
}
|
||||
for src, n := range bySrc {
|
||||
if skipRestrictedSource(src) {
|
||||
continue
|
||||
}
|
||||
sumQ[src] += n
|
||||
if !restricted {
|
||||
ch <- prometheus.MustNewConstMetric(dQCaller, prometheus.CounterValue, float64(n), caller, src)
|
||||
}
|
||||
}
|
||||
}
|
||||
for caller, bySrc := range s.SearchCredits {
|
||||
restricted := false
|
||||
if s.Restricted != nil {
|
||||
_, restricted = s.Restricted[caller]
|
||||
}
|
||||
for src, n := range bySrc {
|
||||
if skipRestrictedSource(src) {
|
||||
continue
|
||||
}
|
||||
sumC[src] += n
|
||||
if !restricted {
|
||||
ch <- prometheus.MustNewConstMetric(dCCaller, prometheus.CounterValue, float64(n), caller, src)
|
||||
}
|
||||
}
|
||||
}
|
||||
for src, n := range sumQ {
|
||||
ch <- prometheus.MustNewConstMetric(dQTotal, prometheus.CounterValue, float64(n), src)
|
||||
}
|
||||
for src, n := range sumC {
|
||||
ch <- prometheus.MustNewConstMetric(dCTotal, prometheus.CounterValue, float64(n), src)
|
||||
}
|
||||
for src, n := range s.SearchDaily {
|
||||
if skipRestrictedSource(src) {
|
||||
continue
|
||||
}
|
||||
ch <- prometheus.MustNewConstMetric(dDaily, prometheus.GaugeValue, float64(n), src)
|
||||
}
|
||||
for src, n := range s.SearchCapped {
|
||||
if skipRestrictedSource(src) {
|
||||
continue
|
||||
}
|
||||
ch <- prometheus.MustNewConstMetric(dCapped, prometheus.CounterValue, float64(n), src)
|
||||
}
|
||||
}
|
||||
|
||||
var (
|
||||
dBlocks = prometheus.NewDesc("sift_guard_blocks_total", "SSRF/guard rejections.", nil, nil)
|
||||
dSelfIP = prometheus.NewDesc("sift_self_ip_failures_total", "Self-IP discovery failures.", nil, nil)
|
||||
dDigestCalls = prometheus.NewDesc("sift_digest_calls_total", "Dixie digest calls started (excludes cache hits).", nil, nil)
|
||||
dDigestFB = prometheus.NewDesc("sift_digest_fallbacks_total", "Digest fallbacks by reason.", []string{"reason"}, nil)
|
||||
dDixieIn = prometheus.NewDesc("sift_dixie_inflight", "Sift-side Dixie permits in use.", nil, nil)
|
||||
dDixieWait = prometheus.NewDesc("sift_dixie_queue_waiters", "Waiters in the sift-side Dixie FIFO.", nil, nil)
|
||||
dCache = prometheus.NewDesc("sift_cache_requests_total", "Cache lookups.", []string{"cache", "result"}, nil)
|
||||
dQTotal = prometheus.NewDesc("sift_search_queries_total", "Search queries by source (all callers).", []string{"source"}, nil)
|
||||
dQCaller = prometheus.NewDesc("sift_search_queries_caller_total", "Search queries by fleet caller and source.", []string{"caller", "source"}, nil)
|
||||
dCTotal = prometheus.NewDesc("sift_search_credits_total", "Search credits by source (all callers).", []string{"source"}, nil)
|
||||
dCCaller = prometheus.NewDesc("sift_search_credits_caller_total", "Search credits by fleet caller and source.", []string{"caller", "source"}, nil)
|
||||
dDaily = prometheus.NewDesc("sift_search_daily_credits", "UTC-day paid-source credits (fleet partition).", []string{"source"}, nil)
|
||||
dCapped = prometheus.NewDesc("sift_search_capped_total", "Times a paid source was skipped for the daily cap.", []string{"source"}, nil)
|
||||
)
|
||||
@@ -0,0 +1,41 @@
|
||||
package metrics
|
||||
|
||||
import (
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestLoopbackOpen(t *testing.T) {
|
||||
req := httptest.NewRequest("GET", "/metrics", nil)
|
||||
if LoopbackOpen(req) {
|
||||
t.Fatal("httptest default must not be open")
|
||||
}
|
||||
req.RemoteAddr = "127.0.0.1:9"
|
||||
req.Host = "127.0.0.1:8890"
|
||||
if !LoopbackOpen(req) {
|
||||
t.Fatal("loopback host+addr")
|
||||
}
|
||||
req.Header.Set("X-Forwarded-For", "100.1.2.3")
|
||||
if LoopbackOpen(req) {
|
||||
t.Fatal("forwarded-for is :8443")
|
||||
}
|
||||
req.Header.Del("X-Forwarded-For")
|
||||
req.Host = "orion.scylla-hammerhead.ts.net:8443"
|
||||
if LoopbackOpen(req) {
|
||||
t.Fatal("tailnet Host is :8443")
|
||||
}
|
||||
req.Host = "localhost:8890"
|
||||
req.RemoteAddr = "[::1]:9"
|
||||
if !LoopbackOpen(req) {
|
||||
t.Fatal("localhost v6")
|
||||
}
|
||||
}
|
||||
|
||||
func TestCoerceFallback(t *testing.T) {
|
||||
if CoerceFallback("queue_full") != "queue_full" {
|
||||
t.Fatal("keep")
|
||||
}
|
||||
if CoerceFallback("blocked: resolve evil.example") != "other" {
|
||||
t.Fatal("raw must not pass")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user