From 23f69fa9e4dacceb4fe85039a3d16b391eafb5ac Mon Sep 17 00:00:00 2001 From: Bella Khizgiyaev Date: Wed, 26 Aug 2026 15:40:54 +0300 Subject: [PATCH] feat: add e2e test for jumpstarter-telemetry Signed-off-by: Bella Khizgiyaev --- controller/Containerfile.telemetry.prebuilt | 6 + controller/Makefile | 18 +- .../jumpstarter/jumpstarter_controller.go | 25 +- .../controller/jumpstarter/telemetry.go | 62 ++- .../controller/jumpstarter/telemetry_test.go | 145 ++++++- .../deploy/operator/test/e2e/e2e_test.go | 26 ++ .../operator/test/e2e/telemetry_e2e_test.go | 397 ++++++++++++++++++ controller/hack/deploy_vars | 2 + controller/hack/deploy_with_operator.sh | 6 + .../internal/service/telemetry_service.go | 28 +- .../service/telemetry_service_test.go | 207 +++++++++ 11 files changed, 895 insertions(+), 27 deletions(-) create mode 100644 controller/Containerfile.telemetry.prebuilt create mode 100644 controller/deploy/operator/test/e2e/telemetry_e2e_test.go diff --git a/controller/Containerfile.telemetry.prebuilt b/controller/Containerfile.telemetry.prebuilt new file mode 100644 index 000000000..49560da38 --- /dev/null +++ b/controller/Containerfile.telemetry.prebuilt @@ -0,0 +1,6 @@ +# CI-only runtime image for the jumpstarter-telemetry binary. +FROM registry.access.redhat.com/ubi9/ubi-micro:9.8-1786321990@sha256:7e7f79ab747bf2b452e3043dd89f388e92be4c7fdcc8b815b58adf6c99c39c95 +WORKDIR / +COPY telemetry /telemetry +USER 65532:65532 +ENTRYPOINT ["/telemetry"] diff --git a/controller/Makefile b/controller/Makefile index 87b4a7727..467de18af 100644 --- a/controller/Makefile +++ b/controller/Makefile @@ -5,6 +5,7 @@ IMG ?= quay.io/jumpstarter-dev/jumpstarter-controller:latest DOCKER_REPO = $(shell echo $(IMG) | cut -d: -f1) DOCKER_TAG = $(shell echo $(IMG) | cut -d: -f2) EXPORTER_SET_CONTROLLER_IMG ?= quay.io/jumpstarter-dev/jumpstarter-exporterset-controller:latest +TELEMETRY_IMG ?= quay.io/jumpstarter-dev/jumpstarter-telemetry:latest QEMU_RUNTIME_IMG ?= quay.io/jumpstarter-dev/virtual/qemu-runtime:latest EXPORTER_IMG ?= quay.io/jumpstarter-dev/jumpstarter:latest # ENVTEST_K8S_VERSION refers to the version of kubebuilder assets to be downloaded by envtest binary. @@ -124,6 +125,7 @@ build: manifests generate fmt vet ## Build manager binary. go build -ldflags "$(LDFLAGS)" -o bin/manager cmd/main.go go build -ldflags "$(LDFLAGS)" -o bin/router ./cmd/router go build -ldflags "$(LDFLAGS)" -o bin/exporter-set-controller cmd/exporter-set-controller/main.go + go build -ldflags "$(LDFLAGS)" -o bin/telemetry cmd/telemetry/main.go .PHONY: run run: manifests generate fmt vet ## Run a controller from your host. @@ -133,6 +135,10 @@ run: manifests generate fmt vet ## Run a controller from your host. run-router: manifests generate fmt vet ## Run a router from your host. go run ./cmd/router +.PHONY: run-telemetry +run-telemetry: manifests generate fmt vet ## Run the telemetry service from your host. + go run ./cmd/telemetry/main.go + # If you wish to build the manager image targeting other platforms you can use the --platform flag. # (i.e. docker build --platform linux/arm64). However, you must enable docker buildKit for it. # More info: https://docs.docker.com/develop/develop-images/build_enhancements/ @@ -146,12 +152,19 @@ docker-build: ## Build docker image with the manager. .PHONY: docker-build-ci docker-build-ci: ## Build docker images from pre-compiled host binaries (fast CI path). - rm -rf bin/ci-stage && mkdir -p bin/ci-stage/controller bin/ci-stage/esc + rm -rf bin/ci-stage && mkdir -p bin/ci-stage/controller bin/ci-stage/esc bin/ci-stage/telemetry CGO_ENABLED=0 GOOS=linux GOARCH=$(GOARCH) go build -ldflags "$(LDFLAGS)" -o bin/ci-stage/controller/manager cmd/main.go CGO_ENABLED=0 GOOS=linux GOARCH=$(GOARCH) go build -ldflags "$(LDFLAGS)" -o bin/ci-stage/controller/router ./cmd/router CGO_ENABLED=0 GOOS=linux GOARCH=$(GOARCH) go build -ldflags "$(LDFLAGS)" -o bin/ci-stage/esc/exporter-set-controller cmd/exporter-set-controller/main.go $(CONTAINER_TOOL) build --build-arg BIN=manager -t $(IMG) -f Containerfile.prebuilt bin/ci-stage/controller $(CONTAINER_TOOL) build --build-arg BIN=exporter-set-controller -t $(EXPORTER_SET_CONTROLLER_IMG) -f Containerfile.prebuilt bin/ci-stage/esc + $(MAKE) docker-build-telemetry-ci + +.PHONY: docker-build-telemetry-ci +docker-build-telemetry-ci: ## CI-optimized: host-compiled telemetry binary, no multi-stage build. + rm -rf bin/ci-stage/telemetry && mkdir -p bin/ci-stage/telemetry + CGO_ENABLED=0 GOOS=linux GOARCH=$(GOARCH) go build -ldflags "$(LDFLAGS)" -o bin/ci-stage/telemetry/telemetry cmd/telemetry/main.go + $(CONTAINER_TOOL) build -t $(TELEMETRY_IMG) -f Containerfile.telemetry.prebuilt bin/ci-stage/telemetry .PHONY: docker-build-exporter-set-controller-ci docker-build-exporter-set-controller-ci: ## CI-optimized: host-compiled ESC binary, no multi-stage build. @@ -226,6 +239,7 @@ deploy: cluster grpcurl ## Deploy controller using the operator. Set SKIP_BUILD= ifeq ($(SKIP_BUILD),) $(MAKE) docker-build $(MAKE) docker-build-exporter-set-controller + $(MAKE) docker-build-telemetry-ci $(MAKE) build-operator endif ./hack/deploy_with_operator.sh @@ -236,7 +250,7 @@ deploy-ci: cluster grpcurl $(if $(SKIP_BUILD),,docker-build-ci build-operator-ci .PHONY: deploy-operator -deploy-operator: docker-build docker-build-exporter-set-controller build-operator cluster grpcurl ## Deploy only the operator (without Jumpstarter CR) +deploy-operator: docker-build docker-build-exporter-set-controller docker-build-telemetry-ci build-operator cluster grpcurl ## Deploy only the operator (without Jumpstarter CR) NETWORKING_MODE=ingress DEPLOY_JUMPSTARTER=false ./hack/deploy_with_operator.sh .PHONY: deploy-operator-ci diff --git a/controller/deploy/operator/internal/controller/jumpstarter/jumpstarter_controller.go b/controller/deploy/operator/internal/controller/jumpstarter/jumpstarter_controller.go index b50af27dc..b1664ed2b 100644 --- a/controller/deploy/operator/internal/controller/jumpstarter/jumpstarter_controller.go +++ b/controller/deploy/operator/internal/controller/jumpstarter/jumpstarter_controller.go @@ -188,6 +188,13 @@ func (r *JumpstarterReconciler) Reconcile(ctx context.Context, req ctrl.Request) return ctrl.Result{}, err } + // Ensure signing secrets exist before any Deployment that references them + // (CONTROLLER_KEY on controller/telemetry, ROUTER_KEY on router). + if err := r.reconcileSecrets(ctx, &jumpstarter); err != nil { + log.Error(err, "Failed to reconcile Secrets") + return ctrl.Result{}, err + } + // Build the desired controller ConfigMap once and compute its hash up front. // The hash is embedded in the controller pod template annotation so that a config // change (e.g. OIDC CA rotation) triggers a rolling restart without waiting for the @@ -242,18 +249,12 @@ func (r *JumpstarterReconciler) Reconcile(ctx context.Context, req ctrl.Request) return ctrl.Result{}, err } - // Reconcile ConfigMaps (after deployments and services, before secrets) + // Reconcile ConfigMaps (after deployments and services) if err := r.reconcileConfigMaps(ctx, &jumpstarter, desiredConfigMap); err != nil { log.Error(err, "Failed to reconcile ConfigMaps") return ctrl.Result{}, err } - // Reconcile Secrets - if err := r.reconcileSecrets(ctx, &jumpstarter); err != nil { - log.Error(err, "Failed to reconcile Secrets") - return ctrl.Result{}, err - } - // Update status if err := r.updateStatus(ctx, &jumpstarter); err != nil { log.Error(err, "Failed to update status") @@ -1322,7 +1323,6 @@ func (r *JumpstarterReconciler) buildConfig(ctx context.Context, jumpstarter *op } // Telemetry configuration. - // Certificate is intentionally omitted until the telemetry binary supports TLS serving. if jumpstarter.Spec.Telemetry != nil && jumpstarter.Spec.Telemetry.Enabled { t := jumpstarter.Spec.Telemetry telemetryCfg := &config.Telemetry{ @@ -1332,6 +1332,15 @@ func (r *JumpstarterReconciler) buildConfig(ctx context.Context, jumpstarter *op if t.Logging.Filter.MinSeverity != "" { telemetryCfg.Logging.Filter.MinSeverity = t.Logging.Filter.MinSeverity } + if jumpstarter.Spec.CertManager.Enabled { + ca, err := r.resolveTelemetryCA(ctx, jumpstarter) + if err != nil { + return config.Config{}, fmt.Errorf("resolve telemetry CA: %w", err) + } + // Empty ca is valid for public external issuers (system trust). Self-signed + // mode returns an error from resolveTelemetryCA until the CA secret is ready. + telemetryCfg.Certificate = ca + } cfg.Telemetry = telemetryCfg } diff --git a/controller/deploy/operator/internal/controller/jumpstarter/telemetry.go b/controller/deploy/operator/internal/controller/jumpstarter/telemetry.go index c1dbcc532..80cb7ef69 100644 --- a/controller/deploy/operator/internal/controller/jumpstarter/telemetry.go +++ b/controller/deploy/operator/internal/controller/jumpstarter/telemetry.go @@ -273,6 +273,31 @@ func createTelemetryDeployment(jumpstarter *operatorv1alpha1.Jumpstarter) *appsv replicas = *t.Replicas } + var tlsEnv []corev1.EnvVar + var volumeMounts []corev1.VolumeMount + var volumes []corev1.Volume + if jumpstarter.Spec.CertManager.Enabled { + tlsEnv = []corev1.EnvVar{ + {Name: "EXTERNAL_CERT_PEM", Value: "/tls/tls.crt"}, + {Name: "EXTERNAL_KEY_PEM", Value: "/tls/tls.key"}, + } + defaultMode := int32(420) + volumeMounts = []corev1.VolumeMount{{ + Name: "tls-certs", + MountPath: "/tls", + ReadOnly: true, + }} + volumes = []corev1.Volume{{ + Name: "tls-certs", + VolumeSource: corev1.VolumeSource{ + Secret: &corev1.SecretVolumeSource{ + SecretName: getTelemetryCertSecretName(jumpstarter), + DefaultMode: &defaultMode, + }, + }, + }} + } + return &appsv1.Deployment{ ObjectMeta: metav1.ObjectMeta{ Name: fmt.Sprintf("%s-telemetry", jumpstarter.Name), @@ -310,7 +335,7 @@ func createTelemetryDeployment(jumpstarter *operatorv1alpha1.Jumpstarter) *appsv Args: []string{ fmt.Sprintf("--grpc-bind=:%d", telemetryPort), }, - Env: []corev1.EnvVar{ + Env: append([]corev1.EnvVar{ { Name: "CONTROLLER_KEY", ValueFrom: &corev1.EnvVarSource{ @@ -322,7 +347,10 @@ func createTelemetryDeployment(jumpstarter *operatorv1alpha1.Jumpstarter) *appsv }, }, }, - }, + // Advertised endpoint for self-signed SAN generation (must match controller ConfigMap). + {Name: "GRPC_TELEMETRY_ENDPOINT", Value: telemetryEndpointFor(jumpstarter.Namespace)}, + }, tlsEnv...), + VolumeMounts: volumeMounts, Ports: []corev1.ContainerPort{ { ContainerPort: int32(telemetryPort), @@ -371,6 +399,7 @@ func createTelemetryDeployment(jumpstarter *operatorv1alpha1.Jumpstarter) *appsv Type: corev1.SeccompProfileTypeRuntimeDefault, }, }, + Volumes: volumes, ServiceAccountName: jumpstarter.Name + telemetrySASuffix, }, }, @@ -425,14 +454,16 @@ func getTelemetryCertSecretName(js *operatorv1alpha1.Jumpstarter) string { } // resolveTelemetryCA reads the CA certificate that exporters need to verify the -// telemetry TLS connection. For self-signed CA mode, the cert is in the CA secret; -// for external issuers, the user-provided caBundle is used. +// telemetry TLS connection. For self-signed CA mode, the cert is in the CA secret. +// For external issuers, the user-provided caBundle is preferred; when absent, ca.crt +// from the issued telemetry TLS secret is used if present. An empty return with no +// error means exporters should rely on the system trust store (public CA issuers). func (r *JumpstarterReconciler) resolveTelemetryCA(ctx context.Context, jumpstarter *operatorv1alpha1.Jumpstarter) (string, error) { if jumpstarter.Spec.CertManager.Server != nil && jumpstarter.Spec.CertManager.Server.IssuerRef != nil { if len(jumpstarter.Spec.CertManager.Server.IssuerRef.CABundle) > 0 { return string(jumpstarter.Spec.CertManager.Server.IssuerRef.CABundle), nil } - return "", nil + return r.telemetryCAFromCertSecret(ctx, jumpstarter) } // Self-signed CA mode — read from the CA secret created by cert-manager @@ -447,6 +478,27 @@ func (r *JumpstarterReconciler) resolveTelemetryCA(ctx context.Context, jumpstar return "", fmt.Errorf("CA secret %s missing tls.crt", caSecretName) } +// telemetryCAFromCertSecret returns ca.crt from the issued telemetry TLS secret, +// when cert-manager includes it. Missing secret or key is not an error: external +// issuers backed by public CAs may not need an explicit bundle in the controller config. +func (r *JumpstarterReconciler) telemetryCAFromCertSecret(ctx context.Context, jumpstarter *operatorv1alpha1.Jumpstarter) (string, error) { + secret := &corev1.Secret{} + err := r.Get(ctx, client.ObjectKey{ + Name: getTelemetryCertSecretName(jumpstarter), + Namespace: jumpstarter.Namespace, + }, secret) + if err != nil { + if errors.IsNotFound(err) { + return "", nil + } + return "", fmt.Errorf("telemetry TLS secret not found: %w", err) + } + if ca, ok := secret.Data["ca.crt"]; ok && len(ca) > 0 { + return string(ca), nil + } + return "", nil +} + // telemetryEndpointFor returns the in-cluster gRPC endpoint for the telemetry service. func telemetryEndpointFor(namespace string) string { return fmt.Sprintf("%s.%s.svc:%d", telemetryServiceName, namespace, telemetryPort) diff --git a/controller/deploy/operator/internal/controller/jumpstarter/telemetry_test.go b/controller/deploy/operator/internal/controller/jumpstarter/telemetry_test.go index 66e86d113..a720179cc 100644 --- a/controller/deploy/operator/internal/controller/jumpstarter/telemetry_test.go +++ b/controller/deploy/operator/internal/controller/jumpstarter/telemetry_test.go @@ -315,6 +315,44 @@ var _ = Describe("Telemetry Lifecycle", func() { "telemetry should not appear in ConfigMap after disabling") }) + It("includes telemetry CA from issued TLS secret when external issuer has no CABundle", func() { + By("creating a Jumpstarter CR with telemetry, cert-manager, and external issuer") + spec := makeJumpstarterSpec() + spec.CertManager = operatorv1alpha1.CertManagerConfig{ + Enabled: true, + Server: &operatorv1alpha1.ServerCertConfig{ + IssuerRef: &operatorv1alpha1.IssuerReference{ + Name: "my-issuer", + Kind: "ClusterIssuer", + }, + }, + } + spec.Telemetry = &operatorv1alpha1.TelemetryConfig{ + Enabled: true, + Image: "quay.io/jumpstarter-dev/jumpstarter-telemetry:latest", + } + js := &operatorv1alpha1.Jumpstarter{ + ObjectMeta: metav1.ObjectMeta{Name: crName, Namespace: crNamespace}, + Spec: spec, + } + + By("pre-creating the issued telemetry TLS secret with ca.crt") + Expect(k8sClient.Create(ctx, &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Name: getTelemetryCertSecretName(js), + Namespace: crNamespace, + }, + Data: map[string][]byte{ + "ca.crt": []byte(testPEM), + }, + })).To(Succeed()) + + cfg, err := newReconciler().buildConfig(ctx, js) + Expect(err).NotTo(HaveOccurred()) + Expect(cfg.Telemetry).NotTo(BeNil()) + Expect(cfg.Telemetry.Certificate).To(ContainSubstring("BEGIN CERTIFICATE")) + }) + It("propagates telemetry config into the controller ConfigMap", func() { By("creating a Jumpstarter CR with telemetry and a custom minSeverity") spec := makeJumpstarterSpec() @@ -402,10 +440,7 @@ var _ = Describe("Telemetry Lifecycle", func() { Expect(cond.Reason).To(Equal("DeploymentAvailable")) }) - It("does not mount TLS certs even when cert-manager is enabled (TLS serving not yet supported by the binary)", func() { - // EXTERNAL_CERT_PEM/EXTERNAL_KEY_PEM and the tls-certs volume are intentionally - // omitted until the telemetry binary is updated to serve TLS. - // CONTROLLER_KEY is always set for token validation (not TLS-related). + It("mounts TLS certs when cert-manager is enabled", func() { js := &operatorv1alpha1.Jumpstarter{ ObjectMeta: metav1.ObjectMeta{Name: "test-tls", Namespace: "default"}, Spec: operatorv1alpha1.JumpstarterSpec{ @@ -421,10 +456,70 @@ var _ = Describe("Telemetry Lifecycle", func() { dep := createTelemetryDeployment(js) container := dep.Spec.Template.Spec.Containers[0] - // CONTROLLER_KEY should be set for token validation - Expect(container.Env).To(HaveLen(1)) - Expect(container.Env[0].Name).To(Equal("CONTROLLER_KEY")) - // TLS-related env vars should NOT be set + envNames := make([]string, len(container.Env)) + for i, env := range container.Env { + envNames[i] = env.Name + } + Expect(envNames).To(ConsistOf( + "CONTROLLER_KEY", + "GRPC_TELEMETRY_ENDPOINT", + "EXTERNAL_CERT_PEM", + "EXTERNAL_KEY_PEM", + )) + for _, env := range container.Env { + switch env.Name { + case "GRPC_TELEMETRY_ENDPOINT": + Expect(env.Value).To(Equal(telemetryEndpointFor("default"))) + case "EXTERNAL_CERT_PEM": + Expect(env.Value).To(Equal("/tls/tls.crt")) + case "EXTERNAL_KEY_PEM": + Expect(env.Value).To(Equal("/tls/tls.key")) + } + } + Expect(container.VolumeMounts).To(ContainElement(corev1.VolumeMount{ + Name: "tls-certs", MountPath: "/tls", ReadOnly: true, + })) + Expect(dep.Spec.Template.Spec.Volumes).To(ContainElement(corev1.Volume{ + Name: "tls-certs", + VolumeSource: corev1.VolumeSource{ + Secret: &corev1.SecretVolumeSource{ + SecretName: "test-tls" + telemetryCertSuffix, + DefaultMode: ptr.To(int32(420)), + }, + }, + })) + }) + + It("does not mount TLS certs when cert-manager is disabled", func() { + js := &operatorv1alpha1.Jumpstarter{ + ObjectMeta: metav1.ObjectMeta{Name: "test-no-tls", Namespace: "default"}, + Spec: operatorv1alpha1.JumpstarterSpec{ + Telemetry: &operatorv1alpha1.TelemetryConfig{ + Enabled: true, + Image: "quay.io/jumpstarter-dev/jumpstarter-telemetry:latest", + ImagePullPolicy: corev1.PullIfNotPresent, + }, + }, + } + + dep := createTelemetryDeployment(js) + + container := dep.Spec.Template.Spec.Containers[0] + envNames := make([]string, len(container.Env)) + for i, env := range container.Env { + envNames[i] = env.Name + } + Expect(envNames).To(ConsistOf("CONTROLLER_KEY", "GRPC_TELEMETRY_ENDPOINT")) + for _, env := range container.Env { + switch env.Name { + case "CONTROLLER_KEY": + Expect(env.ValueFrom).NotTo(BeNil()) + Expect(env.ValueFrom.SecretKeyRef.Name).To(Equal("jumpstarter-controller-secret")) + Expect(env.ValueFrom.SecretKeyRef.Key).To(Equal("key")) + case "GRPC_TELEMETRY_ENDPOINT": + Expect(env.Value).To(Equal(telemetryEndpointFor("default"))) + } + } for _, env := range container.Env { Expect(env.Name).NotTo(Equal("EXTERNAL_CERT_PEM")) Expect(env.Name).NotTo(Equal("EXTERNAL_KEY_PEM")) @@ -704,7 +799,39 @@ var _ = Describe("resolveTelemetryCA", func() { Expect(err.Error()).To(ContainSubstring("missing tls.crt")) }) - It("returns ('', nil) when an external IssuerRef has a nil CABundle", func() { + It("returns ca.crt from the telemetry TLS secret when an external IssuerRef has no CABundle", func() { + js := &operatorv1alpha1.Jumpstarter{ + ObjectMeta: metav1.ObjectMeta{Name: "test-ca-from-secret", Namespace: crNamespace}, + Spec: operatorv1alpha1.JumpstarterSpec{ + CertManager: operatorv1alpha1.CertManagerConfig{ + Enabled: true, + Server: &operatorv1alpha1.ServerCertConfig{ + IssuerRef: &operatorv1alpha1.IssuerReference{ + Name: "my-issuer", + Kind: "ClusterIssuer", + }, + }, + }, + }, + } + + Expect(k8sClient.Create(ctx, &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Name: getTelemetryCertSecretName(js), + Namespace: crNamespace, + }, + Data: map[string][]byte{ + "ca.crt": []byte(testPEM), + }, + })).To(Succeed()) + + r := &JumpstarterReconciler{Client: k8sClient, Scheme: k8sClient.Scheme()} + ca, err := r.resolveTelemetryCA(ctx, js) + Expect(err).NotTo(HaveOccurred()) + Expect(ca).To(ContainSubstring("BEGIN CERTIFICATE")) + }) + + It("returns ('', nil) when an external IssuerRef has a nil CABundle and no telemetry TLS secret", func() { js := &operatorv1alpha1.Jumpstarter{ ObjectMeta: metav1.ObjectMeta{Name: "test-ca-no-bundle", Namespace: crNamespace}, Spec: operatorv1alpha1.JumpstarterSpec{ diff --git a/controller/deploy/operator/test/e2e/e2e_test.go b/controller/deploy/operator/test/e2e/e2e_test.go index 0eb6a96dd..5baed171f 100644 --- a/controller/deploy/operator/test/e2e/e2e_test.go +++ b/controller/deploy/operator/test/e2e/e2e_test.go @@ -2474,6 +2474,32 @@ func verifyDeploymentHasTLSMount(g Gomega, namespace, name string) { g.Expect(hasKeyEnv).To(BeTrue(), fmt.Sprintf("deployment %s missing EXTERNAL_KEY_PEM env var", name)) } +// verifyDeploymentHasControllerKey checks that a deployment sources CONTROLLER_KEY from +// jumpstarter-controller-secret. PushLogs bearer-token verification requires the same +// signing seed as the controller. +func verifyDeploymentHasControllerKey(g Gomega, namespace, name string) { + deployment := &appsv1.Deployment{} + err := k8sClient.Get(ctx, types.NamespacedName{ + Name: name, + Namespace: namespace, + }, deployment) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(deployment.Spec.Template.Spec.Containers).NotTo(BeEmpty()) + + var found bool + for _, env := range deployment.Spec.Template.Spec.Containers[0].Env { + if env.Name != "CONTROLLER_KEY" { + continue + } + g.Expect(env.ValueFrom).NotTo(BeNil()) + g.Expect(env.ValueFrom.SecretKeyRef).NotTo(BeNil()) + g.Expect(env.ValueFrom.SecretKeyRef.Name).To(Equal("jumpstarter-controller-secret")) + g.Expect(env.ValueFrom.SecretKeyRef.Key).To(Equal("key")) + found = true + } + g.Expect(found).To(BeTrue(), fmt.Sprintf("deployment %s missing CONTROLLER_KEY env var", name)) +} + // verifyDeploymentHasNoTLSMount checks that a deployment does NOT have TLS configuration. // This is used with Gomega assertions to verify the deployment has been reconciled without TLS. func verifyDeploymentHasNoTLSMount(g Gomega, namespace, name string) { diff --git a/controller/deploy/operator/test/e2e/telemetry_e2e_test.go b/controller/deploy/operator/test/e2e/telemetry_e2e_test.go new file mode 100644 index 000000000..4a95bff0d --- /dev/null +++ b/controller/deploy/operator/test/e2e/telemetry_e2e_test.go @@ -0,0 +1,397 @@ +/* +Copyright 2026. + +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 e2e + +import ( + "fmt" + "os" + "time" + + certmanagerv1 "github.com/cert-manager/cert-manager/pkg/apis/certmanager/v1" + cmmeta "github.com/cert-manager/cert-manager/pkg/apis/meta/v1" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + + operatorv1alpha1 "github.com/jumpstarter-dev/jumpstarter/controller/deploy/operator/api/v1alpha1" +) + +const ( + telemetryE2EServiceName = "jumpstarter-telemetry" + telemetryE2EPort = 9093 + telemetryE2ECertSuffix = "-telemetry-tls" +) + +func controllerImage() string { + if image := os.Getenv("IMG"); image != "" { + return image + } + return defaultControllerImage +} + +func telemetryImage() string { + if image := os.Getenv("TELEMETRY_IMG"); image != "" { + return image + } + return "quay.io/jumpstarter-dev/jumpstarter-telemetry:latest" +} + +func telemetryDeploymentName(jumpstarterName string) string { + return jumpstarterName + "-telemetry" +} + +func telemetryCertName(jumpstarterName string) string { + return jumpstarterName + telemetryE2ECertSuffix +} + +func telemetryYAMLBlock() string { + return fmt.Sprintf(` telemetry: + enabled: true + image: %s + imagePullPolicy: IfNotPresent + resources: + requests: + cpu: 50m + memory: 128Mi +`, telemetryImage()) +} + +var _ = Describe("Telemetry lifecycle", Ordered, func() { + const baseDomain = "telemetry.127.0.0.1.nip.io" + const jumpstarterName = "jumpstarter-telemetry" + var telemetryTestNamespace string + + BeforeAll(func() { + telemetryTestNamespace = CreateTestNamespace() + }) + + AfterAll(func() { + DeleteTestNamespace(telemetryTestNamespace) + }) + + It("should deploy telemetry deployment and service when enabled", func() { + image := controllerImage() + + jumpstarterYAML := fmt.Sprintf(`apiVersion: operator.jumpstarter.dev/v1alpha1 +kind: Jumpstarter +metadata: + name: %s + namespace: %s +spec: + baseDomain: %s + useCertManager: false + authentication: + internal: + prefix: "internal:" + enabled: true + controller: + image: %s + imagePullPolicy: IfNotPresent + replicas: 1 + resources: + requests: + cpu: 100m + memory: 100Mi + grpc: + endpoints: + - address: grpc.%s:8082 + nodeport: + enabled: true + port: 30080 + routers: + image: %s + imagePullPolicy: IfNotPresent + replicas: 1 + resources: + requests: + cpu: 100m + memory: 100Mi + grpc: + endpoints: + - address: router.%s:8083 + nodeport: + enabled: true + port: 30081 +%s`, jumpstarterName, telemetryTestNamespace, baseDomain, image, baseDomain, image, baseDomain, telemetryYAMLBlock()) + + Expect(applyYAML(jumpstarterYAML)).To(Succeed()) + + Eventually(func(g Gomega) { + dep := &appsv1.Deployment{} + err := k8sClient.Get(ctx, types.NamespacedName{ + Name: telemetryDeploymentName(jumpstarterName), + Namespace: telemetryTestNamespace, + }, dep) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(dep.Spec.Template.Spec.Containers).NotTo(BeEmpty()) + g.Expect(dep.Spec.Template.Spec.Containers[0].Image).To(Equal(telemetryImage())) + verifyDeploymentHasControllerKey(g, telemetryTestNamespace, telemetryDeploymentName(jumpstarterName)) + }, 2*time.Minute).Should(Succeed()) + + Eventually(func(g Gomega) { + svc := &corev1.Service{} + err := k8sClient.Get(ctx, types.NamespacedName{ + Name: telemetryE2EServiceName, + Namespace: telemetryTestNamespace, + }, svc) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(svc.Spec.Ports[0].Port).To(Equal(int32(telemetryE2EPort))) + g.Expect(svc.Spec.Selector).To(HaveKeyWithValue("app", "jumpstarter-telemetry")) + }, 2*time.Minute).Should(Succeed()) + + Eventually(func(g Gomega) { + sa := &corev1.ServiceAccount{} + err := k8sClient.Get(ctx, types.NamespacedName{ + Name: jumpstarterName + "-telemetry", + Namespace: telemetryTestNamespace, + }, sa) + g.Expect(err).NotTo(HaveOccurred()) + }, 2*time.Minute).Should(Succeed()) + }) + + It("should include telemetry configuration in the controller ConfigMap", func() { + Eventually(func(g Gomega) { + cm := &corev1.ConfigMap{} + err := k8sClient.Get(ctx, types.NamespacedName{ + Name: "jumpstarter-controller", + Namespace: telemetryTestNamespace, + }, cm) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(cm.Data["config"]).To(ContainSubstring("telemetry:")) + g.Expect(cm.Data["config"]).To(ContainSubstring("enabled: true")) + g.Expect(cm.Data["config"]).To(ContainSubstring( + fmt.Sprintf("%s.%s.svc:%d", telemetryE2EServiceName, telemetryTestNamespace, telemetryE2EPort))) + }, 2*time.Minute).Should(Succeed()) + }) + + It("should report TelemetryDeploymentReady when the deployment is available", func() { + waitForCondition(telemetryTestNamespace, jumpstarterName, + operatorv1alpha1.ConditionTypeTelemetryDeploymentReady, metav1.ConditionTrue, 5*time.Minute) + }) + + It("should remove telemetry resources when telemetry is disabled", func() { + js := &operatorv1alpha1.Jumpstarter{} + Expect(k8sClient.Get(ctx, types.NamespacedName{ + Name: jumpstarterName, + Namespace: telemetryTestNamespace, + }, js)).To(Succeed()) + + Expect(js.Spec.Telemetry).NotTo(BeNil()) + js.Spec.Telemetry.Enabled = false + Expect(k8sClient.Update(ctx, js)).To(Succeed()) + + Eventually(func(g Gomega) { + dep := &appsv1.Deployment{} + err := k8sClient.Get(ctx, types.NamespacedName{ + Name: telemetryDeploymentName(jumpstarterName), + Namespace: telemetryTestNamespace, + }, dep) + g.Expect(apierrors.IsNotFound(err)).To(BeTrue()) + }, 2*time.Minute).Should(Succeed()) + + Eventually(func(g Gomega) { + svc := &corev1.Service{} + err := k8sClient.Get(ctx, types.NamespacedName{ + Name: telemetryE2EServiceName, + Namespace: telemetryTestNamespace, + }, svc) + g.Expect(apierrors.IsNotFound(err)).To(BeTrue()) + }, 2*time.Minute).Should(Succeed()) + + Eventually(func(g Gomega) { + cm := &corev1.ConfigMap{} + err := k8sClient.Get(ctx, types.NamespacedName{ + Name: "jumpstarter-controller", + Namespace: telemetryTestNamespace, + }, cm) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(cm.Data["config"]).NotTo(ContainSubstring("telemetry:")) + }, 2*time.Minute).Should(Succeed()) + }) +}) + +var _ = Describe("Telemetry cert-manager integration", Ordered, func() { + const baseDomain = "telemetry-tls.127.0.0.1.nip.io" + const jumpstarterName = "jumpstarter-telemetry-tls" + var telemetryTLSTestNamespace string + + BeforeAll(func() { + telemetryTLSTestNamespace = CreateTestNamespace() + }) + + AfterAll(func() { + DeleteTestNamespace(telemetryTLSTestNamespace) + }) + + It("should deploy jumpstarter with telemetry and cert-manager enabled", func() { + image := controllerImage() + + jumpstarterYAML := fmt.Sprintf(`apiVersion: operator.jumpstarter.dev/v1alpha1 +kind: Jumpstarter +metadata: + name: %s + namespace: %s +spec: + baseDomain: %s + certManager: + enabled: true + server: + selfSigned: + enabled: true + authentication: + internal: + prefix: "internal:" + enabled: true + controller: + image: %s + imagePullPolicy: IfNotPresent + replicas: 1 + resources: + requests: + cpu: 100m + memory: 100Mi + grpc: + endpoints: + - address: grpc.%s:8082 + nodeport: + enabled: true + port: 30082 + routers: + image: %s + imagePullPolicy: IfNotPresent + replicas: 1 + resources: + requests: + cpu: 100m + memory: 100Mi + grpc: + endpoints: + - address: router.%s:8083 + nodeport: + enabled: true + port: 30083 +%s`, jumpstarterName, telemetryTLSTestNamespace, baseDomain, image, baseDomain, image, baseDomain, telemetryYAMLBlock()) + + Expect(applyYAML(jumpstarterYAML)).To(Succeed()) + }) + + It("should create the telemetry TLS certificate", func() { + certName := telemetryCertName(jumpstarterName) + Eventually(func(g Gomega) { + cert := &certmanagerv1.Certificate{} + err := k8sClient.Get(ctx, types.NamespacedName{ + Name: certName, + Namespace: telemetryTLSTestNamespace, + }, cert) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(cert.Spec.IsCA).To(BeFalse()) + }, 2*time.Minute).Should(Succeed()) + + Eventually(func(g Gomega) { + cert := &certmanagerv1.Certificate{} + err := k8sClient.Get(ctx, types.NamespacedName{ + Name: certName, + Namespace: telemetryTLSTestNamespace, + }, cert) + g.Expect(err).NotTo(HaveOccurred()) + for _, cond := range cert.Status.Conditions { + if cond.Type == certmanagerv1.CertificateConditionReady { + g.Expect(cond.Status).To(Equal(cmmeta.ConditionTrue), + fmt.Sprintf("Certificate %s is not ready: %s", certName, cond.Message)) + return + } + } + g.Expect(false).To(BeTrue(), fmt.Sprintf("Certificate %s has no Ready condition", certName)) + }, 2*time.Minute, 2*time.Second).Should(Succeed()) + verifyTLSSecret(telemetryTLSTestNamespace, certName) + }) + + It("should mount TLS certificates in telemetry deployment", func() { + deploymentName := telemetryDeploymentName(jumpstarterName) + Eventually(func(g Gomega) { + verifyDeploymentHasTLSMount(g, telemetryTLSTestNamespace, deploymentName) + verifyDeploymentHasControllerKey(g, telemetryTLSTestNamespace, deploymentName) + }, 2*time.Minute, 2*time.Second).Should(Succeed()) + }) + + It("should include telemetry CA certificate in the controller ConfigMap", func() { + Eventually(func(g Gomega) { + cm := &corev1.ConfigMap{} + err := k8sClient.Get(ctx, types.NamespacedName{ + Name: "jumpstarter-controller", + Namespace: telemetryTLSTestNamespace, + }, cm) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(cm.Data["config"]).To(ContainSubstring("telemetry:")) + g.Expect(cm.Data["config"]).To(ContainSubstring("certificate:")) + g.Expect(cm.Data["config"]).To(ContainSubstring("BEGIN CERTIFICATE")) + }, 2*time.Minute).Should(Succeed()) + }) + + It("should report TelemetryDeploymentReady when telemetry is available", func() { + waitForCondition(telemetryTLSTestNamespace, jumpstarterName, + operatorv1alpha1.ConditionTypeTelemetryDeploymentReady, metav1.ConditionTrue, 5*time.Minute) + }) + + It("should delete the telemetry Certificate when telemetry is disabled", func() { + js := &operatorv1alpha1.Jumpstarter{} + Expect(k8sClient.Get(ctx, types.NamespacedName{ + Name: jumpstarterName, + Namespace: telemetryTLSTestNamespace, + }, js)).To(Succeed()) + + Expect(js.Spec.Telemetry).NotTo(BeNil()) + js.Spec.Telemetry.Enabled = false + Expect(k8sClient.Update(ctx, js)).To(Succeed()) + + certName := telemetryCertName(jumpstarterName) + Eventually(func(g Gomega) { + cert := &certmanagerv1.Certificate{} + err := k8sClient.Get(ctx, types.NamespacedName{ + Name: certName, + Namespace: telemetryTLSTestNamespace, + }, cert) + g.Expect(apierrors.IsNotFound(err)).To(BeTrue(), + "telemetry Certificate should be deleted when telemetry is disabled") + }, 2*time.Minute).Should(Succeed()) + + Eventually(func(g Gomega) { + dep := &appsv1.Deployment{} + err := k8sClient.Get(ctx, types.NamespacedName{ + Name: telemetryDeploymentName(jumpstarterName), + Namespace: telemetryTLSTestNamespace, + }, dep) + g.Expect(apierrors.IsNotFound(err)).To(BeTrue()) + }, 2*time.Minute).Should(Succeed()) + + Eventually(func(g Gomega) { + js := &operatorv1alpha1.Jumpstarter{} + err := k8sClient.Get(ctx, types.NamespacedName{ + Name: jumpstarterName, + Namespace: telemetryTLSTestNamespace, + }, js) + g.Expect(err).NotTo(HaveOccurred()) + cond := meta.FindStatusCondition(js.Status.Conditions, operatorv1alpha1.ConditionTypeTelemetryDeploymentReady) + g.Expect(cond).To(BeNil()) + }, 2*time.Minute).Should(Succeed()) + }) +}) diff --git a/controller/hack/deploy_vars b/controller/hack/deploy_vars index ffa707286..d87aeeb2b 100755 --- a/controller/hack/deploy_vars +++ b/controller/hack/deploy_vars @@ -8,6 +8,7 @@ BASEDOMAIN=${BASEDOMAIN:-"jumpstarter.${IP}.nip.io"} IMG=${IMG:-quay.io/jumpstarter-dev/jumpstarter-controller:latest} OPERATOR_IMG=${OPERATOR_IMG:-$(make -C deploy/operator --no-print-directory -s print-img 2>/dev/null || echo "quay.io/jumpstarter-dev/jumpstarter-operator:latest")} EXPORTER_SET_CONTROLLER_IMG=${EXPORTER_SET_CONTROLLER_IMG:-quay.io/jumpstarter-dev/jumpstarter-exporterset-controller:latest} +TELEMETRY_IMG=${TELEMETRY_IMG:-quay.io/jumpstarter-dev/jumpstarter-telemetry:latest} EXPORTER_IMG=${EXPORTER_IMG:-quay.io/jumpstarter-dev/jumpstarter:latest} QEMU_RUNTIME_IMG=${QEMU_RUNTIME_IMG:-quay.io/jumpstarter-dev/virtual/qemu-runtime:latest} @@ -48,6 +49,7 @@ export IMAGE_TAG export IMG export OPERATOR_IMG export EXPORTER_SET_CONTROLLER_IMG +export TELEMETRY_IMG export EXPORTER_IMG export QEMU_RUNTIME_IMG diff --git a/controller/hack/deploy_with_operator.sh b/controller/hack/deploy_with_operator.sh index b8ec7a281..d67cd8703 100755 --- a/controller/hack/deploy_with_operator.sh +++ b/controller/hack/deploy_with_operator.sh @@ -41,6 +41,12 @@ load_image "${OPERATOR_IMG}" & _load_pids+=($!) load_image "${EXPORTER_SET_CONTROLLER_IMG}" & _load_pids+=($!) +if container_image_exists "${TELEMETRY_IMG}"; then + load_image "${TELEMETRY_IMG}" & + _load_pids+=($!) +else + echo -e "${YELLOW}Skipping load of telemetry image (not present locally): ${TELEMETRY_IMG}${NC}" +fi if container_image_exists "${EXPORTER_IMG}"; then load_image "${EXPORTER_IMG}" & diff --git a/controller/internal/service/telemetry_service.go b/controller/internal/service/telemetry_service.go index 076f044c9..d42d9c6e9 100644 --- a/controller/internal/service/telemetry_service.go +++ b/controller/internal/service/telemetry_service.go @@ -24,6 +24,7 @@ import ( "strings" "time" + "github.com/go-logr/logr" "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/recovery" "github.com/jumpstarter-dev/jumpstarter/controller/internal/authentication" "github.com/jumpstarter-dev/jumpstarter/controller/internal/oidc" @@ -73,6 +74,9 @@ type TelemetryService struct { // Tokens are issued by the controller from the same CONTROLLER_KEY seed, // so the telemetry binary can verify them locally without a k8s client. Signer *oidc.Signer + + // Logger overrides log.FromContext when set (used by tests to capture output). + Logger *logr.Logger } // PushLogs receives a batch of structured log entries and writes them via the @@ -103,8 +107,8 @@ func (s *TelemetryService) PushLogs(ctx context.Context, req *pb.PushLogsRequest return nil, status.Errorf(codes.PermissionDenied, "token has incomplete exporter identity") } - // Use context-based logger so tests can inject their own via logf.IntoContext. - logger := log.FromContext(ctx).WithName("telemetry") + // Use context-based logger so tests can inject their own via Logger or logf.IntoContext. + logger := s.pushLogsLogger(ctx) entries := req.Entries var dropped uint32 @@ -187,6 +191,13 @@ func (s *TelemetryService) PushLogs(ctx context.Context, req *pb.PushLogsRequest }, nil } +func (s *TelemetryService) pushLogsLogger(ctx context.Context) logr.Logger { + if s.Logger != nil { + return s.Logger.WithName("telemetry") + } + return log.FromContext(ctx).WithName("telemetry") +} + // truncate returns s truncated to at most n bytes (rune-safe: truncates at rune boundary). func truncate(s string, n int) string { if len(s) <= n { @@ -236,6 +247,14 @@ func (s *TelemetryService) loadTLSCredentials() (credentials.TransportCredential return LoadTLSCredentials("jumpstarter telemetry", dnsnames, ipaddresses) } +// injectLoggerUnaryInterceptor attaches the process-wide logger to each gRPC +// request context. gRPC may carry a discard logger in ctx; PushLogs uses +// log.FromContext, so we re-bind the global logger at the RPC boundary. +func injectLoggerUnaryInterceptor(ctx context.Context, req any, _ *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) { + ctx = log.IntoContext(ctx, log.FromContext(context.Background())) + return handler(ctx, req) +} + // Start starts the TelemetryService gRPC server and blocks until ctx is cancelled. func (s *TelemetryService) Start(ctx context.Context) error { logger := ctrl.Log.WithName("telemetry").WithValues("component", "telemetry") @@ -259,7 +278,10 @@ func (s *TelemetryService) Start(ctx context.Context) error { srv := grpc.NewServer( grpc.Creds(creds), - grpc.ChainUnaryInterceptor(recovery.UnaryServerInterceptor()), + grpc.ChainUnaryInterceptor( + recovery.UnaryServerInterceptor(), + injectLoggerUnaryInterceptor, + ), ) pb.RegisterTelemetryServiceServer(srv, s) reflection.Register(srv) diff --git a/controller/internal/service/telemetry_service_test.go b/controller/internal/service/telemetry_service_test.go index 20740f26f..7f78d84be 100644 --- a/controller/internal/service/telemetry_service_test.go +++ b/controller/internal/service/telemetry_service_test.go @@ -19,17 +19,22 @@ package service import ( "bytes" "context" + "crypto/tls" "crypto/x509" "encoding/pem" "fmt" + "net" "os" "strings" "testing" + "time" "github.com/jumpstarter-dev/jumpstarter/controller/internal/config" "github.com/jumpstarter-dev/jumpstarter/controller/internal/oidc" pb "github.com/jumpstarter-dev/jumpstarter/controller/internal/protocol/jumpstarter/v1" + "google.golang.org/grpc" "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials" "google.golang.org/grpc/metadata" "google.golang.org/grpc/status" logf "sigs.k8s.io/controller-runtime/pkg/log" @@ -426,6 +431,208 @@ func TestTelemetryService_Start_FailsWhenExternalCertFileMissing(t *testing.T) { } } +// startTelemetryGRPCServer starts a TelemetryService on a random local port and +// returns the listen address, a log buffer capturing structured output, and a cleanup func. +func startTelemetryGRPCServer(t *testing.T, signer *oidc.Signer) (addr string, logBuf *bytes.Buffer, cleanup func()) { + t.Helper() + + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + addr = lis.Addr().String() + if err := lis.Close(); err != nil { + t.Fatalf("close listener: %v", err) + } + + t.Setenv("EXTERNAL_CERT_PEM", "") + t.Setenv("EXTERNAL_KEY_PEM", "") + t.Setenv("GRPC_TELEMETRY_ENDPOINT", addr) + + logBuf = &bytes.Buffer{} + testLogger := zap.New(zap.WriteTo(logBuf)) + + svc := &TelemetryService{BindAddr: addr, Signer: signer, Logger: &testLogger} + ctx, cancel := context.WithCancel(context.Background()) + + errCh := make(chan error, 1) + go func() { + errCh <- svc.Start(ctx) + }() + + deadline := time.Now().Add(5 * time.Second) + var client pb.TelemetryServiceClient + var conn *grpc.ClientConn + tlsCreds := credentials.NewTLS(&tls.Config{InsecureSkipVerify: true}) //nolint:gosec // test-only self-signed cert + for time.Now().Before(deadline) { + conn, err = grpc.NewClient(addr, grpc.WithTransportCredentials(tlsCreds)) + if err != nil { + time.Sleep(10 * time.Millisecond) + continue + } + client = pb.NewTelemetryServiceClient(conn) + probeCtx := metadata.NewOutgoingContext( + context.Background(), + metadata.Pairs("authorization", "Bearer "+mustToken(t, signer, "exporter:jumpstarter:probe:1")), + ) + _, err = client.PushLogs(probeCtx, &pb.PushLogsRequest{}) + if err == nil || status.Code(err) != codes.Unavailable { + break + } + _ = conn.Close() + conn = nil + time.Sleep(10 * time.Millisecond) + } + if conn == nil { + cancel() + <-errCh + t.Fatal("telemetry gRPC server did not become ready") + } + + cleanup = func() { + _ = conn.Close() + cancel() + <-errCh + } + _ = client // returned via dialTelemetryClient helper below + return addr, logBuf, cleanup +} + +func mustToken(t *testing.T, signer *oidc.Signer, subject string) string { + t.Helper() + token, err := signer.Token(subject) + if err != nil { + t.Fatalf("sign token: %v", err) + } + return token +} + +func dialTelemetryClient(t *testing.T, addr string) (pb.TelemetryServiceClient, func()) { + t.Helper() + tlsCreds := credentials.NewTLS(&tls.Config{InsecureSkipVerify: true}) //nolint:gosec // test-only self-signed cert + conn, err := grpc.NewClient(addr, grpc.WithTransportCredentials(tlsCreds)) + if err != nil { + t.Fatalf("grpc.NewClient: %v", err) + } + return pb.NewTelemetryServiceClient(conn), func() { _ = conn.Close() } +} + +func TestTelemetryService_PushLogs_OverGRPC(t *testing.T) { + signer := testSigner(t) + addr, logBuf, cleanup := startTelemetryGRPCServer(t, signer) + defer cleanup() + + client, closeConn := dialTelemetryClient(t, addr) + defer closeConn() + + token := mustToken(t, signer, "exporter:jumpstarter:test-exporter:abc123") + ctx := metadata.NewOutgoingContext( + context.Background(), + metadata.Pairs("authorization", "Bearer "+token), + ) + + resp, err := client.PushLogs(ctx, &pb.PushLogsRequest{ + Entries: []*pb.LogEntry{ + {Severity: "info", Message: "grpc integration hello", Component: "exporter", Exporter: "test-exporter"}, + }, + }) + if err != nil { + t.Fatalf("PushLogs over gRPC: %v", err) + } + if resp.GetAccepted() != 1 { + t.Errorf("Accepted = %d, want 1", resp.GetAccepted()) + } + if resp.GetDropped() != 0 { + t.Errorf("Dropped = %d, want 0", resp.GetDropped()) + } + + logged := logBuf.String() + if !strings.Contains(logged, "grpc integration hello") { + t.Errorf("expected log message in output, got:\n%s", logged) + } + if !strings.Contains(logged, `"exporter":"test-exporter"`) { + t.Errorf("expected exporter label in output, got:\n%s", logged) + } +} + +func TestTelemetryService_PushLogs_OverGRPC_RejectsMissingToken(t *testing.T) { + signer := testSigner(t) + addr, _, cleanup := startTelemetryGRPCServer(t, signer) + defer cleanup() + + client, closeConn := dialTelemetryClient(t, addr) + defer closeConn() + + _, err := client.PushLogs(context.Background(), &pb.PushLogsRequest{ + Entries: []*pb.LogEntry{{Severity: "info", Message: "unauthenticated"}}, + }) + if err == nil { + t.Fatal("expected error without authorization metadata") + } + if status.Code(err) != codes.Unauthenticated { + t.Errorf("expected Unauthenticated, got %v", status.Code(err)) + } +} + +func TestTelemetryService_PushLogs_OverGRPC_RejectsInvalidToken(t *testing.T) { + signer := testSigner(t) + addr, _, cleanup := startTelemetryGRPCServer(t, signer) + defer cleanup() + + client, closeConn := dialTelemetryClient(t, addr) + defer closeConn() + + ctx := metadata.NewOutgoingContext( + context.Background(), + metadata.Pairs("authorization", "Bearer not-a-real-token"), + ) + _, err := client.PushLogs(ctx, &pb.PushLogsRequest{ + Entries: []*pb.LogEntry{{Severity: "info", Message: "bad token"}}, + }) + if err == nil { + t.Fatal("expected error with invalid token") + } + if status.Code(err) != codes.Unauthenticated { + t.Errorf("expected Unauthenticated, got %v", status.Code(err)) + } +} + +func TestTelemetryService_PushLogs_OverGRPC_BatchLimit(t *testing.T) { + signer := testSigner(t) + addr, _, cleanup := startTelemetryGRPCServer(t, signer) + defer cleanup() + + client, closeConn := dialTelemetryClient(t, addr) + defer closeConn() + + token := mustToken(t, signer, "exporter:jumpstarter:test-exporter:abc123") + ctx := metadata.NewOutgoingContext( + context.Background(), + metadata.Pairs("authorization", "Bearer "+token), + ) + + entries := make([]*pb.LogEntry, 600) + for i := range entries { + entries[i] = &pb.LogEntry{ + Severity: "info", + Message: fmt.Sprintf("batch entry %d", i), + Exporter: "test-exporter", + Component: "exporter", + } + } + + resp, err := client.PushLogs(ctx, &pb.PushLogsRequest{Entries: entries}) + if err != nil { + t.Fatalf("PushLogs over gRPC: %v", err) + } + if resp.GetAccepted() != 500 { + t.Errorf("Accepted = %d, want 500", resp.GetAccepted()) + } + if resp.GetDropped() != 100 { + t.Errorf("Dropped = %d, want 100", resp.GetDropped()) + } +} + func TestTelemetryService_PushLogs_RequiresAuthentication(t *testing.T) { svc := &TelemetryService{BindAddr: ":0", Signer: testSigner(t)}