Fix deadlock during start (#744)

Author: mariomac

pkg/components/pipe/instrumenter.go

@@ -78,7 +78,12 @@ func newGraphBuilder(
 		DecoratedTraces: tracesReaderToRouter,
 	}), swarm.WithID("ReadFromChannel"))
 
-	routerToKubeDecorator := newQueue("routerToKubeDecorator")
+	routerToKubeDecorator := msg.NewQueue[[]request.Span](
+		msg.ChannelBufferLen(config.ChannelBufferLen),
+		msg.Name("routerToKubeDecorator"),
+		// make sure that we are able to wait for the informer sync timeout before failing the pipeline
+		// if a message gets bocked while the Kube decorator starts
+		msg.SendTimeout(config.Attributes.Kubernetes.InformersSyncTimeout+20*time.Second))
 	swi.Add(transform.RoutesProvider(
 		config.Routes,
 		tracesReaderToRouter,

pkg/export/attributes/attr_selector.go

@@ -111,7 +111,6 @@ func NewCustomAttrSelector(
 	cfg *SelectorConfig,
 	extraDefinitionsProvider func(groups AttrGroups, extraGroupAttributes GroupAttributes) map[Section]AttrReportGroup,
 ) (*AttrSelector, error) {
-	cfg.SelectionCfg.Normalize()
 	extraGroupAttributes := NewGroupAttributes(cfg.ExtraGroupAttributesCfg)
 
 	definitions := getDefinitions(groups, extraGroupAttributes)

pkg/pipe/msg/queue.go

@@ -6,6 +6,7 @@ package msg
 
 import (
 	"fmt"
+	"log/slog"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -14,20 +15,22 @@ import (
 
 // if a Send operation takes more than this time, we panic informing about a deadlock
 // in the user-provide pipeline
-const sendTimeout = 20 * time.Second
+const defaultSendTimeout = 20 * time.Second
 
 const unnamed = "(unnamed)"
 
 type queueConfig struct {
 	channelBufferLen int
 	closingAttempts  int
 	name             string
+	sendTimeout      time.Duration
 }
 
 var defaultQueueConfig = queueConfig{
 	channelBufferLen: 1,
 	closingAttempts:  1,
 	name:             unnamed,
+	sendTimeout:      defaultSendTimeout,
 }
 
 // QueueOpts allow configuring some operation of a queue
@@ -47,6 +50,17 @@ func Name(name string) QueueOpts {
 	}
 }
 
+// SendTimeout sets the timeout for Send operations. This is useful for detecting
+// deadlocks derived from a wrong Pipeline construction. It panics if after
+// a send operation, the channel is blocked for more than this timeout.
+// Some nodes might require too long to initialize. For example the Kubernetes Decorator
+// at start, has to download a whole snapshot
+func SendTimeout(to time.Duration) QueueOpts {
+	return func(c *queueConfig) {
+		c.sendTimeout = to
+	}
+}
+
 // ClosingAttempts sets the number of invocations to MarkCloseable before the channel is
 // effectively closed.
 // This is useful when multiple nodes are sending messages to the same queue, and we want
@@ -88,7 +102,7 @@ func NewQueue[T any](opts ...QueueOpts) *Queue[T] {
 	for _, opt := range opts {
 		opt(&cfg)
 	}
-	return &Queue[T]{cfg: &cfg, remainingClosers: cfg.closingAttempts, sendTimeout: time.NewTimer(sendTimeout)}
+	return &Queue[T]{cfg: &cfg, remainingClosers: cfg.closingAttempts, sendTimeout: time.NewTimer(cfg.sendTimeout)}
 }
 
 // Send a message to all subscribers of this queue.
@@ -114,11 +128,36 @@ func (q *Queue[T]) chainedSend(o T, bypassPath []string) {
 	if len(q.dsts) == 0 {
 		return
 	}
-	q.sendTimeout.Reset(sendTimeout)
+
+	// instead of directly panicking in sendTimeout, we first warn at 0.75*sendTimeout,
+	// to get logged about other blocked senders before panicking
+	q.sendTimeout.Reset(3 * q.cfg.sendTimeout / 4)
+	var blocked []dst[T]
 	for _, d := range q.dsts {
 		select {
 		case d.ch <- o:
-		// good!
+			// good!
+		case <-q.sendTimeout.C:
+			slog.With(
+				"timeout", q.cfg.sendTimeout,
+				"queueLen", len(d.ch), "queueCap", cap(d.ch),
+				"sendPath", strings.Join(bypassPath, "->"),
+				"dstName", d.name).
+				Warn("subscriber channel is taking too long to respond")
+			// reset timeout to a small amount to detect any other possible blocked subscriber
+			q.sendTimeout.Reset(time.Second)
+			blocked = append(blocked, d)
+		}
+	}
+	if len(blocked) == 0 {
+		return
+	}
+	// if we confirm that the blocker candidates are actually blocked, we panic
+	q.sendTimeout.Reset(q.cfg.sendTimeout / 4)
+	for _, d := range blocked {
+		select {
+		case d.ch <- o:
+			// good!
 		case <-q.sendTimeout.C:
 			panic(fmt.Sprintf("sending through queue path %s. Subscriber channel %s is blocked",
 				strings.Join(bypassPath, "->"), d.name))

pkg/pipe/msg/queue_test.go

@@ -403,7 +403,7 @@ func TestBlockingPanics(t *testing.T) {
 			q1.Send(1)
 			q1.Send(2)
 
-			time.Sleep(2 * sendTimeout)
+			time.Sleep(2 * defaultSendTimeout)
 		}, "a deadlock should have been detected")
 	})
 }

pkg/transform/k8s.go

@@ -23,6 +23,7 @@ import (
 	"go.opentelemetry.io/obi/pkg/kubeflags"
 	"go.opentelemetry.io/obi/pkg/pipe/msg"
 	"go.opentelemetry.io/obi/pkg/pipe/swarm"
+	"go.opentelemetry.io/obi/pkg/pipe/swarm/swarms"
 )
 
 var containerInfoForPID = container.InfoForPID
@@ -144,24 +145,13 @@ type metadataDecorator struct {
 func (md *metadataDecorator) nodeLoop(ctx context.Context) {
 	// output channel must be closed so later stages in the pipeline can finish in cascade
 	defer md.output.Close()
-	klog().Debug("starting kubernetes span decoration loop")
-	for {
-		select {
-		case <-ctx.Done():
-			klog().Debug("context done, stopping kubernetes span decoration loop")
-			return
-		case spans, ok := <-md.input:
-			if !ok {
-				klog().Debug("input channel closed, stopping kubernetes span decoration loop")
-				return
-			}
-			// in-place decoration and forwarding
-			for i := range spans {
-				md.do(&spans[i])
-			}
-			md.output.Send(spans)
+	swarms.ForEachInput(ctx, md.input, klog().Debug, func(spans []request.Span) {
+		// in-place decoration and forwarding
+		for i := range spans {
+			md.do(&spans[i])
 		}
-	}
+		md.output.Send(spans)
+	})
 }
 
 func (md *metadataDecorator) do(span *request.Span) {
@@ -172,11 +162,15 @@ func (md *metadataDecorator) do(span *request.Span) {
 		span.Service.Metadata = map[attr.Name]string{}
 	}
 	// override the peer and host names from Kubernetes metadata, if found
-	if name, _ := md.db.ServiceNameNamespaceForIP(span.Host); name != "" {
-		span.HostName = name
+	if span.Host != "" {
+		if name, _ := md.db.ServiceNameNamespaceForIP(span.Host); name != "" {
+			span.HostName = name
+		}
 	}
-	if name, _ := md.db.ServiceNameNamespaceForIP(span.Peer); name != "" {
-		span.PeerName = name
+	if span.Peer != "" {
+		if name, _ := md.db.ServiceNameNamespaceForIP(span.Peer); name != "" {
+			span.PeerName = name
+		}
 	}
 }