Add retry when slave is busy to modbus input (influxdata#7271)

plugins/inputs/modbus/README.md

@@ -20,6 +20,14 @@ The Modbus plugin collects Discrete Inputs, Coils, Input Registers and Holding R
  ## Timeout for each request
  timeout = "1s"
 
+ ## Maximum number of retries and the time to wait between retries
+ ## when a slave-device is busy.
+ ## NOTE: Please make sure that the overall retry time (#retries * wait time)
+ ##       is always smaller than the query interval as otherwise you will get
+ ##       an "did not complete within its interval" warning.
+ #busy_retries = 0
+ #busy_retries_wait = "100ms"
+
  # TCP - connect via Modbus/TCP
  controller = "tcp://localhost:502"
 
@@ -53,7 +61,7 @@ The Modbus plugin collects Discrete Inputs, Coils, Input Registers and Holding R
 
  ## Analog Variables, Input Registers and Holding Registers
  ## measurement - the (optional) measurement name, defaults to "modbus"
- ## name       - the variable name 
+ ## name       - the variable name
  ## byte_order - the ordering of bytes
  ##  |---AB, ABCD   - Big Endian
  ##  |---BA, DCBA   - Little Endian

plugins/inputs/modbus/modbus.go

@@ -3,6 +3,7 @@ package modbus
 import (
 	"encoding/binary"
 	"fmt"
+	"log"
 	"math"
 	"net"
 	"net/url"
@@ -27,6 +28,8 @@ type Modbus struct {
 	StopBits         int               `toml:"stop_bits"`
 	SlaveID          int               `toml:"slave_id"`
 	Timeout          internal.Duration `toml:"timeout"`
+	Retries          int               `toml:"busy_retries"`
+	RetriesWaitTime  internal.Duration `toml:"busy_retries_wait"`
 	DiscreteInputs   []fieldContainer  `toml:"discrete_inputs"`
 	Coils            []fieldContainer  `toml:"coils"`
 	HoldingRegisters []fieldContainer  `toml:"holding_registers"`
@@ -84,6 +87,14 @@ const sampleConfig = `
  ## Timeout for each request
  timeout = "1s"
 
+ ## Maximum number of retries and the time to wait between retries
+ ## when a slave-device is busy.
+ ## NOTE: Please make sure that the overall retry time (#retries * wait time)
+ ##       is always smaller than the query interval as otherwise you will get
+ ##       an "did not complete within its interval" warning.
+ #busy_retries = 0
+ #busy_retries_wait = "100ms"
+
  # TCP - connect via Modbus/TCP
  controller = "tcp://localhost:502"
 
@@ -159,6 +170,10 @@ func (m *Modbus) Init() error {
 		return fmt.Errorf("device name is empty")
 	}
 
+	if m.Retries < 0 {
+		return fmt.Errorf("retries cannot be negative")
+	}
+
 	err := m.InitRegister(m.DiscreteInputs, cDiscreteInputs)
 	if err != nil {
 		return err
@@ -642,11 +657,22 @@ func (m *Modbus) Gather(acc telegraf.Accumulator) error {
 	}
 
 	timestamp := time.Now()
-	err := m.getFields()
-	if err != nil {
-		disconnect(m)
-		m.isConnected = false
-		return err
+	for retry := 0; retry <= m.Retries; retry += 1 {
+		timestamp = time.Now()
+		err := m.getFields()
+		if err != nil {
+			mberr, ok := err.(*mb.ModbusError)
+			if ok && mberr.ExceptionCode == mb.ExceptionCodeServerDeviceBusy && retry < m.Retries {
+				log.Printf("I! [inputs.modbus] device busy! Retrying %d more time(s)...", m.Retries-retry)
+				time.Sleep(m.RetriesWaitTime.Duration)
+				continue
+			}
+			disconnect(m)
+			m.isConnected = false
+			return err
+		}
+		// Reading was successful, leave the retry loop
+		break
 	}
 
 	grouper := metric.NewSeriesGrouper()

plugins/inputs/modbus/modbus_test.go

@@ -494,3 +494,131 @@ func TestHoldingRegisters(t *testing.T) {
 		})
 	}
 }
+
+func TestRetrySuccessful(t *testing.T) {
+	retries := 0
+	maxretries := 2
+	value := 1
+
+	serv := mbserver.NewServer()
+	err := serv.ListenTCP("localhost:1502")
+	assert.NoError(t, err)
+	defer serv.Close()
+
+	// Make read on coil-registers fail for some trials by making the device
+	// to appear busy
+	serv.RegisterFunctionHandler(1,
+		func(s *mbserver.Server, frame mbserver.Framer) ([]byte, *mbserver.Exception) {
+			data := make([]byte, 2)
+			data[0] = byte(1)
+			data[1] = byte(value)
+
+			except := &mbserver.SlaveDeviceBusy
+			if retries >= maxretries {
+				except = &mbserver.Success
+			}
+			retries += 1
+
+			return data, except
+		})
+
+	t.Run("retry_success", func(t *testing.T) {
+		modbus := Modbus{
+			Name:       "TestRetry",
+			Controller: "tcp://localhost:1502",
+			SlaveID:    1,
+			Retries:    maxretries,
+			Coils: []fieldContainer{
+				{
+					Name:    "retry_success",
+					Address: []uint16{0},
+				},
+			},
+		}
+
+		err = modbus.Init()
+		assert.NoError(t, err)
+		var acc testutil.Accumulator
+		err = modbus.Gather(&acc)
+		assert.NoError(t, err)
+		assert.NotEmpty(t, modbus.registers)
+
+		for _, coil := range modbus.registers {
+			assert.Equal(t, uint16(value), coil.Fields[0].value)
+		}
+	})
+}
+
+func TestRetryFail(t *testing.T) {
+	maxretries := 2
+
+	serv := mbserver.NewServer()
+	err := serv.ListenTCP("localhost:1502")
+	assert.NoError(t, err)
+	defer serv.Close()
+
+	// Make the read on coils fail with busy
+	serv.RegisterFunctionHandler(1,
+		func(s *mbserver.Server, frame mbserver.Framer) ([]byte, *mbserver.Exception) {
+			data := make([]byte, 2)
+			data[0] = byte(1)
+			data[1] = byte(0)
+
+			return data, &mbserver.SlaveDeviceBusy
+		})
+
+	t.Run("retry_fail", func(t *testing.T) {
+		modbus := Modbus{
+			Name:       "TestRetryFail",
+			Controller: "tcp://localhost:1502",
+			SlaveID:    1,
+			Retries:    maxretries,
+			Coils: []fieldContainer{
+				{
+					Name:    "retry_fail",
+					Address: []uint16{0},
+				},
+			},
+		}
+
+		err = modbus.Init()
+		assert.NoError(t, err)
+		var acc testutil.Accumulator
+		err = modbus.Gather(&acc)
+		assert.Error(t, err)
+	})
+
+	// Make the read on coils fail with illegal function preventing retry
+	counter := 0
+	serv.RegisterFunctionHandler(1,
+		func(s *mbserver.Server, frame mbserver.Framer) ([]byte, *mbserver.Exception) {
+			counter += 1
+			data := make([]byte, 2)
+			data[0] = byte(1)
+			data[1] = byte(0)
+
+			return data, &mbserver.IllegalFunction
+		})
+
+	t.Run("retry_fail", func(t *testing.T) {
+		modbus := Modbus{
+			Name:       "TestRetryFail",
+			Controller: "tcp://localhost:1502",
+			SlaveID:    1,
+			Retries:    maxretries,
+			Coils: []fieldContainer{
+				{
+					Name:    "retry_fail",
+					Address: []uint16{0},
+				},
+			},
+		}
+
+		err = modbus.Init()
+		assert.NoError(t, err)
+		var acc testutil.Accumulator
+		err = modbus.Gather(&acc)
+		assert.Error(t, err)
+		assert.Equal(t, counter, 1)
+	})
+}