Repository: skupperproject/skupper
Pull request #1714 (2024-10-11 18:48:59)

fixes deadlock found in the network console collector
exposes the flow pair reconcile process to the collector_job_seconds internal metric to aid in debugging related issues in the future.


Patch:
diff --git a/cmd/network-console-collector/internal/collector/collector.go b/cmd/network-console-collector/internal/collector/collector.go
index 50e77d98..29d347ea 100644
--- a/cmd/network-console-collector/internal/collector/collector.go
+++ b/cmd/network-console-collector/internal/collector/collector.go
@@ -48,7 +48,7 @@ func New(logger *slog.Logger, factory session.ContainerFactory, reg *prometheus.
 	collector.graph = NewGraph(collector.Records).(*graph)
 	collector.processManager = newProcessManager(logger, collector.Records, collector.graph, newStableIdentityProvider(), collector.metrics)
 	collector.addressManager = newAddressManager(collector.logger, collector.Records)
-	collector.pairManager = newPairManager(logger, collector.Records, collector.graph)
+	collector.pairManager = newPairManager(logger, collector.Records, collector.graph, collector.metrics)
 	routerCfg := collector.recordRouting
 	for _, typ := range standardRecordTypes {
 		routerCfg[typ.String()] = collector.Records
diff --git a/cmd/network-console-collector/internal/collector/pairs.go b/cmd/network-console-collector/internal/collector/pairs.go
index e8c0a9cb..cd412f86 100644
--- a/cmd/network-console-collector/internal/collector/pairs.go
+++ b/cmd/network-console-collector/internal/collector/pairs.go
@@ -6,22 +6,25 @@ import (
 	"sync"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/skupperproject/skupper/pkg/vanflow/store"
 )
 
 type pairManager struct {
-	logger *slog.Logger
-	stor   store.Interface
-	graph  *graph
-	idp    idProvider
-	source store.SourceRef
+	logger          *slog.Logger
+	stor            store.Interface
+	graph           *graph
+	idp             idProvider
+	source          store.SourceRef
+	reconcileMetric prometheus.Observer
 
 	mu          sync.Mutex
 	pairs       map[string]struct{}
 	updatePairs chan ProcPairRecord
 }
 
-func newPairManager(logger *slog.Logger, stor store.Interface, graph *graph) *pairManager {
+func newPairManager(logger *slog.Logger, stor store.Interface, graph *graph, m metrics) *pairManager {
+
 	return &pairManager{
 		logger:      logger,
 		stor:        stor,
@@ -33,6 +36,7 @@ func newPairManager(logger *slog.Logger, stor store.Interface, graph *graph) *pa
 			Version: "0.1",
 			ID:      "self",
 		},
+		reconcileMetric: m.internal.reconcileTime.WithLabelValues("self", "flowpairs"),
 	}
 }
 
@@ -55,6 +59,10 @@ func (m *pairManager) run(ctx context.Context) func() error {
 				}()
 			case pair := <-m.updatePairs:
 				func() {
+					start := time.Now()
+					defer func() {
+						m.reconcileMetric.Observe(time.Since(start).Seconds())
+					}()
 					m.mu.Lock()
 					defer m.mu.Unlock()
 					var (
@@ -143,13 +151,21 @@ func (m *pairManager) handleChangeEvent(event changeEvent, stor readonly) {
 	}
 	switch r := entry.Record.(type) {
 	case ProcPairRecord:
-		m.mu.Lock()
-		defer m.mu.Unlock()
-		if _, ok := m.pairs[r.ID]; ok {
-			return
+		if !m.hasPair(r.ID) {
+			select {
+			case m.updatePairs <- r:
+			default:
+				// skip if queue is full - will eventually be reconciled
+			}
 		}
-		m.updatePairs <- r
 	default:
 	}
 
 }
+
+func (m *pairManager) hasPair(id string) bool {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	_, ok := m.pairs[id]
+	return ok
+}


Test patch:
diff --git a/cmd/network-console-collector/internal/collector/pairs_test.go b/cmd/network-console-collector/internal/collector/pairs_test.go
new file mode 100644
index 00000000..a12a035e
--- /dev/null
+++ b/cmd/network-console-collector/internal/collector/pairs_test.go
@@ -0,0 +1,97 @@
+package collector
+
+import (
+	"context"
+	"io"
+	"log/slog"
+	"testing"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/skupperproject/skupper/pkg/vanflow"
+	"github.com/skupperproject/skupper/pkg/vanflow/store"
+	"gotest.tools/poll"
+)
+
+func TestPairManagerSitePairs(t *testing.T) {
+	tlog := slog.New(slog.NewTextHandler(io.Discard, nil))
+	stor := store.NewSyncMapStore(store.SyncMapStoreConfig{
+		Indexers: RecordIndexers(),
+	})
+	graph := NewGraph(stor).(*graph)
+	metrics := register(prometheus.NewRegistry())
+
+	source := store.SourceRef{ID: "test"}
+	sites := []vanflow.SiteRecord{
+		{BaseRecord: vanflow.NewBase("s1")},
+		{BaseRecord: vanflow.NewBase("s2")},
+		{BaseRecord: vanflow.NewBase("s3")},
+	}
+	processes := []vanflow.ProcessRecord{
+		{BaseRecord: vanflow.NewBase("p1"), Parent: ptrTo("s1")},
+
+		{BaseRecord: vanflow.NewBase("p2"), Parent: ptrTo("s2")},
+		{BaseRecord: vanflow.NewBase("p3"), Parent: ptrTo("s2")},
+
+		{BaseRecord: vanflow.NewBase("p4"), Parent: ptrTo("s3")},
+		{BaseRecord: vanflow.NewBase("p5"), Parent: ptrTo("s3")},
+	}
+	procPairs := []struct {
+		Pair           ProcPairRecord
+		ExpectedSource string
+		ExpectedDest   string
+	}{
+		{
+			Pair:           ProcPairRecord{ID: "pp1", Source: "p1", Dest: "p2", Protocol: "test"},
+			ExpectedSource: "s1", ExpectedDest: "s2",
+		}, {
+			Pair:           ProcPairRecord{ID: "pp2", Source: "p2", Dest: "p3", Protocol: "test"},
+			ExpectedSource: "s2", ExpectedDest: "s2",
+		}, {
+			Pair:           ProcPairRecord{ID: "pp3", Source: "p2", Dest: "p1", Protocol: "test"},
+			ExpectedSource: "s2", ExpectedDest: "s1",
+		}, {
+			Pair:           ProcPairRecord{ID: "pp1", Source: "p1", Dest: "p2", Protocol: "test"},
+			ExpectedSource: "s1", ExpectedDest: "s2",
+		}, {
+			Pair:           ProcPairRecord{ID: "pp4", Source: "p2", Dest: "p5", Protocol: "test"},
+			ExpectedSource: "s2", ExpectedDest: "s3",
+		},
+	}
+	for _, site := range sites {
+		stor.Add(site, source)
+	}
+	for _, proc := range processes {
+		stor.Add(proc, source)
+	}
+	for _, pp := range procPairs {
+		stor.Add(pp.Pair, source)
+	}
+
+	graph.Reset()
+
+	manager := newPairManager(tlog, stor, graph, metrics)
+	go manager.run(context.TODO())()
+	for _, procPair := range procPairs {
+		manager.handleChangeEvent(addEvent{Record: procPair.Pair}, stor)
+		poll.WaitOn(t, func(t poll.LogT) poll.Result {
+			for _, entry := range stor.Index(store.TypeIndex, store.Entry{Record: SitePairRecord{}}) {
+				actual := entry.Record.(SitePairRecord)
+				if actual.Protocol == "test" &&
+					actual.Source == procPair.ExpectedSource &&
+					actual.Dest == procPair.ExpectedDest {
+					return poll.Success()
+				}
+			}
+			return poll.Continue("missing site pair")
+		}, poll.WithDelay(time.Microsecond*500))
+	}
+	// re-fire events many times - should be a noop
+	for i := 0; i < 10_000; i++ {
+		for _, procPair := range procPairs {
+			manager.handleChangeEvent(addEvent{Record: procPair.Pair}, stor)
+		}
+	}
+}
+
+func ptrTo[T any](obj T) *T { return &obj }