feat(input): add upsd implementation (#9890)

docs/LICENSE_OF_DEPENDENCIES.md

@@ -244,6 +244,7 @@ following works:
 - github.com/rcrowley/go-metrics [MIT License](https://github.com/rcrowley/go-metrics/blob/master/LICENSE)
 - github.com/remyoudompheng/bigfft [BSD 3-Clause "New" or "Revised" License](https://github.com/remyoudompheng/bigfft/blob/master/LICENSE)
 - github.com/riemann/riemann-go-client [MIT License](https://github.com/riemann/riemann-go-client/blob/master/LICENSE)
+- github.com/robbiet480/go.nut [MIT License](https://github.com/robbiet480/go.nut/blob/master/LICENSE)
 - github.com/safchain/ethtool [Apache License 2.0](https://github.com/safchain/ethtool/blob/master/LICENSE)
 - github.com/samuel/go-zookeeper [BSD 3-Clause Clear License](https://github.com/samuel/go-zookeeper/blob/master/LICENSE)
 - github.com/shirou/gopsutil [BSD 3-Clause Clear License](https://github.com/shirou/gopsutil/blob/master/LICENSE)

go.mod

@@ -125,6 +125,7 @@ require (
 	github.com/prometheus/prometheus v1.8.2-0.20210430082741-2a4b8e12bbf2
 	github.com/rabbitmq/amqp091-go v1.3.4
 	github.com/riemann/riemann-go-client v0.5.1-0.20211206220514-f58f10cdce16
+	github.com/robbiet480/go.nut v0.0.0-20220219091450-bd8f121e1fa1
 	github.com/safchain/ethtool v0.0.0-20200218184317-f459e2d13664
 	github.com/sensu/sensu-go/api/core/v2 v2.14.0
 	github.com/shirou/gopsutil/v3 v3.22.4

go.sum

@@ -2067,6 +2067,8 @@ github.com/riemann/riemann-go-client v0.5.1-0.20211206220514-f58f10cdce16 h1:bGX
 github.com/riemann/riemann-go-client v0.5.1-0.20211206220514-f58f10cdce16/go.mod h1:4rS0vfmzOMwfFPhi6Zve4k/59TsBepqd6WESNULE0ho=
 github.com/rivo/tview v0.0.0-20200219210816-cd38d7432498/go.mod h1:6lkG1x+13OShEf0EaOCaTQYyB7d5nSbb181KtjlS+84=
 github.com/rivo/uniseg v0.1.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
+github.com/robbiet480/go.nut v0.0.0-20220219091450-bd8f121e1fa1 h1:YmFqprZILGlF/X3tvMA4Rwn3ySxyE3hGUajBHkkaZbM=
+github.com/robbiet480/go.nut v0.0.0-20220219091450-bd8f121e1fa1/go.mod h1:pL1huxuIlWub46MsMVJg4p7OXkzbPp/APxh9IH0eJjQ=
 github.com/robertkrimen/otto v0.0.0-20191219234010-c382bd3c16ff h1:+6NUiITWwE5q1KO6SAfUX918c+Tab0+tGAM/mtdlUyA=
 github.com/robertkrimen/otto v0.0.0-20191219234010-c382bd3c16ff/go.mod h1:xvqspoSXJTIpemEonrMDFq6XzwHYYgToXWj5eRX1OtY=
 github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=

plugins/inputs/all/all.go

@@ -204,6 +204,7 @@ import (
 	_ "github.com/influxdata/telegraf/plugins/inputs/twemproxy"
 	_ "github.com/influxdata/telegraf/plugins/inputs/udp_listener"
 	_ "github.com/influxdata/telegraf/plugins/inputs/unbound"
+	_ "github.com/influxdata/telegraf/plugins/inputs/upsd"
 	_ "github.com/influxdata/telegraf/plugins/inputs/uwsgi"
 	_ "github.com/influxdata/telegraf/plugins/inputs/varnish"
 	_ "github.com/influxdata/telegraf/plugins/inputs/vault"

plugins/inputs/upsd/README.md

@@ -0,0 +1,61 @@
+# UPSD Input Plugin
+
+This plugin reads data of one or more Uninterruptible Power Supplies
+from an upsd daemon using its NUT network protocol.
+
+## Requirements
+
+upsd should be installed and it's daemon should be running.
+
+## Configuration
+
+```toml
+[[inputs.upsd]]
+  ## A running NUT server to connect to.
+  # If not provided will default to 127.0.0.1
+  # server = "127.0.0.1"
+  
+  ## The default NUT port 3493 can be overridden with:
+  # port = 3493
+  
+  # username = "user"
+  # password = "password"
+```
+
+## Metrics
+
+This implementation tries to maintain compatibility with the apcupsd metrics:
+
+- upsd
+  - tags:
+    - serial
+    - ups_name
+    - model
+  - fields:
+    - status_flags ([status-bits][])
+    - input_voltage
+    - load_percent
+    - battery_charge_percent
+    - time_left_ns
+    - output_voltage
+    - internal_temp
+    - battery_voltage
+    - input_frequency
+    - battery_date
+    - nominal_input_voltage
+    - nominal_battery_voltage
+    - nominal_power
+    - firmware
+
+With the exception of:
+
+- tags:
+  - status (string representing the set status_flags)
+- fields:
+  - time_on_battery_ns
+
+## Example Output
+
+```shell
+upsd,serial=AS1231515,ups_name=name1 load_percent=9.7,time_left_ns=9800000,output_voltage=230.4,internal_temp=32.4,battery_voltage=27.4,input_frequency=50.2,input_voltage=230.4,battery_charge_percent=100,status_flags=8i 1490035922000000000
+```

plugins/inputs/upsd/upsd.go

@@ -0,0 +1,188 @@
+package upsd
+
+import (
+	"fmt"
+	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/internal/choice"
+	"github.com/influxdata/telegraf/plugins/inputs"
+	nut "github.com/robbiet480/go.nut"
+	"strings"
+)
+
+//See: https://networkupstools.org/docs/developer-guide.chunked/index.html
+
+const defaultAddress = "127.0.0.1"
+const defaultPort = 3493
+
+type Upsd struct {
+	Server   string
+	Port     int
+	Username string
+	Password string
+	Log      telegraf.Logger `toml:"-"`
+
+	batteryRuntimeTypeWarningIssued bool
+}
+
+func (*Upsd) Description() string {
+	return "Monitor UPSes connected via Network UPS Tools"
+}
+
+var sampleConfig = `
+ ## A running NUT server to connect to.
+ # server = "127.0.0.1"
+ # port = 3493
+ # username = "user"
+ # password = "password"
+`
+
+func (*Upsd) SampleConfig() string {
+	return sampleConfig
+}
+
+func (u *Upsd) Gather(acc telegraf.Accumulator) error {
+	upsList, err := u.fetchVariables(u.Server, u.Port)
+	if err != nil {
+		return err
+	}
+	for name, variables := range upsList {
+		u.gatherUps(acc, name, variables)
+	}
+	return nil
+}
+
+func (u *Upsd) gatherUps(acc telegraf.Accumulator, name string, variables []nut.Variable) {
+	metrics := make(map[string]interface{})
+	for _, variable := range variables {
+		name := variable.Name
+		value := variable.Value
+		metrics[name] = value
+	}
+
+	tags := map[string]string{
+		"serial":   fmt.Sprintf("%v", metrics["device.serial"]),
+		"ups_name": name,
+		//"variables": variables.Status not sure if it's a good idea to provide this
+		"model": fmt.Sprintf("%v", metrics["device.model"]),
+	}
+
+	// For compatibility with the apcupsd plugin's output we map the status string status into a bit-format
+	status := u.mapStatus(metrics, tags)
+
+	timeLeftS, ok := metrics["battery.runtime"].(int64)
+	if !ok && !u.batteryRuntimeTypeWarningIssued {
+		u.Log.Warnf("'battery.runtime' type is not int64")
+		u.batteryRuntimeTypeWarningIssued = true
+	}
+
+	fields := map[string]interface{}{
+		"status_flags":            status,
+		"ups_status":              metrics["ups.status"],
+		"input_voltage":           metrics["input.voltage"],
+		"load_percent":            metrics["ups.load"],
+		"battery_charge_percent":  metrics["battery.charge"],
+		"time_left_ns":            timeLeftS * 1_000_000_000, //Compatibility with apcupsd metrics format
+		"output_voltage":          metrics["output.voltage"],
+		"internal_temp":           metrics["ups.temperature"],
+		"battery_voltage":         metrics["battery.voltage"],
+		"input_frequency":         metrics["input.frequency"],
+		"nominal_input_voltage":   metrics["input.voltage.nominal"],
+		"nominal_battery_voltage": metrics["battery.voltage.nominal"],
+		"nominal_power":           metrics["ups.realpower.nominal"],
+		"firmware":                metrics["ups.firmware"],
+		"battery_date":            metrics["battery.mfr.date"],
+	}
+
+	acc.AddFields("upsd", fields, tags)
+}
+
+func (u *Upsd) mapStatus(metrics map[string]interface{}, tags map[string]string) uint64 {
+	status := uint64(0)
+	statusString := fmt.Sprintf("%v", metrics["ups.status"])
+	statuses := strings.Fields(statusString)
+	//Source: 1.3.2 at http://rogerprice.org/NUT/ConfigExamples.A5.pdf
+	//apcupsd bits:
+	//0	Runtime calibration occurring (Not reported by Smart UPS v/s and BackUPS Pro)
+	//1	SmartTrim (Not reported by 1st and 2nd generation SmartUPS models)
+	//2	SmartBoost
+	//3	On line (this is the normal condition)
+	//4	On battery
+	//5	Overloaded output
+	//6	Battery low
+	//7	Replace battery
+	if choice.Contains("CAL", statuses) {
+		status |= 1 << 0
+		tags["status_CAL"] = "true"
+	}
+	if choice.Contains("TRIM", statuses) {
+		status |= 1 << 1
+		tags["status_TRIM"] = "true"
+	}
+	if choice.Contains("BOOST", statuses) {
+		status |= 1 << 2
+		tags["status_BOOST"] = "true"
+	}
+	if choice.Contains("OL", statuses) {
+		status |= 1 << 3
+		tags["status_OL"] = "true"
+	}
+	if choice.Contains("OB", statuses) {
+		status |= 1 << 4
+		tags["status_OB"] = "true"
+	}
+	if choice.Contains("OVER", statuses) {
+		status |= 1 << 5
+		tags["status_OVER"] = "true"
+	}
+	if choice.Contains("LB", statuses) {
+		status |= 1 << 6
+		tags["status_LB"] = "true"
+	}
+	if choice.Contains("RB", statuses) {
+		status |= 1 << 7
+		tags["status_RB"] = "true"
+	}
+	return status
+}
+
+func (u *Upsd) fetchVariables(server string, port int) (map[string][]nut.Variable, error) {
+	client, err := nut.Connect(server, port)
+	if err != nil {
+		return nil, fmt.Errorf("connect: %w", err)
+	}
+
+	if u.Username != "" && u.Password != "" {
+		_, err = client.Authenticate(u.Username, u.Password)
+		if err != nil {
+			return nil, fmt.Errorf("auth: %w", err)
+		}
+	}
+
+	upsList, err := client.GetUPSList()
+	if err != nil {
+		return nil, fmt.Errorf("getupslist: %w", err)
+	}
+
+	defer func() {
+		_, disconnectErr := client.Disconnect()
+		if disconnectErr != nil {
+			err = fmt.Errorf("disconnect: %w", disconnectErr)
+		}
+	}()
+
+	result := make(map[string][]nut.Variable)
+	for _, ups := range upsList {
+		result[ups.Name] = ups.Variables
+	}
+
+	return result, err
+}
+
+func init() {
+	inputs.Add("upsd", func() telegraf.Input {
+		return &Upsd{
+			Server: defaultAddress,
+			Port:   defaultPort,
+		}
+	})
+}

plugins/inputs/upsd/upsd_test.go

@@ -0,0 +1,260 @@
+package upsd
+
+import (
+	"context"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/influxdata/telegraf/testutil"
+	"github.com/stretchr/testify/require"
+)
+
+func TestUpsdGather(t *testing.T) {
+	nut := &Upsd{}
+
+	var (
+		tests = []struct {
+			name   string
+			err    bool
+			tags   map[string]string
+			fields map[string]interface{}
+			out    func() []interaction
+		}{
+			{
+				name: "test listening server with output",
+				err:  false,
+				tags: map[string]string{
+					"serial":    "ABC123",
+					"ups_name":  "fake",
+					"model":     "Model 12345",
+					"status_OL": "true",
+				},
+				fields: map[string]interface{}{
+					"status_flags":            uint64(8),
+					"ups_status":              "OL",
+					"battery_charge_percent":  float64(100),
+					"battery_voltage":         float64(13.4),
+					"input_frequency":         nil,
+					"input_voltage":           float64(242),
+					"internal_temp":           nil,
+					"load_percent":            float64(23),
+					"output_voltage":          float64(230),
+					"time_left_ns":            int64(600000000000),
+					"nominal_input_voltage":   float64(230),
+					"nominal_battery_voltage": float64(24),
+					"nominal_power":           int64(700),
+					"firmware":                "CUSTOM_FIRMWARE",
+					"battery_date":            "2016-07-26",
+				},
+				out: genOutput,
+			},
+		}
+
+		acc testutil.Accumulator
+	)
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctx, cancel := context.WithCancel(context.Background())
+
+			lAddr, err := listen(ctx, t, tt.out())
+			require.NoError(t, err)
+
+			nut.Server = (lAddr.(*net.TCPAddr)).IP.String()
+			nut.Port = (lAddr.(*net.TCPAddr)).Port
+
+			err = nut.Gather(&acc)
+			if tt.err {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				acc.AssertContainsTaggedFields(t, "upsd", tt.fields, tt.tags)
+			}
+			cancel()
+		})
+	}
+}
+
+func TestUpsdGatherFail(t *testing.T) {
+	nut := &Upsd{}
+
+	var (
+		tests = []struct {
+			name   string
+			err    bool
+			tags   map[string]string
+			fields map[string]interface{}
+			out    func() []interaction
+		}{
+			{
+				name: "test with bad output",
+				err:  true,
+				out:  genBadOutput,
+			},
+		}
+
+		acc testutil.Accumulator
+	)
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctx, cancel := context.WithCancel(context.Background())
+
+			lAddr, err := listen(ctx, t, tt.out())
+			require.NoError(t, err)
+
+			nut.Server = (lAddr.(*net.TCPAddr)).IP.String()
+			nut.Port = (lAddr.(*net.TCPAddr)).Port
+
+			err = nut.Gather(&acc)
+			if tt.err {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				acc.AssertContainsTaggedFields(t, "upsd", tt.fields, tt.tags)
+			}
+			cancel()
+		})
+	}
+}
+
+func listen(ctx context.Context, t *testing.T, out []interaction) (net.Addr, error) {
+	lc := net.ListenConfig{}
+	ln, err := lc.Listen(ctx, "tcp4", "127.0.0.1:0")
+	if err != nil {
+		return nil, err
+	}
+
+	go func() {
+		defer ln.Close()
+
+		for ctx.Err() == nil {
+			func() {
+				conn, err := ln.Accept()
+				if err != nil {
+					return
+				}
+				defer conn.Close()
+				require.NoError(t, conn.SetReadDeadline(time.Now().Add(time.Minute)))
+
+				in := make([]byte, 128)
+				for _, interaction := range out {
+					n, err := conn.Read(in)
+					require.NoError(t, err, "failed to read from connection")
+
+					expectedBytes := []byte(interaction.Expected)
+					want, got := expectedBytes, in[:n]
+					require.Equal(t, want, got)
+
+					_, err = conn.Write([]byte(interaction.Response))
+					require.NoError(t, err, "failed to respond to LIST UPS")
+				}
+
+				// Append EOF to end of output bytes
+				_, err = conn.Write([]byte{0, 0})
+				require.NoError(t, err, "failed to write EOF")
+			}()
+		}
+	}()
+
+	return ln.Addr(), nil
+}
+
+type interaction struct {
+	Expected string
+	Response string
+}
+
+func genOutput() []interaction {
+	m := make([]interaction, 0)
+	m = append(m, interaction{
+		Expected: "VER\n",
+		Response: "1\n",
+	})
+	m = append(m, interaction{
+		Expected: "NETVER\n",
+		Response: "1\n",
+	})
+	m = append(m, interaction{
+		Expected: "LIST UPS\n",
+		Response: `BEGIN LIST UPS
+UPS fake "fakescription"
+END LIST UPS
+`,
+	})
+	m = append(m, interaction{
+		Expected: "LIST CLIENT fake\n",
+		Response: `BEGIN LIST CLIENT fake
+CLIENT fake 192.168.1.1
+END LIST CLIENT fake
+`,
+	})
+	m = append(m, interaction{
+		Expected: "LIST CMD fake\n",
+		Response: `BEGIN LIST CMD fake
+END LIST CMD fake
+`,
+	})
+	m = append(m, interaction{
+		Expected: "GET UPSDESC fake\n",
+		Response: "UPSDESC fake \"stub-ups-description\"\n",
+	})
+	m = append(m, interaction{
+		Expected: "GET NUMLOGINS fake\n",
+		Response: "NUMLOGINS fake 1\n",
+	})
+	m = append(m, interaction{
+		Expected: "LIST VAR fake\n",
+		Response: `BEGIN LIST VAR fake
+VAR fake device.serial "ABC123"
+VAR fake device.model "Model 12345"
+VAR fake input.voltage "242.0"
+VAR fake ups.load "23.0"
+VAR fake battery.charge "100.0"
+VAR fake battery.runtime "600"
+VAR fake output.voltage "230.0"
+VAR fake battery.voltage "13.4"
+VAR fake input.voltage.nominal "230.0"
+VAR fake battery.voltage.nominal "24.0"
+VAR fake ups.realpower.nominal "700"
+VAR fake ups.firmware "CUSTOM_FIRMWARE"
+VAR fake battery.mfr.date "2016-07-26"
+VAR fake ups.status "OL"
+END LIST VAR fake
+`,
+	})
+	m = appendVariable(m, "device.serial", "STRING:64")
+	m = appendVariable(m, "device.model", "STRING:64")
+	m = appendVariable(m, "input.voltage", "NUMBER")
+	m = appendVariable(m, "ups.load", "NUMBER")
+	m = appendVariable(m, "battery.charge", "NUMBER")
+	m = appendVariable(m, "battery.runtime", "NUMBER")
+	m = appendVariable(m, "output.voltage", "NUMBER")
+	m = appendVariable(m, "battery.voltage", "NUMBER")
+	m = appendVariable(m, "input.voltage.nominal", "NUMBER")
+	m = appendVariable(m, "battery.voltage.nominal", "NUMBER")
+	m = appendVariable(m, "ups.realpower.nominal", "NUMBER")
+	m = appendVariable(m, "ups.firmware", "STRING:64")
+	m = appendVariable(m, "battery.mfr.date", "STRING:64")
+	m = appendVariable(m, "ups.status", "STRING:64")
+
+	return m
+}
+
+func appendVariable(m []interaction, name string, typ string) []interaction {
+	m = append(m, interaction{
+		Expected: "GET DESC fake " + name + "\n",
+		Response: "DESC fake" + name + " \"No description here\"\n",
+	})
+	m = append(m, interaction{
+		Expected: "GET TYPE fake " + name + "\n",
+		Response: "TYPE fake " + name + " " + typ + "\n",
+	})
+	return m
+}
+
+func genBadOutput() []interaction {
+	m := make([]interaction, 0)
+	return m
+}