feat(inputs.opcua): Implement browsing tags on initial connect

plugins/common/opcua/input/input_client.go

@@ -9,6 +9,7 @@ import (
  "strings"
  "time"
 
+ gopcua_id "github.com/gopcua/opcua/id"
  "github.com/gopcua/opcua/ua"
 
  "github.com/influxdata/telegraf"
@@ -57,6 +58,7 @@ type NodeSettings struct {
  TagsSlice [][]string `toml:"tags" deprecated:"1.25.0;use 'default_tags' instead"`
  DefaultTags map[string]string `toml:"default_tags"`
  MonitoringParams MonitoringParameters `toml:"monitoring_params"`
+ Browse bool `toml:"browse"`
 }
 
 // NodeID returns the OPC UA node id
@@ -69,6 +71,7 @@ type NodeGroupSettings struct {
  MetricName string `toml:"name"` // Overrides plugin's setting
  Namespace string `toml:"namespace"` // Can be overridden by node setting
  IdentifierType string `toml:"identifier_type"` // Can be overridden by node setting
+ Browse bool `toml:"browse"` // Can be overridden by node setting, although only to true state
  Nodes []NodeSettings `toml:"nodes"`
  TagsSlice [][]string `toml:"tags" deprecated:"1.26.0;use default_tags"`
  DefaultTags map[string]string `toml:"default_tags"`
@@ -352,6 +355,9 @@ func (o *OpcUAInputClient) InitNodeMetricMapping() error {
  if node.MonitoringParams.SamplingInterval == 0 {
  node.MonitoringParams.SamplingInterval = group.SamplingInterval
  }
+ if node.Browse == false {
+ node.Browse = group.Browse
+ }
 
  nmm, err := NewNodeMetricMapping(group.MetricName, node, groupTags)
  if err != nil {
@@ -368,19 +374,111 @@ func (o *OpcUAInputClient) InitNodeMetricMapping() error {
  return nil
 }
 
-func (o *OpcUAInputClient) InitNodeIDs() error {
+func (o *OpcUAInputClient) InitNodeIDs(ctx context.Context) error {
  o.NodeIDs = make([]*ua.NodeID, 0, len(o.NodeMetricMapping))
  for _, node := range o.NodeMetricMapping {
- nid, err := ua.ParseNodeID(node.Tag.NodeID())
- if err != nil {
- return err
+ var nid *ua.NodeID = nil
+
+ if node.Tag.Browse && node.Tag.IdentifierType == "s" {
+ browsedNodeID, err := o.browseNode(ctx, node.Tag.Namespace, node.Tag.Identifier)
+ if err != nil {
+ o.Log.Errorf("Error browsing node %s: %s", node.Tag.FieldName, err)
+ // continue, other nodes may be browseable
+ // also, add the node as a string identifier. The server will not be able to return data
+ // but we must add a node to the o.NodeIDs array in order for it to match the o.NodeMetricMappings
+ parsedNodeID, err := ua.ParseNodeID(node.Tag.NodeID())
+ if err != nil {
+ return err
+ }
+ nid = parsedNodeID
+ } else {
+ nid = browsedNodeID
+ }
+ } else {
+ parsedNodeID, err := ua.ParseNodeID(node.Tag.NodeID())
+ if err != nil {
+ return err
+ }
+ nid = parsedNodeID
  }
+
  o.NodeIDs = append(o.NodeIDs, nid)
  }
 
  return nil
 }
 
+func parseNamespaceId(ns string) (uint16, error) {
+ nsId, err := strconv.ParseUint(ns, 10, 16)
+ if err != nil {
+ return 0, fmt.Errorf("invalid namespace number: %s", err)
+ }
+
+ return uint16(nsId), nil
+}
+
+// Parse a browse path somewhat as specified in OPC UA Part 4, Annex A.
+// basically a list of segments of ns:BrowseName separated by slashes.
+// for example: 4:Folder/3:Object/Member.Value
+// Use the initialNamespace as the namespace if no ns is specified in the path segment
+func parseBrowsePath(initialNamespace string, browsePath string) ([]*ua.QualifiedName, error) {
+ segmentSplitter := func(r rune) bool {
+ return r == '/' || r == '.'
+ }
+ segments := strings.FieldsFunc(browsePath, segmentSplitter)
+
+ ns, err := parseNamespaceId(initialNamespace)
+ if err != nil {
+ return nil, err
+ }
+
+ pathNames := make([]*ua.QualifiedName, 0, len(segments))
+
+ for _, segment := range segments {
+ browseName := strings.Split(segment, ":")
+
+ segmentNs := ns
+ segmentName := ""
+
+ if len(browseName) == 1 {
+ segmentName = browseName[0]
+ } else if len(browseName) == 2 {
+ segmentNs, err = parseNamespaceId(browseName[0])
+ if err != nil {
+ return nil, fmt.Errorf("Namespace part of segment %s cannot be parsed as an namespace id: %s", segment, err)
+ }
+ segmentName = browseName[1]
+ } else {
+ return nil, fmt.Errorf("Browse name segment %s contains more than one colon", segment)
+ }
+
+ pathNames = append(pathNames, &ua.QualifiedName{
+ NamespaceIndex: segmentNs,
+ Name: segmentName,
+ })
+ }
+
+ return pathNames, nil
+}
+
+func (o *OpcUAInputClient) browseNode(ctx context.Context, ns string, browsePath string) (*ua.NodeID, error) {
+ o.Log.Infof("Browsing node %s", browsePath)
+
+ objectsFolderId := ua.NewNumericNodeID(0, gopcua_id.ObjectsFolder)
+
+ pathNames, err := parseBrowsePath(ns, browsePath)
+ if err != nil {
+ return nil, fmt.Errorf("invalid browse path, %s", err)
+ }
+
+ targetNodeId, err := o.Client.Node(objectsFolderId).TranslateBrowsePathsToNodeIDs(ctx, pathNames)
+ if err != nil {
+ return nil, fmt.Errorf("unable to browse for node, %s", err)
+ }
+
+ return targetNodeId, nil
+}
+
 func (o *OpcUAInputClient) initLastReceivedValues() {
  o.LastReceivedData = make([]NodeValue, len(o.NodeMetricMapping))
  for nodeIdx, nmm := range o.NodeMetricMapping {

plugins/inputs/opcua/README.md

@@ -89,13 +89,19 @@ to use them.
  ## identifier - OPC UA ID (tag as shown in opcua browser)
  ## tags - extra tags to be added to the output metric (optional); deprecated in 1.25.0; use default_tags
  ## default_tags - extra tags to be added to the output metric (optional)
+ ## browse - browse the OPC tag using BrowseName attributes (optional).
+ ## identifier_type must be "s" and identifier must be set to a valid browse path according to OPC UA Part 4, Annex A,
+ ## e.g.: 3:Folder/4:SubFolder/4:Object.Member.Value 
+ ## Objects are browsed starting from the ObjectsFolder folder.
+ ## The namespace in the namespace field is used as the initial namespace, but may be overridden by the browse path
  ##
  ## Use either the inline notation or the bracketed notation, not both.
  #
  ## Inline notation (default_tags not supported yet)
  # nodes = [
  # {name="", namespace="", identifier_type="", identifier="", tags=[["tag1", "value1"], ["tag2", "value2"]},
  # {name="", namespace="", identifier_type="", identifier=""},
+ # {name="", namespace="", identifier_type="s", identifier="3:Folder/4:SubFolder/4:Object.Member.Value", browse=true}
  # ]
  #
  ## Bracketed notation
@@ -104,6 +110,7 @@ to use them.
  # namespace = ""
  # identifier_type = ""
  # identifier = ""
+ # browse = false
  # default_tags = { tag1 = "value1", tag2 = "value2" }
  #
  # [[inputs.opcua.nodes]]

plugins/inputs/opcua/read_client.go

@@ -61,7 +61,7 @@ func (o *ReadClient) Connect() error {
 
  // Make sure we setup the node-ids correctly after reconnect
  // as the server might be restarted and IDs changed
- if err := o.OpcUAInputClient.InitNodeIDs(); err != nil {
+ if err := o.OpcUAInputClient.InitNodeIDs(o.ctx); err != nil {
  return fmt.Errorf("initializing node IDs failed: %w", err)
  }
 

plugins/inputs/opcua/sample.conf

@@ -61,13 +61,19 @@
  ## identifier - OPC UA ID (tag as shown in opcua browser)
  ## tags - extra tags to be added to the output metric (optional); deprecated in 1.25.0; use default_tags
  ## default_tags - extra tags to be added to the output metric (optional)
+ ## browse - browse the OPC tag using BrowseName attributes (optional).
+ ## identifier_type must be "s" and identifier must be set to a valid browse path according to OPC UA Part 4, Annex A,
+ ## e.g.: 3:Folder/4:SubFolder/4:Object.Member.Value 
+ ## Objects are browsed starting from the ObjectsFolder folder.
+ ## The namespace in the namespace field is used as the initial namespace, but may be overridden by the browse path
  ##
  ## Use either the inline notation or the bracketed notation, not both.
  #
  ## Inline notation (default_tags not supported yet)
  # nodes = [
  # {name="", namespace="", identifier_type="", identifier="", tags=[["tag1", "value1"], ["tag2", "value2"]},
  # {name="", namespace="", identifier_type="", identifier=""},
+ # {name="", namespace="", identifier_type="s", identifier="3:Folder/4:SubFolder/4:Object.Member.Value", browse=true}
  # ]
  #
  ## Bracketed notation
@@ -76,6 +82,7 @@
  # namespace = ""
  # identifier_type = ""
  # identifier = ""
+ # browse = false
  # default_tags = { tag1 = "value1", tag2 = "value2" }
  #
  # [[inputs.opcua.nodes]]

plugins/inputs/opcua_listener/README.md

@@ -218,6 +218,7 @@ to use them.
  # nodes = [
  # {name="node1", namespace="", identifier_type="", identifier="",}
  # {name="node2", namespace="", identifier_type="", identifier="", monitoring_params={sampling_interval="0s", queue_size=10, discard_oldest=true, data_change_filter={trigger="Status", deadband_type="Absolute", deadband_value=0.0}}},
+ # {name="node3", namespace="3", identifier_type="s" identifier=3:Folder/4.Object.Member.Field"}
  #]
  #
  ## Bracketed notation

plugins/inputs/opcua_listener/opcua_listener_test.go

@@ -800,6 +800,7 @@ func TestSubscribeClientConfigValidMonitoringParams(t *testing.T) {
  })
 
  subClient, err := subscribeConfig.CreateSubscribeClient(testutil.Logger{})
+ subClient.Connect()
  require.NoError(t, err)
  require.Equal(t, &ua.MonitoringParameters{
  SamplingInterval: 50,

plugins/inputs/opcua_listener/sample.conf

@@ -179,6 +179,7 @@
  # nodes = [
  # {name="node1", namespace="", identifier_type="", identifier="",}
  # {name="node2", namespace="", identifier_type="", identifier="", monitoring_params={sampling_interval="0s", queue_size=10, discard_oldest=true, data_change_filter={trigger="Status", deadband_type="Absolute", deadband_value=0.0}}},
+ # {name="node3", namespace="3", identifier_type="s" identifier=3:Folder/4.Object.Member.Field"}
  #]
  #
  ## Bracketed notation

plugins/inputs/opcua_listener/subscribe_client.go

@@ -87,15 +87,11 @@ func (sc *SubscribeClientConfig) CreateSubscribeClient(log telegraf.Logger) (*Su
  return nil, err
  }
 
- if err := client.InitNodeIDs(); err != nil {
- return nil, err
- }
-
  processingCtx, processingCancel := context.WithCancel(context.Background())
  subClient := &SubscribeClient{
  OpcUAInputClient: client,
  Config: *sc,
- monitoredItemsReqs: make([]*ua.MonitoredItemCreateRequest, len(client.NodeIDs)),
+ monitoredItemsReqs: make([]*ua.MonitoredItemCreateRequest, len(client.NodeMetricMapping)),
  // 100 was chosen to make sure that the channels will not block when multiple changes come in at the same time.
  // The channel size should be increased if reports come in on Telegraf blocking when many changes come in at
  // the same time. It could be made dependent on the number of nodes subscribed to and the subscription interval.
@@ -105,16 +101,6 @@ func (sc *SubscribeClientConfig) CreateSubscribeClient(log telegraf.Logger) (*Su
  cancel: processingCancel,
  }
 
- log.Debugf("Creating monitored items")
- for i, nodeID := range client.NodeIDs {
- // The node id index (i) is used as the handle for the monitored item
- req := opcua.NewMonitoredItemCreateRequestWithDefaults(nodeID, ua.AttributeIDValue, uint32(i))
- if err := assignConfigValuesToRequest(req, &client.NodeMetricMapping[i].Tag.MonitoringParams); err != nil {
- return nil, err
- }
- subClient.monitoredItemsReqs[i] = req
- }
-
  return subClient, nil
 }
 
@@ -124,6 +110,22 @@ func (o *SubscribeClient) Connect() error {
  return err
  }
 
+ // Make sure we setup the node-ids correctly after reconnect
+ // as the server might be restarted and IDs changed
+ o.Log.Debugf("Creating monitored items on first connection")
+ if err := o.OpcUAInputClient.InitNodeIDs(o.ctx); err != nil {
+ return err
+ }
+
+ for i, nodeID := range o.OpcUAInputClient.NodeIDs {
+ // The node id index (i) is used as the handle for the monitored item
+ req := opcua.NewMonitoredItemCreateRequestWithDefaults(nodeID, ua.AttributeIDValue, uint32(i))
+ if err := assignConfigValuesToRequest(req, &o.OpcUAInputClient.NodeMetricMapping[i].Tag.MonitoringParams); err != nil {
+ return err
+ }
+ o.monitoredItemsReqs[i] = req
+ }
+
  o.Log.Debugf("Creating OPC UA subscription")
  o.sub, err = o.Client.Subscribe(o.ctx, &opcua.SubscriptionParameters{
  Interval: time.Duration(o.Config.SubscriptionInterval),