Repository: skupperproject/skupper
Pull request #1678 (2024-09-23 23:28:47)

Collector api bugfixes
Quick followup to the updated network-console-collector fixing a few bugs.

* Fix several bugs in "pair" manager - including double sitepair records and fixing processgrouppairs.
* Update our sortBy logic to work with the types generated by openapi-spec instead of only primitives.
* /routers should have parent that points to a site.

bartoval: It seems correct to me 👍🏼 .  Perhaps I would add the peername to the links, given that in this PR you add the sourcePGName and destPGName 
c-kruse: Will do a followup for our changes to routerlinks! That's a little larger in scope.

Patch:
diff --git a/cmd/network-console-collector/internal/collector/connections.go b/cmd/network-console-collector/internal/collector/connections.go
index 80ae2ae3..691e99cc 100644
--- a/cmd/network-console-collector/internal/collector/connections.go
+++ b/cmd/network-console-collector/internal/collector/connections.go
@@ -580,38 +580,6 @@ func (c *connectionManager) run(ctx context.Context) {
 						continue
 					}
 
-					sn := c.graph.Process(pair.Source)
-					sSiteID := sn.Parent().ID()
-					dn := c.graph.Process(pair.Dest)
-					dSiteID := dn.Parent().ID()
-
-					sourceProc, ok := sn.GetRecord()
-					if !ok {
-						return
-					}
-					destProc, ok := dn.GetRecord()
-					if !ok {
-						return
-					}
-
-					var (
-						sourceGroupID string
-						destGroupID   string
-					)
-					sGroup, dGroup := dref(sourceProc.Group), dref(destProc.Group)
-					sGroupEntries := c.records.Index(IndexByTypeName, store.Entry{Record: ProcessGroupRecord{
-						Name: sGroup,
-					}})
-					if len(sGroupEntries) > 0 {
-						sourceGroupID = sGroupEntries[0].Record.Identity()
-					}
-					dGroupEntries := c.records.Index(IndexByTypeName, store.Entry{Record: ProcessGroupRecord{
-						Name: dGroup,
-					}})
-					if len(dGroupEntries) > 0 {
-						destGroupID = dGroupEntries[0].Record.Identity()
-					}
-
 					id := c.idp.ID("processpair", pair.Source, pair.Dest, pair.Protocol)
 					if _, ok := c.records.Get(id); !ok {
 						record := ProcPairRecord{
@@ -624,37 +592,6 @@ func (c *connectionManager) run(ctx context.Context) {
 						c.logger.Info("Adding process pairs", slog.Any("id", id))
 						c.records.Add(record, store.SourceRef{ID: "self"})
 					}
-
-					if sSiteID != "" && dSiteID != "" {
-						id := c.idp.ID("sitepair", sSiteID, dSiteID, pair.Protocol)
-						if _, ok := c.records.Get(id); !ok {
-							record := SitePairRecord{
-								ID:       id,
-								Source:   sSiteID,
-								Dest:     dSiteID,
-								Protocol: pair.Protocol,
-								Start:    time.Now(),
-							}
-							c.logger.Info("Adding site pairs", slog.Any("id", id))
-							c.records.Add(record, store.SourceRef{ID: "self"})
-						}
-					}
-
-					if sourceGroupID != "" && destGroupID != "" {
-						id := c.idp.ID("processgrouppair", sourceGroupID, destGroupID, pair.Protocol)
-						if _, ok := c.records.Get(id); !ok {
-							record := ProcGroupPairRecord{
-								ID:       id,
-								Source:   sourceGroupID,
-								Dest:     destGroupID,
-								Protocol: pair.Protocol,
-								Start:    time.Now(),
-							}
-							c.logger.Info("Adding process group pairs", slog.Any("id", id))
-							c.records.Add(record, store.SourceRef{ID: "self"})
-						}
-					}
-
 					c.processPairs[pair] = false
 				}
 			}()
diff --git a/cmd/network-console-collector/internal/collector/pairs.go b/cmd/network-console-collector/internal/collector/pairs.go
index ef12db50..e8c0a9cb 100644
--- a/cmd/network-console-collector/internal/collector/pairs.go
+++ b/cmd/network-console-collector/internal/collector/pairs.go
@@ -38,14 +38,29 @@ func newPairManager(logger *slog.Logger, stor store.Interface, graph *graph) *pa
 
 func (m *pairManager) run(ctx context.Context) func() error {
 	return func() error {
+		refresh := time.NewTicker(time.Second * 20)
+		defer refresh.Stop()
 		for {
 			select {
 			case <-ctx.Done():
 				return nil
+			case <-refresh.C:
+				go func() {
+					processPairs := m.stor.Index(store.TypeIndex, store.Entry{Record: ProcPairRecord{}})
+					for _, entry := range processPairs {
+						if pair, ok := entry.Record.(ProcPairRecord); ok {
+							m.updatePairs <- pair
+						}
+					}
+				}()
 			case pair := <-m.updatePairs:
 				func() {
 					m.mu.Lock()
 					defer m.mu.Unlock()
+					var (
+						hasSitePair  bool
+						hasGroupPair bool
+					)
 					if _, ok := m.pairs[pair.ID]; ok {
 						return
 					}
@@ -55,8 +70,9 @@ func (m *pairManager) run(ctx context.Context) func() error {
 
 					// add site and group pairs
 					if sourceID, destID := sourceN.ID(), destN.ID(); sourceID != "" && destID != "" {
-						spid := m.idp.ID("sp", sourceID, destID, pair.Protocol)
-						m.stor.Add(SitePairRecord{
+						hasSitePair = true
+						spid := m.idp.ID("sitepair", sourceID, destID, pair.Protocol)
+						added := m.stor.Add(SitePairRecord{
 							ID:       spid,
 							Start:    time.Now(),
 							Protocol: pair.Protocol,
@@ -65,34 +81,52 @@ func (m *pairManager) run(ctx context.Context) func() error {
 						},
 							m.source,
 						)
+						if added {
+							m.logger.Info("Added site pair", slog.String("id", spid), slog.String("source", sourceID), slog.String("dest", destID))
+						}
 					}
 
-					pgIDByProcess := func(node Process) (string, bool) {
+					pgIDByProcess := func(node Process) (string, string, bool) {
 						record, ok := node.GetRecord()
 						if !ok {
-							return "", false
+							return "", "", false
 						}
-						groups := m.stor.Index(IndexByTypeName, store.Entry{Record: ProcessGroupRecord{Name: dref(record.Name)}})
+						groups := m.stor.Index(IndexByTypeName, store.Entry{Record: ProcessGroupRecord{Name: dref(record.Group)}})
 						if len(groups) > 0 {
-							return groups[0].Record.Identity(), true
+							if pg, ok := groups[0].Record.(ProcessGroupRecord); ok {
+								return pg.ID, pg.Name, true
+							}
 						}
-						return "", false
+						return "", "", false
 					}
-					sourcePG, sourceFound := pgIDByProcess(m.graph.Process(pair.Source))
-					destPG, destFound := pgIDByProcess(m.graph.Process(pair.Dest))
+					sourcePG, sourcePGName, sourceFound := pgIDByProcess(m.graph.Process(pair.Source))
+					destPG, destPGName, destFound := pgIDByProcess(m.graph.Process(pair.Dest))
 					if sourceFound && destFound {
-						pgpid := m.idp.ID("sp", sourcePG, destPG, pair.Protocol)
-						m.stor.Add(SitePairRecord{
-							ID:       pgpid,
-							Start:    time.Now(),
-							Protocol: pair.Protocol,
-							Source:   sourcePG,
-							Dest:     destPG,
+						hasGroupPair = true
+						pgpid := m.idp.ID("grouppair", sourcePG, destPG, pair.Protocol)
+						added := m.stor.Add(ProcGroupPairRecord{
+							ID:         pgpid,
+							Start:      time.Now(),
+							Protocol:   pair.Protocol,
+							Source:     sourcePG,
+							SourceName: sourcePGName,
+							Dest:       destPG,
+							DestName:   destPGName,
 						},
 							m.source,
 						)
+						if added {
+							m.logger.Info(
+								"Added process group pair",
+								slog.String("id", pgpid),
+								slog.String("source", sourcePG),
+								slog.String("dest", destPG))
+						}
 
 					}
+					if hasSitePair && hasGroupPair {
+						m.pairs[pair.ID] = struct{}{}
+					}
 				}()
 			}
 		}
diff --git a/cmd/network-console-collector/internal/collector/records.go b/cmd/network-console-collector/internal/collector/records.go
index 9f6399b7..380f4c3b 100644
--- a/cmd/network-console-collector/internal/collector/records.go
+++ b/cmd/network-console-collector/internal/collector/records.go
@@ -62,11 +62,13 @@ func (r SitePairRecord) GetTypeMeta() vanflow.TypeMeta {
 }
 
 type ProcGroupPairRecord struct {
-	ID       string
-	Protocol string
-	Source   string
-	Dest     string
-	Start    time.Time
+	ID         string
+	Protocol   string
+	Source     string
+	SourceName string
+	Dest       string
+	DestName   string
+	Start      time.Time
 }
 
 func (r ProcGroupPairRecord) Identity() string {
diff --git a/cmd/network-console-collector/internal/server/query.go b/cmd/network-console-collector/internal/server/query.go
index a5477378..8ae7f309 100644
--- a/cmd/network-console-collector/internal/server/query.go
+++ b/cmd/network-console-collector/internal/server/query.go
@@ -159,10 +159,9 @@ func (m fieldIndex[T]) Compare(x, y T) int {
 		}
 		vx, vy = vx.Elem(), vy.Elem()
 	}
-	xx, yy := vx.Interface(), vy.Interface()
-	switch xx := xx.(type) {
-	case string:
-		yy := yy.(string)
+	switch vx.Kind() {
+	case reflect.String:
+		xx, yy := vx.String(), vy.String()
 		if xx == yy {
 			return 0
 		}
@@ -170,14 +169,18 @@ func (m fieldIndex[T]) Compare(x, y T) int {
 			return -1
 		}
 		return 1
-	case uint64:
-		return int(xx - yy.(uint64))
-	case int32:
-		return int(xx - yy.(int32))
-	case int64:
-		return int(xx - yy.(int64))
-	case int:
-		return int(xx - yy.(int))
+	case reflect.Uint32:
+		fallthrough
+	case reflect.Uint64:
+		xx, yy := vx.Uint(), vy.Uint()
+		return int(xx - yy)
+	case reflect.Int32:
+		fallthrough
+	case reflect.Int:
+		fallthrough
+	case reflect.Int64:
+		xx, yy := vx.Int(), vy.Int()
+		return int(xx - yy)
 	}
 	return 0
 }
diff --git a/cmd/network-console-collector/internal/server/views/views.go b/cmd/network-console-collector/internal/server/views/views.go
index dcc45742..7e79df11 100644
--- a/cmd/network-console-collector/internal/server/views/views.go
+++ b/cmd/network-console-collector/internal/server/views/views.go
@@ -105,7 +105,9 @@ func NewProcessGroupPairProvider() func(collector.ProcGroupPairRecord) api.FlowA
 		out.StartTime = uint64(record.Start.UnixMicro())
 		out.PairType = api.PROCESSGROUP
 		out.SourceId = record.Source
+		out.SourceName = record.SourceName
 		out.DestinationId = record.Dest
+		out.DestinationName = record.DestName
 		out.Protocol = record.Protocol
 		return out
 	}
@@ -553,6 +555,7 @@ func Router(record vanflow.RouterRecord) api.RouterRecord {
 	out := defaultRouter(record.ID)
 	out.StartTime, out.EndTime = vanflowTimes(record.BaseRecord)
 	out.Namespace = record.Namespace
+	setOpt(&out.Parent, record.Parent)
 	setOpt(&out.HostName, record.Hostname)
 	setOpt(&out.ImageName, record.ImageName)
 	setOpt(&out.ImageVersion, record.ImageVersion)


Test patch:
diff --git a/cmd/network-console-collector/internal/server/process_test.go b/cmd/network-console-collector/internal/server/process_test.go
index b480b18d..4f243891 100644
--- a/cmd/network-console-collector/internal/server/process_test.go
+++ b/cmd/network-console-collector/internal/server/process_test.go
@@ -75,6 +75,17 @@ func TestProcesses(t *testing.T) {
 					SourceHost:     "10.99.99.2",
 				})
 			},
+		}, {
+			Records:     append(exProcessWithAddresses(), wrapRecords(vanflow.ProcessRecord{BaseRecord: vanflow.NewBase("0")})...),
+			ExpectOK:    true,
+			ExpectCount: 2,
+			Parameters: map[string][]string{
+				"sortBy": {"processBinding.asc"},
+			},
+			ExpectResults: func(t *testing.T, results []api.ProcessRecord) {
+				assert.Equal(t, results[0].Identity, "1")
+				assert.Equal(t, results[1].Identity, "0")
+			},
 		}, {
 			Records: exProcessWithAddresses(),
 			Parameters: map[string][]string{
diff --git a/cmd/network-console-collector/internal/server/router_test.go b/cmd/network-console-collector/internal/server/router_test.go
new file mode 100644
index 00000000..8473ef0e
--- /dev/null
+++ b/cmd/network-console-collector/internal/server/router_test.go
@@ -0,0 +1,94 @@
+package server
+
+import (
+	"context"
+	"log/slog"
+	"testing"
+
+	"github.com/skupperproject/skupper/cmd/network-console-collector/internal/api"
+	"github.com/skupperproject/skupper/cmd/network-console-collector/internal/collector"
+	"github.com/skupperproject/skupper/pkg/vanflow"
+	"github.com/skupperproject/skupper/pkg/vanflow/store"
+	"gotest.tools/assert"
+)
+
+func TestRouters(t *testing.T) {
+	tlog := slog.Default()
+	stor := store.NewSyncMapStore(store.SyncMapStoreConfig{Indexers: collector.RecordIndexers()})
+	graph := collector.NewGraph(stor)
+	srv, c := requireTestClient(t, New(tlog, stor, graph))
+	defer srv.Close()
+
+	van := []vanflow.Record{
+		vanflow.SiteRecord{BaseRecord: vanflow.NewBase("site-a"), Name: ptrTo("site a")},
+		vanflow.RouterRecord{BaseRecord: vanflow.NewBase("router-a-1"), Name: ptrTo("router a.1"), Parent: ptrTo("site-a"),
+			Mode: ptrTo("inter-router"),
+		},
+
+		vanflow.SiteRecord{BaseRecord: vanflow.NewBase("site-b"), Name: ptrTo("site b")},
+		vanflow.RouterRecord{BaseRecord: vanflow.NewBase("router-b-1"), Name: ptrTo("router b.1"), Parent: ptrTo("site-b")},
+		vanflow.RouterRecord{BaseRecord: vanflow.NewBase("router-b-2"), Name: ptrTo("router b.2"), Parent: ptrTo("site-b")},
+	}
+
+	testcases := []struct {
+		Records              []store.Entry
+		Parameters           map[string][]string
+		ExpectOK             bool
+		ExpectCount          int
+		ExpectTimeRangeCount int
+		ExpectResults        func(t *testing.T, results []api.RouterRecord)
+	}{
+		{ExpectOK: true},
+		{
+			ExpectOK:    true,
+			Records:     wrapRecords(van...),
+			ExpectCount: 3,
+			ExpectResults: func(t *testing.T, results []api.RouterRecord) {
+				for _, result := range results {
+					switch result.Identity {
+					case "router-a-1":
+						assert.DeepEqual(
+							t, result, api.RouterRecord{
+								Identity:     "router-a-1",
+								Name:         "router a.1",
+								HostName:     "unknown",
+								ImageName:    "unknown",
+								ImageVersion: "unknown",
+								Mode:         "inter-router",
+								Parent:       "site-a",
+							},
+						)
+
+					}
+
+				}
+			},
+		},
+	}
+
+	for _, tc := range testcases {
+		t.Run("", func(t *testing.T) {
+			stor.Replace(tc.Records)
+			graph.(reset).Reset()
+			for _, r := range tc.Records {
+				graph.(reset).Reindex(r.Record)
+			}
+			resp, err := c.RoutersWithResponse(context.TODO(), WithParameters(tc.Parameters))
+			assert.Check(t, err)
+			if tc.ExpectOK {
+				assert.Equal(t, resp.StatusCode(), 200)
+				assert.Equal(t, resp.JSON200.Count, int64(tc.ExpectCount))
+				assert.Assert(t, resp.JSON200.Results != nil)
+				assert.Equal(t, len(resp.JSON200.Results), tc.ExpectCount)
+				if tc.ExpectTimeRangeCount != 0 {
+					assert.Equal(t, resp.JSON200.TimeRangeCount, int64(tc.ExpectTimeRangeCount))
+				}
+				if tc.ExpectResults != nil {
+					tc.ExpectResults(t, resp.JSON200.Results)
+				}
+			} else {
+				assert.Check(t, resp.JSON400 != nil)
+			}
+		})
+	}
+}