fix loop chip config, clean up chip emitter impl

Author: krehermann

pkg/beholder/chip_ingress_emitter.go

@@ -3,25 +3,53 @@ package beholder
 import (
 	"context"
 	"fmt"
+	"time"
 
 	ch "github.com/smartcontractkit/chainlink-common/pkg/chipingress"
 	chpb "github.com/smartcontractkit/chainlink-common/pkg/chipingress/pb"
 )
 
+const (
+	defaultTimeout = 3 * time.Second
+)
+
 type ChipIngressEmitter struct {
-	client chpb.ChipIngressClient
+	client  chpb.ChipIngressClient
+	timeout time.Duration
+}
+
+type Opt func(*ChipIngressEmitter)
+
+func WithTimeout(timeout time.Duration) Opt {
+	return func(e *ChipIngressEmitter) {
+		e.timeout = timeout
+	}
 }
 
-func NewChipIngressEmitter(client chpb.ChipIngressClient) (Emitter, error) {
+func NewChipIngressEmitter(client chpb.ChipIngressClient, opts ...Opt) (Emitter, error) {
 
 	if client == nil {
 		return nil, fmt.Errorf("chip ingress client is nil")
 	}
-
-	return &ChipIngressEmitter{client: client}, nil
+	e := &ChipIngressEmitter{client: client, timeout: defaultTimeout}
+	for _, opt := range opts {
+		opt(e)
+	}
+	return e, nil
 }
 
 func (c *ChipIngressEmitter) Emit(ctx context.Context, body []byte, attrKVs ...any) error {
+	// newCtx returns a ctx with a timeout min(c.timeout, ctx.Deadline())
+	newCtx := func(ctx context.Context) (context.Context, context.CancelFunc) {
+		// check if ctx has a deadline and it's less than the emitter timeout,
+		// then we need a new ctx with the emitter timeout
+		if dl, ok := ctx.Deadline(); ok && time.Until(dl) < c.timeout {
+			return context.WithTimeout(context.Background(), c.timeout)
+		}
+		return context.WithTimeout(ctx, c.timeout)
+	}
+	ctx, cancel := newCtx(ctx)
+	defer cancel()
 
 	sourceDomain, entityType, err := ExtractSourceAndType(attrKVs...)
 	if err != nil {

pkg/beholder/client.go

@@ -6,7 +6,6 @@ import (
 	"fmt"
 	"net"
 
-	"github.com/smartcontractkit/chainlink-common/pkg/chipingress"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc"
 	"go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp"
@@ -21,6 +20,8 @@ import (
 	oteltrace "go.opentelemetry.io/otel/trace"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/chipingress"
 )
 
 type Emitter interface {
@@ -249,7 +250,7 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (*Client, erro
 			return nil, err
 		}
 
-		chipIngressEmitter, err := NewChipIngressEmitter(chipIngressClient)
+		chipIngressEmitter, err := NewChipIngressEmitter(chipIngressClient, WithTimeout(cfg.EmitterExportTimeout))
 		if err != nil {
 			return nil, fmt.Errorf("failed to create chip ingress emitter: %w", err)
 		}

pkg/beholder/dual_source_emitter.go

@@ -3,7 +3,6 @@ package beholder
 import (
 	"context"
 	"fmt"
-	"time"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 )
@@ -49,11 +48,8 @@ func (d *DualSourceEmitter) Emit(ctx context.Context, body []byte, attrKVs ...an
 
 	// Emit via chip ingress async
 	go func() {
-		d.log.Debugw("emitting to chip ingress", "body", string(body), "attributes", attrKVs)
-		d.log.Debugw("overriding context for chip ingress emission", "timeout", 5*time.Second)
-		ctx2, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-		defer cancel()
-		if err := d.chipIngressEmitter.Emit(ctx2, body, attrKVs...); err != nil {
+
+		if err := d.chipIngressEmitter.Emit(ctx, body, attrKVs...); err != nil {
 			// If the chip ingress emitter fails, we ONLY log the error
 			// because we still want to send the data to the OTLP collector and not cause disruption
 			d.log.Errorw("failed to emit to chip ingress", "error", err)

pkg/loop/server.go

@@ -126,6 +126,7 @@ func (s *Server) start() error {
 			EmitterMaxQueueSize:            s.EnvConfig.TelemetryEmitterMaxQueueSize,
 			ChipIngressEmitterEnabled:      s.EnvConfig.ChipIngressEndpoint != "",
 			ChipIngressEmitterGRPCEndpoint: s.EnvConfig.ChipIngressEndpoint,
+			ChipIngressInsecureConnection:  s.EnvConfig.TelemetryInsecureConnection,
 		}
 
 		if tracingConfig.Enabled {