feat: add ChipIngress batch emitter support

core/cmd/shell.go

@@ -83,7 +83,7 @@ func metricViews() []sdkmetric.View {
 	)
 }
 
-func initGlobals(cfgProm config.Prometheus, cfgTracing config.Tracing, cfgTelemetry config.Telemetry, lggr logger.Logger, csaPubKeyHex string, beholderAuthHeaders map[string]string) error {
+func initGlobals(cfgProm config.Prometheus, cfgTelemetry config.Telemetry, cfgTracing config.Tracing, lggr logger.Logger, beholderClient *beholder.Client) error {
 	// Avoid double initializations, but does not prevent relay methods from being called multiple times.
 	var err error
 	initGlobalsOnce.Do(func() {
@@ -95,71 +95,97 @@ func initGlobals(cfgProm config.Prometheus, cfgTracing config.Tracing, cfgTeleme
 				lggr.Errorw("Telemetry error", "err", err)
 			}))
 
-			tracingCfg := loop.TracingConfig{
-				Enabled:         cfgTracing.Enabled(),
-				CollectorTarget: cfgTracing.CollectorTarget(),
-				NodeAttributes:  cfgTracing.Attributes(),
-				SamplingRatio:   cfgTracing.SamplingRatio(),
-				TLSCertPath:     cfgTracing.TLSCertPath(),
-				OnDialError:     func(error) { lggr.Errorw("Failed to dial", "err", err) },
-			}
 			if !cfgTelemetry.Enabled() {
-				return loop.SetupTracing(tracingCfg)
-			}
-
-			var attributes []attribute.KeyValue
-			if tracingCfg.Enabled {
-				attributes = tracingCfg.Attributes()
-			}
-			for k, v := range cfgTelemetry.ResourceAttributes() {
-				attributes = append(attributes, attribute.String(k, v))
+				return loop.SetupTracing(tracingConfig(cfgTracing, lggr))
 			}
 
-			clientCfg := beholder.Config{
-				InsecureConnection:             cfgTelemetry.InsecureConnection(),
-				CACertFile:                     cfgTelemetry.CACertFile(),
-				OtelExporterGRPCEndpoint:       cfgTelemetry.OtelExporterGRPCEndpoint(),
-				ResourceAttributes:             attributes,
-				TraceSampleRatio:               cfgTelemetry.TraceSampleRatio(),
-				EmitterBatchProcessor:          cfgTelemetry.EmitterBatchProcessor(),
-				EmitterExportTimeout:           cfgTelemetry.EmitterExportTimeout(),
-				AuthPublicKeyHex:               csaPubKeyHex,
-				AuthHeaders:                    beholderAuthHeaders,
-				AuthHeadersTTL:                 cfgTelemetry.AuthHeadersTTL(),
-				ChipIngressEmitterEnabled:      cfgTelemetry.ChipIngressEndpoint() != "",
-				ChipIngressEmitterGRPCEndpoint: cfgTelemetry.ChipIngressEndpoint(),
-				ChipIngressInsecureConnection:  cfgTelemetry.ChipIngressInsecureConnection(),
-				LogStreamingEnabled:            cfgTelemetry.LogStreamingEnabled(),
-				LogLevel:                       cfgTelemetry.LogLevel(),
-				LogBatchProcessor:              cfgTelemetry.LogBatchProcessor(),
-				LogExportTimeout:               cfgTelemetry.LogExportTimeout(),
-				LogExportMaxBatchSize:          cfgTelemetry.LogExportMaxBatchSize(),
-				LogExportInterval:              cfgTelemetry.LogExportInterval(),
-				LogMaxQueueSize:                cfgTelemetry.LogMaxQueueSize(),
-				// note: due to the OTEL specification, all histogram buckets
-				// must be defined when the beholder client is created
-				MetricViews: metricViews(),
+			if beholderClient != nil {
+				beholder.SetClient(beholderClient)
+				beholder.SetGlobalOtelProviders()
 			}
-
-			if tracingCfg.Enabled {
-				clientCfg.TraceSpanExporter, err = tracingCfg.NewSpanExporter()
-				if err != nil {
-					return err
-				}
-			}
-			var beholderClient *beholder.Client
-			beholderClient, err = beholder.NewClient(clientCfg)
-			if err != nil {
-				return err
-			}
-			beholder.SetClient(beholderClient)
-			beholder.SetGlobalOtelProviders()
 			return nil
 		}()
 	})
 	return err
 }
 
+func tracingConfig(cfgTracing config.Tracing, lggr logger.Logger) loop.TracingConfig {
+	return loop.TracingConfig{
+		Enabled:         cfgTracing.Enabled(),
+		CollectorTarget: cfgTracing.CollectorTarget(),
+		NodeAttributes:  cfgTracing.Attributes(),
+		SamplingRatio:   cfgTracing.SamplingRatio(),
+		TLSCertPath:     cfgTracing.TLSCertPath(),
+		OnDialError:     func(e error) { lggr.Errorw("Failed to dial", "err", e) },
+	}
+}
+
+// newBeholderClient builds a Beholder client from tracing/telemetry config
+// and sets the CSA signer used for auth header refresh.
+func newBeholderClient(
+	lggr logger.Logger,
+	keyStore keystore.Master,
+	cfgTracing config.Tracing,
+	cfgTelemetry config.Telemetry,
+	csaPubKeyHex string,
+	beholderAuthHeaders map[string]string,
+) (*beholder.Client, error) {
+	attributes := make([]attribute.KeyValue, 0, len(cfgTelemetry.ResourceAttributes()))
+	for k, v := range cfgTelemetry.ResourceAttributes() {
+		attributes = append(attributes, attribute.String(k, v))
+	}
+
+	clientCfg := beholder.Config{
+		InsecureConnection:             cfgTelemetry.InsecureConnection(),
+		CACertFile:                     cfgTelemetry.CACertFile(),
+		OtelExporterGRPCEndpoint:       cfgTelemetry.OtelExporterGRPCEndpoint(),
+		ResourceAttributes:             attributes,
+		TraceSampleRatio:               cfgTelemetry.TraceSampleRatio(),
+		EmitterBatchProcessor:          cfgTelemetry.EmitterBatchProcessor(),
+		EmitterExportTimeout:           cfgTelemetry.EmitterExportTimeout(),
+		AuthPublicKeyHex:               csaPubKeyHex,
+		AuthHeaders:                    beholderAuthHeaders,
+		AuthHeadersTTL:                 cfgTelemetry.AuthHeadersTTL(),
+		ChipIngressEmitterEnabled:      cfgTelemetry.ChipIngressEndpoint() != "",
+		ChipIngressEmitterGRPCEndpoint: cfgTelemetry.ChipIngressEndpoint(),
+		ChipIngressInsecureConnection:  cfgTelemetry.ChipIngressInsecureConnection(),
+		ChipIngressBatchEmitterEnabled: cfgTelemetry.ChipIngressBatchEmitterEnabled(),
+		ChipIngressLogger:              lggr,
+		LogStreamingEnabled:            cfgTelemetry.LogStreamingEnabled(),
+		LogLevel:                       cfgTelemetry.LogLevel(),
+		LogBatchProcessor:              cfgTelemetry.LogBatchProcessor(),
+		LogExportTimeout:               cfgTelemetry.LogExportTimeout(),
+		LogExportMaxBatchSize:          cfgTelemetry.LogExportMaxBatchSize(),
+		LogExportInterval:              cfgTelemetry.LogExportInterval(),
+		LogMaxQueueSize:                cfgTelemetry.LogMaxQueueSize(),
+		// Due to OpenTelemetry semantics, histogram bucket boundaries must be set
+		// when the Beholder client is constructed.
+		MetricViews: metricViews(),
+	}
+
+	if cfgTracing.Enabled() {
+		tracingCfg := tracingConfig(cfgTracing, lggr)
+		// add tracing attributes to resource attributes
+		clientCfg.ResourceAttributes = append(clientCfg.ResourceAttributes, tracingCfg.Attributes()...)
+
+		var err error
+		clientCfg.TraceSpanExporter, err = tracingCfg.NewSpanExporter()
+		if err != nil {
+			return nil, err
+		}
+	}
+	beholderClient, err := beholder.NewClient(clientCfg)
+	if err != nil {
+		return nil, err
+	}
+
+	// Set the signer used to refresh auth headers when AuthHeadersTTL is non-zero.
+	// TTL 0 means static headers only; the signer will not run.
+	beholderClient.SetSigner(&keystore.CSASigner{CSA: keyStore.CSA()})
+
+	return beholderClient, nil
+}
+
 // ErrNoAPICredentialsAvailable is returned when not run from a terminal
 // and no API credentials have been provided
 var ErrNoAPICredentialsAvailable = errors.New("API credentials must be supplied")
@@ -188,9 +214,10 @@ type Shell struct {
 	secretsFiles     []string
 	secretsFileIsSet bool
 
-	LDB      pg.LockedDB        // initialized in BeforeNode
-	DS       sqlutil.DataSource // initialized in BeforeNode
-	KeyStore keystore.Master    // initialized in BeforeNode
+	LDB            pg.LockedDB        // initialized in BeforeNode
+	DS             sqlutil.DataSource // initialized in BeforeNode
+	KeyStore       keystore.Master    // initialized in BeforeNode
+	BeholderClient *beholder.Client   // initialized in BeforeNode
 
 	CleanupOnce sync.Once // ensures cleanup happens exactly once
 }

core/cmd/shell_local.go

@@ -1196,33 +1196,55 @@ func (s *Shell) beforeNode(c *cli.Context) error {
 		return fmt.Errorf("failed to build Beholder auth: %w", err)
 	}
 
-	// Initialize globals with beholder and telemetry
-	err = initGlobals(s.Config.Prometheus(), s.Config.Tracing(), s.Config.Telemetry(), s.Logger, csaPubKeyHex, beholderAuthHeaders)
-	if err != nil {
+	// Build Beholder client: a real one when telemetry is enabled, otherwise a
+	// no-op so that downstream code never needs to nil-check.
+	if s.Config.Telemetry().Enabled() {
+		var beholderErr error
+		s.BeholderClient, beholderErr = newBeholderClient(s.Logger, keyStore, s.Config.Tracing(), s.Config.Telemetry(), csaPubKeyHex, beholderAuthHeaders)
+		if beholderErr != nil {
+			return fmt.Errorf("failed creating beholder client: %w", beholderErr)
+		}
+	} else {
+		s.BeholderClient = beholder.NewNoopClient()
+	}
+
+	// Prometheus, grpc, tracing, and (when telemetry is on) Beholder OTel globals.
+	if err = initGlobals(s.Config.Prometheus(), s.Config.Telemetry(), s.Config.Tracing(), s.Logger, s.BeholderClient); err != nil {
 		return fmt.Errorf("failed initializing globals: %w", err)
 	}
 
-	// Set the signing mechanism for beholder auth headers
-	// if the TTL is 0, we will use the static headers, and this signer will never be called.
-	beholder.GetClient().SetSigner(&keystore.CSASigner{CSA: keyStore.CSA()})
-	// Emit node configuration through beholder
-	s.EmitNodeConfig(ctx)
-	// If log streaming is enabled swap core to add Otel
-	if s.Config.Telemetry().LogStreamingEnabled() {
-		if s.SetOtelCore == nil {
-			return errors.New("Shell.SetOtelCore is nil")
-		}
-		otelLogger := beholder.GetLogger()
-		logLevel := s.Config.Telemetry().LogLevel()
-		otelCore := otelzap.NewCore(otelLogger, otelzap.WithLevel(logLevel))
+	// Wire log streaming before Start so ChipIngressLogger and other internals
+	// see the OTel-backed logger core.
+	if err = s.setupLogStreaming(); err != nil {
+		return err
+	}
+
+	// Start the beholder client after log streaming is wired.
+	if err = s.BeholderClient.Start(ctx); err != nil {
+		return fmt.Errorf("failed to start beholder client: %w", err)
+	}
 
-		s.SetOtelCore(otelCore)
-		lggr.Info("Log streaming enabled")
+	if s.Config.Telemetry().Enabled() {
+		s.EmitNodeConfig(ctx)
 	}
 
 	return nil
 }
 
+func (s *Shell) setupLogStreaming() error {
+	if !s.Config.Telemetry().LogStreamingEnabled() {
+		return nil
+	}
+	if s.SetOtelCore == nil {
+		return errors.New("Shell.SetOtelCore is nil")
+	}
+	logLevel := s.Config.Telemetry().LogLevel()
+	otelCore := otelzap.NewCore(s.BeholderClient.Logger, otelzap.WithLevel(logLevel))
+	s.SetOtelCore(otelCore)
+	s.Logger.Info("Log streaming enabled")
+	return nil
+}
+
 // BeforeNode initializes database, keystore, logger, and beholder for node startup.
 // This is used as a Before hook in CLI commands that require these components.
 func (s *Shell) BeforeNode(c *cli.Context) error {
@@ -1247,6 +1269,9 @@ func (s *Shell) afterNode(lggr logger.SugaredLogger) {
 				log.Printf("Failed to close Logger: %v", err)
 			}
 		}
+		if err := s.BeholderClient.Close(); err != nil {
+			log.Printf("Failed to close Beholder client: %v", err)
+		}
 	})
 }