diff --git a/.dockerignore b/.dockerignore index 80c0faa1..7bc06bf2 100644 --- a/.dockerignore +++ b/.dockerignore @@ -1,9 +1,11 @@ .git .github +.cache coverage dist bin build +otelcol/bin **/*.test **/*.out **/*.log diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 846fab7e..e8933f8e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -177,6 +177,40 @@ jobs: path: fleetint-${{ matrix.goos }}-${{ matrix.goarch }} retention-days: 7 + build_check_otelcol: + name: Build Check (otelcol) + runs-on: ubuntu-latest + needs: [lint] + strategy: + matrix: + goarch: [amd64, arm64] + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Set up Go + uses: actions/setup-go@v4 + with: + go-version: ${{ env.GO_VERSION }} + + - name: Install OTel Collector Builder + run: go install go.opentelemetry.io/collector/cmd/builder@v0.156.0 + + - name: Build fleetint-otelcol + working-directory: otelcol + env: + GOOS: linux + GOARCH: ${{ matrix.goarch }} + CGO_ENABLED: 0 + run: builder --config=otelcol-builder.yaml + + - name: Run OTel gateway integration test + if: matrix.goarch == 'amd64' + working-directory: otelcol/auth/sakauth + env: + FLEETINT_OTELCOL_INTEGRATION: "1" + run: go test -race -run '^TestCollectorGatewayEndToEnd$' . + codeql: name: CodeQL Analysis runs-on: ubuntu-latest diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index aee56e3d..4b8fb88f 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -97,6 +97,21 @@ jobs: fi echo "EOF" >> "$GITHUB_OUTPUT" + - name: Prepare fleetint-otelcol image tags + id: otelcol_image + run: | + set -euo pipefail + VERSION="${{ steps.version.outputs.version }}" + OWNER_LC="$(echo "${{ github.repository_owner }}" | tr '[:upper:]' '[:lower:]')" + GHCR_IMAGE="ghcr.io/${OWNER_LC}/fleetint-otelcol" + echo "ghcr_image=${GHCR_IMAGE}" >> "$GITHUB_OUTPUT" + echo "tags<> "$GITHUB_OUTPUT" + echo "${GHCR_IMAGE}:${VERSION}" >> "$GITHUB_OUTPUT" + if [ "${{ steps.version.outputs.is_prerelease }}" = "false" ]; then + echo "${GHCR_IMAGE}:latest" >> "$GITHUB_OUTPUT" + fi + echo "EOF" >> "$GITHUB_OUTPUT" + - name: Set up QEMU uses: docker/setup-qemu-action@v3 @@ -124,6 +139,15 @@ jobs: REVISION=${{ steps.version.outputs.commit }} BUILD_TIMESTAMP=${{ steps.version.outputs.build_timestamp }} + - name: Build and push fleetint-otelcol image + uses: docker/build-push-action@v6 + with: + context: . + file: ./otelcol/Dockerfile + platforms: linux/amd64,linux/arm64 + push: true + tags: ${{ steps.otelcol_image.outputs.tags }} + - name: Package Helm chart run: | helm package deployments/helm/fleet-intelligence-agent \ @@ -191,6 +215,9 @@ jobs: echo "**fleet-intelligence-agent image tags:**" >> $GITHUB_STEP_SUMMARY echo "${{ steps.agent_image_ghcr.outputs.tags }}" >> $GITHUB_STEP_SUMMARY echo "" >> $GITHUB_STEP_SUMMARY + echo "**fleetint-otelcol image tags:**" >> $GITHUB_STEP_SUMMARY + echo "${{ steps.otelcol_image.outputs.tags }}" >> $GITHUB_STEP_SUMMARY + echo "" >> $GITHUB_STEP_SUMMARY echo "**Helm chart (GHCR OCI):**" >> $GITHUB_STEP_SUMMARY echo "repo: oci://ghcr.io/${{ steps.agent_image_ghcr.outputs.owner_lc }}/charts" >> $GITHUB_STEP_SUMMARY echo "chart: fleet-intelligence-agent:${{ steps.version.outputs.version }}" >> $GITHUB_STEP_SUMMARY diff --git a/.gitignore b/.gitignore index 7f60a97c..73ca3a57 100644 --- a/.gitignore +++ b/.gitignore @@ -94,3 +94,4 @@ AGENTS.md .worktrees/ docs/plans/ +otelcol/bin/ diff --git a/Makefile b/Makefile index a92c61b4..a2a109eb 100644 --- a/Makefile +++ b/Makefile @@ -17,7 +17,7 @@ ROOTDIR=$(dir $(abspath $(lastword $(MAKEFILE_LIST)))) BUILD_TIMESTAMP ?= $(shell date -u +"%Y-%m-%dT%H:%M:%SZ") VERSION ?= $(shell git describe --match 'v[0-9]*' --dirty='.m' --always) -REVISION=$(shell git rev-parse HEAD)$(shell if ! git diff --no-ext-diff --quiet --exit-code; then echo .m; fi) +REVISION ?= $(shell git rev-parse HEAD)$(shell if ! git diff --no-ext-diff --quiet --exit-code; then echo .m; fi) PACKAGE=github.com/NVIDIA/fleet-intelligence-agent ifneq "$(strip $(shell command -v $(GO) 2>/dev/null))" "" @@ -118,7 +118,7 @@ docker-test: ## build test image and run tests in container -t $(TEST_IMAGE) \ . @echo "Running tests..." - @$(DOCKER) run --rm $(TEST_IMAGE) + @$(DOCKER) run --rm -e VERSION=$(VERSION) -e REVISION=$(REVISION) $(TEST_IMAGE) # Specific target for fleetint (your main binary) fleetint: bin/fleetint ## build fleetint binary diff --git a/cmd/fleetint/gateway_test.go b/cmd/fleetint/gateway_test.go new file mode 100644 index 00000000..0808c056 --- /dev/null +++ b/cmd/fleetint/gateway_test.go @@ -0,0 +1,51 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "testing" + + pkgmetadata "github.com/NVIDIA/fleet-intelligence-sdk/pkg/metadata" + "github.com/stretchr/testify/require" + + "github.com/NVIDIA/fleet-intelligence-agent/internal/config" +) + +func TestConfigureHealthExporterFromEnvCollectorEndpoint(t *testing.T) { + t.Setenv("FLEETINT_COLLECTOR_ENDPOINT", "http://fleetint-otel-gateway:4318") + cfg := &config.Config{HealthExporter: &config.HealthExporterConfig{}} + + require.NoError(t, configureHealthExporterFromEnv(cfg)) + require.Equal(t, "http://fleetint-otel-gateway:4318", cfg.HealthExporter.CollectorEndpoint) +} + +func TestConfigureHealthExporterFromEnvPreservesCollectorEndpointWhenUnset(t *testing.T) { + t.Setenv("FLEETINT_COLLECTOR_ENDPOINT", "") + cfg := &config.Config{HealthExporter: &config.HealthExporterConfig{ + CollectorEndpoint: "https://collector.example", + }} + + require.NoError(t, configureHealthExporterFromEnv(cfg)) + require.Equal(t, "https://collector.example", cfg.HealthExporter.CollectorEndpoint) +} + +func TestMaskMetadataValue(t *testing.T) { + const secret = "secret-token-value" + + require.Equal(t, pkgmetadata.MaskToken(secret), maskMetadataValue(pkgmetadata.MetadataKeyToken, secret)) + require.Equal(t, pkgmetadata.MaskToken(secret), maskMetadataValue("sak_token", secret)) + require.Equal(t, secret, maskMetadataValue("backend_base_url", secret)) +} diff --git a/cmd/fleetint/metadata.go b/cmd/fleetint/metadata.go index 6f814f10..036ece50 100644 --- a/cmd/fleetint/metadata.go +++ b/cmd/fleetint/metadata.go @@ -73,11 +73,7 @@ func metadataCommand(cliContext *cli.Context) error { log.Logger.Debugw("successfully read metadata") for k, v := range metadata { - // Mask sensitive tokens (JWT and SAK) - if k == pkgmetadata.MetadataKeyToken || k == "sak_token" { - v = pkgmetadata.MaskToken(v) - } - fmt.Printf("%s: %s\n", k, v) + fmt.Printf("%s: %s\n", k, maskMetadataValue(k, v)) } setKey := cliContext.String("set-key") @@ -114,3 +110,10 @@ func metadataCommand(cliContext *cli.Context) error { fmt.Printf("%s successfully updated metadata\n", cmdutil.CheckMark) return nil } + +func maskMetadataValue(key, value string) string { + if key == pkgmetadata.MetadataKeyToken || key == "sak_token" { + return pkgmetadata.MaskToken(value) + } + return value +} diff --git a/cmd/fleetint/run.go b/cmd/fleetint/run.go index 442aa47b..bd0d8adc 100644 --- a/cmd/fleetint/run.go +++ b/cmd/fleetint/run.go @@ -219,6 +219,14 @@ func configureHealthExporterFromEnv(cfg *config.Config) error { return err } + // OTel gateway collector mode: when set, metrics/logs are routed to the gateway + // instead of the backend directly. Enrollment remains independent for inventory + // and attestation; backend credentials are not forwarded to the gateway. + if val := os.Getenv("FLEETINT_COLLECTOR_ENDPOINT"); val != "" { + he.CollectorEndpoint = val + log.Logger.Infow("set OTel gateway collector endpoint from env", "collector_endpoint", val) + } + return nil } diff --git a/deployments/helm/fleet-intelligence-agent/templates/daemonset.yaml b/deployments/helm/fleet-intelligence-agent/templates/daemonset.yaml index 1e738e43..089ab710 100644 --- a/deployments/helm/fleet-intelligence-agent/templates/daemonset.yaml +++ b/deployments/helm/fleet-intelligence-agent/templates/daemonset.yaml @@ -148,6 +148,10 @@ spec: fieldPath: spec.nodeName - name: IS_RUNTIME_K8S value: "true" + {{- if .Values.otelGateway.enabled }} + - name: FLEETINT_COLLECTOR_ENDPOINT + value: {{ printf "http://%s-otel-gateway:%d" (include "fleet-intelligence-agent.fullname" .) (int .Values.otelGateway.service.port) | quote }} + {{- end }} ports: - name: http containerPort: {{ .Values.ports.http }} diff --git a/deployments/helm/fleet-intelligence-agent/templates/otel-gateway-configmap.yaml b/deployments/helm/fleet-intelligence-agent/templates/otel-gateway-configmap.yaml new file mode 100644 index 00000000..460986dc --- /dev/null +++ b/deployments/helm/fleet-intelligence-agent/templates/otel-gateway-configmap.yaml @@ -0,0 +1,85 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +{{- if .Values.otelGateway.enabled }} +apiVersion: v1 +kind: ConfigMap +metadata: + name: {{ include "fleet-intelligence-agent.fullname" . }}-otel-gateway + labels: + {{- include "fleet-intelligence-agent.labels" . | nindent 4 }} +data: + config.yaml: | + extensions: + # Fleet Intelligence auth: enrolls with the backend using the SAK token, extracts + # the customer ID from the returned JWT, and refreshes the JWT proactively via + # response headers and on 401 as a fallback. + sakauth: + enroll_endpoint: ${env:ENROLL_ENDPOINT} + sak_token: ${env:SAK_TOKEN} + health_check: + endpoint: 0.0.0.0:13133 + + receivers: + otlp: + protocols: + http: + # No inbound auth: the gateway is a ClusterIP service accessible only + # within the cluster. Cluster network isolation is the security boundary. + # The gateway uses its own SAK JWT when forwarding to the backend. + # Agent identity flows through OTLP resource attributes (machine.id, GPU UUIDs). + endpoint: 0.0.0.0:{{ .Values.otelGateway.service.port }} + + processors: + memory_limiter: + check_interval: 5s + limit_percentage: {{ .Values.otelGateway.memoryLimiter.limitPercentage }} + spike_limit_percentage: {{ .Values.otelGateway.memoryLimiter.spikeLimitPercentage }} + batch: + timeout: 10s + send_batch_size: 1000 + + exporters: + otlp_http/backend: + metrics_endpoint: ${env:BACKEND_ENDPOINT}/metrics + logs_endpoint: ${env:BACKEND_ENDPOINT}/logs + compression: none + auth: + authenticator: sakauth + retry_on_failure: + enabled: true + initial_interval: 5s + max_interval: 30s + max_elapsed_time: 300s + sending_queue: + enabled: true + num_consumers: 4 + queue_size: 1000 + + service: + extensions: [sakauth, health_check] + pipelines: + metrics: + receivers: [otlp] + processors: [memory_limiter, batch] + exporters: [otlp_http/backend] + logs: + receivers: [otlp] + processors: [memory_limiter, batch] + exporters: [otlp_http/backend] + telemetry: + logs: + level: warn +{{- end }} diff --git a/deployments/helm/fleet-intelligence-agent/templates/otel-gateway-deployment.yaml b/deployments/helm/fleet-intelligence-agent/templates/otel-gateway-deployment.yaml new file mode 100644 index 00000000..0cfc17de --- /dev/null +++ b/deployments/helm/fleet-intelligence-agent/templates/otel-gateway-deployment.yaml @@ -0,0 +1,124 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +{{- if .Values.otelGateway.enabled }} +{{- if and (not .Values.otelGateway.enroll.sakToken) (not .Values.otelGateway.enroll.existingSecret) }} +{{- fail "otelGateway.enroll.sakToken or otelGateway.enroll.existingSecret is required when otelGateway.enabled=true" }} +{{- end }} +apiVersion: apps/v1 +kind: Deployment +metadata: + name: {{ include "fleet-intelligence-agent.fullname" . }}-otel-gateway + labels: + {{- include "fleet-intelligence-agent.labels" . | nindent 4 }} + app.kubernetes.io/component: otel-gateway +spec: + replicas: {{ .Values.otelGateway.replicas }} + strategy: + type: RollingUpdate + rollingUpdate: + maxUnavailable: 0 + maxSurge: 1 + selector: + matchLabels: + {{- include "fleet-intelligence-agent.selectorLabels" . | nindent 6 }} + app.kubernetes.io/component: otel-gateway + template: + metadata: + labels: + {{- include "fleet-intelligence-agent.selectorLabels" . | nindent 8 }} + app.kubernetes.io/component: otel-gateway + annotations: + # Force pod restart when the collector config changes. + checksum/config: {{ include (print $.Template.BasePath "/otel-gateway-configmap.yaml") . | sha256sum }} + {{- if not .Values.otelGateway.enroll.existingSecret }} + checksum/sak-secret: {{ include (print $.Template.BasePath "/otel-gateway-secret.yaml") . | sha256sum }} + {{- end }} + spec: + automountServiceAccountToken: false + {{- with .Values.imagePullSecrets }} + imagePullSecrets: + {{- toYaml . | nindent 8 }} + {{- end }} + securityContext: + runAsNonRoot: true + seccompProfile: + type: RuntimeDefault + terminationGracePeriodSeconds: {{ .Values.otelGateway.terminationGracePeriodSeconds }} + topologySpreadConstraints: + - maxSkew: 1 + topologyKey: kubernetes.io/hostname + whenUnsatisfiable: ScheduleAnyway + labelSelector: + matchLabels: + {{- include "fleet-intelligence-agent.selectorLabels" . | nindent 14 }} + app.kubernetes.io/component: otel-gateway + containers: + - name: otel-gateway + image: "{{ .Values.otelGateway.image.repository }}:{{ default .Chart.AppVersion .Values.otelGateway.image.tag }}" + imagePullPolicy: {{ .Values.otelGateway.image.pullPolicy }} + securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: + drop: ["ALL"] + args: + - --config=/etc/otelcol/config.yaml + env: + - name: BACKEND_ENDPOINT + value: {{ required "otelGateway.backendEndpoint is required when otelGateway.enabled=true" .Values.otelGateway.backendEndpoint | trimSuffix "/" | quote }} + - name: ENROLL_ENDPOINT + value: {{ required "otelGateway.enroll.endpoint is required when otelGateway.enabled=true" .Values.otelGateway.enroll.endpoint | quote }} + - name: SAK_TOKEN + valueFrom: + secretKeyRef: + name: {{ .Values.otelGateway.enroll.existingSecret | default (printf "%s-otel-gateway-sak" (include "fleet-intelligence-agent.fullname" .)) }} + key: sak-token + ports: + - name: otlp-http + containerPort: {{ .Values.otelGateway.service.port }} + protocol: TCP + livenessProbe: + httpGet: + path: / + port: 13133 + periodSeconds: 30 + timeoutSeconds: 3 + failureThreshold: 3 + readinessProbe: + httpGet: + path: / + port: 13133 + periodSeconds: 10 + timeoutSeconds: 3 + failureThreshold: 3 + startupProbe: + httpGet: + path: / + port: 13133 + periodSeconds: 2 + timeoutSeconds: 3 + failureThreshold: 30 + resources: + {{- toYaml .Values.otelGateway.resources | nindent 12 }} + volumeMounts: + - name: config + mountPath: /etc/otelcol + readOnly: true + volumes: + - name: config + configMap: + name: {{ include "fleet-intelligence-agent.fullname" . }}-otel-gateway +{{- end }} diff --git a/deployments/helm/fleet-intelligence-agent/templates/otel-gateway-networkpolicy.yaml b/deployments/helm/fleet-intelligence-agent/templates/otel-gateway-networkpolicy.yaml new file mode 100644 index 00000000..aba52db4 --- /dev/null +++ b/deployments/helm/fleet-intelligence-agent/templates/otel-gateway-networkpolicy.yaml @@ -0,0 +1,39 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +{{- if and .Values.otelGateway.enabled .Values.otelGateway.networkPolicy.enabled }} +apiVersion: networking.k8s.io/v1 +kind: NetworkPolicy +metadata: + name: {{ include "fleet-intelligence-agent.fullname" . }}-otel-gateway + labels: + {{- include "fleet-intelligence-agent.labels" . | nindent 4 }} + app.kubernetes.io/component: otel-gateway +spec: + podSelector: + matchLabels: + {{- include "fleet-intelligence-agent.selectorLabels" . | nindent 6 }} + app.kubernetes.io/component: otel-gateway + policyTypes: + - Ingress + ingress: + - from: + - podSelector: + matchLabels: + {{- include "fleet-intelligence-agent.selectorLabels" . | nindent 14 }} + ports: + - protocol: TCP + port: {{ .Values.otelGateway.service.port }} +{{- end }} diff --git a/deployments/helm/fleet-intelligence-agent/templates/otel-gateway-pdb.yaml b/deployments/helm/fleet-intelligence-agent/templates/otel-gateway-pdb.yaml new file mode 100644 index 00000000..be80b1e1 --- /dev/null +++ b/deployments/helm/fleet-intelligence-agent/templates/otel-gateway-pdb.yaml @@ -0,0 +1,30 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +{{- if and .Values.otelGateway.enabled .Values.otelGateway.podDisruptionBudget.enabled }} +apiVersion: policy/v1 +kind: PodDisruptionBudget +metadata: + name: {{ include "fleet-intelligence-agent.fullname" . }}-otel-gateway + labels: + {{- include "fleet-intelligence-agent.labels" . | nindent 4 }} + app.kubernetes.io/component: otel-gateway +spec: + minAvailable: {{ .Values.otelGateway.podDisruptionBudget.minAvailable }} + selector: + matchLabels: + {{- include "fleet-intelligence-agent.selectorLabels" . | nindent 6 }} + app.kubernetes.io/component: otel-gateway +{{- end }} diff --git a/deployments/helm/fleet-intelligence-agent/templates/otel-gateway-secret.yaml b/deployments/helm/fleet-intelligence-agent/templates/otel-gateway-secret.yaml new file mode 100644 index 00000000..40e187f8 --- /dev/null +++ b/deployments/helm/fleet-intelligence-agent/templates/otel-gateway-secret.yaml @@ -0,0 +1,19 @@ +{{- /* +SPDX-FileCopyrightText: Copyright (c) 2025, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +SPDX-License-Identifier: Apache-2.0 +*/}} +{{- if and .Values.otelGateway.enabled (not .Values.otelGateway.enroll.existingSecret) }} +{{- if not .Values.otelGateway.enroll.sakToken }} +{{- fail "otelGateway.enroll.sakToken or otelGateway.enroll.existingSecret is required when otelGateway.enabled=true" }} +{{- end }} +apiVersion: v1 +kind: Secret +metadata: + name: {{ include "fleet-intelligence-agent.fullname" . }}-otel-gateway-sak + labels: + {{- include "fleet-intelligence-agent.labels" . | nindent 4 }} + app.kubernetes.io/component: otel-gateway +type: Opaque +stringData: + sak-token: {{ .Values.otelGateway.enroll.sakToken | quote }} +{{- end }} diff --git a/deployments/helm/fleet-intelligence-agent/templates/otel-gateway-service.yaml b/deployments/helm/fleet-intelligence-agent/templates/otel-gateway-service.yaml new file mode 100644 index 00000000..ae781e04 --- /dev/null +++ b/deployments/helm/fleet-intelligence-agent/templates/otel-gateway-service.yaml @@ -0,0 +1,34 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +{{- if .Values.otelGateway.enabled }} +apiVersion: v1 +kind: Service +metadata: + name: {{ include "fleet-intelligence-agent.fullname" . }}-otel-gateway + labels: + {{- include "fleet-intelligence-agent.labels" . | nindent 4 }} + app.kubernetes.io/component: otel-gateway +spec: + type: ClusterIP + selector: + {{- include "fleet-intelligence-agent.selectorLabels" . | nindent 4 }} + app.kubernetes.io/component: otel-gateway + ports: + - name: otlp-http + port: {{ .Values.otelGateway.service.port }} + targetPort: otlp-http + protocol: TCP +{{- end }} diff --git a/deployments/helm/fleet-intelligence-agent/values.yaml b/deployments/helm/fleet-intelligence-agent/values.yaml index e256b524..1ae37e7e 100644 --- a/deployments/helm/fleet-intelligence-agent/values.yaml +++ b/deployments/helm/fleet-intelligence-agent/values.yaml @@ -52,7 +52,11 @@ logLevel: warn listenAddress: 127.0.0.1:15133 retentionPeriod: 24h components: all +deployEnv: prod +# Agent enrollment remains independent of the optional telemetry gateway. +# Enrollment is still required for backend inventory and attestation workflows. +# Requires enroll.endpoint and either enroll.tokenValue or enroll.tokenSecretName. enroll: enabled: false unenroll: false @@ -94,3 +98,64 @@ serviceAccount: create: true name: "" automountToken: false + +# OTel gateway collector deployed alongside the agents in the same cluster. +# When enabled: +# - A Deployment + Service for the fleetint-otelcol collector is created. +# - Agent OTLP traffic is sent to the gateway without backend credentials. +# Agent enrollment remains direct because inventory and attestation use the +# agent's backend identity independently of OTLP export. +# - Each agent sends OTLP to the gateway via FLEETINT_COLLECTOR_ENDPOINT. +# - The gateway authenticates outbound requests to the backend via its JWT. +# - Agent identity flows through OTLP resource attributes (machine.id, GPU UUIDs). +otelGateway: + enabled: false + + image: + repository: ghcr.io/nvidia/fleetint-otelcol + tag: "" + pullPolicy: IfNotPresent + + replicas: 2 + terminationGracePeriodSeconds: 30 + + # Enrollment credentials for gateway→backend authentication. + # The gateway is the only component that holds these; agents never see them. + enroll: + # endpoint is the backend enrollment URL. + # Example: https://backend/api/v1/agent/enroll + endpoint: "" + # sakToken is the Service API Key used to authenticate with the API gateway + # on enrollment. Stored in a Kubernetes Secret; never put in plain values files. + # Use --set otelGateway.enroll.sakToken=nvapi-... or reference an existingSecret. + sakToken: "" + # existingSecret: if set, SAK_TOKEN is read from this secret instead of + # creating one from sakToken. The secret must have a key named "sak-token". + existingSecret: "" + + # Backend OTLP HTTP base URL (e.g. https://backend/api/v1/health). + backendEndpoint: "" + + service: + port: 4318 + + # Restrict unauthenticated OTLP ingress to this Helm release's pods. + # Requires a CNI plugin that enforces Kubernetes NetworkPolicy. + networkPolicy: + enabled: true + + podDisruptionBudget: + enabled: true + minAvailable: 1 + + memoryLimiter: + limitPercentage: 80 + spikeLimitPercentage: 20 + + resources: + limits: + cpu: "500m" + memory: "512Mi" + requests: + cpu: "100m" + memory: "128Mi" diff --git a/deployments/otel-collector/config.yaml b/deployments/otel-collector/config.yaml new file mode 100644 index 00000000..8d9d0f21 --- /dev/null +++ b/deployments/otel-collector/config.yaml @@ -0,0 +1,106 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# OTel Gateway Collector configuration for fleet-intelligence-agent. +# +# Deployment model: +# +# Agents ──► Gateway Collector (SAK→JWT) ──► Backend +# │ +# ├──► Prometheus remote write (optional) +# └──► Any other OTLP exporter (optional) +# +# The gateway holds a service-level SAK and handles all backend authentication. +# It enrolls once with a service-level identity and forwards all OTLP with that +# identity. Agents remain identifiable through machine.id and other OTLP +# resource attributes. +# The gateway relies on Kubernetes ClusterIP network isolation for inbound security. +# +# Required environment variables: +# ENROLL_ENDPOINT - Backend enrollment URL (e.g. https://backend/api/v1/agent/enroll) +# SAK_TOKEN - Service API Key sent as Authorization: Bearer on enrollment +# BACKEND_ENDPOINT - Backend OTLP HTTP base URL (e.g. https://backend/api/v1/health) + +extensions: + # Fleet Intelligence auth: enrolls with the backend using the SAK token, extracts + # the customer ID from the returned JWT, and refreshes the JWT proactively via + # response headers and on 401 as a fallback. + sakauth: + enroll_endpoint: ${env:ENROLL_ENDPOINT} + sak_token: ${env:SAK_TOKEN} + + # Health check endpoint used by liveness/readiness probes. + health_check: + endpoint: 0.0.0.0:13133 + +receivers: + otlp: + protocols: + http: + # No inbound auth: the gateway is a ClusterIP service accessible only + # within the cluster. Cluster network isolation is the security boundary. + # The gateway uses its own SAK JWT when forwarding to the backend. + # Agent identity flows through OTLP resource attributes (machine.id, GPU UUIDs). + endpoint: 0.0.0.0:4318 + +processors: + memory_limiter: + check_interval: 5s + limit_mib: 400 + spike_limit_mib: 100 + + batch: + timeout: 10s + send_batch_size: 1000 + +exporters: + otlp_http/backend: + metrics_endpoint: ${env:BACKEND_ENDPOINT}/metrics + logs_endpoint: ${env:BACKEND_ENDPOINT}/logs + compression: none + auth: + authenticator: sakauth + retry_on_failure: + enabled: true + initial_interval: 5s + max_interval: 30s + max_elapsed_time: 300s + sending_queue: + enabled: true + num_consumers: 4 + queue_size: 1000 + + # Optional: uncomment to also fan out to Prometheus remote write. + # prometheusremotewrite/monitoring: + # endpoint: ${env:PROMETHEUS_REMOTE_WRITE_URL} + + # Optional: uncomment to fan out logs to Loki. + # loki/monitoring: + # endpoint: ${env:LOKI_ENDPOINT} + +service: + extensions: [sakauth, health_check] + pipelines: + metrics: + receivers: [otlp] + processors: [memory_limiter, batch] + exporters: [otlp_http/backend] + logs: + receivers: [otlp] + processors: [memory_limiter, batch] + exporters: [otlp_http/backend] + telemetry: + logs: + level: warn diff --git a/internal/config/config.go b/internal/config/config.go index 2c1f45dc..81ce91ab 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -19,6 +19,7 @@ import ( "encoding/json" "errors" "fmt" + "net/url" "reflect" "strings" "time" @@ -135,6 +136,12 @@ type HealthExporterConfig struct { // RetryMaxAttempts is the maximum number of retry attempts for failed requests RetryMaxAttempts int `json:"retry_max_attempts"` + // CollectorEndpoint is the base URL of an OTel gateway collector (e.g. http://otel-gateway:4318). + // When set, the agent sends OTLP to {CollectorEndpoint}/v1/metrics and /v1/logs. + // Agent enrollment remains independent, and backend credentials are not forwarded + // to the gateway. Agent identity flows through OTLP resource attributes. + CollectorEndpoint string `json:"collector_endpoint,omitempty"` + // Offline mode configuration // OfflineMode controls whether to use offline mode (write to files instead of HTTP endpoint) OfflineMode bool `json:"offline_mode"` @@ -171,6 +178,16 @@ func (config *Config) Validate() error { } // Validate health exporter configuration if present if config.HealthExporter != nil { + if endpoint := config.HealthExporter.CollectorEndpoint; endpoint != "" { + parsed, err := url.Parse(endpoint) + if err != nil || (parsed.Scheme != "http" && parsed.Scheme != "https") || parsed.Host == "" { + return fmt.Errorf("collector_endpoint must be an absolute HTTP(S) URL") + } + if parsed.User != nil || parsed.RawQuery != "" || parsed.Fragment != "" { + return fmt.Errorf("collector_endpoint must not contain credentials, a query, or a fragment") + } + } + // Validate health check interval if config.HealthExporter.HealthCheckInterval.Duration != 0 { if config.HealthExporter.HealthCheckInterval.Duration < time.Second { @@ -416,8 +433,9 @@ func extractHealthExporterEntries(cfg *HealthExporterConfig) []ConfigEntry { continue } key := strings.Split(jsonTag, ",")[0] - // Skip sensitive/enrollment-assigned field - if key == "auth_token" || key == "metrics_endpoint" || key == "logs_endpoint" { + // Skip sensitive/enrollment-assigned fields + if key == "auth_token" || key == "metrics_endpoint" || key == "logs_endpoint" || + key == "collector_endpoint" { continue } diff --git a/internal/config/config_test.go b/internal/config/config_test.go index b4adc884..69dc91a8 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -293,6 +293,36 @@ func TestSecureStateFilePermissions(t *testing.T) { } func TestValidateHealthExporter(t *testing.T) { + t.Run("valid collector endpoint", func(t *testing.T) { + cfg := &Config{ + Address: ":8080", + RetentionPeriod: metav1.Duration{Duration: time.Hour}, + HealthExporter: &HealthExporterConfig{ + CollectorEndpoint: "http://fleetint-otel-gateway:4318", + }, + } + require.NoError(t, cfg.Validate()) + }) + + for _, endpoint := range []string{ + "collector:4318", + "ftp://collector.example", + "http://user:password@collector.example", + "http://collector.example?token=secret", + "http://collector.example#fragment", + } { + t.Run("invalid collector endpoint "+endpoint, func(t *testing.T) { + cfg := &Config{ + Address: ":8080", + RetentionPeriod: metav1.Duration{Duration: time.Hour}, + HealthExporter: &HealthExporterConfig{ + CollectorEndpoint: endpoint, + }, + } + require.ErrorContains(t, cfg.Validate(), "collector_endpoint") + }) + } + t.Run("valid health check interval", func(t *testing.T) { cfg := &Config{ Address: ":8080", diff --git a/internal/exporter/exporter.go b/internal/exporter/exporter.go index cccabad5..eee02e0a 100644 --- a/internal/exporter/exporter.go +++ b/internal/exporter/exporter.go @@ -25,6 +25,7 @@ package exporter import ( "context" "fmt" + "strings" "time" "github.com/NVIDIA/fleet-intelligence-sdk/pkg/log" @@ -272,6 +273,17 @@ func (e *healthExporter) exportToHTTP(ctx context.Context, data *collector.Healt // refreshConfigFromMetadata updates the exporter configuration from metadata table func (e *healthExporter) refreshConfigFromMetadata(ctx context.Context) { + // Collector mode is explicit runtime configuration and must take precedence + // over enrollment metadata, which points directly at the backend. + if collectorEndpoint := strings.TrimRight(e.options.config.CollectorEndpoint, "/"); collectorEndpoint != "" { + e.options.config.MetricsEndpoint = collectorEndpoint + "/v1/metrics" + e.options.config.LogsEndpoint = collectorEndpoint + "/v1/logs" + // Authentication to the backend is owned by the gateway. Do not forward + // a node JWT to the cluster-local collector. + e.options.config.AuthToken = "" + return + } + // Use the passed database connection instead of opening a new one if e.options.dbRO == nil { log.Logger.Debugw("no database connection available for metadata refresh") diff --git a/internal/exporter/exporter_test.go b/internal/exporter/exporter_test.go index c85304ea..8b6cfc29 100644 --- a/internal/exporter/exporter_test.go +++ b/internal/exporter/exporter_test.go @@ -786,6 +786,36 @@ func TestExportNow(t *testing.T) { // TestRefreshConfigFromMetadata tests the refreshConfigFromMetadata function func TestRefreshConfigFromMetadata(t *testing.T) { + t.Run("collector endpoint overrides backend metadata and credentials", func(t *testing.T) { + tmpDB := setupTestDB(t) + defer tmpDB.Close() + + ctx := context.Background() + require.NoError(t, pkgmetadata.SetMetadata(ctx, tmpDB, "backend_base_url", "https://backend.example.com")) + require.NoError(t, pkgmetadata.SetMetadata(ctx, tmpDB, pkgmetadata.MetadataKeyToken, "backend-jwt")) + + cfg := &config.HealthExporterConfig{ + Interval: metav1.Duration{Duration: time.Minute}, + Timeout: metav1.Duration{Duration: 30 * time.Second}, + CollectorEndpoint: "http://fleetint-otel-gateway:4318/", + AuthToken: "stale-token", + } + exporter, err := New(ctx, + WithConfig(cfg), + WithDatabaseConnections(tmpDB, tmpDB), + WithMachineID("test-machine-id"), + ) + require.NoError(t, err) + + he := exporter.(*healthExporter) + he.refreshConfigFromMetadata(ctx) + + assert.Equal(t, "http://fleetint-otel-gateway:4318/v1/metrics", he.options.config.MetricsEndpoint) + assert.Equal(t, "http://fleetint-otel-gateway:4318/v1/logs", he.options.config.LogsEndpoint) + assert.Empty(t, he.options.config.AuthToken) + require.NoError(t, exporter.Stop()) + }) + t.Run("skips when no database connection", func(t *testing.T) { cfg := &config.HealthExporterConfig{ Interval: metav1.Duration{Duration: 1 * time.Minute}, diff --git a/otelcol/Dockerfile b/otelcol/Dockerfile new file mode 100644 index 00000000..55cfefd1 --- /dev/null +++ b/otelcol/Dockerfile @@ -0,0 +1,33 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Builds the fleetint-otelcol custom collector binary using the OTel Collector Builder (ocb). +# +# Build from the repository root: +# docker build -f otelcol/Dockerfile -t fleetint-otelcol:latest . + +FROM golang:1.26.5 AS builder + +# Install the OTel Collector Builder at the same version as our components. +RUN go install go.opentelemetry.io/collector/cmd/builder@v0.156.0 +ENV CGO_ENABLED=0 +ENV GOOS=linux + +WORKDIR /src +# Copy only the collector sources required by the local module replacement. +COPY otelcol/otelcol-builder.yaml ./otelcol/otelcol-builder.yaml +COPY otelcol/auth/sakauth/ ./otelcol/auth/sakauth/ + +WORKDIR /src/otelcol +RUN builder --config=otelcol-builder.yaml --skip-compilation=false +RUN cp bin/fleetint-otelcol /bin/fleetint-otelcol + +# --- + +FROM gcr.io/distroless/static:nonroot + +COPY --from=builder /bin/fleetint-otelcol /fleetint-otelcol + +EXPOSE 4318 13133 + +ENTRYPOINT ["/fleetint-otelcol"] diff --git a/otelcol/auth/sakauth/collector_integration_test.go b/otelcol/auth/sakauth/collector_integration_test.go new file mode 100644 index 00000000..b1988f42 --- /dev/null +++ b/otelcol/auth/sakauth/collector_integration_test.go @@ -0,0 +1,336 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package sakauth + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "strconv" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + "go.opentelemetry.io/collector/pdata/pcommon" + "go.opentelemetry.io/collector/pdata/plog" + "go.opentelemetry.io/collector/pdata/plog/plogotlp" + "go.opentelemetry.io/collector/pdata/pmetric" + "go.opentelemetry.io/collector/pdata/pmetric/pmetricotlp" +) + +const collectorIntegrationEnv = "FLEETINT_OTELCOL_INTEGRATION" + +type backendRequest struct { + authorization string + actorID string + contentType string + metrics pmetric.Metrics + logs plog.Logs + err error +} + +func TestCollectorGatewayEndToEnd(t *testing.T) { + if os.Getenv(collectorIntegrationEnv) != "1" { + t.Skipf("set %s=1 after building otelcol/bin/fleetint-otelcol", collectorIntegrationEnv) + } + + collectorBinary := filepath.Clean(filepath.Join("..", "..", "bin", "fleetint-otelcol")) + info, err := os.Stat(collectorBinary) + require.NoError(t, err, "build the collector before running this integration test") + require.False(t, info.IsDir()) + + jwt := testJWT(t, "integration-customer") + enrollmentRequests := make(chan backendRequest, 1) + metricRequests := make(chan backendRequest, 1) + logRequests := make(chan backendRequest, 1) + backend := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/enroll": + enrollmentRequests <- backendRequest{ + authorization: r.Header.Get("Authorization"), + contentType: r.Header.Get("Content-Type"), + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(enrollResponse{JWTAssertion: jwt}) + case "/metrics": + payload, readErr := io.ReadAll(io.LimitReader(r.Body, 1<<20)) + request := pmetricotlp.NewExportRequest() + if readErr == nil { + readErr = request.UnmarshalProto(payload) + } + metricRequests <- backendRequest{ + authorization: r.Header.Get("Authorization"), + actorID: r.Header.Get("Nv-Actor-Id"), + contentType: r.Header.Get("Content-Type"), + metrics: request.Metrics(), + err: readErr, + } + if readErr != nil { + http.Error(w, "invalid OTLP payload", http.StatusBadRequest) + return + } + responsePayload, marshalErr := pmetricotlp.NewExportResponse().MarshalProto() + if marshalErr != nil { + http.Error(w, "failed to encode OTLP response", http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "application/x-protobuf") + _, _ = w.Write(responsePayload) + case "/logs": + payload, readErr := io.ReadAll(io.LimitReader(r.Body, 1<<20)) + request := plogotlp.NewExportRequest() + if readErr == nil { + readErr = request.UnmarshalProto(payload) + } + logRequests <- backendRequest{ + authorization: r.Header.Get("Authorization"), + actorID: r.Header.Get("Nv-Actor-Id"), + contentType: r.Header.Get("Content-Type"), + logs: request.Logs(), + err: readErr, + } + if readErr != nil { + http.Error(w, "invalid OTLP payload", http.StatusBadRequest) + return + } + responsePayload, marshalErr := plogotlp.NewExportResponse().MarshalProto() + if marshalErr != nil { + http.Error(w, "failed to encode OTLP response", http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "application/x-protobuf") + _, _ = w.Write(responsePayload) + default: + http.NotFound(w, r) + } + })) + defer backend.Close() + + receiverPort := availableTCPPort(t) + healthPort := availableTCPPort(t) + configPath := filepath.Join(t.TempDir(), "collector.yaml") + require.NoError(t, os.WriteFile(configPath, []byte(fmt.Sprintf(` +extensions: + sakauth: + enroll_endpoint: %[1]s/enroll + sak_token: integration-sak + tls: + insecure_skip_verify: true + health_check: + endpoint: 127.0.0.1:%[2]d +receivers: + otlp: + protocols: + http: + endpoint: 127.0.0.1:%[3]d +processors: + batch: + timeout: 100ms + send_batch_size: 1 +exporters: + otlp_http/backend: + metrics_endpoint: %[1]s/metrics + logs_endpoint: %[1]s/logs + compression: none + tls: + insecure_skip_verify: true + auth: + authenticator: sakauth +service: + extensions: [sakauth, health_check] + pipelines: + metrics: + receivers: [otlp] + processors: [batch] + exporters: [otlp_http/backend] + logs: + receivers: [otlp] + processors: [batch] + exporters: [otlp_http/backend] +`, backend.URL, healthPort, receiverPort)), 0o600)) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + logPath := filepath.Join(t.TempDir(), "collector.log") + logFile, err := os.Create(logPath) + require.NoError(t, err) + cmd := exec.CommandContext(ctx, collectorBinary, "--config="+configPath) + cmd.Env = os.Environ() + cmd.Stdout = logFile + cmd.Stderr = logFile + require.NoError(t, cmd.Start()) + waitCh := make(chan error, 1) + go func() { + waitCh <- cmd.Wait() + }() + var stopOnce sync.Once + stopCollector := func() { + stopOnce.Do(func() { + cancel() + select { + case <-waitCh: + case <-time.After(5 * time.Second): + _ = cmd.Process.Kill() + <-waitCh + } + _ = logFile.Close() + }) + } + t.Cleanup(stopCollector) + + healthURL := "http://127.0.0.1:" + strconv.Itoa(healthPort) + "/" + if err := waitForCollector(ctx, healthURL); err != nil { + stopCollector() + logs, _ := os.ReadFile(logPath) + t.Fatalf("collector failed to become healthy: %v\n%s", err, logs) + } + + select { + case request := <-enrollmentRequests: + require.Equal(t, "Bearer integration-sak", request.authorization) + require.Empty(t, request.contentType) + case <-ctx.Done(): + t.Fatal("timed out waiting for gateway enrollment") + } + + payload := marshalTestMetrics(t) + request, err := http.NewRequestWithContext( + ctx, + http.MethodPost, + "http://127.0.0.1:"+strconv.Itoa(receiverPort)+"/v1/metrics", + bytes.NewReader(payload), + ) + require.NoError(t, err) + request.Header.Set("Content-Type", "application/x-protobuf") + response, err := http.DefaultClient.Do(request) + require.NoError(t, err) + defer response.Body.Close() + require.Equal(t, http.StatusOK, response.StatusCode) + + select { + case backendRequest := <-metricRequests: + require.NoError(t, backendRequest.err) + require.Equal(t, "Bearer "+jwt, backendRequest.authorization) + require.Equal(t, "integration-customer", backendRequest.actorID) + require.Equal(t, "application/x-protobuf", backendRequest.contentType) + require.Equal(t, 1, backendRequest.metrics.MetricCount()) + resourceMetrics := backendRequest.metrics.ResourceMetrics() + require.Equal(t, 1, resourceMetrics.Len()) + machineID, ok := resourceMetrics.At(0).Resource().Attributes().Get("machine.id") + require.True(t, ok) + require.Equal(t, "integration-node", machineID.Str()) + case <-ctx.Done(): + t.Fatal("timed out waiting for gateway OTLP export") + } + + request, err = http.NewRequestWithContext( + ctx, + http.MethodPost, + "http://127.0.0.1:"+strconv.Itoa(receiverPort)+"/v1/logs", + bytes.NewReader(marshalTestLogs(t)), + ) + require.NoError(t, err) + request.Header.Set("Content-Type", "application/x-protobuf") + response, err = http.DefaultClient.Do(request) + require.NoError(t, err) + defer response.Body.Close() + require.Equal(t, http.StatusOK, response.StatusCode) + + select { + case backendRequest := <-logRequests: + require.NoError(t, backendRequest.err) + require.Equal(t, "Bearer "+jwt, backendRequest.authorization) + require.Equal(t, "integration-customer", backendRequest.actorID) + require.Equal(t, "application/x-protobuf", backendRequest.contentType) + require.Equal(t, 1, backendRequest.logs.LogRecordCount()) + resourceLogs := backendRequest.logs.ResourceLogs() + require.Equal(t, 1, resourceLogs.Len()) + machineID, ok := resourceLogs.At(0).Resource().Attributes().Get("machine.id") + require.True(t, ok) + require.Equal(t, "integration-node", machineID.Str()) + case <-ctx.Done(): + t.Fatal("timed out waiting for gateway OTLP log export") + } +} + +func availableTCPPort(t *testing.T) int { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + defer listener.Close() + return listener.Addr().(*net.TCPAddr).Port +} + +func waitForCollector(ctx context.Context, healthURL string) error { + ticker := time.NewTicker(50 * time.Millisecond) + defer ticker.Stop() + for { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, healthURL, nil) + if err != nil { + return err + } + resp, err := http.DefaultClient.Do(req) + if err == nil { + _ = resp.Body.Close() + if resp.StatusCode == http.StatusOK { + return nil + } + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-ticker.C: + } + } +} + +func marshalTestMetrics(t *testing.T) []byte { + t.Helper() + metrics := pmetric.NewMetrics() + resourceMetrics := metrics.ResourceMetrics().AppendEmpty() + resourceMetrics.Resource().Attributes().PutStr("machine.id", "integration-node") + metric := resourceMetrics.ScopeMetrics().AppendEmpty().Metrics().AppendEmpty() + metric.SetName("fleetint.gateway.integration") + dataPoint := metric.SetEmptyGauge().DataPoints().AppendEmpty() + dataPoint.SetIntValue(1) + dataPoint.SetTimestamp(pcommon.Timestamp(time.Now().UnixNano())) + + payload, err := pmetricotlp.NewExportRequestFromMetrics(metrics).MarshalProto() + require.NoError(t, err) + return payload +} + +func marshalTestLogs(t *testing.T) []byte { + t.Helper() + logs := plog.NewLogs() + resourceLogs := logs.ResourceLogs().AppendEmpty() + resourceLogs.Resource().Attributes().PutStr("machine.id", "integration-node") + logRecord := resourceLogs.ScopeLogs().AppendEmpty().LogRecords().AppendEmpty() + logRecord.SetTimestamp(pcommon.Timestamp(time.Now().UnixNano())) + logRecord.Body().SetStr("gateway integration test") + + payload, err := plogotlp.NewExportRequestFromLogs(logs).MarshalProto() + require.NoError(t, err) + return payload +} diff --git a/otelcol/auth/sakauth/config.go b/otelcol/auth/sakauth/config.go new file mode 100644 index 00000000..ce8e12e8 --- /dev/null +++ b/otelcol/auth/sakauth/config.go @@ -0,0 +1,61 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package sakauth + +import ( + "fmt" + "net/url" + + "go.opentelemetry.io/collector/config/configopaque" + "go.opentelemetry.io/collector/config/configtls" +) + +// Config holds the configuration for the Fleet Intelligence auth extension. +type Config struct { + // EnrollEndpoint is the backend enrollment URL used to obtain a JWT. + // Example: https://backend/api/v1/agent/enroll + EnrollEndpoint string `mapstructure:"enroll_endpoint"` + + // SAKToken is the Service API Key sent as Authorization: Bearer on enrollment + // requests. Required to pass the API gateway that sits in front of the backend + // receiver. The customer identity is derived by the backend from the SAK and + // embedded in the returned JWT's assertion.customer_id claim. + // Injected from a Kubernetes Secret via environment variable. + SAKToken configopaque.String `mapstructure:"sak_token"` + + // TLS configures certificate verification for the enrollment endpoint. + TLS configtls.ClientConfig `mapstructure:"tls"` +} + +func (c *Config) Validate() error { + if c.EnrollEndpoint == "" { + return fmt.Errorf("sakauth: enroll_endpoint is required") + } + endpoint, err := url.Parse(c.EnrollEndpoint) + if err != nil || endpoint.Scheme != "https" || endpoint.Host == "" { + return fmt.Errorf("sakauth: enroll_endpoint must be an absolute HTTPS URL") + } + if endpoint.User != nil || endpoint.Fragment != "" { + return fmt.Errorf("sakauth: enroll_endpoint must not contain credentials or a fragment") + } + if c.SAKToken == "" { + return fmt.Errorf("sakauth: sak_token is required") + } + if c.TLS.Insecure { + return fmt.Errorf("sakauth: tls.insecure is not allowed; enroll_endpoint must use HTTPS") + } + return nil +} diff --git a/otelcol/auth/sakauth/extension.go b/otelcol/auth/sakauth/extension.go new file mode 100644 index 00000000..5377b714 --- /dev/null +++ b/otelcol/auth/sakauth/extension.go @@ -0,0 +1,287 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package sakauth implements an OpenTelemetry Collector auth.Client extension +// that enrolls with the Fleet Intelligence backend using a SAK token +// (Authorization: Bearer) to obtain a short-lived JWT, and automatically +// refreshes it via response headers or on 401 responses. +// +// The customer ID is not configured explicitly — it is extracted from the +// JWT's assertion.customer_id claim after enrollment and forwarded as +// Nv-Actor-Id on outbound OTLP requests. +package sakauth + +import ( + "context" + "encoding/base64" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + "sync" + "time" + + "go.opentelemetry.io/collector/component" + "go.opentelemetry.io/collector/config/configtls" + "go.opentelemetry.io/collector/extension/extensionauth" +) + +// enrollResponse mirrors the backend enrollment response. +type enrollResponse struct { + JWTAssertion string `json:"jwtAssertion"` + LegacyJWTAssertion string `json:"jwt_assertion"` +} + +func (r enrollResponse) token() string { + if r.JWTAssertion != "" { + return r.JWTAssertion + } + return r.LegacyJWTAssertion +} + +// sakAuthExtension implements HTTP client authentication using the SAK→JWT enrollment flow. +type sakAuthExtension struct { + cfg *Config + client *http.Client + jwt string + customerID string // extracted from JWT assertion.customer_id after enrollment + mu sync.RWMutex + refreshMu sync.Mutex +} + +// Ensure the extension satisfies the HTTP client authenticator interface. +var _ extensionauth.HTTPClient = (*sakAuthExtension)(nil) + +func newSAKAuthExtension(cfg *Config) (*sakAuthExtension, error) { + if err := cfg.Validate(); err != nil { + return nil, err + } + client, err := newEnrollmentHTTPClient(&cfg.TLS) + if err != nil { + return nil, fmt.Errorf("sakauth: invalid TLS configuration: %w", err) + } + return &sakAuthExtension{ + cfg: cfg, + client: client, + }, nil +} + +func newEnrollmentHTTPClient(tlsConfig *configtls.ClientConfig) (*http.Client, error) { + transport := http.DefaultTransport.(*http.Transport).Clone() + loadedTLSConfig, err := tlsConfig.LoadTLSConfig(context.Background()) + if err != nil { + return nil, err + } + transport.TLSClientConfig = loadedTLSConfig + return &http.Client{ + Timeout: 30 * time.Second, + Transport: transport, + // Never forward the SAK to a redirect target. + CheckRedirect: func(*http.Request, []*http.Request) error { + return http.ErrUseLastResponse + }, + }, nil +} + +// Start fetches the initial JWT and extracts the customer ID from it. +func (e *sakAuthExtension) Start(ctx context.Context, _ component.Host) error { + jwt, err := e.performEnrollment(ctx) + if err != nil { + return fmt.Errorf("sakauth: initial enrollment failed: %w", err) + } + e.mu.Lock() + e.jwt = jwt + e.customerID = extractCustomerID(jwt) + e.mu.Unlock() + return nil +} + +// Shutdown is a no-op for this extension. +func (e *sakAuthExtension) Shutdown(_ context.Context) error { + return nil +} + +// RoundTripper returns an http.RoundTripper that injects the current JWT into +// each outbound request and refreshes it on a 401 response. +func (e *sakAuthExtension) RoundTripper(base http.RoundTripper) (http.RoundTripper, error) { + if base == nil { + base = http.DefaultTransport + } + return &sakRoundTripper{base: base, ext: e}, nil +} + +func (e *sakAuthExtension) getJWT() string { + e.mu.RLock() + defer e.mu.RUnlock() + return e.jwt +} + +func (e *sakAuthExtension) snapshot() (jwt, customerID string) { + e.mu.RLock() + defer e.mu.RUnlock() + return e.jwt, e.customerID +} + +func (e *sakAuthExtension) refreshJWT(ctx context.Context, staleJWT string) (string, error) { + e.refreshMu.Lock() + defer e.refreshMu.Unlock() + + // Another request may already have refreshed the shared token while this + // request was waiting. + if current := e.getJWT(); current != "" && current != staleJWT { + return current, nil + } + + jwt, err := e.performEnrollment(ctx) + if err != nil { + return "", err + } + e.mu.Lock() + e.jwt = jwt + e.customerID = extractCustomerID(jwt) + e.mu.Unlock() + return jwt, nil +} + +// performEnrollment calls the backend enrollment endpoint with the SAK token +// and returns the JWT from the response. +func (e *sakAuthExtension) performEnrollment(ctx context.Context) (string, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, e.cfg.EnrollEndpoint, nil) + if err != nil { + return "", fmt.Errorf("failed to build enrollment request: %w", err) + } + req.Header.Set("Authorization", "Bearer "+string(e.cfg.SAKToken)) + req.Header.Set("User-Agent", "fleetint-otelcol") + + resp, err := e.client.Do(req) + if err != nil { + return "", fmt.Errorf("enrollment request failed: %w", err) + } + defer resp.Body.Close() + + const maxEnrollmentResponseSize = 1 << 20 + body, err := io.ReadAll(io.LimitReader(resp.Body, maxEnrollmentResponseSize+1)) + if err != nil { + return "", fmt.Errorf("failed to read enrollment response: %w", err) + } + if len(body) > maxEnrollmentResponseSize { + return "", fmt.Errorf("enrollment response exceeds %d bytes", maxEnrollmentResponseSize) + } + + if resp.StatusCode != http.StatusOK { + return "", fmt.Errorf("enrollment returned HTTP %d", resp.StatusCode) + } + + var result enrollResponse + if err := json.Unmarshal(body, &result); err != nil { + return "", fmt.Errorf("failed to parse enrollment response: %w", err) + } + jwt := result.token() + if jwt == "" { + return "", fmt.Errorf("enrollment response missing jwtAssertion field") + } + return jwt, nil +} + +// extractCustomerID decodes the JWT payload and returns assertion.customer_id. +// Returns an empty string if the claim is absent or the token is malformed. +func extractCustomerID(token string) string { + parts := strings.SplitN(token, ".", 3) + if len(parts) != 3 { + return "" + } + payload := parts[1] + if r := len(payload) % 4; r != 0 { + payload += strings.Repeat("=", 4-r) + } + b, err := base64.URLEncoding.DecodeString(payload) + if err != nil { + return "" + } + var claims struct { + Assertion struct { + CustomerID string `json:"customer_id"` + } `json:"assertion"` + } + if err := json.Unmarshal(b, &claims); err != nil { + return "" + } + return claims.Assertion.CustomerID +} + +// sakRoundTripper injects the JWT and handles transparent token refresh on 401. +type sakRoundTripper struct { + base http.RoundTripper + ext *sakAuthExtension +} + +func (rt *sakRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) { + req = req.Clone(req.Context()) + usedJWT, customerID := rt.ext.snapshot() + req.Header.Set("Authorization", "Bearer "+usedJWT) + if customerID != "" { + req.Header.Set("Nv-Actor-Id", customerID) + } else { + req.Header.Del("Nv-Actor-Id") + } + + resp, err := rt.base.RoundTrip(req) + if err != nil { + return nil, err + } + + // The backend proactively sends a refreshed JWT in the response header when + // the current token is near expiry (>80% of max age). Store it immediately. + if newJWT := resp.Header.Get("jwt_assertion"); newJWT != "" { + rt.ext.mu.Lock() + rt.ext.jwt = newJWT + rt.ext.customerID = extractCustomerID(newJWT) + rt.ext.mu.Unlock() + } + + if resp.StatusCode != http.StatusUnauthorized { + return resp, nil + } + + // 401 fallback: re-enroll and retry once. + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 4096)) + resp.Body.Close() + + // Restore the request body for the retry. The OTel exporter sets GetBody + // so we can get a fresh reader; if it didn't, we can't retry safely. + if req.GetBody == nil { + return nil, fmt.Errorf("sakauth: 401 received but request body is not replayable; cannot retry") + } + newBody, err := req.GetBody() + if err != nil { + return nil, fmt.Errorf("sakauth: failed to replay request body for retry: %w", err) + } + + newJWT, err := rt.ext.refreshJWT(req.Context(), usedJWT) + if err != nil { + return nil, fmt.Errorf("sakauth: token refresh after 401 failed: %w", err) + } + + req = req.Clone(req.Context()) + req.Body = newBody + req.Header.Set("Authorization", "Bearer "+newJWT) + if customerID := extractCustomerID(newJWT); customerID != "" { + req.Header.Set("Nv-Actor-Id", customerID) + } else { + req.Header.Del("Nv-Actor-Id") + } + return rt.base.RoundTrip(req) +} diff --git a/otelcol/auth/sakauth/extension_test.go b/otelcol/auth/sakauth/extension_test.go new file mode 100644 index 00000000..c7ac2eeb --- /dev/null +++ b/otelcol/auth/sakauth/extension_test.go @@ -0,0 +1,479 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package sakauth + +import ( + "bytes" + "context" + "encoding/base64" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + + "github.com/stretchr/testify/require" + "go.opentelemetry.io/collector/component" + "go.opentelemetry.io/collector/config/configopaque" + "go.opentelemetry.io/collector/config/configtls" + "go.opentelemetry.io/collector/extension" +) + +type roundTripperFunc func(*http.Request) (*http.Response, error) + +func (f roundTripperFunc) RoundTrip(req *http.Request) (*http.Response, error) { + return f(req) +} + +func testJWT(t *testing.T, customerID string) string { + t.Helper() + payload, err := json.Marshal(map[string]any{ + "assertion": map[string]string{"customer_id": customerID}, + }) + require.NoError(t, err) + return "header." + base64.RawURLEncoding.EncodeToString(payload) + ".signature" +} + +func TestConfigValidateRequiresSecureEndpoint(t *testing.T) { + require.ErrorContains(t, (&Config{}).Validate(), "enroll_endpoint is required") + require.ErrorContains(t, (&Config{ + EnrollEndpoint: "https://backend.example/api/v1/agent/enroll", + }).Validate(), "sak_token is required") + + tests := []struct { + name string + endpoint string + }{ + {name: "HTTP", endpoint: "http://backend.example/enroll"}, + {name: "relative", endpoint: "/enroll"}, + {name: "credentials", endpoint: "https://user:pass@backend.example/enroll"}, + {name: "fragment", endpoint: "https://backend.example/enroll#secret"}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + cfg := &Config{EnrollEndpoint: tc.endpoint, SAKToken: configopaque.String("sak")} + require.Error(t, cfg.Validate()) + }) + } + + require.NoError(t, (&Config{ + EnrollEndpoint: "https://backend.example/api/v1/health/enroll", + SAKToken: configopaque.String("sak"), + }).Validate()) + require.ErrorContains(t, (&Config{ + EnrollEndpoint: "https://backend.example/api/v1/health/enroll", + SAKToken: configopaque.String("sak"), + TLS: configtls.ClientConfig{Insecure: true}, + }).Validate(), "tls.insecure is not allowed") + + cfg := &Config{SAKToken: configopaque.String("sak-token")} + require.Equal(t, "[REDACTED]", cfg.SAKToken.String()) + require.NotContains(t, cfg.SAKToken.String(), "sak-token") +} + +func TestExtractCustomerID(t *testing.T) { + require.Equal(t, "customer-1", extractCustomerID(testJWT(t, "customer-1"))) + require.Empty(t, extractCustomerID("")) + require.Empty(t, extractCustomerID("not.a.jwt.with.too.many.parts")) + require.Empty(t, extractCustomerID("header.%%%.signature")) + require.Empty(t, extractCustomerID("header.e30.signature")) +} + +func TestEnrollResponseToken(t *testing.T) { + require.Equal(t, "current", (enrollResponse{ + JWTAssertion: "current", + LegacyJWTAssertion: "legacy", + }).token()) + require.Equal(t, "legacy", (enrollResponse{ + LegacyJWTAssertion: "legacy", + }).token()) +} + +func TestFactoryCreatesExtension(t *testing.T) { + cfg := createDefaultConfig() + require.IsType(t, &Config{}, cfg) + require.NotNil(t, NewFactory()) + + authExtension, err := createExtension(context.Background(), extension.Settings{}, &Config{ + EnrollEndpoint: "https://backend.example/api/v1/agent/enroll", + SAKToken: configopaque.String("sak-token"), + }) + require.NoError(t, err) + require.IsType(t, &sakAuthExtension{}, authExtension) + require.NoError(t, authExtension.Shutdown(context.Background())) + + _, err = createExtension(context.Background(), extension.Settings{}, &Config{}) + require.Error(t, err) +} + +func TestRoundTripperDefaultsBaseTransport(t *testing.T) { + ext := &sakAuthExtension{} + transport, err := ext.RoundTripper(nil) + require.NoError(t, err) + require.NotNil(t, transport) + require.Implements(t, (*component.Component)(nil), ext) +} + +func TestEnrollmentAndUnauthorizedRefresh(t *testing.T) { + firstJWT := testJWT(t, "customer-1") + secondJWT := testJWT(t, "customer-2") + var enrollments atomic.Int32 + var exports atomic.Int32 + enrollmentAuth := make(chan string, 2) + exportHeaders := make(chan [2]string, 2) + handlerErrors := make(chan error, 2) + + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/enroll": + enrollmentAuth <- r.Header.Get("Authorization") + count := enrollments.Add(1) + token := firstJWT + if count > 1 { + token = secondJWT + } + if err := json.NewEncoder(w).Encode(enrollResponse{JWTAssertion: token}); err != nil { + handlerErrors <- err + } + case "/v1/metrics": + count := exports.Add(1) + exportHeaders <- [2]string{ + r.Header.Get("Authorization"), + r.Header.Get("Nv-Actor-Id"), + } + if count == 1 { + w.WriteHeader(http.StatusUnauthorized) + return + } + w.WriteHeader(http.StatusOK) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + + ext, err := newSAKAuthExtension(&Config{ + EnrollEndpoint: server.URL + "/enroll", + SAKToken: configopaque.String("sak-token"), + }) + require.NoError(t, err) + ext.client = server.Client() + ext.client.CheckRedirect = func(*http.Request, []*http.Request) error { + return http.ErrUseLastResponse + } + require.NoError(t, ext.Start(context.Background(), nil)) + + transport, err := ext.RoundTripper(server.Client().Transport) + require.NoError(t, err) + client := &http.Client{Transport: transport} + req, err := http.NewRequest(http.MethodPost, server.URL+"/v1/metrics", bytes.NewReader([]byte("payload"))) + require.NoError(t, err) + + resp, err := client.Do(req) + require.NoError(t, err) + defer resp.Body.Close() + require.Equal(t, http.StatusOK, resp.StatusCode) + require.Equal(t, int32(2), enrollments.Load()) + require.Equal(t, int32(2), exports.Load()) + require.Equal(t, "Bearer sak-token", <-enrollmentAuth) + require.Equal(t, "Bearer sak-token", <-enrollmentAuth) + require.Equal(t, [2]string{"Bearer " + firstJWT, "customer-1"}, <-exportHeaders) + require.Equal(t, [2]string{"Bearer " + secondJWT, "customer-2"}, <-exportHeaders) + select { + case handlerErr := <-handlerErrors: + require.NoError(t, handlerErr) + default: + } +} + +func TestEnrollmentDoesNotFollowRedirects(t *testing.T) { + var redirectTargetCalls atomic.Int32 + target := httptest.NewTLSServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { + redirectTargetCalls.Add(1) + })) + defer target.Close() + + redirector := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Redirect(w, r, target.URL, http.StatusTemporaryRedirect) + })) + defer redirector.Close() + + ext, err := newSAKAuthExtension(&Config{ + EnrollEndpoint: redirector.URL, + SAKToken: configopaque.String("sak-token"), + }) + require.NoError(t, err) + ext.client = redirector.Client() + ext.client.CheckRedirect = func(*http.Request, []*http.Request) error { + return http.ErrUseLastResponse + } + + _, err = ext.performEnrollment(context.Background()) + require.ErrorContains(t, err, "HTTP 307") + require.Zero(t, redirectTargetCalls.Load()) +} + +func TestEnrollmentRejectsOversizedResponse(t *testing.T) { + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write(bytes.Repeat([]byte("x"), (1<<20)+1)) + })) + defer server.Close() + + ext, err := newSAKAuthExtension(&Config{ + EnrollEndpoint: server.URL, + SAKToken: configopaque.String("sak-token"), + }) + require.NoError(t, err) + ext.client = server.Client() + + _, err = ext.performEnrollment(context.Background()) + require.ErrorContains(t, err, "exceeds 1048576 bytes") +} + +func TestEnrollmentResponseErrors(t *testing.T) { + tests := []struct { + name string + status int + body string + errorContains string + }{ + { + name: "HTTP status does not expose body", + status: http.StatusUnauthorized, + body: "sensitive backend details", + errorContains: "HTTP 401", + }, + { + name: "invalid JSON", + status: http.StatusOK, + body: "{", + errorContains: "failed to parse enrollment response", + }, + { + name: "missing assertion", + status: http.StatusOK, + body: "{}", + errorContains: "missing jwtAssertion", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(tc.status) + _, _ = io.WriteString(w, tc.body) + })) + defer server.Close() + + ext, err := newSAKAuthExtension(&Config{ + EnrollEndpoint: server.URL, + SAKToken: configopaque.String("sak-token"), + }) + require.NoError(t, err) + ext.client = server.Client() + + _, err = ext.performEnrollment(context.Background()) + require.ErrorContains(t, err, tc.errorContains) + require.NotContains(t, err.Error(), "sensitive backend details") + }) + } +} + +func TestEnrollmentHonorsCanceledContext(t *testing.T) { + server := httptest.NewTLSServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {})) + defer server.Close() + + ext, err := newSAKAuthExtension(&Config{ + EnrollEndpoint: server.URL, + SAKToken: configopaque.String("sak-token"), + }) + require.NoError(t, err) + ext.client = server.Client() + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + _, err = ext.performEnrollment(ctx) + require.ErrorIs(t, err, context.Canceled) +} + +func TestUnauthorizedNonReplayableRequestDoesNotRefresh(t *testing.T) { + ext := &sakAuthExtension{jwt: "jwt-token"} + transport, err := ext.RoundTripper(roundTripperFunc(func(*http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: http.StatusUnauthorized, + Header: http.Header{}, + Body: http.NoBody, + }, nil + })) + require.NoError(t, err) + + req, err := http.NewRequest( + http.MethodPost, + "https://backend.example/v1/metrics", + io.NopCloser(strings.NewReader("payload")), + ) + require.NoError(t, err) + require.Nil(t, req.GetBody) + + _, err = transport.RoundTrip(req) + require.ErrorContains(t, err, "request body is not replayable") +} + +func TestSecondUnauthorizedResponseIsReturnedWithoutRefreshLoop(t *testing.T) { + firstJWT := testJWT(t, "customer-1") + secondJWT := testJWT(t, "customer-2") + var enrollments atomic.Int32 + var exports atomic.Int32 + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/enroll": + token := firstJWT + if enrollments.Add(1) > 1 { + token = secondJWT + } + _ = json.NewEncoder(w).Encode(enrollResponse{JWTAssertion: token}) + case "/v1/metrics": + exports.Add(1) + w.WriteHeader(http.StatusUnauthorized) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + + ext, err := newSAKAuthExtension(&Config{ + EnrollEndpoint: server.URL + "/enroll", + SAKToken: configopaque.String("sak-token"), + }) + require.NoError(t, err) + ext.client = server.Client() + require.NoError(t, ext.Start(context.Background(), nil)) + transport, err := ext.RoundTripper(server.Client().Transport) + require.NoError(t, err) + + req, err := http.NewRequest(http.MethodPost, server.URL+"/v1/metrics", bytes.NewReader([]byte("payload"))) + require.NoError(t, err) + resp, err := transport.RoundTrip(req) + require.NoError(t, err) + defer resp.Body.Close() + require.Equal(t, http.StatusUnauthorized, resp.StatusCode) + require.Equal(t, int32(2), enrollments.Load()) + require.Equal(t, int32(2), exports.Load()) +} + +func TestResponseHeaderRefreshesJWT(t *testing.T) { + firstJWT := testJWT(t, "customer-1") + secondJWT := testJWT(t, "customer-2") + ext := &sakAuthExtension{jwt: firstJWT, customerID: "customer-1"} + + var requests atomic.Int32 + base := roundTripperFunc(func(req *http.Request) (*http.Response, error) { + count := requests.Add(1) + if count == 1 { + require.Equal(t, "Bearer "+firstJWT, req.Header.Get("Authorization")) + return &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Jwt_assertion": []string{secondJWT}}, + Body: http.NoBody, + }, nil + } + require.Equal(t, "Bearer "+secondJWT, req.Header.Get("Authorization")) + require.Equal(t, "customer-2", req.Header.Get("Nv-Actor-Id")) + return &http.Response{StatusCode: http.StatusOK, Header: http.Header{}, Body: http.NoBody}, nil + }) + + transport, err := ext.RoundTripper(base) + require.NoError(t, err) + for range 2 { + req, err := http.NewRequest(http.MethodPost, "https://backend.example/v1/metrics", bytes.NewReader([]byte("payload"))) + require.NoError(t, err) + resp, err := transport.RoundTrip(req) + require.NoError(t, err) + require.NoError(t, resp.Body.Close()) + } + require.Equal(t, int32(2), requests.Load()) +} + +func TestConcurrentUnauthorizedResponsesSingleRefresh(t *testing.T) { + firstJWT := testJWT(t, "customer-1") + secondJWT := testJWT(t, "customer-2") + var enrollments atomic.Int32 + + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/enroll": + count := enrollments.Add(1) + token := firstJWT + if count > 1 { + token = secondJWT + } + _ = json.NewEncoder(w).Encode(enrollResponse{JWTAssertion: token}) + case "/v1/metrics": + if r.Header.Get("Authorization") == "Bearer "+firstJWT { + w.WriteHeader(http.StatusUnauthorized) + return + } + w.WriteHeader(http.StatusOK) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + + ext, err := newSAKAuthExtension(&Config{ + EnrollEndpoint: server.URL + "/enroll", + SAKToken: configopaque.String("sak-token"), + }) + require.NoError(t, err) + ext.client = server.Client() + require.NoError(t, ext.Start(context.Background(), nil)) + transport, err := ext.RoundTripper(server.Client().Transport) + require.NoError(t, err) + + const concurrency = 8 + start := make(chan struct{}) + errs := make(chan error, concurrency) + var wg sync.WaitGroup + for range concurrency { + wg.Add(1) + go func() { + defer wg.Done() + <-start + req, err := http.NewRequest(http.MethodPost, server.URL+"/v1/metrics", bytes.NewReader([]byte("payload"))) + if err != nil { + errs <- err + return + } + resp, err := transport.RoundTrip(req) + if err == nil { + err = resp.Body.Close() + } + errs <- err + }() + } + close(start) + wg.Wait() + close(errs) + + for err := range errs { + require.NoError(t, err) + } + require.Equal(t, int32(2), enrollments.Load()) +} diff --git a/otelcol/auth/sakauth/factory.go b/otelcol/auth/sakauth/factory.go new file mode 100644 index 00000000..702d61a2 --- /dev/null +++ b/otelcol/auth/sakauth/factory.go @@ -0,0 +1,48 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package sakauth + +import ( + "context" + + "go.opentelemetry.io/collector/component" + "go.opentelemetry.io/collector/config/configtls" + "go.opentelemetry.io/collector/extension" +) + +const ( + // typeStr is the identifier registered with the OTel Collector registry. + typeStr = "sakauth" +) + +// NewFactory returns the factory for the sakauth extension. +func NewFactory() extension.Factory { + return extension.NewFactory( + component.MustNewType(typeStr), + createDefaultConfig, + createExtension, + component.StabilityLevelDevelopment, + ) +} + +func createDefaultConfig() component.Config { + return &Config{TLS: configtls.NewDefaultClientConfig()} +} + +//nolint:gocritic // extension.CreateFunc requires Settings by value. +func createExtension(_ context.Context, _ extension.Settings, cfg component.Config) (extension.Extension, error) { + return newSAKAuthExtension(cfg.(*Config)) +} diff --git a/otelcol/auth/sakauth/go.mod b/otelcol/auth/sakauth/go.mod new file mode 100644 index 00000000..ee06bbaf --- /dev/null +++ b/otelcol/auth/sakauth/go.mod @@ -0,0 +1,53 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +module github.com/NVIDIA/fleet-intelligence-agent/otelcol/auth/sakauth + +go 1.25.0 + +require ( + github.com/stretchr/testify v1.11.1 + go.opentelemetry.io/collector/component v1.62.0 + go.opentelemetry.io/collector/config/configopaque v1.62.0 + go.opentelemetry.io/collector/config/configtls v1.62.0 + go.opentelemetry.io/collector/extension v1.62.0 + go.opentelemetry.io/collector/extension/extensionauth v1.62.0 + go.opentelemetry.io/collector/pdata v1.62.0 +) + +require ( + github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/davecgh/go-spew v1.1.1 // indirect + github.com/foxboron/go-tpm-keyfiles v0.0.0-20250903184740-5d135037bd4d // indirect + github.com/fsnotify/fsnotify v1.10.1 // indirect + github.com/go-viper/mapstructure/v2 v2.5.0 // indirect + github.com/gobwas/glob v0.2.3 // indirect + github.com/google/go-tpm v0.9.8 // indirect + github.com/hashicorp/go-version v1.9.0 // indirect + github.com/json-iterator/go v1.1.12 // indirect + github.com/knadh/koanf/maps v0.1.2 // indirect + github.com/knadh/koanf/providers/confmap v1.0.0 // indirect + github.com/knadh/koanf/v2 v2.3.5 // indirect + github.com/mitchellh/copystructure v1.2.0 // indirect + github.com/mitchellh/reflectwalk v1.0.2 // indirect + github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect + github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect + go.opentelemetry.io/collector/confmap v1.62.0 // indirect + go.opentelemetry.io/collector/featuregate v1.62.0 // indirect + go.opentelemetry.io/collector/internal/componentalias v0.156.0 // indirect + go.opentelemetry.io/otel v1.44.0 // indirect + go.opentelemetry.io/otel/metric v1.44.0 // indirect + go.opentelemetry.io/otel/trace v1.44.0 // indirect + go.uber.org/multierr v1.11.0 // indirect + go.uber.org/zap v1.28.0 // indirect + go.yaml.in/yaml/v3 v3.0.4 // indirect + golang.org/x/crypto v0.52.0 // indirect + golang.org/x/net v0.55.0 // indirect + golang.org/x/sys v0.45.0 // indirect + golang.org/x/text v0.37.0 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478 // indirect + google.golang.org/grpc v1.82.0 // indirect + google.golang.org/protobuf v1.36.11 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect +) diff --git a/otelcol/auth/sakauth/go.sum b/otelcol/auth/sakauth/go.sum new file mode 100644 index 00000000..3e1425ef --- /dev/null +++ b/otelcol/auth/sakauth/go.sum @@ -0,0 +1,129 @@ +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/foxboron/go-tpm-keyfiles v0.0.0-20250903184740-5d135037bd4d h1:EdO/NMMuCZfxhdzTZLuKAciQSnI2DV+Ppg8+vAYrnqA= +github.com/foxboron/go-tpm-keyfiles v0.0.0-20250903184740-5d135037bd4d/go.mod h1:uAyTlAUxchYuiFjTHmuIEJ4nGSm7iOPaGcAyA81fJ80= +github.com/foxboron/swtpm_test v0.0.0-20230726224112-46aaafdf7006 h1:50sW4r0PcvlpG4PV8tYh2RVCapszJgaOLRCS2subvV4= +github.com/foxboron/swtpm_test v0.0.0-20230726224112-46aaafdf7006/go.mod h1:eIXCMsMYCaqq9m1KSSxXwQG11krpuNPGP3k0uaWrbas= +github.com/fsnotify/fsnotify v1.10.1 h1:b0/UzAf9yR5rhf3RPm9gf3ehBPpf0oZKIjtpKrx59Ho= +github.com/fsnotify/fsnotify v1.10.1/go.mod h1:TLheqan6HD6GBK6PrDWyDPBaEV8LspOxvPSjC+bVfgo= +github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= +github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/go-viper/mapstructure/v2 v2.5.0 h1:vM5IJoUAy3d7zRSVtIwQgBj7BiWtMPfmPEgAXnvj1Ro= +github.com/go-viper/mapstructure/v2 v2.5.0/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM= +github.com/gobwas/glob v0.2.3 h1:A4xDbljILXROh+kObIiy5kIaPYD8e96x1tgBhUI5J+Y= +github.com/gobwas/glob v0.2.3/go.mod h1:d3Ez4x06l9bZtSvzIay5+Yzi0fmZzPgnTbPcKjJAkT8= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= +github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY= +github.com/google/go-tpm-tools v0.4.7 h1:J3ycC8umYxM9A4eF73EofRZu4BxY0jjQnUnkhIBbvws= +github.com/google/go-tpm-tools v0.4.7/go.mod h1:gSyXTZHe3fgbzb6WEGd90QucmsnT1SRdlye82gH8QjQ= +github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/hashicorp/go-version v1.9.0 h1:CeOIz6k+LoN3qX9Z0tyQrPtiB1DFYRPfCIBtaXPSCnA= +github.com/hashicorp/go-version v1.9.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA= +github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= +github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= +github.com/knadh/koanf/maps v0.1.2 h1:RBfmAW5CnZT+PJ1CVc1QSJKf4Xu9kxfQgYVQSu8hpbo= +github.com/knadh/koanf/maps v0.1.2/go.mod h1:npD/QZY3V6ghQDdcQzl1W4ICNVTkohC8E73eI2xW4yI= +github.com/knadh/koanf/providers/confmap v1.0.0 h1:mHKLJTE7iXEys6deO5p6olAiZdG5zwp8Aebir+/EaRE= +github.com/knadh/koanf/providers/confmap v1.0.0/go.mod h1:txHYHiI2hAtF0/0sCmcuol4IDcuQbKTybiB1nOcUo1A= +github.com/knadh/koanf/v2 v2.3.5 h1:2dXJUYaKGm4SGYeoAtBviq9+02JZo/pxQ2ssOd60rJg= +github.com/knadh/koanf/v2 v2.3.5/go.mod h1:gRb40VRAbd4iJMYYD5IxZ6hfuopFcXBpc9bbQpZwo28= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/mitchellh/copystructure v1.2.0 h1:vpKXTN4ewci03Vljg/q9QvCGUDttBOGBIa15WveJJGw= +github.com/mitchellh/copystructure v1.2.0/go.mod h1:qLl+cE2AmVv+CoeAwDPye/v+N2HKCj9FbZEVFJRxO9s= +github.com/mitchellh/reflectwalk v1.0.2 h1:G2LzWKi524PWgd3mLHV8Y5k7s6XUvT0Gef6zxSIeXaQ= +github.com/mitchellh/reflectwalk v1.0.2/go.mod h1:mSTlrgnPZtwu0c4WaC2kGObEpuNDbx0jmZXqmk4esnw= +github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee h1:W5t00kpgFdJifH4BDsTlE89Zl93FEloxaWZfGcifgq8= +github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/rogpeppe/go-internal v1.13.1 h1:KvO1DLK/DRN07sQ1LQKScxyZJuNnedQ5/wKSR38lUII= +github.com/rogpeppe/go-internal v1.13.1/go.mod h1:uMEvuHeurkdAXX61udpOXGD/AzZDWNMNyH2VO9fmH0o= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= +go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +go.opentelemetry.io/collector/component v1.62.0 h1:F1MHUlUEjSJgwcumsCbbH2rRmTK4dC8m/ipp9v4vFh0= +go.opentelemetry.io/collector/component v1.62.0/go.mod h1:NqdVWse4diWnlqh5WurI2KncJuBXe1zzYtxuC9Mmew0= +go.opentelemetry.io/collector/config/configopaque v1.62.0 h1:E64BPiumLcJO501g6XETf/vX6r+AK1ytqBc5UEcmkmI= +go.opentelemetry.io/collector/config/configopaque v1.62.0/go.mod h1:z4FPFfKiO83yJz/DqzjlGofUYF9u1A5U/s9NLaa6L1w= +go.opentelemetry.io/collector/config/configtls v1.62.0 h1:C4WywYuIhIHMkAcWmK19gHxub9KjHdxUREv281bKrvU= +go.opentelemetry.io/collector/config/configtls v1.62.0/go.mod h1:2r+Hlr7RXBs9u03HSd4eYJCLi6hukRQv7o36WrgzNkY= +go.opentelemetry.io/collector/confmap v1.62.0 h1:JF1hNjXeZGDKKyK0QBa9yAtGUado+zj4hLHM0BCag40= +go.opentelemetry.io/collector/confmap v1.62.0/go.mod h1:4rRpkbOkE/LvUSmrMX+jCr94i8P4JtYf93TBvfR5LUA= +go.opentelemetry.io/collector/extension v1.62.0 h1:otGURB9mCfpmRrBr+aI2NS/RjwZr2TZ4Crbqi1N3D7w= +go.opentelemetry.io/collector/extension v1.62.0/go.mod h1:EmaC0bqQ6cc4cEkiR29r04UZWQLVT7KLJTfzfycLEEQ= +go.opentelemetry.io/collector/extension/extensionauth v1.62.0 h1:2yhRG9OFxUSCrc+0GqgON+WKVciV65s+rrnOoWLR4V4= +go.opentelemetry.io/collector/extension/extensionauth v1.62.0/go.mod h1:bJV7oxY/JWRDXrZDbjuv9DjU0NNNs6r+YQcYkWVzf7o= +go.opentelemetry.io/collector/featuregate v1.62.0 h1:pYY7RlulSCTOS9mFWxasMLwYJCfNXHtnOkZlv3jg/V4= +go.opentelemetry.io/collector/featuregate v1.62.0/go.mod h1:4ga1QBMPEejXXmpyJS8lmaRpknJ3Lb9Bvk6e420bUFU= +go.opentelemetry.io/collector/internal/componentalias v0.156.0 h1:Ku9pTxb4imQME35PoR0mzXv+v3jLtbGxRT0PiH4j034= +go.opentelemetry.io/collector/internal/componentalias v0.156.0/go.mod h1:1YJUCQ6Her24ZhJnYgKSuov7AaFB1jEPawvEAjrp1ms= +go.opentelemetry.io/collector/internal/testutil v0.156.0 h1:Nu02vhHA2UQ3Yjyjisk3N24HHxwvw7PQiTz9O1PuiUY= +go.opentelemetry.io/collector/internal/testutil v0.156.0/go.mod h1:Jkjs6rkqs973LqgZ0Fe3zrokQRKULYXPIf4HuqStiEE= +go.opentelemetry.io/collector/pdata v1.62.0 h1:xGdwl2Cs5Rq5nKs0nYvAxm3Qq20HcySVAmUElATS8Es= +go.opentelemetry.io/collector/pdata v1.62.0/go.mod h1:WFy5R6XGpz2Q4MaekeEm+qc4GY5V3+BhQIwGPkp+fj0= +go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU= +go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc= +go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc= +go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo= +go.opentelemetry.io/otel/sdk v1.43.0 h1:pi5mE86i5rTeLXqoF/hhiBtUNcrAGHLKQdhg4h4V9Dg= +go.opentelemetry.io/otel/sdk v1.43.0/go.mod h1:P+IkVU3iWukmiit/Yf9AWvpyRDlUeBaRg6Y+C58QHzg= +go.opentelemetry.io/otel/sdk/metric v1.43.0 h1:S88dyqXjJkuBNLeMcVPRFXpRw2fuwdvfCGLEo89fDkw= +go.opentelemetry.io/otel/sdk/metric v1.43.0/go.mod h1:C/RJtwSEJ5hzTiUz5pXF1kILHStzb9zFlIEe85bhj6A= +go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk= +go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE= +go.opentelemetry.io/proto/slim/otlp v1.10.0 h1:iR97Vs/ZDR+y9TfuP9b1XBtdPWeC+OMslIBmhcLU7jM= +go.opentelemetry.io/proto/slim/otlp v1.10.0/go.mod h1:lV9250stpjYLPNA5viFabIgP2QlUGRT1GdTgAf8SIUk= +go.opentelemetry.io/proto/slim/otlp/collector/profiles/v1development v0.3.0 h1:RUF5rO0hAlgiJt1fzQVzcVs3vZVNHIcMLgOgG4rWNcQ= +go.opentelemetry.io/proto/slim/otlp/collector/profiles/v1development v0.3.0/go.mod h1:I89cynRj8y+383o7tEQVg2SVA6SRgDVIouWPUVXjx0U= +go.opentelemetry.io/proto/slim/otlp/profiles/v1development v0.3.0 h1:CQvJSldHRUN6Z8jsUeYv8J0lXRvygALXIzsmAeCcZE0= +go.opentelemetry.io/proto/slim/otlp/profiles/v1development v0.3.0/go.mod h1:xSQ+mEfJe/GjK1LXEyVOoSI1N9JV9ZI923X5kup43W4= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= +go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0= +go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y= +go.uber.org/zap v1.28.0 h1:IZzaP1Fv73/T/pBMLk4VutPl36uNC+OSUh3JLG3FIjo= +go.uber.org/zap v1.28.0/go.mod h1:rDLpOi171uODNm/mxFcuYWxDsqWSAVkFdX4XojSKg/Q= +go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc= +go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= +golang.org/x/crypto v0.52.0 h1:RMs7fP2rXdep0CftQlK8Uf+kibLm7qkCcradZWYz988= +golang.org/x/crypto v0.52.0/go.mod h1:1QgfPxDqh0T2M/elOJtp9RvuR95kVjir0e6/BvEmGbc= +golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8= +golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww= +golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY= +golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc= +golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38= +gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= +gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478 h1:RmoJA1ujG+/lRGNfUnOMfhCy5EipVMyvUE+KNbPbTlw= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/grpc v1.82.0 h1:vguDnZUPjE26w09A63VoxZPnvPjB5Riyc0mkXPFmAIU= +google.golang.org/grpc v1.82.0/go.mod h1:yzTZ1TB1Z3SG+LIYaI+WiE8D5+PZ3ArnrSp8zF3+/ZA= +google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= +google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/otelcol/otelcol-builder.yaml b/otelcol/otelcol-builder.yaml new file mode 100644 index 00000000..d69c6f16 --- /dev/null +++ b/otelcol/otelcol-builder.yaml @@ -0,0 +1,34 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# OpenTelemetry Collector Builder manifest for fleetint-otelcol. +# +# Build: +# builder --config=otelcol-builder.yaml +# +# The builder generates main.go + go.mod into a temp dir and compiles the binary +# to dist.output_path. The sakauth extension is built from the local module. + +dist: + name: fleetint-otelcol + description: "Fleet Intelligence OTel Collector (SAK auth)" + output_path: ./bin + +extensions: + - gomod: github.com/NVIDIA/fleet-intelligence-agent/otelcol/auth/sakauth v0.0.0 + path: ./auth/sakauth + - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/extension/healthcheckextension v0.156.0 + +receivers: + - gomod: go.opentelemetry.io/collector/receiver/otlpreceiver v0.156.0 + +processors: + - gomod: go.opentelemetry.io/collector/processor/batchprocessor v0.156.0 + - gomod: go.opentelemetry.io/collector/processor/memorylimiterprocessor v0.156.0 + +exporters: + - gomod: go.opentelemetry.io/collector/exporter/otlphttpexporter v0.156.0 + +providers: + - gomod: go.opentelemetry.io/collector/confmap/provider/envprovider v1.62.0 + - gomod: go.opentelemetry.io/collector/confmap/provider/fileprovider v1.62.0