feat(inputs.modbus): Add workaround for devices with smaller batch size

plugins/inputs/modbus/README.md

@@ -362,11 +362,6 @@ plugin ordering. See [CONFIGURATION.md][CONFIGURATION.md] for more details.
     ## stay connected during gather and disconnect afterwards.
     # close_connection_after_gather = false
 
-    ## Force the plugin to read each field in a separate request.
-    ## This might be necessary for devices not conforming to the spec,
-    ## see https://github.com/influxdata/telegraf/issues/12071.
-    # one_request_per_field = false
-
     ## Enforce the starting address to be zero for the first request on
     ## coil registers. This is necessary for some devices see
     ## https://github.com/influxdata/telegraf/issues/8905
@@ -381,6 +376,21 @@ plugin ordering. See [CONFIGURATION.md][CONFIGURATION.md] for more details.
     ##   upper -- use only upper byte of the register (XX00 XX00 XX00 XX00)
     ## By default both bytes of the register are used (XXXX XXXX).
     # string_register_location = ""
+
+    ## Force the plugin to read each field in a separate request.
+    ## This might be necessary for devices not conforming to the spec,
+    ## see https://github.com/influxdata/telegraf/issues/12071.
+    # one_request_per_field = false
+
+    ## Maximum number of coil or discrete-input registers to read per request
+    ## The specification defines this limit as 2000 but some devices only
+    ## support a smaller number per request
+    # max_bit_registers_per_request = 2000
+
+    ## Maximum number of input or holding registers to read per request
+    ## The specification defines this limits to 125 but some devices only
+    ## support a smaller number per request
+    # max_word_registers_per_request = 125
 ```
 
 ## Notes
@@ -399,8 +409,8 @@ collection interval. Note that pauses add up if multiple requests are sent!
 The modbus plugin supports multiple configuration styles that can be set using
 the `configuration_type` setting. The different styles are described
 below. Please note that styles cannot be mixed.
-Only the settings belonging to the configured `configuration_type` are used for constructing _modbus_
-requests and creation of metrics.
+Only the settings belonging to the configured `configuration_type` are used for
+constructing _modbus_ requests and creation of metrics.
 
 Directly jump to the styles:
 

plugins/inputs/modbus/configuration_metric.go

@@ -218,6 +218,8 @@ func (c *configurationPerMetric) process() (map[byte]requestSet, error) {
 				params.maxBatchSize = maxQuantityCoils
 				if c.workarounds.OnRequestPerField {
 					params.maxBatchSize = 1
+				} else if c.workarounds.MaxBitRegistersPerRequest > 0 {
+					params.maxBatchSize = c.workarounds.MaxBitRegistersPerRequest
 				}
 				params.enforceFromZero = c.workarounds.ReadCoilsStartingAtZero
 				requests := groupFieldsToRequests(fields, params)
@@ -226,20 +228,26 @@ func (c *configurationPerMetric) process() (map[byte]requestSet, error) {
 				params.maxBatchSize = maxQuantityDiscreteInput
 				if c.workarounds.OnRequestPerField {
 					params.maxBatchSize = 1
+				} else if c.workarounds.MaxBitRegistersPerRequest > 0 {
+					params.maxBatchSize = c.workarounds.MaxBitRegistersPerRequest
 				}
 				requests := groupFieldsToRequests(fields, params)
 				set.discrete = append(set.discrete, requests...)
 			case "holding":
 				params.maxBatchSize = maxQuantityHoldingRegisters
 				if c.workarounds.OnRequestPerField {
 					params.maxBatchSize = 1
+				} else if c.workarounds.MaxWordRegistersPerRequest > 0 {
+					params.maxBatchSize = c.workarounds.MaxWordRegistersPerRequest
 				}
 				requests := groupFieldsToRequests(fields, params)
 				set.holding = append(set.holding, requests...)
 			case "input":
 				params.maxBatchSize = maxQuantityInputRegisters
 				if c.workarounds.OnRequestPerField {
 					params.maxBatchSize = 1
+				} else if c.workarounds.MaxWordRegistersPerRequest > 0 {
+					params.maxBatchSize = c.workarounds.MaxWordRegistersPerRequest
 				}
 				requests := groupFieldsToRequests(fields, params)
 				set.input = append(set.input, requests...)

plugins/inputs/modbus/configuration_metric_test.go

@@ -1,6 +1,7 @@
 package modbus
 
 import (
+	"fmt"
 	"testing"
 	"time"
 
@@ -391,3 +392,72 @@ func TestMetricAddressOverflow(t *testing.T) {
 	}
 	require.ErrorIs(t, plugin.Init(), errAddressOverflow)
 }
+
+func TestMetricMaxRegistersWorkaround(t *testing.T) {
+	fields := make([]metricFieldDefinition, 0, 40)
+	for i := range 10 {
+		fields = append(fields,
+			metricFieldDefinition{
+				Name:         fmt.Sprintf("field-coil-%d", i),
+				Address:      uint16(i),
+				RegisterType: "coil",
+			},
+		)
+	}
+	for i := range 10 {
+		fields = append(fields,
+			metricFieldDefinition{
+				Name:         fmt.Sprintf("field-discrete-%d", i),
+				Address:      uint16(i),
+				RegisterType: "discrete",
+			},
+		)
+	}
+	for i := range 10 {
+		fields = append(fields,
+			metricFieldDefinition{
+				Name:         fmt.Sprintf("field-holding-%d", i),
+				Address:      uint16(i * 4),
+				InputType:    "UINT64",
+				RegisterType: "holding",
+			},
+		)
+	}
+	for i := range 10 {
+		fields = append(fields,
+			metricFieldDefinition{
+				Name:         fmt.Sprintf("field-input-%d", i),
+				Address:      uint16(i * 4),
+				InputType:    "UINT64",
+				RegisterType: "input",
+			},
+		)
+	}
+	plugin := &Modbus{
+		Name:              "Test",
+		Controller:        "tcp://localhost:1502",
+		ConfigurationType: "metric",
+		Log:               &testutil.Logger{},
+		Workarounds: workarounds{
+			MaxBitRegistersPerRequest:  6,
+			MaxWordRegistersPerRequest: 8,
+		},
+	}
+	plugin.Metrics = []metricDefinition{
+		{
+			SlaveID:     1,
+			ByteOrder:   "ABCD",
+			Measurement: "test",
+			Fields:      fields,
+		},
+	}
+	require.NoError(t, plugin.Init())
+
+	// Check the resulting requests
+	require.Len(t, plugin.requests, 1)
+	require.Contains(t, plugin.requests, byte(1))
+	require.Len(t, plugin.requests[1].coil, 2, "coil")
+	require.Len(t, plugin.requests[1].discrete, 2, "discrete")
+	require.Len(t, plugin.requests[1].holding, 5, "holding")
+	require.Len(t, plugin.requests[1].input, 5, "input")
+}

plugins/inputs/modbus/configuration_register.go

@@ -58,33 +58,44 @@ func (c *configurationOriginal) check() error {
 }
 
 func (c *configurationOriginal) process() (map[byte]requestSet, error) {
-	maxQuantity := uint16(1)
-	if !c.workarounds.OnRequestPerField {
-		maxQuantity = maxQuantityCoils
+	maxQuantity := maxQuantityCoils
+	if c.workarounds.OnRequestPerField {
+		maxQuantity = 1
+	} else if c.workarounds.MaxBitRegistersPerRequest > 0 {
+		maxQuantity = c.workarounds.MaxBitRegistersPerRequest
 	}
 	coil, err := c.initRequests(c.Coils, maxQuantity, false)
 	if err != nil {
 		return nil, err
 	}
 
-	if !c.workarounds.OnRequestPerField {
-		maxQuantity = maxQuantityDiscreteInput
+	maxQuantity = maxQuantityDiscreteInput
+	if c.workarounds.OnRequestPerField {
+		maxQuantity = 1
+	} else if c.workarounds.MaxBitRegistersPerRequest > 0 {
+		maxQuantity = c.workarounds.MaxBitRegistersPerRequest
 	}
 	discrete, err := c.initRequests(c.DiscreteInputs, maxQuantity, false)
 	if err != nil {
 		return nil, err
 	}
 
-	if !c.workarounds.OnRequestPerField {
-		maxQuantity = maxQuantityHoldingRegisters
+	maxQuantity = maxQuantityHoldingRegisters
+	if c.workarounds.OnRequestPerField {
+		maxQuantity = 1
+	} else if c.workarounds.MaxWordRegistersPerRequest > 0 {
+		maxQuantity = c.workarounds.MaxWordRegistersPerRequest
 	}
 	holding, err := c.initRequests(c.HoldingRegisters, maxQuantity, true)
 	if err != nil {
 		return nil, err
 	}
 
-	if !c.workarounds.OnRequestPerField {
-		maxQuantity = maxQuantityInputRegisters
+	maxQuantity = maxQuantityInputRegisters
+	if c.workarounds.OnRequestPerField {
+		maxQuantity = 1
+	} else if c.workarounds.MaxWordRegistersPerRequest > 0 {
+		maxQuantity = c.workarounds.MaxWordRegistersPerRequest
 	}
 	input, err := c.initRequests(c.InputRegisters, maxQuantity, true)
 	if err != nil {

plugins/inputs/modbus/configuration_register_test.go

@@ -1193,3 +1193,72 @@ func TestRegisterHighAddresses(t *testing.T) {
 	require.NoError(t, modbus.Gather(&acc))
 	testutil.RequireMetricsEqual(t, expected, acc.GetTelegrafMetrics(), testutil.IgnoreTime())
 }
+
+func TestRegisterMaxRegistersWorkaround(t *testing.T) {
+	plugin := &Modbus{
+		Name:                  "Test",
+		Controller:            "tcp://localhost:1502",
+		ConfigurationType:     "register",
+		configurationOriginal: configurationOriginal{SlaveID: 1},
+		Log:                   &testutil.Logger{},
+		Workarounds: workarounds{
+			MaxBitRegistersPerRequest:  6,
+			MaxWordRegistersPerRequest: 8,
+		},
+	}
+	plugin.Coils = make([]fieldDefinition, 0, 10)
+	for i := range uint16(10) {
+		plugin.Coils = append(plugin.Coils,
+			fieldDefinition{
+				Measurement: "test",
+				Name:        fmt.Sprintf("field-coil-%d", i),
+				Address:     []uint16{i},
+			},
+		)
+	}
+	plugin.DiscreteInputs = make([]fieldDefinition, 0, 10)
+	for i := range uint16(10) {
+		plugin.DiscreteInputs = append(plugin.DiscreteInputs,
+			fieldDefinition{
+				Measurement: "test",
+				Name:        fmt.Sprintf("field-discrete-%d", i),
+				Address:     []uint16{i},
+			},
+		)
+	}
+	plugin.HoldingRegisters = make([]fieldDefinition, 0, 10)
+	for i := range uint16(10) {
+		plugin.HoldingRegisters = append(plugin.HoldingRegisters,
+			fieldDefinition{
+				Measurement: "test",
+				Name:        fmt.Sprintf("field-holding-%d", i),
+				Address:     []uint16{4 * i, 4*i + 1, 4*i + 2, 4*i + 3},
+				DataType:    "UINT64",
+				ByteOrder:   "ABCDEFGH",
+				Scale:       1.0,
+			},
+		)
+	}
+	plugin.InputRegisters = make([]fieldDefinition, 0, 10)
+	for i := range uint16(10) {
+		plugin.InputRegisters = append(plugin.InputRegisters,
+			fieldDefinition{
+				Measurement: "test",
+				Name:        fmt.Sprintf("field-input-%d", i),
+				Address:     []uint16{4 * i, 4*i + 1, 4*i + 2, 4*i + 3},
+				DataType:    "UINT64",
+				ByteOrder:   "ABCDEFGH",
+				Scale:       1.0,
+			},
+		)
+	}
+	require.NoError(t, plugin.Init())
+
+	// Check the resulting requests
+	require.Len(t, plugin.requests, 1)
+	require.Contains(t, plugin.requests, byte(1))
+	require.Len(t, plugin.requests[1].coil, 2, "coil")
+	require.Len(t, plugin.requests[1].discrete, 2, "discrete")
+	require.Len(t, plugin.requests[1].holding, 5, "holding")
+	require.Len(t, plugin.requests[1].input, 5, "input")
+}

plugins/inputs/modbus/configuration_request.go

@@ -233,6 +233,8 @@ func (c *configurationPerRequest) process() (map[byte]requestSet, error) {
 			params.maxBatchSize = maxQuantityCoils
 			if c.workarounds.OnRequestPerField {
 				params.maxBatchSize = 1
+			} else if c.workarounds.MaxBitRegistersPerRequest > 0 {
+				params.maxBatchSize = c.workarounds.MaxBitRegistersPerRequest
 			}
 			params.enforceFromZero = c.workarounds.ReadCoilsStartingAtZero
 			requests := groupFieldsToRequests(fields, params)
@@ -241,20 +243,26 @@ func (c *configurationPerRequest) process() (map[byte]requestSet, error) {
 			params.maxBatchSize = maxQuantityDiscreteInput
 			if c.workarounds.OnRequestPerField {
 				params.maxBatchSize = 1
+			} else if c.workarounds.MaxBitRegistersPerRequest > 0 {
+				params.maxBatchSize = c.workarounds.MaxBitRegistersPerRequest
 			}
 			requests := groupFieldsToRequests(fields, params)
 			set.discrete = append(set.discrete, requests...)
 		case "holding":
 			params.maxBatchSize = maxQuantityHoldingRegisters
 			if c.workarounds.OnRequestPerField {
 				params.maxBatchSize = 1
+			} else if c.workarounds.MaxWordRegistersPerRequest > 0 {
+				params.maxBatchSize = c.workarounds.MaxWordRegistersPerRequest
 			}
 			requests := groupFieldsToRequests(fields, params)
 			set.holding = append(set.holding, requests...)
 		case "input":
 			params.maxBatchSize = maxQuantityInputRegisters
 			if c.workarounds.OnRequestPerField {
 				params.maxBatchSize = 1
+			} else if c.workarounds.MaxWordRegistersPerRequest > 0 {
+				params.maxBatchSize = c.workarounds.MaxWordRegistersPerRequest
 			}
 			requests := groupFieldsToRequests(fields, params)
 			set.input = append(set.input, requests...)