diff --git a/chart/templates/network-policy.yaml b/chart/templates/network-policy.yaml index c6573e7..03731e9 100644 --- a/chart/templates/network-policy.yaml +++ b/chart/templates/network-policy.yaml @@ -202,6 +202,32 @@ spec: - port: 443 protocol: TCP --- +# Agent metrics ingress: the runner extension serves Prometheus metrics on the +# pod's named "metrics" port (see chart/templates/k8s-ai-agent-operator/ +# podmonitor-agents.yaml). Without an ingress allow, the default-deny policy +# silently refuses Prometheus scrapes and every agent metric (notably +# agent_tokens_used_total) disappears from the observability dashboard. +# Selector matches the PodMonitor (managed-by, not component) so pods created +# before the component label existed are covered too. +apiVersion: networking.k8s.io/v1 +kind: NetworkPolicy +metadata: + name: agent-metrics + namespace: {{ .Values.namespace }} + labels: + {{- include "ainsel.labels" . | nindent 4 }} + app.kubernetes.io/component: agents +spec: + podSelector: + matchLabels: + app.kubernetes.io/managed-by: agent-operator + policyTypes: + - Ingress + ingress: + # Allow metrics scraping (same open posture as the hub-backend metrics rule). + - ports: + - port: metrics +--- # Webhook ingress for connector receiver pods. Connectors are created # dynamically by the event-source-gateway operator; external webhooks reach # them through the ingress controller, and in-cluster producers (e.g. a diff --git a/docs/network-policies.md b/docs/network-policies.md index 487c54c..3c398a6 100644 --- a/docs/network-policies.md +++ b/docs/network-policies.md @@ -54,6 +54,7 @@ All policies live in the release namespace (`.Values.namespace`). | `postgres` | `component: postgres` | Ingress | `hub-backend` only, port 5432 | | `qdrant` | `component: qdrant` | Ingress | agent pods (`component: agents`) and `hub-backend` pods, ports 6333/6334 | | `agent-egress` | `component: agents` | Egress | qdrant (6333), hub-backend (8080), DNS (53), any destination on 443/TCP (LLM APIs; FQDN scoping tracked in #652) | +| `agent-metrics` | `managed-by: agent-operator` | Ingress | any source, named port `metrics` (Prometheus scrapes of `agent_tokens_used_total` etc.; selector matches the `ainsel-agents` PodMonitor) | | `connectors-webhook-ingress` | `managed-by: connector-operator` | Ingress | ingress controller's namespace (external webhooks) plus any peers listed in `networkPolicy.connectorWebhookSources`, port `http` | Notes: @@ -96,6 +97,7 @@ These are the flows that break most often when a policy is missing: | external → connector | webhook delivery through the ingress controller | chart `connectors-webhook-ingress` (ingressNamespace) | | in-cluster producer → connector | e.g. a Forgejo in the cluster delivering directly via the `*-webhook` service | chart `connectors-webhook-ingress` (`connectorWebhookSources`) | | agent → qdrant | vector memory | chart `qdrant` + `agent-egress` | +| Prometheus → agent pods | scraping `agent_tokens_used_total` and other runner metrics | chart `agent-metrics` policy (named port `metrics`) | | **hub-backend → MCP servers** | "Refresh MCP Tools" on AgentImages calls every configured MCP server directly from hub-backend to discover tools; the mcp gateway proxies MCP traffic | **not covered by the chart** for in-cluster servers outside the release — see below | | agent → MCP servers | tools declared in the AgentImage | chart `agent-egress` covers 443/TCP; in-cluster servers on other ports need extension policies | diff --git a/pi/pi-extensions/ainsel-runner/index.ts b/pi/pi-extensions/ainsel-runner/index.ts index fdd7582..edee190 100644 --- a/pi/pi-extensions/ainsel-runner/index.ts +++ b/pi/pi-extensions/ainsel-runner/index.ts @@ -1062,6 +1062,10 @@ async function processTask( nak_delay_ms: NAK_DELAY_MS, task_id: task.id, }); + // Report the failure to the hub as an error-level task log so it + // appears in the observability errors feed. Console logError output + // never leaves the pod; without this the errors view stays empty. + await postTaskLog(hubUrl, token, agentName, "error", `Event failed: ${errMsg}`, correlationId, invocationId, { event_type: evCtx.type, task_id: task.id, duration_ms: durationMs }); } } diff --git a/services/hub/internal/api/handlers_observability.go b/services/hub/internal/api/handlers_observability.go index ddfe95a..5a35db4 100644 --- a/services/hub/internal/api/handlers_observability.go +++ b/services/hub/internal/api/handlers_observability.go @@ -3,6 +3,7 @@ package api import ( "context" "fmt" + "math" "net/http" "sort" "strings" @@ -11,6 +12,7 @@ import ( agentv1alpha1 "github.com/DominikPinsel/ainsel/shared/api/api/v1alpha1" "github.com/DominikPinsel/ainsel/services/hub/internal/prometheus" + "github.com/DominikPinsel/ainsel/services/hub/internal/tasklogs" "sigs.k8s.io/controller-runtime/pkg/client" ) @@ -58,6 +60,12 @@ var hubMetrics = []hubMetric{ } // MetricsSummary holds the current values of the hub-internal counters. +// +// RoutingErrors is named for its original source (hub_routing_errors_total), +// but with a range set it reports error-level task logs within the window — +// the same entries the errors page lists. The routing-errors counter only +// tracks router failures and was effectively always zero, which left the +// Errors KPI card stuck at 0 even when events were failing. type MetricsSummary struct { EventsConsumed float64 `json:"eventsConsumed"` TriggersMatched float64 `json:"triggersMatched"` @@ -293,6 +301,19 @@ func (s *Server) getMetricsSummary(w http.ResponseWriter, r *http.Request) { summary := MetricsSummary{UpdatedAt: time.Now().UTC()} for _, m := range hubMetrics { + // With a range set, the Errors card counts error-level task logs in + // the window (what the errors page lists) instead of the + // hub_routing_errors_total counter. The counter remains the source on + // the legacy range-less path and when no log store is configured. + if m.Name == "routing_errors" && rng != nil && s.taskLogs != nil { + count, err := s.taskLogs.CountByLevelSince(r.Context(), tasklogs.LevelError, time.Now().UTC().Add(-rng.Duration)) + if err != nil { + writeError(w, http.StatusBadGateway, fmt.Sprintf("failed to count error task logs: %s", err.Error())) + return + } + summary.RoutingErrors = float64(count) + continue + } query := m.PromQL if rng != nil { // Use increase() over the requested range so the card shows @@ -306,6 +327,10 @@ func (s *Server) getMetricsSummary(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadGateway, fmt.Sprintf("failed to query %s: %s", m.Name, err.Error())) return } + // These metrics are event counts. increase() extrapolates fractional + // values (e.g. 64.7826), which surfaced on the dashboard KPI cards as + // long decimals; round to the nearest whole count before returning. + val = math.Round(val) switch m.Name { case "events_consumed": summary.EventsConsumed = val diff --git a/services/hub/internal/api/handlers_observability_test.go b/services/hub/internal/api/handlers_observability_test.go index ab72986..9f2cdc3 100644 --- a/services/hub/internal/api/handlers_observability_test.go +++ b/services/hub/internal/api/handlers_observability_test.go @@ -592,6 +592,45 @@ func TestObservability_SummaryWithRangeUsesIncreaseQueries(t *testing.T) { } } +func TestObservability_SummaryWithRangeRoundsFractionalCounts(t *testing.T) { + // increase() extrapolates fractional values (e.g. 64.7826); the summary + // reports event counts, so values must be rounded to whole numbers before + // reaching the dashboard KPI cards. + srv := fakePromServer(t, func(path string, params url.Values) interface{} { + q := params.Get("query") + switch { + case strings.Contains(q, "hub_events_consumed_total"): + return vectorResponse([]vectorSample{{Labels: map[string]string{}, Value: "64.7826"}}) + case strings.Contains(q, "hub_triggers_matched_total"): + return vectorResponse([]vectorSample{{Labels: map[string]string{}, Value: "12.3"}}) + case strings.Contains(q, "hub_events_routed_total"): + return vectorResponse([]vectorSample{{Labels: map[string]string{}, Value: "0.4"}}) + case strings.Contains(q, "hub_routing_errors_total"): + return vectorResponse([]vectorSample{{Labels: map[string]string{}, Value: "1.5"}}) + } + t.Fatalf("unexpected query: %s", q) + return nil + }) + defer srv.Close() + + s := newServerWithProm(t, prometheus.NewClient(srv.URL, nil)) + + req := httptest.NewRequest(http.MethodGet, "/api/v1/observability/metrics/summary?range=24h", nil) + rec := httptest.NewRecorder() + s.mux.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", rec.Code, rec.Body.String()) + } + var body MetricsSummary + if err := json.NewDecoder(rec.Body).Decode(&body); err != nil { + t.Fatalf("decode: %v", err) + } + if body.EventsConsumed != 65 || body.TriggersMatched != 12 || body.EventsRouted != 0 || body.RoutingErrors != 2 { + t.Errorf("expected rounded counts, got %+v", body) + } +} + func TestObservability_SummaryRejectsInvalidRange(t *testing.T) { srv := fakePromServer(t, func(string, url.Values) interface{} { return nil }) defer srv.Close()