Merge pull request #117 from united-manufacturing-hub/eng-2364-fix-loops

[ENG-2364] Avoid Looping of nodes that are visited already while browsing OPCUA server
united-manufacturing-hub · Jan 29, 2025 · 4e410e6 · 4e410e6
2 parents 86150a2 + 845c9fb
commit 4e410e6
Show file tree

Hide file tree

Showing 6 changed files with 36 additions and 25 deletions.
diff --git a/opcua_plugin/browse.go b/opcua_plugin/browse.go
@@ -62,8 +62,8 @@ type Logger interface {
 
 // Browse is a public wrapper function for the browse function
 // Avoid using this function directly, use it only for testing
-func Browse(ctx context.Context, n NodeBrowser, path string, level int, logger Logger, parentNodeId string, nodeChan chan NodeDef, errChan chan error, wg *TrackedWaitGroup, opcuaBrowserChan chan NodeDef) {
-	browse(ctx, n, path, level, logger, parentNodeId, nodeChan, errChan, wg, opcuaBrowserChan)
+func Browse(ctx context.Context, n NodeBrowser, path string, level int, logger Logger, parentNodeId string, nodeChan chan NodeDef, errChan chan error, wg *TrackedWaitGroup, opcuaBrowserChan chan NodeDef, visited *sync.Map) {
+	browse(ctx, n, path, level, logger, parentNodeId, nodeChan, errChan, wg, opcuaBrowserChan, visited)
 }
 
 // browse recursively explores OPC UA nodes to build a comprehensive list of NodeDefs.
@@ -92,9 +92,15 @@ func Browse(ctx context.Context, n NodeBrowser, path string, level int, logger L
 // - `wg` (`*sync.WaitGroup`): WaitGroup to synchronize the completion of goroutines.
 // **Returns:**
 // - `void`: Errors are sent through `errChan`, and discovered nodes are sent through `nodeChan`.
-func browse(ctx context.Context, n NodeBrowser, path string, level int, logger Logger, parentNodeId string, nodeChan chan NodeDef, errChan chan error, wg *TrackedWaitGroup, opcuaBrowserChan chan NodeDef) {
+func browse(ctx context.Context, n NodeBrowser, path string, level int, logger Logger, parentNodeId string, nodeChan chan NodeDef, errChan chan error, wg *TrackedWaitGroup, opcuaBrowserChan chan NodeDef, visited *sync.Map) {
 	defer wg.Done()
 
+	// Check if the current node is already visited else proceed
+	if _, exists := visited.LoadOrStore(n.ID(), struct{}{}); exists {
+		logger.Debugf("node %s is visited already, hence doing an early exit from the browse routine", n.ID().String())
+		return
+	}
+
 	// Limits browsing depth to a maximum of 25 levels in the node hierarchy.
 	// Performance impact is minimized since most browse operations terminate earlier
 	// due to other exit conditions before reaching this maximum depth.
@@ -287,7 +293,7 @@ func browse(ctx context.Context, n NodeBrowser, path string, level int, logger L
 		}
 		for _, child := range children {
 			wg.Add(1)
-			go browse(ctx, child, def.Path, level+1, logger, def.NodeID.String(), nodeChan, errChan, wg, opcuaBrowserChan)
+			go browse(ctx, child, def.Path, level+1, logger, def.NodeID.String(), nodeChan, errChan, wg, opcuaBrowserChan, visited)
 		}
 		return nil
 	}
@@ -403,7 +409,7 @@ func (g *OPCUAInput) discoverNodes(ctx context.Context) ([]NodeDef, map[string]s
 		g.Log.Debugf("Browsing nodeID: %s", nodeID.String())
 		wg.Add(1)
 		wrapperNodeID := NewOpcuaNodeWrapper(g.Client.Node(nodeID))
-		go browse(timeoutCtx, wrapperNodeID, "", 0, g.Log, nodeID.String(), nodeChan, errChan, &wg, opcuaBrowserChan)
+		go browse(timeoutCtx, wrapperNodeID, "", 0, g.Log, nodeID.String(), nodeChan, errChan, &wg, opcuaBrowserChan, &g.visited)
 	}
 
 	go func() {
@@ -479,7 +485,7 @@ func (g *OPCUAInput) BrowseAndSubscribeIfNeeded(ctx context.Context) error {
 
 			wgHeartbeat.Add(1)
 			wrapperNodeID := NewOpcuaNodeWrapper(g.Client.Node(heartbeatNodeID))
-			go browse(ctx, wrapperNodeID, "", 1, g.Log, heartbeatNodeID.String(), nodeHeartbeatChan, errChanHeartbeat, &wgHeartbeat, opcuaBrowserChanHeartbeat)
+			go browse(ctx, wrapperNodeID, "", 1, g.Log, heartbeatNodeID.String(), nodeHeartbeatChan, errChanHeartbeat, &wgHeartbeat, opcuaBrowserChanHeartbeat, &g.visited)
 
 			wgHeartbeat.Wait()
 			close(nodeHeartbeatChan)

diff --git a/opcua_plugin/browse_frontend.go b/opcua_plugin/browse_frontend.go
@@ -41,7 +41,7 @@ func (g *OPCUAInput) GetNodeTree(ctx context.Context, msgChan chan<- string, roo
 
 	var wg TrackedWaitGroup
 	wg.Add(1)
-	browse(ctx, NewOpcuaNodeWrapper(g.Client.Node(rootNode.NodeId)), "", 0, g.Log, rootNode.NodeId.String(), nodeChan, errChan, &wg, opcuaBrowserChan)
+	browse(ctx, NewOpcuaNodeWrapper(g.Client.Node(rootNode.NodeId)), "", 0, g.Log, rootNode.NodeId.String(), nodeChan, errChan, &wg, opcuaBrowserChan, &g.visited)
 	go logBrowseStatus(ctx, nodeChan, msgChan, &wg)
 	go logErrors(ctx, errChan, g.Log)
 	go collectNodes(ctx, opcuaBrowserChan, nodeIDMap, &nodes)

diff --git a/opcua_plugin/opcua.go b/opcua_plugin/opcua.go
@@ -213,6 +213,7 @@ type OPCUAInput struct {
 	browseErrorChan              chan error
 	AutoReconnect                bool
 	ReconnectIntervalInSeconds   int
+	visited                      sync.Map
 }
 
 // cleanupBrowsing ensures the browsing goroutine is properly stopped and cleaned up

diff --git a/opcua_plugin/opcua_opc-plc_test.go b/opcua_plugin/opcua_opc-plc_test.go
@@ -1425,11 +1425,12 @@ opcua:
 			errChan := make(chan error, 100)
 			opcuaBrowserChan := make(chan NodeDef, 100)
 			var wg TrackedWaitGroup
+			var visited sync.Map
 
 			// Browse the node
 			wg.Add(1)
 			wrapperNodeID := NewOpcuaNodeWrapper(input.Client.Node(parsedNodeIDs[0]))
-			go Browse(ctx, wrapperNodeID, "", 0, input.Log, parsedNodeIDs[0].String(), nodeChan, errChan, &wg, opcuaBrowserChan)
+			go Browse(ctx, wrapperNodeID, "", 0, input.Log, parsedNodeIDs[0].String(), nodeChan, errChan, &wg, opcuaBrowserChan, &visited)
 
 			wg.Wait()
 			close(nodeChan)

diff --git a/opcua_plugin/opcua_unittest_test.go b/opcua_plugin/opcua_unittest_test.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/gopcua/opcua/id"
@@ -66,6 +67,7 @@ var _ = Describe("Unit Tests", func() {
 			errChan          chan error
 			wg               *TrackedWaitGroup
 			opcuaBrowserChan chan NodeDef
+			visited          sync.Map
 		)
 		BeforeEach(func() {
 			ctx, cncl = context.WithTimeout(context.Background(), 180*time.Second)
@@ -80,6 +82,7 @@ var _ = Describe("Unit Tests", func() {
 		})
 		AfterEach(func() {
 			cncl()
+			visited.Clear()
 		})
 
 		Context("When browsing nodes with a node class value nil", func() {
@@ -100,7 +103,7 @@ var _ = Describe("Unit Tests", func() {
 				nodeBrowser = rootNodeWithNilNodeClass
 				wg.Add(1)
 				go func() {
-					Browse(ctx, nodeBrowser, path, level, logger, parentNodeId, nodeChan, errChan, wg, opcuaBrowserChan)
+					Browse(ctx, nodeBrowser, path, level, logger, parentNodeId, nodeChan, errChan, wg, opcuaBrowserChan, &visited)
 				}()
 				wg.Wait()
 				close(nodeChan)
@@ -127,7 +130,7 @@ var _ = Describe("Unit Tests", func() {
 				nodeBrowser = rootNode
 				wg.Add(1)
 				go func() {
-					Browse(ctx, nodeBrowser, path, level, logger, parentNodeId, nodeChan, errChan, wg, opcuaBrowserChan)
+					Browse(ctx, nodeBrowser, path, level, logger, parentNodeId, nodeChan, errChan, wg, opcuaBrowserChan, &visited)
 				}()
 				wg.Wait()
 				close(nodeChan)
@@ -165,7 +168,7 @@ var _ = Describe("Unit Tests", func() {
 				nodeBrowser = rootNode
 				wg.Add(1)
 				go func() {
-					Browse(ctx, nodeBrowser, path, level, logger, parentNodeId, nodeChan, errChan, wg, opcuaBrowserChan)
+					Browse(ctx, nodeBrowser, path, level, logger, parentNodeId, nodeChan, errChan, wg, opcuaBrowserChan, &visited)
 				}()
 				wg.Wait()
 				close(nodeChan)
@@ -205,7 +208,7 @@ var _ = Describe("Unit Tests", func() {
 				nodeBrowser = rootNode
 				wg.Add(1)
 				go func() {
-					Browse(ctx, nodeBrowser, path, level, logger, parentNodeId, nodeChan, errChan, wg, opcuaBrowserChan)
+					Browse(ctx, nodeBrowser, path, level, logger, parentNodeId, nodeChan, errChan, wg, opcuaBrowserChan, &visited)
 				}()
 				wg.Wait()
 				close(nodeChan)
@@ -245,7 +248,7 @@ var _ = Describe("Unit Tests", func() {
 				nodeBrowser = rootNode
 				wg.Add(1)
 				go func() {
-					Browse(ctx, nodeBrowser, path, level, logger, parentNodeId, nodeChan, errChan, wg, opcuaBrowserChan)
+					Browse(ctx, nodeBrowser, path, level, logger, parentNodeId, nodeChan, errChan, wg, opcuaBrowserChan, &visited)
 				}()
 				wg.Wait()
 				close(nodeChan)
@@ -288,7 +291,7 @@ var _ = Describe("Unit Tests", func() {
 				nodeBrowser = rootNode
 				wg.Add(1)
 				go func() {
-					Browse(ctx, nodeBrowser, path, level, logger, parentNodeId, nodeChan, errChan, wg, opcuaBrowserChan)
+					Browse(ctx, nodeBrowser, path, level, logger, parentNodeId, nodeChan, errChan, wg, opcuaBrowserChan, &visited)
 				}()
 				wg.Wait()
 				close(nodeChan)
@@ -364,7 +367,7 @@ var _ = Describe("Unit Tests", func() {
 				nodeBrowser = abcFolder
 				wg.Add(1)
 				go func() {
-					Browse(ctx, nodeBrowser, path, level, logger, parentNodeId, nodeChan, errChan, wg, opcuaBrowserChan)
+					Browse(ctx, nodeBrowser, path, level, logger, parentNodeId, nodeChan, errChan, wg, opcuaBrowserChan, &visited)
 				}()
 				wg.Wait()
 				close(nodeChan)
@@ -393,7 +396,7 @@ var _ = Describe("Unit Tests", func() {
 				childNode := createMockNode(85, "child", ua.NodeClassVariable)
 				rootNode.AddReferenceNode(id.HasChild, childNode)
 
-				nodes, errs := startBrowsing(ctx, rootNode, path, level, logger, parentNodeId, nodeChan, errChan, wg, opcuaBrowserChan)
+				nodes, errs := startBrowsing(ctx, rootNode, path, level, logger, parentNodeId, nodeChan, errChan, wg, opcuaBrowserChan, &visited)
 
 				Expect(errs).Should(BeEmpty())
 				Expect(nodes).Should(HaveLen(1))
@@ -406,7 +409,7 @@ var _ = Describe("Unit Tests", func() {
 				childNode := createMockNode(85, "child", ua.NodeClassVariable)
 				rootNode.AddReferenceNode(id.HasComponent, childNode)
 
-				nodes, errs := startBrowsing(ctx, rootNode, path, level, logger, parentNodeId, nodeChan, errChan, wg, opcuaBrowserChan)
+				nodes, errs := startBrowsing(ctx, rootNode, path, level, logger, parentNodeId, nodeChan, errChan, wg, opcuaBrowserChan, &visited)
 				Expect(errs).Should(BeEmpty())
 				Expect(nodes).Should(HaveLen(1))
 				Expect(nodes[0].NodeID.String()).To(Equal("i=85"))
@@ -418,7 +421,7 @@ var _ = Describe("Unit Tests", func() {
 				childNode := createMockNode(85, "child", ua.NodeClassVariable)
 				rootNode.AddReferenceNode(id.Organizes, childNode)
 
-				nodes, errs := startBrowsing(ctx, rootNode, path, level, logger, parentNodeId, nodeChan, errChan, wg, opcuaBrowserChan)
+				nodes, errs := startBrowsing(ctx, rootNode, path, level, logger, parentNodeId, nodeChan, errChan, wg, opcuaBrowserChan, &visited)
 				Expect(errs).Should(BeEmpty())
 				Expect(nodes).Should(HaveLen(1))
 				Expect(nodes[0].NodeID.String()).To(Equal("i=85"))
@@ -430,7 +433,7 @@ var _ = Describe("Unit Tests", func() {
 				childNode := createMockNode(85, "child", ua.NodeClassVariable)
 				rootNode.AddReferenceNode(id.FolderType, childNode)
 
-				nodes, errs := startBrowsing(ctx, rootNode, path, level, logger, parentNodeId, nodeChan, errChan, wg, opcuaBrowserChan)
+				nodes, errs := startBrowsing(ctx, rootNode, path, level, logger, parentNodeId, nodeChan, errChan, wg, opcuaBrowserChan, &visited)
 				Expect(errs).Should(BeEmpty())
 				Expect(nodes).Should(HaveLen(1))
 				Expect(nodes[0].NodeID.String()).To(Equal("i=85"))
@@ -442,7 +445,7 @@ var _ = Describe("Unit Tests", func() {
 				childNode := createMockNode(85, "child", ua.NodeClassVariable)
 				rootNode.AddReferenceNode(id.HasNotifier, childNode)
 
-				nodes, errs := startBrowsing(ctx, rootNode, path, level, logger, parentNodeId, nodeChan, errChan, wg, opcuaBrowserChan)
+				nodes, errs := startBrowsing(ctx, rootNode, path, level, logger, parentNodeId, nodeChan, errChan, wg, opcuaBrowserChan, &visited)
 				Expect(errs).Should(BeEmpty())
 				Expect(nodes).Should(HaveLen(1))
 				Expect(nodes[0].NodeID.String()).To(Equal("i=85"))
@@ -686,10 +689,10 @@ func createMockNode(id uint32, name string, nodeClass ua.NodeClass) *MockOpcuaNo
 // Ensure that the MockOpcuaNodeWraper implements the NodeBrowser interface
 var _ NodeBrowser = &MockOpcuaNodeWraper{}
 
-func startBrowsing(ctx context.Context, rootNode NodeBrowser, path string, level int, logger Logger, parentNodeId string, nodeChan chan NodeDef, errChan chan error, wg *TrackedWaitGroup, opcuaBrowserChan chan NodeDef) ([]NodeDef, []error) {
+func startBrowsing(ctx context.Context, rootNode NodeBrowser, path string, level int, logger Logger, parentNodeId string, nodeChan chan NodeDef, errChan chan error, wg *TrackedWaitGroup, opcuaBrowserChan chan NodeDef, visited *sync.Map) ([]NodeDef, []error) {
 	wg.Add(1)
 	go func() {
-		Browse(ctx, rootNode, path, level, logger, parentNodeId, nodeChan, errChan, wg, opcuaBrowserChan)
+		Browse(ctx, rootNode, path, level, logger, parentNodeId, nodeChan, errChan, wg, opcuaBrowserChan, visited)
 	}()
 	wg.Wait()
 	close(nodeChan)

diff --git a/opcua_plugin/server_info.go b/opcua_plugin/server_info.go
@@ -37,9 +37,9 @@ func (g *OPCUAInput) GetOPCUAServerInformation(ctx context.Context) (ServerInfo,
 	var wg TrackedWaitGroup
 
 	wg.Add(3)
-	go browse(ctx, NewOpcuaNodeWrapper(g.Client.Node(manufacturerNameNodeID)), "", 0, g.Log, manufacturerNameNodeID.String(), nodeChan, errChan, &wg, opcuaBrowserChan)
-	go browse(ctx, NewOpcuaNodeWrapper(g.Client.Node(productNameNodeID)), "", 0, g.Log, productNameNodeID.String(), nodeChan, errChan, &wg, opcuaBrowserChan)
-	go browse(ctx, NewOpcuaNodeWrapper(g.Client.Node(softwareVersionNodeID)), "", 0, g.Log, softwareVersionNodeID.String(), nodeChan, errChan, &wg, opcuaBrowserChan)
+	go browse(ctx, NewOpcuaNodeWrapper(g.Client.Node(manufacturerNameNodeID)), "", 0, g.Log, manufacturerNameNodeID.String(), nodeChan, errChan, &wg, opcuaBrowserChan, &g.visited)
+	go browse(ctx, NewOpcuaNodeWrapper(g.Client.Node(productNameNodeID)), "", 0, g.Log, productNameNodeID.String(), nodeChan, errChan, &wg, opcuaBrowserChan, &g.visited)
+	go browse(ctx, NewOpcuaNodeWrapper(g.Client.Node(softwareVersionNodeID)), "", 0, g.Log, softwareVersionNodeID.String(), nodeChan, errChan, &wg, opcuaBrowserChan, &g.visited)
 	wg.Wait()
 
 	close(nodeChan)