feat: K8S API client integration

Author: zeroc0d3

CHANGELOG.md

@@ -107,6 +107,44 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   - `workload_generations: true` — Deployment/StatefulSet observed vs desired generation drift
 - **`collectors.remote_write_receiver` config section**: Added to all config files (`tfo-agent.yaml`, `tfo-agent.default.yaml`, `tfo-agent-one-for-all.yaml`), default: `enabled: false`, `port: 9091`
 - **Apache License 2.0 headers**: Full license boilerplate + package documentation added to all 187 `.go` files across all packages; property-based test files previously missing headers now covered
+- **K8s NetworkPolicy Collector** (`internal/collector/kubernetes/network_policies.go`): Full NetworkPolicy resource collection with ingress/egress rule detail
+  - Gathers all NetworkPolicy resources across namespaces; respects `shouldCollectNamespace` filter
+  - Extracts policy types (Ingress/Egress), pod selectors, ingress/egress rule counts
+  - Parses ingress rules: ports (protocol, port), peers (podSelector, namespaceSelector, IPBlock with CIDR and except ranges)
+  - Parses egress rules: same structure as ingress — ports + to-peers with full IPBlock support
+  - Emits `k8s.networkpolicy.count` gauge metric per namespace with cluster label
+  - New data types: `NetworkPolicyState`, `NetworkPolicyRule`, `NetworkPolicyPort`, `NetworkPolicyPeer`, `NetworkPolicyIPBlock`
+  - Added to `ClusterState.NetworkPolicies` for sync to TFO Platform
+  - Configurable via `collectors.kubernetes.network_policies: true`
+- **Network Flow Exporter** (`internal/exporter/network_flows.go`): New exporter that batches and sends pod-to-pod network flow events to the TFO Platform
+  - `NetworkFlowRecord` struct aligned with Cilium Hubble flow model: source/target namespace, pod, IP, port, labels, protocol, direction, verdict, bytes/packets, retransmits, RTT, HTTP status code, DNS query, external flag
+  - `NetworkFlowExporter` with thread-safe buffer, periodic flush loop (default: 10s), configurable max batch size (default: 500)
+  - POSTs `NetworkFlowBatch` to `/api/v2/monitoring/network-map/k8s/flows` with API key authentication headers
+  - Graceful shutdown with final flush of remaining buffered flows
+- **Ingress Collector Separated** (`internal/collector/kubernetes/ingresses.go`): Extracted Ingress collection into its own file for clarity
+  - Previously inlined; now a standalone `collectIngresses()` function returning `([]Metric, []IngressState, error)`
+  - Collects alongside Services since they share networking context
+- **Services Collector Enhanced** (`internal/collector/kubernetes/services.go`): Expanded to return full `EndpointState` objects alongside `ServiceState`
+  - `collectServices()` signature changed from `→ (metrics, []ServiceState, error)` to `→ (metrics, []ServiceState, []EndpointState, error)`
+  - Pre-fetched endpoints now produce full `EndpointSubset` with ready/not-ready addresses, node names, target refs, and ports
+  - Services now include `ServicePort` detail (name, protocol, port, target_port, node_port), external IPs from both spec and LoadBalancer status
+  - Added describe-level fields: `SessionAffinity`, `ExternalTrafficPolicy`, `HealthCheckNodePort`, `LoadBalancerSourceRanges`
+- **Node Network Metrics Expanded** (`internal/collector/kubernetes/nodes.go`): Three new node-level network metrics from Kubelet summary
+  - `k8s.node.network.receive_bytes` (Counter) — network bytes received per node
+  - `k8s.node.network.transmit_bytes` (Counter) — network bytes transmitted per node
+  - `k8s.node.network.receive_drop_total` (Counter) — network receive errors/drops per node
+  - Also now tracks `totalRxDrop` and `totalTxDrop` across all node interfaces
+- **Pod QoS & Status Metrics** (`internal/collector/kubernetes/pods.go`): Two new pod-level metrics
+  - `k8s.pod.qos_class` (Gauge, label: `qos_class`) — exposes pod QoS class (Guaranteed, Burstable, BestEffort)
+  - `k8s.pod.status_reason` (Gauge, label: `reason`) — exposes pod status reason when present (Evicted, NodeLost, etc.)
+- **Test Exports** (`internal/collector/kubernetes/exports.go`): New file exposing internal parse functions for unit testing
+  - `ParseApiServerMetricsExported()`, `ParseCoreDNSMetricsExported()`, `ParsePromLineExported()` — test-only wrappers
+- **K8s Unit Tests** (`tests/unit/domain/kubernetes/`): Four new test files for recently added sub-collectors
+  - `apiserver_test.go` — API Server metrics parser validation
+  - `coredns_test.go` — CoreDNS metrics parser validation
+  - `ingresses_test.go` — Ingress collection with rules, TLS, LoadBalancer IPs
+  - `services_test.go` — Service + Endpoint collection with describe-level fields
+- **Container Build Script** (`run-container.sh`): New unified container build/run script (221 lines) replacing the previous `run-build-container.sh`
 
 ### Fixed
 
@@ -124,6 +162,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **Helm chart path**: Renamed `deploy/helm/tfo-agent/` → `deploy/helm/telemetryflow-agent/` for naming consistency with other TelemetryFlow Helm charts
 - **Platform monolith configs** (`config/tfo-agent/`): All three deployment configs (`tfo-agent.yaml`, `tfo-agent.k8s.yaml`, `tfo-agent.container.yaml`) updated with KSM gap fields and `remote_write_receiver` section
 - **All config files updated** (`configs/`, `deploy/`): Added extended K8s metrics fields (`apiserver_metrics`, `coredns_metrics`, `container_extended_metrics`, `pv_io_stats`) to all config variants — `configs/tfo-agent.yaml`, `configs/tfo-agent.default.yaml`, `configs/tfo-agent-one-for-all.yaml`, `deploy/helm/values.yaml`, `deploy/helm/values-one-for-all.yaml`, `deploy/kubernetes/configmap.yaml`
+- **K8s config**: Added `network_policies: true` to all config files under `collectors.kubernetes`
+- **`ClusterState` type**: Added `NetworkPolicies []NetworkPolicyState` field for platform sync
+- **RBAC ClusterRole**: Added `networkpolicies` (networking.k8s.io) resource permission for NetworkPolicy collector
+
+### Dependencies
+
+- `google.golang.org/grpc`: Bumped from v1.79.1 to v1.79.3
 
 ## [1.1.8] - 2026-03-09
 
@@ -540,7 +585,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 | Version | Date       | OTEL SDK | Description                                                                                                                                                                                                                            |
 | ------- | ---------- | -------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
-| 1.1.9   | 2026-03-20 | v1.40.0  | K8s network resources (Services/Endpoints/Ingresses); API Server & CoreDNS metrics scrapers; Fluent Bit log collector; Prometheus Remote Write Receiver; KSM gap fields (5); license headers; eBPF build constraint fixes; Helm rename |
+| 1.1.9   | 2026-03-20 | v1.47.0  | K8s network resources (Services/Endpoints/Ingresses); NetworkPolicy collector + Network Flow Exporter; API Server & CoreDNS metrics scrapers; Fluent Bit log collector; Prometheus Remote Write Receiver; KSM gap fields (5); Pod QoS/status metrics; Node network rx/tx/drop metrics; 4 new K8s test files; license headers; eBPF build constraint fixes; Helm rename; gRPC v1.79.3 |
 | 1.1.8   | 2026-03-09 | v1.40.0  | HPA/PDB/pod-logs sub-collectors; Kubelet summary ephemeral + working set; Go 1.26 + security fixes; 17 collector docs                                                                                                                  |
 | 1.1.7   | 2026-03-08 | v1.40.0  | Stable agent identity via UUIDv5 host fingerprint; K8s provider detection (15 providers); fix SyncKubernetesState                                                                                                                      |
 | 1.1.6   | 2026-02-21 | v1.40.0  | Go 1.25.7, OTEL SDK v1.40.0, build-tag lint fixes, errcheck/staticcheck cleanup                                                                                                                                                        |

deploy/helm/telemetryflow-agent/templates/deployment-k8s.yaml

@@ -93,13 +93,24 @@ spec:
               value: "true"
             - name: TELEMETRYFLOW_PROMETHEUS_PORT
               value: {{ .Values.ports.metrics | quote }}
+            {{- if .Values.agentApi.enabled }}
+            - name: TELEMETRYFLOW_AGENT_API_ENABLED
+              value: "true"
+            - name: TELEMETRYFLOW_AGENT_API_PORT
+              value: {{ .Values.agentApi.port | default 8889 | quote }}
+            {{- end }}
             {{- with .Values.extraEnv }}
             {{- toYaml . | nindent 12 }}
             {{- end }}
           ports:
             - containerPort: {{ .Values.ports.metrics }}
               name: metrics
               protocol: TCP
+            {{- if .Values.agentApi.enabled }}
+            - containerPort: {{ .Values.agentApi.port | default 8889 }}
+              name: agent-api
+              protocol: TCP
+            {{- end }}
           # startupProbe: 90s budget for first-boot cluster auto-registration call
           startupProbe:
             httpGet:

deploy/helm/telemetryflow-agent/templates/service-k8s-api.yaml

@@ -0,0 +1,19 @@
+{{- if .Values.agentApi.enabled }}
+apiVersion: v1
+kind: Service
+metadata:
+  name: {{ include "tfo-agent.fullname" . }}-k8s-api
+  namespace: {{ .Release.Namespace }}
+  labels:
+    {{- include "tfo-agent.k8sCollector.labels" . | nindent 4 }}
+    app.kubernetes.io/component: k8s-api
+spec:
+  type: ClusterIP
+  selector:
+    {{- include "tfo-agent.k8sCollector.selectorLabels" . | nindent 4 }}
+  ports:
+    - name: agent-api
+      port: {{ .Values.agentApi.port | default 8889 }}
+      targetPort: {{ .Values.agentApi.port | default 8889 }}
+      protocol: TCP
+{{- end }}

deploy/helm/telemetryflow-agent/values.yaml

@@ -39,6 +39,12 @@ ports:
   otlpHttp: 4318
   metrics: 8888
   health: 13133
+  agentApi: 8889
+
+# -- Agent HTTP API server (real-time K8s queries: pod log streaming)
+agentApi:
+  enabled: true
+  port: 8889
 
 # -- ServiceAccount settings
 serviceAccount:

internal/agent/agent.go

@@ -29,6 +29,7 @@ import (
 
 	"go.uber.org/zap"
 
+	agentapi "github.com/telemetryflow/telemetryflow-agent/internal/api"
 	"github.com/telemetryflow/telemetryflow-agent/internal/collector"
 	cadvisorcollector "github.com/telemetryflow/telemetryflow-agent/internal/collector/cadvisor"
 	dockercollector "github.com/telemetryflow/telemetryflow-agent/internal/collector/docker"
@@ -58,6 +59,7 @@ type Agent struct {
 	k8sCollector     *kubernetes.KubernetesCollector // kept for registration retry
 	collectors       []collector.Collector
 	prometheusServer *exporter.PrometheusServer
+	agentAPIServer   *agentapi.Server
 
 	// State
 	mu      sync.RWMutex
@@ -367,6 +369,23 @@ func New(cfg *config.Config, logger *zap.Logger) (*Agent, error) {
 		)
 	}
 
+	// Create Agent API server if enabled (for real-time K8s queries like pod log streaming)
+	var apiServer *agentapi.Server
+	if cfg.AgentAPI.Enabled && k8sCollector != nil {
+		apiServer = agentapi.NewServer(
+			agentapi.Config{
+				Enabled: cfg.AgentAPI.Enabled,
+				Port:    cfg.AgentAPI.Port,
+				APIKey:  cfg.AgentAPI.APIKey,
+			},
+			k8sCollector.Clientset(),
+			logger,
+		)
+		logger.Info("Agent API server enabled",
+			zap.Int("port", cfg.AgentAPI.Port),
+		)
+	}
+
 	return &Agent{
 		id:               agentID,
 		config:           cfg,
@@ -377,6 +396,7 @@ func New(cfg *config.Config, logger *zap.Logger) (*Agent, error) {
 		k8sCollector:     k8sCollector,
 		collectors:       collectors,
 		prometheusServer: promServer,
+		agentAPIServer:   apiServer,
 	}, nil
 }
 
@@ -425,6 +445,15 @@ func (a *Agent) Run(ctx context.Context) error {
 		}()
 	}
 
+	// Start Agent API server (pod log streaming)
+	if a.agentAPIServer != nil {
+		go func() {
+			if err := a.agentAPIServer.Start(ctx); err != nil && err != context.Canceled {
+				errChan <- fmt.Errorf("agent API server error: %w", err)
+			}
+		}()
+	}
+
 	// Start heartbeat
 	go func() {
 		if err := a.heartbeat.Start(ctx); err != nil && err != context.Canceled {
@@ -536,6 +565,19 @@ func (a *Agent) shutdown() error {
 		}()
 	}
 
+	// Stop Agent API server
+	if a.agentAPIServer != nil {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			if err := a.agentAPIServer.Stop(); err != nil {
+				errMu.Lock()
+				errs = append(errs, fmt.Errorf("agent API server stop: %w", err))
+				errMu.Unlock()
+			}
+		}()
+	}
+
 	// Stop heartbeat
 	wg.Add(1)
 	go func() {

internal/api/handlers.go

@@ -0,0 +1,140 @@
+package api
+
+import (
+	"bufio"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"strconv"
+
+	"go.uber.org/zap"
+	corev1 "k8s.io/api/core/v1"
+)
+
+// handleHealth returns a simple health check response.
+func (s *Server) handleHealth(w http.ResponseWriter, _ *http.Request) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(http.StatusOK)
+	_ = json.NewEncoder(w).Encode(map[string]string{"status": "ok"})
+}
+
+// handlePodLogs streams pod container logs via SSE (Server-Sent Events).
+// Query params: container, follow, tailLines, sinceSeconds, previous, timestamps
+func (s *Server) handlePodLogs(w http.ResponseWriter, r *http.Request) {
+	namespace := r.PathValue("namespace")
+	podName := r.PathValue("pod")
+
+	if namespace == "" || podName == "" {
+		http.Error(w, `{"error":"namespace and pod are required"}`, http.StatusBadRequest)
+		return
+	}
+
+	// Parse query parameters
+	container := r.URL.Query().Get("container")
+	follow := r.URL.Query().Get("follow") == "true"
+	timestamps := r.URL.Query().Get("timestamps") != "false" // default true
+	previous := r.URL.Query().Get("previous") == "true"
+
+	var tailLines *int64
+	if tl := r.URL.Query().Get("tailLines"); tl != "" {
+		if n, err := strconv.ParseInt(tl, 10, 64); err == nil {
+			tailLines = &n
+		}
+	}
+	// Default to 100 tail lines if not specified and not following
+	if tailLines == nil {
+		defaultTail := int64(100)
+		tailLines = &defaultTail
+	}
+
+	var sinceSeconds *int64
+	if ss := r.URL.Query().Get("sinceSeconds"); ss != "" {
+		if n, err := strconv.ParseInt(ss, 10, 64); err == nil {
+			sinceSeconds = &n
+		}
+	}
+
+	// Build PodLogOptions
+	opts := &corev1.PodLogOptions{
+		Follow:       follow,
+		Timestamps:   timestamps,
+		Previous:     previous,
+		TailLines:    tailLines,
+		SinceSeconds: sinceSeconds,
+	}
+	if container != "" {
+		opts.Container = container
+	}
+
+	s.logger.Debug("Pod log request",
+		zap.String("namespace", namespace),
+		zap.String("pod", podName),
+		zap.String("container", container),
+		zap.Bool("follow", follow),
+	)
+
+	// Get log stream from K8s API
+	stream, err := s.clientset.CoreV1().Pods(namespace).GetLogs(podName, opts).Stream(r.Context())
+	if err != nil {
+		s.logger.Error("Failed to get pod log stream",
+			zap.String("namespace", namespace),
+			zap.String("pod", podName),
+			zap.Error(err),
+		)
+		http.Error(w, fmt.Sprintf(`{"error":"failed to get pod logs: %s"}`, err.Error()), http.StatusInternalServerError)
+		return
+	}
+	defer func() { _ = stream.Close() }()
+
+	if follow {
+		// SSE streaming mode
+		w.Header().Set("Content-Type", "text/event-stream")
+		w.Header().Set("Cache-Control", "no-cache")
+		w.Header().Set("Connection", "keep-alive")
+		w.Header().Set("X-Accel-Buffering", "no") // Disable nginx buffering
+
+		flusher, ok := w.(http.Flusher)
+		if !ok {
+			http.Error(w, `{"error":"streaming not supported"}`, http.StatusInternalServerError)
+			return
+		}
+
+		scanner := bufio.NewScanner(stream)
+		// Increase buffer size for long log lines (1MB)
+		scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
+
+		for scanner.Scan() {
+			select {
+			case <-r.Context().Done():
+				return
+			default:
+			}
+			line := scanner.Text()
+			_, _ = fmt.Fprintf(w, "data: %s\n\n", line)
+			flusher.Flush()
+		}
+
+		if err := scanner.Err(); err != nil {
+			s.logger.Debug("Pod log stream ended", zap.Error(err))
+		}
+	} else {
+		// Non-streaming mode: return all lines as JSON array
+		w.Header().Set("Content-Type", "application/json")
+
+		var lines []string
+		scanner := bufio.NewScanner(stream)
+		scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
+
+		for scanner.Scan() {
+			lines = append(lines, scanner.Text())
+		}
+
+		_ = json.NewEncoder(w).Encode(map[string]interface{}{
+			"namespace": namespace,
+			"pod":       podName,
+			"container": container,
+			"lines":     lines,
+			"count":     len(lines),
+		})
+	}
+}

internal/api/middleware.go

@@ -0,0 +1,45 @@
+package api
+
+import (
+	"net/http"
+
+	"go.uber.org/zap"
+)
+
+// authMiddleware validates API key from request headers.
+// Accepts X-TelemetryFlow-Key-ID or X-API-Key-ID headers.
+func (s *Server) authMiddleware(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		// Health endpoint is always public
+		if r.URL.Path == "/api/v1/health" {
+			next.ServeHTTP(w, r)
+			return
+		}
+
+		// Skip auth if no API key is configured (development mode)
+		if s.config.APIKey == "" {
+			next.ServeHTTP(w, r)
+			return
+		}
+
+		// Check API key from headers
+		apiKey := r.Header.Get("X-TelemetryFlow-Key-ID")
+		if apiKey == "" {
+			apiKey = r.Header.Get("X-API-Key-ID")
+		}
+		if apiKey == "" {
+			apiKey = r.Header.Get("Authorization")
+		}
+
+		if apiKey != s.config.APIKey {
+			s.logger.Debug("Agent API auth failed",
+				zap.String("remote", r.RemoteAddr),
+				zap.String("path", r.URL.Path),
+			)
+			http.Error(w, `{"error":"unauthorized"}`, http.StatusUnauthorized)
+			return
+		}
+
+		next.ServeHTTP(w, r)
+	})
+}