import "github.com/kinorai/omnifeed/internal/observability"
Package observability wires structured logging, Prometheus metrics, and Kubernetes-style health endpoints.
Server names, the values of the omnifeed_http_* `server` label.
const (
ServerLoader = "loader"
ServerMCP = "mcp"
ServerObs = "obs"
)
Cache results as the omnifeed_requests_total `cache` label. CacheOff means the response cache is disabled for this deployment.
const (
CacheHit = "hit"
CacheMiss = "miss"
CacheBypass = "bypass"
CacheOff = "off"
)
Panic sites, the values of omnifeed_panics_total’s `where` label. A fixed set: every recover point in the binary names one of these.
const (
PanicHTTP = "http" // an HTTP handler (any server)
PanicMCPTool = "mcp_tool" // an MCP tool handler (stdio or HTTP)
PanicLoaderCrawl = "loader_crawl" // one URL of a POST /crawl batch
PanicGitHubFanout = "github_fanout" // one parallel GitHub API call
PanicBackground = "background" // a background loop (dependency checks, SearXNG init)
)
RequestIDHeader is the header a caller may send to name its request, and the header every response carries back, so one id joins the caller’s logs, omnifeed’s logs and the MCP result (_meta.request_id).
const RequestIDHeader = "X-Request-Id"
RouteOther is the route label of every path outside the fixed set.
const RouteOther = "other"
TracerName is the instrumentation scope of omnifeed’s own spans.
const TracerName = "github.com/kinorai/omnifeed"
func Explain(err error) string
Explain renders a failed crawl/search error as a short, caller-safe explanation: the classified reason, the upstream HTTP status when the error carries one, and the root cause with internal endpoints redacted. The back-off (FetchError.RetryAfter) is not repeated here: each transport reports it once, from Classify (MCP: “Retryable after Ns.” plus retry_after_s in structuredContent; loader/REST: retry_after_s fields). Transports prefix their own context (“fetch_url failed: “ + Explain(err)) so an MCP or HTTP client sees what metrics already know instead of an opaque failure. Returns “” only when err is nil.
func NewLogger(level, format string) *slog.Logger
| NewLogger returns a slog.Logger configured per the given level and format. Level: debug | info | warn | error (default info). Format: json | text (default json). |
func NewQueryID() string
NewQueryID returns a correlation id for one search. It only has to be unique among the searches a log store holds at once and is never a secret; crypto/rand costs nothing here and keeps the scanners quiet.
func NewRequestID() string
NewRequestID returns a fresh random request id (16 bytes, hex).
func ObservePanic(where string, v any)
ObservePanic counts and logs one recovered panic. v is what recover() returned. Callers that recover themselves (middleware that must also write a 500) use this; everyone else defers Recover.
func QueryIDFrom(ctx context.Context) string
QueryIDFrom returns the search correlation id carried by ctx, or “”.
func Reason(err error) string
Reason maps a crawl error to the bounded `reason` label recorded on omnifeed_requests_total, so metrics and alerts can tell a 403 from a CAPTCHA from a timeout. It reads the classified cause from a domain.FetchError that the engines attach as data — no error-string parsing. Errors that aren’t a FetchError fall back to canceled (caller abort), timeout (context deadline), or “error”.
func Recover(where string, errp *error)
Recover is deferred at the top of a goroutine (or a call that must not take the process down): it recovers a panic, counts it in omnifeed_panics_total, logs it with its stack and, when errp is non-nil, turns it into an error.
defer observability.Recover(observability.PanicGitHubFanout, &err)
It must be deferred directly — recover() only stops a panic when called by the deferred function itself.
func RegisterPprof(mux *http.ServeMux)
RegisterPprof attaches /debug/pprof/* to mux. Opt-in via OMNIFEED_ENABLE_PPROF.
func RequestID(next http.Handler) http.Handler
RequestID is the HTTP middleware that adopts the caller’s X-Request-Id when it is valid (see ValidRequestID) or generates one, puts it in the request context and returns it in the response header.
func RequestIDFrom(ctx context.Context) string
RequestIDFrom returns the request id carried by ctx, or “”.
func RouteOf(path string) string
RouteOf maps a request path to its bounded route label.
func ScrubText(s string) string
ScrubText replaces every absolute URL in s with its scheme and host.
func SetupTracing(ctx context.Context, cfg TracingConfig) (shutdown func(context.Context) error, enabled bool, err error)
SetupTracing installs an OTLP/HTTP tracer provider (parent-based, always-on sampling) and the W3C trace-context propagator as the globals. With an empty Endpoint it installs nothing and returns enabled=false and a no-op shutdown. shutdown flushes buffered spans; call it on exit.
The exporter reads the rest of the standard OTEL_EXPORTER_OTLP_* variables (headers, timeout, …) itself. Every exported span passes through scrubExporter first, which removes full URLs, queries and paths: the HTTP instrumentation records them by default, and they carry the caller’s URLs.
func SpanError(span trace.Span, err error)
SpanError marks span failed with err (scrubbed at export). A nil err is a no-op, so callers can defer it unconditionally.
func StatusOf(err error) string
StatusOf reports the coarse request status recorded alongside Reason on the crawl/search metrics: “ok” when err is nil, “error” otherwise.
func Tracer() trace.Tracer
Tracer returns omnifeed’s tracer from the global provider: a no-op until SetupTracing installs a real one, so instrumented code costs ~nothing when tracing is off.
func ValidRequestID(id string) bool
ValidRequestID reports whether a caller-supplied id is safe to adopt: 1-128 characters from [A-Za-z0-9._:-]. Anything else (empty, too long, spaces, quotes, control bytes) is replaced by a generated id rather than echoed into logs and response headers.
func WithQueryID(ctx context.Context, id string) context.Context
WithQueryID returns ctx carrying the search correlation id (query_id) the searcher should use for its audit lines, so the caller can log the same id on its own completion line.
func WithRequestID(ctx context.Context, id string) context.Context
WithRequestID returns ctx carrying id as the request id.
ContextHandler decorates a slog.Handler with the request-scoped fields the context carries: request_id (see RequestID) and, when a span is recording, trace_id and span_id — so a log line joins its request and its trace. Only the *Context logging calls (InfoContext, WarnContext, …) pass a context; the plain ones log without these fields, exactly as before.
type ContextHandler struct {
// contains filtered or unexported fields
}
func NewContextHandler(inner slog.Handler) *ContextHandler
NewContextHandler wraps inner.
func (h *ContextHandler) Enabled(ctx context.Context, l slog.Level) bool
Enabled implements slog.Handler.
func (h *ContextHandler) Handle(ctx context.Context, r slog.Record) error
Handle implements slog.Handler.
func (h *ContextHandler) WithAttrs(attrs []slog.Attr) slog.Handler
WithAttrs implements slog.Handler.
func (h *ContextHandler) WithGroup(name string) slog.Handler
WithGroup implements slog.Handler.
DependencyCheck probes one dependency; nil means up.
type DependencyCheck func(ctx context.Context) error
DependencyChecker probes each configured dependency on an interval and publishes omnifeed_dependency_up{dependency}. It replaces gating /readyz on crawl4ai and SearXNG: a dependency outage fails the calls that need it (and says so, classified), while the pod stays in rotation for everything else — the Reddit quota, the GitHub/HN/Bluesky engines and the cache keep serving.
type DependencyChecker struct {
Checks map[string]DependencyCheck
Interval time.Duration // between rounds; defaults to 30s
Timeout time.Duration // per check; defaults to 5s
Metrics *Metrics
Logger *slog.Logger
// contains filtered or unexported fields
}
func (d *DependencyChecker) CheckOnce(ctx context.Context)
CheckOnce runs every check once and publishes the results.
func (d *DependencyChecker) Run(ctx context.Context)
Run checks immediately, then every Interval, until ctx ends.
ErrorCode is the stable, caller-facing error code every transport reports for a failed fetch/search (MCP structuredContent.code, the Open WebUI loader’s error metadata, the REST error body). It is deliberately coarser than FailureKind: FailureKind is the metric taxonomy and may grow, while these codes are a contract clients branch on. Documented in docs/errors.md.
type ErrorCode string
The complete set of caller-facing error codes.
const (
CodeRateLimited ErrorCode = "rate_limited"
CodeBlocked ErrorCode = "blocked"
CodeCaptcha ErrorCode = "captcha"
CodeTimeout ErrorCode = "timeout"
CodeThinContent ErrorCode = "thin_content"
CodeNotFound ErrorCode = "not_found"
CodeUpstreamError ErrorCode = "upstream_error"
CodeQuotaExhausted ErrorCode = "quota_exhausted"
CodeInvalidRequest ErrorCode = "invalid_request"
CodeDNSBlocked ErrorCode = "dns_blocked"
)
Failure is the caller-facing classification of a failed call.
type Failure struct {
Code ErrorCode
// Retryable says whether the same call may succeed if repeated later
// (after RetryAfter, when set). False means retrying is wasted work.
Retryable bool
// RetryAfter is the back-off the upstream or omnifeed's own pacing asked
// for; 0 when unknown.
RetryAfter time.Duration
// UpstreamStatus is the upstream HTTP status, 0 when none applies.
UpstreamStatus int
}
func Classify(err error) Failure
Classify maps a failed call’s error onto the caller-facing Failure. A domain.InvalidRequestError anywhere in the chain is the caller’s own mistake (invalid_request, not retryable); otherwise the FailureKind (see Reason) decides. Returns the zero Failure for a nil err.
func (f Failure) Fields() map[string]any
Fields renders the Failure as the machine-readable object transports embed in an error response: code and retryable always, retry_after_s and upstream_status only when known. Callers add their own context (url, …).
func (f Failure) RetryAfterSeconds() int
RetryAfterSeconds renders RetryAfter as whole seconds, rounded UP (a caller told “retry in 12s” for a 12.4s wait would be refused again); 0 when unknown.
Health exposes Kubernetes-style health endpoints.
/livez — process is up; always 200 unless we're shutting down
/readyz — runs all ReadyCheck functions; 200 only if all pass
/healthz — alias of /readyz for backwards compat
type Health struct {
// contains filtered or unexported fields
}
func NewHealth(cacheTTL time.Duration, checks ...ReadyCheck) *Health
NewHealth returns a Health with the given readiness checks. Results are cached for `cacheTTL` to avoid hammering upstreams on every probe.
func (h *Health) MarkShuttingDown()
MarkShuttingDown causes /livez to return 503 so the LB stops sending traffic.
func (h *Health) Register(mux *http.ServeMux)
Register attaches the endpoints to mux.
Metrics holds Prometheus collectors emitted by the proxy.
type Metrics struct {
RequestsTotal *prometheus.CounterVec // engine, tenant, status, reason, cache
RequestAttempts *prometheus.CounterVec // upstream, attempt (first|retry)
RequestSecs *prometheus.HistogramVec // engine, status, reason
UpstreamSecs *prometheus.HistogramVec // upstream, op, status
LimiterWaitSecs *prometheus.HistogramVec // engine
RatelimitErrors *prometheus.CounterVec // op
RatelimitPenalties *prometheus.CounterVec // upstream
RatelimitDegraded *prometheus.GaugeVec // scope
ResponseChars *prometheus.HistogramVec // engine
EngineFallbacks *prometheus.CounterVec // from_engine, reason
SearxngUnresponsive *prometheus.CounterVec // engine, error
SearxngEngineHits *prometheus.CounterVec // engine
SearxngEmpty *prometheus.CounterVec // scoped
SearxngQueries *prometheus.CounterVec // scoped
SearxngEngineZero *prometheus.CounterVec // engine
RedditRounds prometheus.Histogram
RedditMethodCalls *prometheus.CounterVec // method, outcome
RedditMethodUp *prometheus.GaugeVec // method
RedditMethodLastSuccess *prometheus.GaugeVec // method
RedditRemaining *prometheus.GaugeVec // method
RedditSessionExpiry prometheus.Gauge
RedditInPageRequests *prometheus.CounterVec // kind
SearchesTotal *prometheus.CounterVec // searcher, status, reason, scoped
SearchSecs *prometheus.HistogramVec // searcher, status
SearchFallbacks *prometheus.CounterVec // from, to, trigger, reason, outcome
SearchEnginePos *prometheus.HistogramVec // engine
SearchEngineUnique *prometheus.CounterVec // engine
CacheRequests *prometheus.CounterVec // result
CacheBytes prometheus.Gauge
CacheBackend *prometheus.GaugeVec // backend
CacheBackendErrors *prometheus.CounterVec // op, kind
HTTPRequests *prometheus.CounterVec // server, route, code
HTTPSecs *prometheus.HistogramVec // server, route
HTTPInFlight *prometheus.GaugeVec // server
MCPToolCalls *prometheus.CounterVec // tool, outcome, client
MCPToolSecs *prometheus.HistogramVec // tool, outcome
DependencyUp *prometheus.GaugeVec // dependency
BuildInfo *prometheus.GaugeVec // version
// contains filtered or unexported fields
}
func NewMetrics() *Metrics
NewMetrics builds and registers all collectors.
func (m *Metrics) HTTPMiddleware(server string, next http.Handler) http.Handler
HTTPMiddleware records RED metrics for one server — requests by route and status code, latency by route, requests in flight — and recovers a handler panic: it is counted (omnifeed_panics_total{where=”http”}), logged with its stack, and answered with a 500 when nothing was written yet, instead of net/http’s bare connection reset (after a partial write, the connection is aborted instead, so a truncated body never passes for a complete one).
func (m *Metrics) InitSearxngEngine(engine string, errorTypes []string)
InitSearxngEngine creates every per-engine search series for engine at zero: results, zero-result searches, unique results, the rank histogram, and the unresponsive counter for each of errorTypes. increase() and rate() cannot count a counter’s first sample, so a series born at its first value hides that value — after each restart, a pod’s first results per engine. Minted at startup, the series already sits at 0 when the first search lands. Idempotent.
func (m *Metrics) Observe(engine, tenant, status, reason, cache string, duration time.Duration)
Observe records a single crawl result. reason is a bounded classification of WHY a request failed (see Reason); it is “ok” on success. cache is the response-cache result (CacheHit, CacheMiss, CacheBypass, CacheOff). A hit is counted but not timed: it never reached an engine, and its microseconds would make the latency histogram claim the engines got faster as the hit ratio rose.
func (m *Metrics) ObserveAttempt(upstream string, retry bool)
ObserveAttempt records one HTTP attempt from the retrying client against the named upstream. retry is false for the first try and true for each retry, so attempt=”retry” counts the re-drives that #2’s RetryableStatus veto removes for non-transient blocks.
func (m *Metrics) ObserveCache(result string)
| ObserveCache counts one fetch_url response-cache lookup by result (hit | miss | bypass | error). |
func (m *Metrics) ObserveCacheBackendError(op, kind string)
| ObserveCacheBackendError counts one response-cache Redis error by op (get | set | probe) and kind (noperm | transient | other). |
func (m *Metrics) ObserveEmptySearch(scoped bool)
ObserveEmptySearch counts a search that came back with no results and no failure report — the shape a silently blocked pool produces, and the shape an honest zero-hit query produces. scoped says whether a `site:` filter was applied, because the site-scoped path is where silent blocks concentrate.
func (m *Metrics) ObserveEngineRank(engine string, rank int, unique bool)
ObserveEngineRank records the rank an engine gave one of its own results, and whether that result was unique to it. Both are aggregates only — the URL and the query stay in the audit log, never in a label.
func (m *Metrics) ObserveEngineResults(engine string, rows int)
ObserveEngineResults counts the rows one SearXNG engine contributed to a search response. Engines that block silently — HTTP 200, no error, empty results — are invisible in every other metric, and this is what makes them visible: their series stops advancing while the rest of the pool continues.
func (m *Metrics) ObserveEngineZeroResults(engine string)
ObserveEngineZeroResults counts one search an engine contributed nothing to while the search itself returned rows. One increment per search per engine, never per row: it measures how OFTEN an engine sits out, not how much it missed.
func (m *Metrics) ObserveFallback(fromEngine, reason string)
ObserveFallback counts one engine→generic-fallback handoff: fromEngine is the dedicated engine that failed, reason its classified failure (see Reason).
func (m *Metrics) ObserveLimiterWait(engine, outcome string, duration time.Duration)
ObserveLimiterWait records the time an engine spent blocked acquiring the per-domain limiter. Wired as the DomainLimiter’s OnWait hook. outcome is “acquired”, “canceled” (the wait died in the queue) or “budget_exceeded” (the wait was longer than the caller’s budget, so the limiter refused to queue).
func (m *Metrics) ObserveMCPToolCall(tool, outcome, client string, duration time.Duration)
ObserveMCPToolCall records one MCP tools/call: tool and client must already be bounded (known tool names; ClientLabel output).
func (m *Metrics) ObserveRatelimitBackendError(op string)
ObserveRatelimitBackendError counts one failed Redis operation in the distributed limiter. op is “acquire” or “release”. Wired as the redis limiter’s OnError hook.
func (m *Metrics) ObserveRatelimitPenalty(upstream string)
ObserveRatelimitPenalty counts one upstream Retry-After fed back into the limiters as a hold on that host. upstream is the client’s own upstream label, never the host — hosts are unbounded.
func (m *Metrics) ObserveResponseChars(engine string, chars int)
ObserveResponseChars records the character count of a successful crawl’s extracted content, PRE-truncation (the engine’s output, before any transport max_chars clipping) — the quality guard for scrape-option changes. Recorded at the registry choke point so the label names the engine that actually produced the document, fallbacks included.
func (m *Metrics) ObserveSearch(searcher, status, reason string, scoped bool, duration time.Duration)
ObserveSearch records a single search query result. reason classifies WHY a search failed (see Reason); it is “ok” on success. scoped says whether the caller asked for a `site:` filter — site-scoped search runs on a different engine pool and fails for different reasons, and every existing sum() over this counter collapses the label anyway. SearchSecs stays keyed on searcher+status only — adding reason or scoped would just inflate histogram cardinality.
func (m *Metrics) ObserveSearchFallback(from, to, trigger, reason, outcome string)
ObserveSearchFallback counts one search a primary searcher handed to its backup: trigger says why (see failover.Trigger), reason the primary’s classified failure (“ok” unless trigger is error — quota_exhausted there is omnifeed’s own pacing, not an upstream fault), outcome how the backup did (“served”, “empty” when it found nothing, “failed”, “canceled” when the caller hung up while the backup ran).
func (m *Metrics) ObserveSearxngQuery(scoped bool)
ObserveSearxngQuery counts one query actually sent to SearXNG — past the limiter, before the HTTP call. This is the rate the engines behind SearXNG see, which is what their own quotas are measured against; the caller-side counter (ObserveSearch) includes queries pacing refused.
func (m *Metrics) ObserveUnresponsiveEngine(engine, errType string)
ObserveUnresponsiveEngine counts one engine SearXNG reported unresponsive on a search response, by engine name and error type.
func (m *Metrics) ObserveUpstream(upstream, op, status string, duration time.Duration)
ObserveUpstream records one upstream HTTP round-trip attempt (request start until the response body is fully read, or transport error). status is “ok” for 2xx and “error” otherwise. Wired as the httpx client’s OnUpstream hook.
func (m *Metrics) RegisterMetrics(mux *http.ServeMux)
RegisterMetrics attaches /metrics to mux.
func (m *Metrics) SetBuildInfo(version string)
SetBuildInfo publishes omnifeed_build_info{version} = 1.
func (m *Metrics) SetCacheBackend(name string)
| SetCacheBackend marks name (redis | memory) as the response cache’s active backend. |
func (m *Metrics) SetCacheBytes(n int)
SetCacheBytes publishes the in-process response cache’s approximate size.
func (m *Metrics) SetDependencyUp(dependency string, up bool)
SetDependencyUp records the result of one periodic dependency check.
func (m *Metrics) SetRatelimitDegraded(scope string, down bool)
SetRatelimitDegraded records whether pacing is currently degraded to per-pod limits because the Redis backend is unreachable. Wired as the FallbackLimiter’s OnDegraded hook, which fires on transitions only.
Keyed by scope (“domain” or “searxng”): each scope has its own FallbackLimiter and degrades on its own, so one recovering must not zero the other’s signal.
PrintfLogger routes a library’s Printf-style logger into slog, so its lines share the app’s format (JSON in production) and carry a level. go-redis is the case: without redis.SetLogger it prints “redis: 2026/10/26 … pool.go:” text lines to stderr, which log collectors store with no level. Its Printf(ctx, format, v…) signature matches go-redis’s Logging interface, so this package does not import go-redis.
type PrintfLogger struct {
// contains filtered or unexported fields
}
func NewPrintfLogger(l *slog.Logger, component string) *PrintfLogger
NewPrintfLogger logs through l, tagged with component.
func (p *PrintfLogger) Printf(ctx context.Context, format string, v ...any)
Printf logs at warn. go-redis passes no level: most calls are errors it retries or recovers from (dial failures, pool and push-notification errors, recovered panics), a few are state changes (sentinel failover, ring shards).
ReadyCheck is a function that reports whether the binary is ready to serve. Return nil when ready, an error describing why not otherwise.
type ReadyCheck func(context.Context) error
TracingConfig configures SetupTracing. Endpoint is OTEL_EXPORTER_OTLP_ENDPOINT (a base URL; the exporter appends /v1/traces); empty disables tracing.
type TracingConfig struct {
Endpoint string
ServiceName string
Version string
}
Generated by gomarkdoc