diff --git a/controller/cmd/telemetry/main.go b/controller/cmd/telemetry/main.go index 08ecb5468..4e16de89f 100644 --- a/controller/cmd/telemetry/main.go +++ b/controller/cmd/telemetry/main.go @@ -18,6 +18,18 @@ limitations under the License. // via the PushLogs gRPC RPC and writes them to structured stdout for downstream // log shippers (Promtail, Grafana Alloy, Vector) to forward to Loki. // +// TLS: always enabled. Set EXTERNAL_CERT_PEM and EXTERNAL_KEY_PEM to file paths of +// operator-mounted cert/key (e.g. from a cert-manager Secret); when absent a +// self-signed certificate is generated. The self-signed cert PEM is logged at +// startup — copy it into the controller ConfigMap's telemetry.certificate field +// so exporters can verify the TLS connection. +// +// Endpoint: GRPC_TELEMETRY_ENDPOINT must be set on BOTH this pod and the controller +// pod to the same value (e.g. "jumpstarter-telemetry.jumpstarter.svc:9093"). +// The telemetry service uses it to generate the correct SAN in the self-signed +// certificate; the controller uses it to advertise the address to exporters via +// GetServiceEndpoints. A mismatch causes TLS hostname verification failures. +// // Future phases will add direct Loki push and MetricsStream for reverse-scrape // of exporter prometheus_client registries. package main diff --git a/controller/internal/config/config.go b/controller/internal/config/config.go index 6b0b2170e..e8dbd2d34 100644 --- a/controller/internal/config/config.go +++ b/controller/internal/config/config.go @@ -127,12 +127,6 @@ func LoadConfiguration( if err := config.Telemetry.Validate(); err != nil { return nil, err } - // Auto-derive the gRPC address when the operator has not overridden it. - // The well-known service name follows the same pattern as the controller - // and router: ..svc (in-cluster DNS). - if config.Telemetry.Endpoint == "" { - config.Telemetry.Endpoint = "jumpstarter-telemetry." + key.Namespace + ":9093" - } telemetry = config.Telemetry } diff --git a/controller/internal/config/types.go b/controller/internal/config/types.go index 3a52f5041..890f10ac2 100644 --- a/controller/internal/config/types.go +++ b/controller/internal/config/types.go @@ -30,15 +30,22 @@ type Telemetry struct { // When true the controller advertises the endpoint returned by GetServiceEndpoints. Enabled bool `json:"enabled,omitempty" yaml:"enabled,omitempty"` - // Endpoint is an optional override for the telemetry gRPC address. - // When empty and Enabled is true, defaults to - // "jumpstarter-telemetry.:9093" derived from the controller namespace. + // Endpoint is an optional override for the telemetry gRPC address advertised + // to exporters. When empty the controller reads GRPC_TELEMETRY_ENDPOINT from its + // own environment (set by the operator on the controller Deployment). Endpoint string `json:"endpoint,omitempty" yaml:"endpoint,omitempty"` - // Certificate is reserved for a future phase where the telemetry service manages - // its own TLS credentials. Leave empty for Phase 1 deployments — the telemetry - // server listens on plaintext gRPC and exporters that receive a certificate here - // will fail to connect. + // Certificate is the PEM-encoded CA certificate that exporters use to verify + // the telemetry server's TLS certificate. + // + // When the operator provisions the telemetry service with a cert-manager-issued + // certificate, set this to the issuer's CA certificate. + // + // When the telemetry service runs in self-signed mode (no EXTERNAL_CERT_PEM / + // EXTERNAL_KEY_PEM set), it logs the generated certificate PEM at startup under + // the key "certPEM". Copy that value here so exporters can pin and verify it. + // A self-signed certificate is not trusted by the system CA pool, so leaving + // this field empty means exporters cannot establish a verified TLS connection. Certificate string `json:"certificate,omitempty" yaml:"certificate,omitempty"` // Logging configures the log ingestion path to the telemetry service. diff --git a/controller/internal/service/controller_service.go b/controller/internal/service/controller_service.go index 2e42adb9b..262fe6afd 100644 --- a/controller/internal/service/controller_service.go +++ b/controller/internal/service/controller_service.go @@ -327,8 +327,11 @@ func (s *ControllerService) GetServiceEndpoints( resp := &pb.GetServiceEndpointsResponse{} if s.TelemetryConfig != nil && s.TelemetryConfig.Enabled { + // Prefer the explicit ConfigMap endpoint; fall back to GRPC_TELEMETRY_ENDPOINT + // so the operator can pass the address via env var without touching the ConfigMap. + ep := cmp.Or(s.TelemetryConfig.Endpoint, telemetryEndpoint()) resp.TelemetryEndpoints = append(resp.TelemetryEndpoints, &pb.TelemetryEndpoint{ - Endpoint: s.TelemetryConfig.Endpoint, + Endpoint: ep, Certificate: s.TelemetryConfig.Certificate, MinSeverity: cmp.Or(s.TelemetryConfig.Logging.Filter.MinSeverity, "info"), }) diff --git a/controller/internal/service/endpoints.go b/controller/internal/service/endpoints.go index 93d0cc379..d79aac5f2 100644 --- a/controller/internal/service/endpoints.go +++ b/controller/internal/service/endpoints.go @@ -21,6 +21,10 @@ func routerEndpoint() string { return ep } +func telemetryEndpoint() string { + return os.Getenv("GRPC_TELEMETRY_ENDPOINT") +} + func endpointToSAN(endpoint string) ([]string, []net.IP, error) { host, _, err := net.SplitHostPort(endpoint) if err != nil { diff --git a/controller/internal/service/telemetry_service.go b/controller/internal/service/telemetry_service.go index f9eeb5b4a..679d69361 100644 --- a/controller/internal/service/telemetry_service.go +++ b/controller/internal/service/telemetry_service.go @@ -18,9 +18,12 @@ package service import ( "context" + "crypto/tls" + "encoding/pem" "errors" "fmt" "net" + "os" "strings" "time" @@ -29,6 +32,7 @@ import ( 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/reflection" "google.golang.org/grpc/status" ctrl "sigs.k8s.io/controller-runtime" @@ -55,12 +59,11 @@ var reservedExtraFieldKeys = map[string]struct{}{ // TelemetryService receives structured log entries from exporters and clients, // logs them via structured stdout, and will forward them to Loki in a future phase. // -// Phase 1 design: the server listens on plaintext gRPC only. TLS termination is -// expected to be handled by a sidecar (e.g. Envoy) or service mesh in production -// deployments. The Certificate field advertised via GetServiceEndpoints is reserved -// for a future phase where the telemetry binary manages its own TLS credentials. -// Do NOT configure the Certificate field in the ConfigMap for Phase 1 deployments — -// exporters that receive a certificate will attempt a TLS handshake that will fail. +// TLS: the server always uses TLS. When EXTERNAL_CERT_PEM and EXTERNAL_KEY_PEM +// env vars point to certificate/key files (mounted by the operator from a Secret), +// those are loaded. Otherwise a self-signed certificate is generated — traffic is +// still encrypted, but clients cannot verify the server identity without the CA cert +// in the ConfigMap telemetry.certificate field. type TelemetryService struct { pb.UnimplementedTelemetryServiceServer @@ -197,16 +200,88 @@ func truncate(s string, n int) string { return s[:b] } +// loadTLSCredentials loads TLS credentials for the gRPC server. +// It reads EXTERNAL_CERT_PEM and EXTERNAL_KEY_PEM env vars (file paths set by the +// operator via Secret volume mounts). When either is absent it falls back to a +// self-signed certificate so that traffic is always encrypted. +// +// selfSignedPEM is non-empty only when a self-signed certificate was generated. +// Callers should log it so the operator can copy it into the ConfigMap's +// telemetry.certificate field — exporters need this PEM to verify the TLS connection. +func (s *TelemetryService) loadTLSCredentials() (creds credentials.TransportCredentials, selfSignedPEM string, err error) { + certPEMPath := os.Getenv("EXTERNAL_CERT_PEM") + keyPEMPath := os.Getenv("EXTERNAL_KEY_PEM") + + var cert *tls.Certificate + if certPEMPath != "" && keyPEMPath != "" { + certPEMBytes, readErr := os.ReadFile(certPEMPath) + if readErr != nil { + return nil, "", fmt.Errorf("failed to read external certificate file: %w", readErr) + } + keyPEMBytes, readErr := os.ReadFile(keyPEMPath) + if readErr != nil { + return nil, "", fmt.Errorf("failed to read external key file: %w", readErr) + } + parsedCert, parseErr := tls.X509KeyPair(certPEMBytes, keyPEMBytes) + if parseErr != nil { + return nil, "", fmt.Errorf("failed to parse external certificate: %w", parseErr) + } + cert = &parsedCert + } else { + // Derive the TLS SAN from the advertised endpoint (what clients connect to), + // not from the bind address (which is a local port like ":9093"). + // Same pattern as the router and controller services. + // IMPORTANT: GRPC_TELEMETRY_ENDPOINT must be set on the telemetry pod itself + // so the SAN matches the endpoint the controller advertises to exporters. + advertised := telemetryEndpoint() + var dnsnames []string + var ipaddresses []net.IP + if advertised != "" { + var sanErr error + dnsnames, ipaddresses, sanErr = endpointToSAN(advertised) + if sanErr != nil { + dnsnames = []string{"localhost"} + } + } else { + // No advertised endpoint configured — development/local mode. + dnsnames = []string{"localhost"} + } + var genErr error + cert, genErr = NewSelfSignedCertificate("jumpstarter telemetry", dnsnames, ipaddresses) + if genErr != nil { + return nil, "", genErr + } + // Encode the leaf cert as PEM so the caller can log it for the operator. + selfSignedPEM = string(pem.EncodeToMemory(&pem.Block{ + Type: "CERTIFICATE", + Bytes: cert.Certificate[0], + })) + } + return credentials.NewServerTLSFromCert(cert), selfSignedPEM, nil +} + // 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") + creds, selfSignedPEM, err := s.loadTLSCredentials() + if err != nil { + return fmt.Errorf("telemetry: load TLS credentials: %w", err) + } + if selfSignedPEM != "" { + // Log the self-signed cert so the operator can copy it into the controller + // ConfigMap's telemetry.certificate field. Exporters need this PEM to verify + // the TLS connection — a self-signed cert is not trusted by the system CA pool. + logger.Info("Using self-signed TLS certificate; copy certPEM into the controller ConfigMap telemetry.certificate so exporters can verify TLS", + "certPEM", selfSignedPEM) + } + lis, err := net.Listen("tcp", s.BindAddr) if err != nil { return fmt.Errorf("telemetry: listen %s: %w", s.BindAddr, err) } - srv := grpc.NewServer() + srv := grpc.NewServer(grpc.Creds(creds)) 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 5eacca8d7..988759ba8 100644 --- a/controller/internal/service/telemetry_service_test.go +++ b/controller/internal/service/telemetry_service_test.go @@ -18,7 +18,10 @@ package service import ( "context" + "crypto/x509" + "encoding/pem" "fmt" + "os" "strings" "testing" @@ -91,8 +94,12 @@ func buildTelemetryEndpointsResponse(cfg *config.Telemetry) *pb.GetServiceEndpoi if minSev == "" { minSev = "info" } + ep := cfg.Endpoint + if ep == "" { + ep = telemetryEndpoint() + } resp.TelemetryEndpoints = append(resp.TelemetryEndpoints, &pb.TelemetryEndpoint{ - Endpoint: cfg.Endpoint, + Endpoint: ep, Certificate: cfg.Certificate, MinSeverity: minSev, }) @@ -153,6 +160,246 @@ func TestGetServiceEndpoints_DefaultsMinSeverityToInfo(t *testing.T) { } } +func TestGetServiceEndpoints_UsesEnvVarWhenEndpointEmpty(t *testing.T) { + t.Setenv("GRPC_TELEMETRY_ENDPOINT", "telemetry.jumpstarter.svc:9093") + + resp := buildTelemetryEndpointsResponse(&config.Telemetry{ + Enabled: true, + // Endpoint intentionally left empty — should fall back to env var. + }) + + if len(resp.TelemetryEndpoints) != 1 { + t.Fatalf("expected 1 endpoint, got %d", len(resp.TelemetryEndpoints)) + } + if resp.TelemetryEndpoints[0].Endpoint != "telemetry.jumpstarter.svc:9093" { + t.Errorf("Endpoint = %q, want %q (from env)", resp.TelemetryEndpoints[0].Endpoint, "telemetry.jumpstarter.svc:9093") + } +} + +func TestGetServiceEndpoints_BothEndpointAndEnvVarEmpty_ReturnsEmptyEndpoint(t *testing.T) { + t.Setenv("GRPC_TELEMETRY_ENDPOINT", "") + + resp := buildTelemetryEndpointsResponse(&config.Telemetry{ + Enabled: true, + // Both Endpoint and GRPC_TELEMETRY_ENDPOINT are empty. + }) + + if len(resp.TelemetryEndpoints) != 1 { + t.Fatalf("expected 1 endpoint entry, got %d", len(resp.TelemetryEndpoints)) + } + // An empty endpoint is returned; the caller must handle this gracefully. + if resp.TelemetryEndpoints[0].Endpoint != "" { + t.Errorf("Endpoint = %q, want empty string when nothing is configured", resp.TelemetryEndpoints[0].Endpoint) + } +} + +func TestGetServiceEndpoints_ConfigEndpointTakesPrecedenceOverEnvVar(t *testing.T) { + t.Setenv("GRPC_TELEMETRY_ENDPOINT", "env-telemetry.svc:9093") + + resp := buildTelemetryEndpointsResponse(&config.Telemetry{ + Enabled: true, + Endpoint: "config-telemetry.svc:9093", + }) + + if len(resp.TelemetryEndpoints) != 1 { + t.Fatalf("expected 1 endpoint, got %d", len(resp.TelemetryEndpoints)) + } + if resp.TelemetryEndpoints[0].Endpoint != "config-telemetry.svc:9093" { + t.Errorf("Endpoint = %q, want config value to win over env var", resp.TelemetryEndpoints[0].Endpoint) + } +} + +// writeTLSPEMFiles generates a self-signed cert, marshals it to PEM, and writes +// cert and key to temporary files. Returns (certPath, keyPath). +func writeTLSPEMFiles(t *testing.T) (certPath, keyPath string) { + t.Helper() + + tlsCert, err := NewSelfSignedCertificate("test", []string{"localhost"}, nil) + if err != nil { + t.Fatalf("NewSelfSignedCertificate: %v", err) + } + + certPEM := pem.EncodeToMemory(&pem.Block{ + Type: "CERTIFICATE", + Bytes: tlsCert.Certificate[0], + }) + + keyDER, err := x509.MarshalPKCS8PrivateKey(tlsCert.PrivateKey) + if err != nil { + t.Fatalf("MarshalPKCS8PrivateKey: %v", err) + } + keyPEM := pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: keyDER}) + + dir := t.TempDir() + certPath = dir + "/tls.crt" + keyPath = dir + "/tls.key" + if err := os.WriteFile(certPath, certPEM, 0o600); err != nil { + t.Fatalf("write cert: %v", err) + } + if err := os.WriteFile(keyPath, keyPEM, 0o600); err != nil { + t.Fatalf("write key: %v", err) + } + return certPath, keyPath +} + +// selfSignedSANs generates a self-signed certificate with the same logic as +// loadTLSCredentials for a given GRPC_TELEMETRY_ENDPOINT value, and returns +// its DNS SANs for assertion. +func selfSignedSANs(t *testing.T, advertised string) []string { + t.Helper() + var dnsnames []string + if advertised != "" { + dns, _, err := endpointToSAN(advertised) + if err != nil { + dnsnames = []string{"localhost"} + } else { + dnsnames = dns + } + } else { + dnsnames = []string{"localhost"} + } + cert, err := NewSelfSignedCertificate("test", dnsnames, nil) + if err != nil { + t.Fatalf("NewSelfSignedCertificate: %v", err) + } + leaf, err := x509.ParseCertificate(cert.Certificate[0]) + if err != nil { + t.Fatalf("ParseCertificate: %v", err) + } + return leaf.DNSNames +} + +func TestTelemetryService_LoadTLSCredentials_SelfSigned(t *testing.T) { + t.Setenv("EXTERNAL_CERT_PEM", "") + t.Setenv("EXTERNAL_KEY_PEM", "") + + svc := &TelemetryService{BindAddr: ":9093", Signer: testSigner(t)} + creds, selfSignedPEM, err := svc.loadTLSCredentials() + if err != nil { + t.Fatalf("loadTLSCredentials() with self-signed cert failed: %v", err) + } + if creds == nil { + t.Fatal("expected non-nil credentials") + } + if selfSignedPEM == "" { + t.Error("expected non-empty selfSignedPEM when no external cert is configured") + } + // Must be parseable PEM. + block, _ := pem.Decode([]byte(selfSignedPEM)) + if block == nil { + t.Errorf("selfSignedPEM is not valid PEM: %s", selfSignedPEM) + } +} + +func TestTelemetryService_LoadTLSCredentials_SelfSignedUsesAdvertisedEndpointForSAN(t *testing.T) { + // When GRPC_TELEMETRY_ENDPOINT is set, the self-signed cert SAN should derive + // from the advertised hostname — not from the bind address. + // We test the SAN derivation logic directly (same code path as loadTLSCredentials). + sans := selfSignedSANs(t, "telemetry.jumpstarter.svc:9093") + if len(sans) != 1 || sans[0] != "telemetry.jumpstarter.svc" { + t.Errorf("expected SAN [telemetry.jumpstarter.svc], got %v", sans) + } +} + +func TestTelemetryService_LoadTLSCredentials_SelfSignedFallsBackToLocalhostWhenNoEndpoint(t *testing.T) { + // When GRPC_TELEMETRY_ENDPOINT is empty, SAN defaults to "localhost". + sans := selfSignedSANs(t, "") + if len(sans) != 1 || sans[0] != "localhost" { + t.Errorf("expected SAN [localhost], got %v", sans) + } +} + +func TestTelemetryService_LoadTLSCredentials_OnlyCertEnvVarFallsBackToSelfSigned(t *testing.T) { + certPath, _ := writeTLSPEMFiles(t) + // Only cert set, key is missing — should fall back to self-signed, not error. + t.Setenv("EXTERNAL_CERT_PEM", certPath) + t.Setenv("EXTERNAL_KEY_PEM", "") + + svc := &TelemetryService{BindAddr: ":9093", Signer: testSigner(t)} + creds, selfSignedPEM, err := svc.loadTLSCredentials() + if err != nil { + t.Fatalf("expected self-signed fallback, got error: %v", err) + } + if creds == nil { + t.Fatal("expected non-nil credentials") + } + // Partial env vars → self-signed fallback, so PEM must be non-empty. + if selfSignedPEM == "" { + t.Error("expected non-empty selfSignedPEM on self-signed fallback") + } +} + +func TestTelemetryService_LoadTLSCredentials_WithValidPEMFiles(t *testing.T) { + certPath, keyPath := writeTLSPEMFiles(t) + t.Setenv("EXTERNAL_CERT_PEM", certPath) + t.Setenv("EXTERNAL_KEY_PEM", keyPath) + + svc := &TelemetryService{BindAddr: ":9093", Signer: testSigner(t)} + creds, selfSignedPEM, err := svc.loadTLSCredentials() + if err != nil { + t.Fatalf("loadTLSCredentials() with valid PEM files failed: %v", err) + } + if creds == nil { + t.Fatal("expected non-nil credentials") + } + // External cert provided — selfSignedPEM must be empty. + if selfSignedPEM != "" { + t.Errorf("expected empty selfSignedPEM when external cert is provided, got non-empty") + } +} + +func TestTelemetryService_LoadTLSCredentials_BadCertFileReturnsError(t *testing.T) { + certFile, err := os.CreateTemp(t.TempDir(), "tls-*.crt") + if err != nil { + t.Fatalf("CreateTemp: %v", err) + } + if err := certFile.Close(); err != nil { + t.Fatalf("close: %v", err) + } + _, keyPath := writeTLSPEMFiles(t) + + t.Setenv("EXTERNAL_CERT_PEM", certFile.Name()) + t.Setenv("EXTERNAL_KEY_PEM", keyPath) + + svc := &TelemetryService{BindAddr: ":9093", Signer: testSigner(t)} + _, _, err = svc.loadTLSCredentials() + if err == nil { + t.Fatal("expected error parsing empty cert file") + } + // Must be a parse error, not a file-not-found. + if strings.Contains(err.Error(), "no such file") { + t.Errorf("expected parse error, got: %v", err) + } +} + +func TestTelemetryService_LoadTLSCredentials_MissingCertFileReturnsError(t *testing.T) { + _, keyPath := writeTLSPEMFiles(t) + t.Setenv("EXTERNAL_CERT_PEM", "/does/not/exist/tls.crt") + t.Setenv("EXTERNAL_KEY_PEM", keyPath) + + svc := &TelemetryService{BindAddr: ":9093", Signer: testSigner(t)} + _, _, err := svc.loadTLSCredentials() + if err == nil { + t.Fatal("expected error reading missing cert file") + } +} + +func TestTelemetryService_Start_FailsWhenExternalCertFileMissing(t *testing.T) { + t.Setenv("EXTERNAL_CERT_PEM", "/no/such/cert.pem") + t.Setenv("EXTERNAL_KEY_PEM", "/no/such/key.pem") + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + svc := &TelemetryService{BindAddr: ":0", Signer: testSigner(t)} + err := svc.Start(ctx) + if err == nil { + t.Fatal("expected Start to fail with missing cert files") + } + if !strings.Contains(err.Error(), "TLS") { + t.Errorf("expected 'TLS' in error, got: %v", err) + } +} + func TestTelemetryService_PushLogs_RequiresAuthentication(t *testing.T) { svc := &TelemetryService{BindAddr: ":0", Signer: testSigner(t)}