CL-Softwares
/
telegraf
8
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: Linter fixes for plugins/inputs/s* (#10104) 

Co-authored-by: Pawel Zak <Pawel Zak>
This commit is contained in:
Paweł Żak 2021-11-18 17:04:52 +01:00 committed by GitHub
parent 146fff3183
commit 2a0c3059a1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
20 changed files with 199 additions and 170 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
plugins/inputs/sensors/sensors_test.go
View File

@ -367,7 +367,7 @@ Vcore Voltage:
// Previous arguments are tests stuff, that looks like : // Previous arguments are tests stuff, that looks like :
// /tmp/go-build970079519/…/_test/integration.test -test.run=TestHelperProcess -- // /tmp/go-build970079519/…/_test/integration.test -test.run=TestHelperProcess --
cmd, args := args[3], args[4:] cmd, _ := args[3], args[4:]
if cmd == "sensors" { if cmd == "sensors" {
//nolint:errcheck,revive //nolint:errcheck,revive
@ -375,7 +375,9 @@ Vcore Voltage:
} else { } else {
//nolint:errcheck,revive //nolint:errcheck,revive
fmt.Fprint(os.Stdout, "command not found") fmt.Fprint(os.Stdout, "command not found")
//nolint:revive // error code is important for this "test"
os.Exit(1) os.Exit(1)
} }
//nolint:revive // error code is important for this "test"
os.Exit(0) os.Exit(0)
} }

10
plugins/inputs/snmp/snmp.go
View File

@ -16,6 +16,7 @@ import (
"time" "time"
"github.com/gosnmp/gosnmp" "github.com/gosnmp/gosnmp"
"github.com/influxdata/telegraf" "github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/config" "github.com/influxdata/telegraf/config"
"github.com/influxdata/telegraf/internal/snmp" "github.com/influxdata/telegraf/internal/snmp"
@ -679,7 +680,7 @@ func fieldConvert(conv string, v interface{}) (interface{}, error) {
case float32: case float32:
v = float64(vt) / math.Pow10(d) v = float64(vt) / math.Pow10(d)
case float64: case float64:
v = float64(vt) / math.Pow10(d) v = vt / math.Pow10(d)
case int: case int:
v = float64(vt) / math.Pow10(d) v = float64(vt) / math.Pow10(d)
case int8: case int8:
@ -766,7 +767,8 @@ func fieldConvert(conv string, v interface{}) (interface{}, error) {
return v, nil return v, nil
} }
if endian == "LittleEndian" { switch endian {
case "LittleEndian":
switch bit { switch bit {
case "uint64": case "uint64":
v = binary.LittleEndian.Uint64(bv) v = binary.LittleEndian.Uint64(bv)
@ -777,7 +779,7 @@ func fieldConvert(conv string, v interface{}) (interface{}, error) {
default: default:
return nil, fmt.Errorf("invalid bit value (%s) for hex to int conversion", bit) return nil, fmt.Errorf("invalid bit value (%s) for hex to int conversion", bit)
} }
} else if endian == "BigEndian" { case "BigEndian":
switch bit { switch bit {
case "uint64": case "uint64":
v = binary.BigEndian.Uint64(bv) v = binary.BigEndian.Uint64(bv)
@ -788,7 +790,7 @@ func fieldConvert(conv string, v interface{}) (interface{}, error) {
default: default:
return nil, fmt.Errorf("invalid bit value (%s) for hex to int conversion", bit) return nil, fmt.Errorf("invalid bit value (%s) for hex to int conversion", bit)
} }
} else { default:
return nil, fmt.Errorf("invalid Endian value (%s) for hex to int conversion", endian) return nil, fmt.Errorf("invalid Endian value (%s) for hex to int conversion", endian)
} }

3
plugins/inputs/snmp/snmp_mocks_test.go
View File

@ -46,6 +46,7 @@ func TestMockExecCommand(_ *testing.T) {
cv := fmt.Sprintf("%#v", cmd)[8:] // trim `[]string` prefix cv := fmt.Sprintf("%#v", cmd)[8:] // trim `[]string` prefix
//nolint:errcheck,revive //nolint:errcheck,revive
fmt.Fprintf(os.Stderr, "Unmocked command. Please add the following to `mockedCommands` in snmp_mocks_generate.go, and then run `go generate`:\n\t%s,\n", cv) fmt.Fprintf(os.Stderr, "Unmocked command. Please add the following to `mockedCommands` in snmp_mocks_generate.go, and then run `go generate`:\n\t%s,\n", cv)
//nolint:revive // error code is important for this "test"
os.Exit(1) os.Exit(1)
} }
//nolint:errcheck,revive //nolint:errcheck,revive
@ -53,8 +54,10 @@ func TestMockExecCommand(_ *testing.T) {
//nolint:errcheck,revive //nolint:errcheck,revive
fmt.Fprintf(os.Stderr, "%s", mcr.stderr) fmt.Fprintf(os.Stderr, "%s", mcr.stderr)
if mcr.exitError { if mcr.exitError {
//nolint:revive // error code is important for this "test"
os.Exit(1) os.Exit(1)
} }
//nolint:revive // error code is important for this "test"
os.Exit(0) os.Exit(0)
} }

3
plugins/inputs/snmp/snmp_test.go
View File

@ -850,11 +850,12 @@ func TestFieldConvert(t *testing.T) {
conv string conv string
expected interface{} expected interface{}
}{ }{
{[]byte("foo"), "", string("foo")}, {[]byte("foo"), "", "foo"},
{"0.123", "float", float64(0.123)}, {"0.123", "float", float64(0.123)},
{[]byte("0.123"), "float", float64(0.123)}, {[]byte("0.123"), "float", float64(0.123)},
{float32(0.123), "float", float64(float32(0.123))}, {float32(0.123), "float", float64(float32(0.123))},
{float64(0.123), "float", float64(0.123)}, {float64(0.123), "float", float64(0.123)},
{float64(0.123123123123), "float", float64(0.123123123123)},
{123, "float", float64(123)}, {123, "float", float64(123)},
{123, "float(0)", float64(123)}, {123, "float(0)", float64(123)},
{123, "float(4)", float64(0.0123)}, {123, "float(4)", float64(0.0123)},

30
plugins/inputs/snmp_legacy/snmp_legacy.go
View File

@ -8,10 +8,10 @@ import (
"strings" "strings"
"time" "time"
"github.com/gosnmp/gosnmp"
"github.com/influxdata/telegraf" "github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/plugins/inputs" "github.com/influxdata/telegraf/plugins/inputs"
"github.com/gosnmp/gosnmp"
) )
// Snmp is a snmp plugin // Snmp is a snmp plugin
@ -46,9 +46,9 @@ type Host struct {
// Table // Table
Table []HostTable Table []HostTable
// Oids // Oids
getOids []Data internalGetOids []Data
bulkOids []Data bulkOids []Data
tables []HostTable tables []HostTable
// array of processed oids // array of processed oids
// to skip oid duplication // to skip oid duplication
processedOids []string processedOids []string
@ -250,7 +250,7 @@ func fillnode(parentNode Node, oidName string, ids []string) {
} }
} }
func findnodename(node Node, ids []string) (string, string) { func findNodeName(node Node, ids []string) (oidName string, instance string) {
// ids = ["1", "3", "6", ...] // ids = ["1", "3", "6", ...]
if len(ids) == 1 { if len(ids) == 1 {
return node.name, ids[0] return node.name, ids[0]
@ -259,7 +259,7 @@ func findnodename(node Node, ids []string) (string, string) {
// Get node // Get node
subnode, ok := node.subnodes[id] subnode, ok := node.subnodes[id]
if ok { if ok {
return findnodename(subnode, ids) return findNodeName(subnode, ids)
} }
// We got a node // We got a node
// Get node name // Get node name
@ -345,7 +345,7 @@ func (s *Snmp) Gather(acc telegraf.Accumulator) error {
oid.rawOid = oidstring oid.rawOid = oidstring
} }
} }
host.getOids = append(host.getOids, oid) host.internalGetOids = append(host.internalGetOids, oid)
} }
for _, oidName := range host.Collect { for _, oidName := range host.Collect {
@ -362,7 +362,7 @@ func (s *Snmp) Gather(acc telegraf.Accumulator) error {
} else { } else {
oid.rawOid = oid.Oid oid.rawOid = oid.Oid
} }
host.getOids = append(host.getOids, oid) host.internalGetOids = append(host.internalGetOids, oid)
} }
} }
// Get GETBULK oids // Get GETBULK oids
@ -463,7 +463,7 @@ func (h *Host) SNMPMap(
} }
// TODO check oid validity // TODO check oid validity
// Add the new oid to getOids list // Add the new oid to bulkOids list
h.bulkOids = append(h.bulkOids, oid) h.bulkOids = append(h.bulkOids, oid)
} }
} }
@ -569,8 +569,8 @@ func (h *Host) SNMPMap(
} }
// TODO check oid validity // TODO check oid validity
// Add the new oid to getOids list // Add the new oid to internalGetOids list
h.getOids = append(h.getOids, oid) h.internalGetOids = append(h.internalGetOids, oid)
} }
} }
default: default:
@ -606,7 +606,7 @@ func (h *Host) SNMPGet(acc telegraf.Accumulator, initNode Node) error {
defer snmpClient.Conn.Close() defer snmpClient.Conn.Close()
// Prepare OIDs // Prepare OIDs
oidsList := make(map[string]Data) oidsList := make(map[string]Data)
for _, oid := range h.getOids { for _, oid := range h.internalGetOids {
oidsList[oid.rawOid] = oid oidsList[oid.rawOid] = oid
} }
oidsNameList := make([]string, 0, len(oidsList)) oidsNameList := make([]string, 0, len(oidsList))
@ -701,7 +701,7 @@ func (h *Host) GetSNMPClient() (*gosnmp.GoSNMP, error) {
// Prepare host and port // Prepare host and port
host, portStr, err := net.SplitHostPort(h.Address) host, portStr, err := net.SplitHostPort(h.Address)
if err != nil { if err != nil {
portStr = string("161") portStr = "161"
} }
// convert port_str to port in uint16 // convert port_str to port in uint16
port64, err := strconv.ParseUint(portStr, 10, 16) port64, err := strconv.ParseUint(portStr, 10, 16)
@ -763,7 +763,7 @@ func (h *Host) HandleResponse(
var oidName string var oidName string
var instance string var instance string
// Get oidname and instance from translate file // Get oidname and instance from translate file
oidName, instance = findnodename(initNode, oidName, instance = findNodeName(initNode,
strings.Split(variable.Name[1:], ".")) strings.Split(variable.Name[1:], "."))
// Set instance tag // Set instance tag
// From mapping table // From mapping table

6
plugins/inputs/snmp_trap/snmp_trap_test.go
View File

@ -10,11 +10,10 @@ import (
"time" "time"
"github.com/gosnmp/gosnmp" "github.com/gosnmp/gosnmp"
"github.com/stretchr/testify/require"
"github.com/influxdata/telegraf" "github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/testutil" "github.com/influxdata/telegraf/testutil"
"github.com/stretchr/testify/require"
) )
func newMsgFlagsV3(secLevel string) gosnmp.SnmpV3MsgFlags { func newMsgFlagsV3(secLevel string) gosnmp.SnmpV3MsgFlags {
@ -1267,7 +1266,7 @@ func TestReceiveTrap(t *testing.T) {
return mibEntry{entry.e.mibName, entry.e.oidText}, nil return mibEntry{entry.e.mibName, entry.e.oidText}, nil
} }
} }
return mibEntry{}, fmt.Errorf("Unexpected oid") return mibEntry{}, fmt.Errorf("unexpected oid")
}, },
//if cold start be answer otherwise err //if cold start be answer otherwise err
Log: testutil.Logger{}, Log: testutil.Logger{},
@ -1311,7 +1310,6 @@ func TestReceiveTrap(t *testing.T) {
testutil.SortMetrics()) testutil.SortMetrics())
}) })
} }
} }
func TestGosmiSingleMib(t *testing.T) { func TestGosmiSingleMib(t *testing.T) {

59
plugins/inputs/socket_listener/socket_listener_test.go
View File

@ -12,19 +12,19 @@ import (
"testing" "testing"
"time" "time"
"github.com/stretchr/testify/require"
"github.com/influxdata/telegraf/config" "github.com/influxdata/telegraf/config"
"github.com/influxdata/telegraf/internal" "github.com/influxdata/telegraf/internal"
"github.com/influxdata/telegraf/testutil" "github.com/influxdata/telegraf/testutil"
"github.com/influxdata/wlog" "github.com/influxdata/wlog"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
) )
var pki = testutil.NewPKI("../../../testutil/pki") var pki = testutil.NewPKI("../../../testutil/pki")
// testEmptyLog is a helper function to ensure no data is written to log. // prepareLog is a helper function to ensure no data is written to log.
// Should be called at the start of the test, and returns a function which should run at the end. // Should be called at the start of the test, and returns a function which should run at the end.
func testEmptyLog(t *testing.T) func() { func prepareLog(t *testing.T) func() {
buf := bytes.NewBuffer(nil) buf := bytes.NewBuffer(nil)
log.SetOutput(wlog.NewWriter(buf)) log.SetOutput(wlog.NewWriter(buf))
@ -37,16 +37,17 @@ func testEmptyLog(t *testing.T) func() {
for { for {
line, err := buf.ReadBytes('\n') line, err := buf.ReadBytes('\n')
if err != nil { if err != nil {
assert.Equal(t, io.EOF, err) require.Equal(t, io.EOF, err)
break break
} }
assert.Empty(t, string(line), "log not empty") require.Empty(t, string(line), "log not empty")
} }
} }
} }
func TestSocketListener_tcp_tls(t *testing.T) { func TestSocketListener_tcp_tls(t *testing.T) {
defer testEmptyLog(t)() testEmptyLog := prepareLog(t)
defer testEmptyLog()
sl := newSocketListener() sl := newSocketListener()
sl.Log = testutil.Logger{} sl.Log = testutil.Logger{}
@ -84,8 +85,8 @@ func TestSocketListener_unix_tls(t *testing.T) {
defer sl.Stop() defer sl.Stop()
tlsCfg, err := pki.TLSClientConfig().TLSConfig() tlsCfg, err := pki.TLSClientConfig().TLSConfig()
tlsCfg.InsecureSkipVerify = true
require.NoError(t, err) require.NoError(t, err)
tlsCfg.InsecureSkipVerify = true
secureClient, err := tls.Dial("unix", sock, tlsCfg) secureClient, err := tls.Dial("unix", sock, tlsCfg)
require.NoError(t, err) require.NoError(t, err)
@ -94,7 +95,8 @@ func TestSocketListener_unix_tls(t *testing.T) {
} }
func TestSocketListener_tcp(t *testing.T) { func TestSocketListener_tcp(t *testing.T) {
defer testEmptyLog(t)() testEmptyLog := prepareLog(t)
defer testEmptyLog()
sl := newSocketListener() sl := newSocketListener()
sl.Log = testutil.Logger{} sl.Log = testutil.Logger{}
@ -113,7 +115,8 @@ func TestSocketListener_tcp(t *testing.T) {
} }
func TestSocketListener_udp(t *testing.T) { func TestSocketListener_udp(t *testing.T) {
defer testEmptyLog(t)() testEmptyLog := prepareLog(t)
defer testEmptyLog()
sl := newSocketListener() sl := newSocketListener()
sl.Log = testutil.Logger{} sl.Log = testutil.Logger{}
@ -137,7 +140,8 @@ func TestSocketListener_unix(t *testing.T) {
defer os.RemoveAll(tmpdir) defer os.RemoveAll(tmpdir)
sock := filepath.Join(tmpdir, "sl.TestSocketListener_unix.sock") sock := filepath.Join(tmpdir, "sl.TestSocketListener_unix.sock")
defer testEmptyLog(t)() testEmptyLog := prepareLog(t)
defer testEmptyLog()
f, _ := os.Create(sock) f, _ := os.Create(sock)
require.NoError(t, f.Close()) require.NoError(t, f.Close())
@ -167,7 +171,8 @@ func TestSocketListener_unixgram(t *testing.T) {
defer os.RemoveAll(tmpdir) defer os.RemoveAll(tmpdir)
sock := filepath.Join(tmpdir, "sl.TestSocketListener_unixgram.sock") sock := filepath.Join(tmpdir, "sl.TestSocketListener_unixgram.sock")
defer testEmptyLog(t)() testEmptyLog := prepareLog(t)
defer testEmptyLog()
_, err = os.Create(sock) _, err = os.Create(sock)
require.NoError(t, err) require.NoError(t, err)
@ -188,7 +193,8 @@ func TestSocketListener_unixgram(t *testing.T) {
} }
func TestSocketListenerDecode_tcp(t *testing.T) { func TestSocketListenerDecode_tcp(t *testing.T) {
defer testEmptyLog(t)() testEmptyLog := prepareLog(t)
defer testEmptyLog()
sl := newSocketListener() sl := newSocketListener()
sl.Log = testutil.Logger{} sl.Log = testutil.Logger{}
@ -208,7 +214,8 @@ func TestSocketListenerDecode_tcp(t *testing.T) {
} }
func TestSocketListenerDecode_udp(t *testing.T) { func TestSocketListenerDecode_udp(t *testing.T) {
defer testEmptyLog(t)() testEmptyLog := prepareLog(t)
defer testEmptyLog()
sl := newSocketListener() sl := newSocketListener()
sl.Log = testutil.Logger{} sl.Log = testutil.Logger{}
@ -256,18 +263,18 @@ func testSocketListener(t *testing.T, sl *SocketListener, client net.Conn) {
m3 := acc.Metrics[2] m3 := acc.Metrics[2]
acc.Unlock() acc.Unlock()
assert.Equal(t, "test", m1.Measurement) require.Equal(t, "test", m1.Measurement)
assert.Equal(t, map[string]string{"foo": "bar"}, m1.Tags) require.Equal(t, map[string]string{"foo": "bar"}, m1.Tags)
assert.Equal(t, map[string]interface{}{"v": int64(1)}, m1.Fields) require.Equal(t, map[string]interface{}{"v": int64(1)}, m1.Fields)
assert.True(t, time.Unix(0, 123456789).Equal(m1.Time)) require.True(t, time.Unix(0, 123456789).Equal(m1.Time))
assert.Equal(t, "test", m2.Measurement) require.Equal(t, "test", m2.Measurement)
assert.Equal(t, map[string]string{"foo": "baz"}, m2.Tags) require.Equal(t, map[string]string{"foo": "baz"}, m2.Tags)
assert.Equal(t, map[string]interface{}{"v": int64(2)}, m2.Fields) require.Equal(t, map[string]interface{}{"v": int64(2)}, m2.Fields)
assert.True(t, time.Unix(0, 123456790).Equal(m2.Time)) require.True(t, time.Unix(0, 123456790).Equal(m2.Time))
assert.Equal(t, "test", m3.Measurement) require.Equal(t, "test", m3.Measurement)
assert.Equal(t, map[string]string{"foo": "zab"}, m3.Tags) require.Equal(t, map[string]string{"foo": "zab"}, m3.Tags)
assert.Equal(t, map[string]interface{}{"v": int64(3)}, m3.Fields) require.Equal(t, map[string]interface{}{"v": int64(3)}, m3.Fields)
assert.True(t, time.Unix(0, 123456791).Equal(m3.Time)) require.True(t, time.Unix(0, 123456791).Equal(m3.Time))
} }

20
plugins/inputs/solr/solr.go
View File

@ -202,7 +202,7 @@ func getCoresFromStatus(adminCoresStatus *AdminCoresStatus) []string {
// Add core metrics from admin to accumulator // Add core metrics from admin to accumulator
// This is the only point where size_in_bytes is available (as far as I checked) // This is the only point where size_in_bytes is available (as far as I checked)
func addAdminCoresStatusToAcc(acc telegraf.Accumulator, adminCoreStatus *AdminCoresStatus, time time.Time) { func addAdminCoresStatusToAcc(acc telegraf.Accumulator, adminCoreStatus *AdminCoresStatus, measurementTime time.Time) {
for core, metrics := range adminCoreStatus.Status { for core, metrics := range adminCoreStatus.Status {
coreFields := map[string]interface{}{ coreFields := map[string]interface{}{
"deleted_docs": metrics.Index.DeletedDocs, "deleted_docs": metrics.Index.DeletedDocs,
@ -214,13 +214,13 @@ func addAdminCoresStatusToAcc(acc telegraf.Accumulator, adminCoreStatus *AdminCo
"solr_admin", "solr_admin",
coreFields, coreFields,
map[string]string{"core": core}, map[string]string{"core": core},
time, measurementTime,
) )
} }
} }
// Add core metrics section to accumulator // Add core metrics section to accumulator
func addCoreMetricsToAcc(acc telegraf.Accumulator, core string, mBeansData *MBeansData, time time.Time) error { func addCoreMetricsToAcc(acc telegraf.Accumulator, core string, mBeansData *MBeansData, measurementTime time.Time) error {
var coreMetrics map[string]Core var coreMetrics map[string]Core
if len(mBeansData.SolrMbeans) < 2 { if len(mBeansData.SolrMbeans) < 2 {
return fmt.Errorf("no core metric data to unmarshal") return fmt.Errorf("no core metric data to unmarshal")
@ -243,14 +243,14 @@ func addCoreMetricsToAcc(acc telegraf.Accumulator, core string, mBeansData *MBea
map[string]string{ map[string]string{
"core": core, "core": core,
"handler": name}, "handler": name},
time, measurementTime,
) )
} }
return nil return nil
} }
// Add query metrics section to accumulator // Add query metrics section to accumulator
func addQueryHandlerMetricsToAcc(acc telegraf.Accumulator, core string, mBeansData *MBeansData, time time.Time) error { func addQueryHandlerMetricsToAcc(acc telegraf.Accumulator, core string, mBeansData *MBeansData, measurementTime time.Time) error {
var queryMetrics map[string]QueryHandler var queryMetrics map[string]QueryHandler
if len(mBeansData.SolrMbeans) < 4 { if len(mBeansData.SolrMbeans) < 4 {
@ -284,7 +284,7 @@ func addQueryHandlerMetricsToAcc(acc telegraf.Accumulator, core string, mBeansDa
map[string]string{ map[string]string{
"core": core, "core": core,
"handler": name}, "handler": name},
time, measurementTime,
) )
} }
return nil return nil
@ -324,7 +324,7 @@ func convertQueryHandlerMap(value map[string]interface{}) map[string]interface{}
} }
// Add update metrics section to accumulator // Add update metrics section to accumulator
func addUpdateHandlerMetricsToAcc(acc telegraf.Accumulator, core string, mBeansData *MBeansData, time time.Time) error { func addUpdateHandlerMetricsToAcc(acc telegraf.Accumulator, core string, mBeansData *MBeansData, measurementTime time.Time) error {
var updateMetrics map[string]UpdateHandler var updateMetrics map[string]UpdateHandler
if len(mBeansData.SolrMbeans) < 6 { if len(mBeansData.SolrMbeans) < 6 {
@ -363,7 +363,7 @@ func addUpdateHandlerMetricsToAcc(acc telegraf.Accumulator, core string, mBeansD
map[string]string{ map[string]string{
"core": core, "core": core,
"handler": name}, "handler": name},
time, measurementTime,
) )
} }
return nil return nil
@ -404,7 +404,7 @@ func getInt(unk interface{}) int64 {
} }
// Add cache metrics section to accumulator // Add cache metrics section to accumulator
func addCacheMetricsToAcc(acc telegraf.Accumulator, core string, mBeansData *MBeansData, time time.Time) error { func addCacheMetricsToAcc(acc telegraf.Accumulator, core string, mBeansData *MBeansData, measurementTime time.Time) error {
if len(mBeansData.SolrMbeans) < 8 { if len(mBeansData.SolrMbeans) < 8 {
return fmt.Errorf("no cache metric data to unmarshal") return fmt.Errorf("no cache metric data to unmarshal")
} }
@ -444,7 +444,7 @@ func addCacheMetricsToAcc(acc telegraf.Accumulator, core string, mBeansData *MBe
map[string]string{ map[string]string{
"core": core, "core": core,
"handler": name}, "handler": name},
time, measurementTime,
) )
} }
return nil return nil

48
plugins/inputs/sqlserver/sqlserver.go
View File

@ -4,13 +4,13 @@ import (
"database/sql" "database/sql"
"errors" "errors"
"fmt" "fmt"
"log"
"strings" "strings"
"sync" "sync"
"time" "time"
"github.com/Azure/go-autorest/autorest/adal" "github.com/Azure/go-autorest/autorest/adal"
mssql "github.com/denisenkom/go-mssqldb" mssql "github.com/denisenkom/go-mssqldb"
"github.com/influxdata/telegraf" "github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/filter" "github.com/influxdata/telegraf/filter"
"github.com/influxdata/telegraf/plugins/inputs" "github.com/influxdata/telegraf/plugins/inputs"
@ -18,18 +18,20 @@ import (
// SQLServer struct // SQLServer struct
type SQLServer struct { type SQLServer struct {
Servers []string `toml:"servers"` Servers []string `toml:"servers"`
AuthMethod string `toml:"auth_method"` AuthMethod string `toml:"auth_method"`
QueryVersion int `toml:"query_version"` QueryVersion int `toml:"query_version"`
AzureDB bool `toml:"azuredb"` AzureDB bool `toml:"azuredb"`
DatabaseType string `toml:"database_type"` DatabaseType string `toml:"database_type"`
IncludeQuery []string `toml:"include_query"` IncludeQuery []string `toml:"include_query"`
ExcludeQuery []string `toml:"exclude_query"` ExcludeQuery []string `toml:"exclude_query"`
HealthMetric bool `toml:"health_metric"` HealthMetric bool `toml:"health_metric"`
pools []*sql.DB Log telegraf.Logger `toml:"-"`
queries MapQuery
adalToken *adal.Token pools []*sql.DB
muCacheLock sync.RWMutex queries MapQuery
adalToken *adal.Token
muCacheLock sync.RWMutex
} }
// Query struct // Query struct
@ -142,10 +144,10 @@ type scanner interface {
Scan(dest ...interface{}) error Scan(dest ...interface{}) error
} }
func initQueries(s *SQLServer) error { func (s *SQLServer) initQueries() error {
s.queries = make(MapQuery) s.queries = make(MapQuery)
queries := s.queries queries := s.queries
log.Printf("I! [inputs.sqlserver] Config: database_type: %s , query_version:%d , azuredb: %t", s.DatabaseType, s.QueryVersion, s.AzureDB) s.Log.Infof("Config: database_type: %s , query_version:%d , azuredb: %t", s.DatabaseType, s.QueryVersion, s.AzureDB)
// New config option database_type // New config option database_type
// To prevent query definition conflicts // To prevent query definition conflicts
@ -202,7 +204,7 @@ func initQueries(s *SQLServer) error {
} }
// Decide if we want to run version 1 or version 2 queries // Decide if we want to run version 1 or version 2 queries
if s.QueryVersion == 2 { if s.QueryVersion == 2 {
log.Println("W! DEPRECATION NOTICE: query_version=2 is being deprecated in favor of database_type.") s.Log.Warn("DEPRECATION NOTICE: query_version=2 is being deprecated in favor of database_type.")
queries["PerformanceCounters"] = Query{ScriptName: "PerformanceCounters", Script: sqlPerformanceCountersV2, ResultByRow: true} queries["PerformanceCounters"] = Query{ScriptName: "PerformanceCounters", Script: sqlPerformanceCountersV2, ResultByRow: true}
queries["WaitStatsCategorized"] = Query{ScriptName: "WaitStatsCategorized", Script: sqlWaitStatsCategorizedV2, ResultByRow: false} queries["WaitStatsCategorized"] = Query{ScriptName: "WaitStatsCategorized", Script: sqlWaitStatsCategorizedV2, ResultByRow: false}
queries["DatabaseIO"] = Query{ScriptName: "DatabaseIO", Script: sqlDatabaseIOV2, ResultByRow: false} queries["DatabaseIO"] = Query{ScriptName: "DatabaseIO", Script: sqlDatabaseIOV2, ResultByRow: false}
@ -213,7 +215,7 @@ func initQueries(s *SQLServer) error {
queries["VolumeSpace"] = Query{ScriptName: "VolumeSpace", Script: sqlServerVolumeSpaceV2, ResultByRow: false} queries["VolumeSpace"] = Query{ScriptName: "VolumeSpace", Script: sqlServerVolumeSpaceV2, ResultByRow: false}
queries["Cpu"] = Query{ScriptName: "Cpu", Script: sqlServerCPUV2, ResultByRow: false} queries["Cpu"] = Query{ScriptName: "Cpu", Script: sqlServerCPUV2, ResultByRow: false}
} else { } else {
log.Println("W! DEPRECATED: query_version=1 has been deprecated in favor of database_type.") s.Log.Warn("DEPRECATED: query_version=1 has been deprecated in favor of database_type.")
queries["PerformanceCounters"] = Query{ScriptName: "PerformanceCounters", Script: sqlPerformanceCounters, ResultByRow: true} queries["PerformanceCounters"] = Query{ScriptName: "PerformanceCounters", Script: sqlPerformanceCounters, ResultByRow: true}
queries["WaitStatsCategorized"] = Query{ScriptName: "WaitStatsCategorized", Script: sqlWaitStatsCategorized, ResultByRow: false} queries["WaitStatsCategorized"] = Query{ScriptName: "WaitStatsCategorized", Script: sqlWaitStatsCategorized, ResultByRow: false}
queries["CPUHistory"] = Query{ScriptName: "CPUHistory", Script: sqlCPUHistory, ResultByRow: false} queries["CPUHistory"] = Query{ScriptName: "CPUHistory", Script: sqlCPUHistory, ResultByRow: false}
@ -242,7 +244,7 @@ func initQueries(s *SQLServer) error {
for query := range queries { for query := range queries {
querylist = append(querylist, query) querylist = append(querylist, query)
} }
log.Printf("I! [inputs.sqlserver] Config: Effective Queries: %#v\n", querylist) s.Log.Infof("Config: Effective Queries: %#v\n", querylist)
return nil return nil
} }
@ -283,7 +285,7 @@ func (s *SQLServer) Gather(acc telegraf.Accumulator) error {
// Start initialize a list of connection pools // Start initialize a list of connection pools
func (s *SQLServer) Start(acc telegraf.Accumulator) error { func (s *SQLServer) Start(acc telegraf.Accumulator) error {
if err := initQueries(s); err != nil { if err := s.initQueries(); err != nil {
acc.AddError(err) acc.AddError(err)
return err return err
} }
@ -355,11 +357,11 @@ func (s *SQLServer) gatherServer(pool *sql.DB, query Query, acc telegraf.Accumul
// Error msg based on the format in SSMS. SQLErrorClass() is another term for severity/level: http://msdn.microsoft.com/en-us/library/dd304156.aspx // Error msg based on the format in SSMS. SQLErrorClass() is another term for severity/level: http://msdn.microsoft.com/en-us/library/dd304156.aspx
if sqlerr, ok := err.(mssql.Error); ok { if sqlerr, ok := err.(mssql.Error); ok {
return fmt.Errorf("Query %s failed for server: %s and database: %s with Msg %d, Level %d, State %d:, Line %d, Error: %w", query.ScriptName, return fmt.Errorf("query %s failed for server: %s and database: %s with Msg %d, Level %d, State %d:, Line %d, Error: %w", query.ScriptName,
serverName, databaseName, sqlerr.SQLErrorNumber(), sqlerr.SQLErrorClass(), sqlerr.SQLErrorState(), sqlerr.SQLErrorLineNo(), err) serverName, databaseName, sqlerr.SQLErrorNumber(), sqlerr.SQLErrorClass(), sqlerr.SQLErrorState(), sqlerr.SQLErrorLineNo(), err)
} }
return fmt.Errorf("Query %s failed for server: %s and database: %s with Error: %w", query.ScriptName, serverName, databaseName, err) return fmt.Errorf("query %s failed for server: %s and database: %s with Error: %w", query.ScriptName, serverName, databaseName, err)
} }
defer rows.Close() defer rows.Close()
@ -425,7 +427,7 @@ func (s *SQLServer) accRow(query Query, acc telegraf.Accumulator, row scanner) e
// values // values
for header, val := range columnMap { for header, val := range columnMap {
if _, ok := (*val).(string); !ok { if _, ok := (*val).(string); !ok {
fields[header] = (*val) fields[header] = *val
} }
} }
// add fields to Accumulator // add fields to Accumulator
@ -476,7 +478,7 @@ func (s *SQLServer) getDatabaseTypeToLog() string {
func (s *SQLServer) Init() error { func (s *SQLServer) Init() error {
if len(s.Servers) == 0 { if len(s.Servers) == 0 {
log.Println("W! Warning: Server list is empty.") s.Log.Warn("Warning: Server list is empty.")
} }
return nil return nil

20
plugins/inputs/sqlserver/sqlserver_test.go
View File

@ -32,8 +32,9 @@ func TestSqlServer_QueriesInclusionExclusion(t *testing.T) {
QueryVersion: 2, QueryVersion: 2,
IncludeQuery: test["IncludeQuery"].([]string), IncludeQuery: test["IncludeQuery"].([]string),
ExcludeQuery: test["ExcludeQuery"].([]string), ExcludeQuery: test["ExcludeQuery"].([]string),
Log: testutil.Logger{},
} }
require.NoError(t, initQueries(&s)) require.NoError(t, s.initQueries())
require.Equal(t, len(s.queries), test["queriesTotal"].(int)) require.Equal(t, len(s.queries), test["queriesTotal"].(int))
for _, query := range test["queries"].([]string) { for _, query := range test["queries"].([]string) {
require.Contains(t, s.queries, query) require.Contains(t, s.queries, query)
@ -116,10 +117,12 @@ func TestSqlServer_MultipleInstanceIntegration(t *testing.T) {
s := &SQLServer{ s := &SQLServer{
Servers: []string{testServer}, Servers: []string{testServer},
ExcludeQuery: []string{"MemoryClerk"}, ExcludeQuery: []string{"MemoryClerk"},
Log: testutil.Logger{},
} }
s2 := &SQLServer{ s2 := &SQLServer{
Servers: []string{testServer}, Servers: []string{testServer},
ExcludeQuery: []string{"DatabaseSize"}, ExcludeQuery: []string{"DatabaseSize"},
Log: testutil.Logger{},
} }
var acc, acc2 testutil.Accumulator var acc, acc2 testutil.Accumulator
@ -151,11 +154,13 @@ func TestSqlServer_MultipleInstanceWithHealthMetricIntegration(t *testing.T) {
s := &SQLServer{ s := &SQLServer{
Servers: []string{testServer}, Servers: []string{testServer},
ExcludeQuery: []string{"MemoryClerk"}, ExcludeQuery: []string{"MemoryClerk"},
Log: testutil.Logger{},
} }
s2 := &SQLServer{ s2 := &SQLServer{
Servers: []string{testServer}, Servers: []string{testServer},
ExcludeQuery: []string{"DatabaseSize"}, ExcludeQuery: []string{"DatabaseSize"},
HealthMetric: true, HealthMetric: true,
Log: testutil.Logger{},
} }
var acc, acc2 testutil.Accumulator var acc, acc2 testutil.Accumulator
@ -192,12 +197,14 @@ func TestSqlServer_HealthMetric(t *testing.T) {
IncludeQuery: []string{"DatabaseSize", "MemoryClerk"}, IncludeQuery: []string{"DatabaseSize", "MemoryClerk"},
HealthMetric: true, HealthMetric: true,
AuthMethod: "connection_string", AuthMethod: "connection_string",
Log: testutil.Logger{},
} }
s2 := &SQLServer{ s2 := &SQLServer{
Servers: []string{fakeServer1}, Servers: []string{fakeServer1},
IncludeQuery: []string{"DatabaseSize"}, IncludeQuery: []string{"DatabaseSize"},
AuthMethod: "connection_string", AuthMethod: "connection_string",
Log: testutil.Logger{},
} }
// acc1 should have the health metric because it is specified in the config // acc1 should have the health metric because it is specified in the config
@ -225,16 +232,17 @@ func TestSqlServer_HealthMetric(t *testing.T) {
} }
func TestSqlServer_MultipleInit(t *testing.T) { func TestSqlServer_MultipleInit(t *testing.T) {
s := &SQLServer{} s := &SQLServer{Log: testutil.Logger{}}
s2 := &SQLServer{ s2 := &SQLServer{
ExcludeQuery: []string{"DatabaseSize"}, ExcludeQuery: []string{"DatabaseSize"},
Log: testutil.Logger{},
} }
require.NoError(t, initQueries(s)) require.NoError(t, s.initQueries())
_, ok := s.queries["DatabaseSize"] _, ok := s.queries["DatabaseSize"]
require.True(t, ok) require.True(t, ok)
require.NoError(t, initQueries(s2)) require.NoError(t, s.initQueries())
_, ok = s2.queries["DatabaseSize"] _, ok = s2.queries["DatabaseSize"]
require.False(t, ok) require.False(t, ok)
s.Stop() s.Stop()
@ -335,11 +343,13 @@ func TestSqlServer_AGQueriesApplicableForDatabaseTypeSQLServer(t *testing.T) {
Servers: []string{testServer}, Servers: []string{testServer},
DatabaseType: "SQLServer", DatabaseType: "SQLServer",
IncludeQuery: []string{"SQLServerAvailabilityReplicaStates", "SQLServerDatabaseReplicaStates"}, IncludeQuery: []string{"SQLServerAvailabilityReplicaStates", "SQLServerDatabaseReplicaStates"},
Log: testutil.Logger{},
} }
s2 := &SQLServer{ s2 := &SQLServer{
Servers: []string{testServer}, Servers: []string{testServer},
DatabaseType: "AzureSQLDB", DatabaseType: "AzureSQLDB",
IncludeQuery: []string{"SQLServerAvailabilityReplicaStates", "SQLServerDatabaseReplicaStates"}, IncludeQuery: []string{"SQLServerAvailabilityReplicaStates", "SQLServerDatabaseReplicaStates"},
Log: testutil.Logger{},
} }
var acc, acc2 testutil.Accumulator var acc, acc2 testutil.Accumulator
@ -376,11 +386,13 @@ func TestSqlServer_AGQueryFieldsOutputBasedOnSQLServerVersion(t *testing.T) {
Servers: []string{testServer2019}, Servers: []string{testServer2019},
DatabaseType: "SQLServer", DatabaseType: "SQLServer",
IncludeQuery: []string{"SQLServerAvailabilityReplicaStates", "SQLServerDatabaseReplicaStates"}, IncludeQuery: []string{"SQLServerAvailabilityReplicaStates", "SQLServerDatabaseReplicaStates"},
Log: testutil.Logger{},
} }
s2012 := &SQLServer{ s2012 := &SQLServer{
Servers: []string{testServer2012}, Servers: []string{testServer2012},
DatabaseType: "SQLServer", DatabaseType: "SQLServer",
IncludeQuery: []string{"SQLServerAvailabilityReplicaStates", "SQLServerDatabaseReplicaStates"}, IncludeQuery: []string{"SQLServerAvailabilityReplicaStates", "SQLServerDatabaseReplicaStates"},
Log: testutil.Logger{},
} }
var acc2019, acc2012 testutil.Accumulator var acc2019, acc2012 testutil.Accumulator

44
plugins/inputs/stackdriver/stackdriver.go
View File

@ -10,18 +10,19 @@ import (
"time" "time"
monitoring "cloud.google.com/go/monitoring/apiv3/v2" monitoring "cloud.google.com/go/monitoring/apiv3/v2"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/config"
"github.com/influxdata/telegraf/internal/limiter"
"github.com/influxdata/telegraf/metric"
"github.com/influxdata/telegraf/plugins/inputs" // Imports the Stackdriver Monitoring client package.
"github.com/influxdata/telegraf/selfstat"
"google.golang.org/api/iterator" "google.golang.org/api/iterator"
distributionpb "google.golang.org/genproto/googleapis/api/distribution" distributionpb "google.golang.org/genproto/googleapis/api/distribution"
metricpb "google.golang.org/genproto/googleapis/api/metric" metricpb "google.golang.org/genproto/googleapis/api/metric"
monitoringpb "google.golang.org/genproto/googleapis/monitoring/v3" monitoringpb "google.golang.org/genproto/googleapis/monitoring/v3"
"google.golang.org/protobuf/types/known/durationpb" "google.golang.org/protobuf/types/known/durationpb"
"google.golang.org/protobuf/types/known/timestamppb" "google.golang.org/protobuf/types/known/timestamppb"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/config"
"github.com/influxdata/telegraf/internal/limiter"
"github.com/influxdata/telegraf/metric"
"github.com/influxdata/telegraf/plugins/inputs" // Imports the Stackdriver Monitoring client package.
"github.com/influxdata/telegraf/selfstat"
) )
const ( const (
@ -312,8 +313,8 @@ func (s *Stackdriver) Gather(acc telegraf.Accumulator) error {
} }
wg.Wait() wg.Wait()
for _, metric := range grouper.Metrics() { for _, groupedMetric := range grouper.Metrics() {
acc.AddMetric(metric) acc.AddMetric(groupedMetric)
} }
return nil return nil
@ -643,35 +644,34 @@ func (s *Stackdriver) gatherTimeSeries(
} }
// AddDistribution adds metrics from a distribution value type. // AddDistribution adds metrics from a distribution value type.
func (s *Stackdriver) addDistribution( func (s *Stackdriver) addDistribution(dist *distributionpb.Distribution, tags map[string]string, ts time.Time,
metric *distributionpb.Distribution, grouper *lockedSeriesGrouper, tsConf *timeSeriesConf,
tags map[string]string, ts time.Time, grouper *lockedSeriesGrouper, tsConf *timeSeriesConf,
) error { ) error {
field := tsConf.fieldKey field := tsConf.fieldKey
name := tsConf.measurement name := tsConf.measurement
if err := grouper.Add(name, tags, ts, field+"_count", metric.Count); err != nil { if err := grouper.Add(name, tags, ts, field+"_count", dist.Count); err != nil {
return err return err
} }
if err := grouper.Add(name, tags, ts, field+"_mean", metric.Mean); err != nil { if err := grouper.Add(name, tags, ts, field+"_mean", dist.Mean); err != nil {
return err return err
} }
if err := grouper.Add(name, tags, ts, field+"_sum_of_squared_deviation", metric.SumOfSquaredDeviation); err != nil { if err := grouper.Add(name, tags, ts, field+"_sum_of_squared_deviation", dist.SumOfSquaredDeviation); err != nil {
return err return err
} }
if metric.Range != nil { if dist.Range != nil {
if err := grouper.Add(name, tags, ts, field+"_range_min", metric.Range.Min); err != nil { if err := grouper.Add(name, tags, ts, field+"_range_min", dist.Range.Min); err != nil {
return err return err
} }
if err := grouper.Add(name, tags, ts, field+"_range_max", metric.Range.Max); err != nil { if err := grouper.Add(name, tags, ts, field+"_range_max", dist.Range.Max); err != nil {
return err return err
} }
} }
linearBuckets := metric.BucketOptions.GetLinearBuckets() linearBuckets := dist.BucketOptions.GetLinearBuckets()
exponentialBuckets := metric.BucketOptions.GetExponentialBuckets() exponentialBuckets := dist.BucketOptions.GetExponentialBuckets()
explicitBuckets := metric.BucketOptions.GetExplicitBuckets() explicitBuckets := dist.BucketOptions.GetExplicitBuckets()
var numBuckets int32 var numBuckets int32
if linearBuckets != nil { if linearBuckets != nil {
@ -704,8 +704,8 @@ func (s *Stackdriver) addDistribution(
// Add to the cumulative count; trailing buckets with value 0 are // Add to the cumulative count; trailing buckets with value 0 are
// omitted from the response. // omitted from the response.
if i < int32(len(metric.BucketCounts)) { if i < int32(len(dist.BucketCounts)) {
count += metric.BucketCounts[i] count += dist.BucketCounts[i]
} }
if err := grouper.Add(name, tags, ts, field+"_bucket", count); err != nil { if err := grouper.Add(name, tags, ts, field+"_bucket", count); err != nil {
return err return err

5
plugins/inputs/statsd/datadog.go
View File

@ -120,11 +120,10 @@ func (s *Statsd) parseEventMessage(now time.Time, message string, defaultHostnam
case "s:": case "s:":
fields["source_type_name"] = rawMetadataFields[i][2:] fields["source_type_name"] = rawMetadataFields[i][2:]
default: default:
if rawMetadataFields[i][0] == '#' { if rawMetadataFields[i][0] != '#' {
parseDataDogTags(tags, rawMetadataFields[i][1:])
} else {
return fmt.Errorf("unknown metadata type: '%s'", rawMetadataFields[i]) return fmt.Errorf("unknown metadata type: '%s'", rawMetadataFields[i])
} }
parseDataDogTags(tags, rawMetadataFields[i][1:])
} }
} }
// Use source tag because host is reserved tag key in Telegraf. // Use source tag because host is reserved tag key in Telegraf.

16
plugins/inputs/statsd/statsd.go
View File

@ -11,13 +11,14 @@ import (
"sync" "sync"
"time" "time"
"github.com/pkg/errors"
"github.com/influxdata/telegraf" "github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/config" "github.com/influxdata/telegraf/config"
"github.com/influxdata/telegraf/internal" "github.com/influxdata/telegraf/internal"
"github.com/influxdata/telegraf/plugins/inputs" "github.com/influxdata/telegraf/plugins/inputs"
"github.com/influxdata/telegraf/plugins/parsers/graphite" "github.com/influxdata/telegraf/plugins/parsers/graphite"
"github.com/influxdata/telegraf/selfstat" "github.com/influxdata/telegraf/selfstat"
"github.com/pkg/errors"
) )
const ( const (
@ -745,10 +746,10 @@ func (s *Statsd) parseStatsdLine(line string) error {
// config file. If there is a match, it will parse the name of the metric and // config file. If there is a match, it will parse the name of the metric and
// map of tags. // map of tags.
// Return values are (<name>, <field>, <tags>) // Return values are (<name>, <field>, <tags>)
func (s *Statsd) parseName(bucket string) (string, string, map[string]string) { func (s *Statsd) parseName(bucket string) (name string, field string, tags map[string]string) {
s.Lock() s.Lock()
defer s.Unlock() defer s.Unlock()
tags := make(map[string]string) tags = make(map[string]string)
bucketparts := strings.Split(bucket, ",") bucketparts := strings.Split(bucket, ",")
// Parse out any tags in the bucket // Parse out any tags in the bucket
@ -761,8 +762,7 @@ func (s *Statsd) parseName(bucket string) (string, string, map[string]string) {
} }
} }
var field string name = bucketparts[0]
name := bucketparts[0]
p := s.graphiteParser p := s.graphiteParser
var err error var err error
@ -789,10 +789,8 @@ func (s *Statsd) parseName(bucket string) (string, string, map[string]string) {
} }
// Parse the key,value out of a string that looks like "key=value" // Parse the key,value out of a string that looks like "key=value"
func parseKeyValue(keyvalue string) (string, string) { func parseKeyValue(keyValue string) (key string, val string) {
var key, val string split := strings.Split(keyValue, "=")
split := strings.Split(keyvalue, "=")
// Must be exactly 2 to get anything meaningful out of them // Must be exactly 2 to get anything meaningful out of them
if len(split) == 2 { if len(split) == 2 {
key = split[0] key = split[0]

39
plugins/inputs/statsd/statsd_test.go
View File

@ -7,11 +7,10 @@ import (
"testing" "testing"
"time" "time"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/config"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/config"
"github.com/influxdata/telegraf/testutil" "github.com/influxdata/telegraf/testutil"
) )
@ -53,19 +52,19 @@ func TestConcurrentConns(t *testing.T) {
time.Sleep(time.Millisecond * 250) time.Sleep(time.Millisecond * 250)
_, err := net.Dial("tcp", "127.0.0.1:8125") _, err := net.Dial("tcp", "127.0.0.1:8125")
assert.NoError(t, err) require.NoError(t, err)
_, err = net.Dial("tcp", "127.0.0.1:8125") _, err = net.Dial("tcp", "127.0.0.1:8125")
assert.NoError(t, err) require.NoError(t, err)
// Connection over the limit: // Connection over the limit:
conn, err := net.Dial("tcp", "127.0.0.1:8125") conn, err := net.Dial("tcp", "127.0.0.1:8125")
assert.NoError(t, err) require.NoError(t, err)
_, err = net.Dial("tcp", "127.0.0.1:8125") _, err = net.Dial("tcp", "127.0.0.1:8125")
assert.NoError(t, err) require.NoError(t, err)
_, err = conn.Write([]byte(testMsg)) _, err = conn.Write([]byte(testMsg))
assert.NoError(t, err) require.NoError(t, err)
time.Sleep(time.Millisecond * 100) time.Sleep(time.Millisecond * 100)
assert.Zero(t, acc.NFields()) require.Zero(t, acc.NFields())
} }
// Test that MaxTCPConnections is respected when max==1 // Test that MaxTCPConnections is respected when max==1
@ -84,17 +83,17 @@ func TestConcurrentConns1(t *testing.T) {
time.Sleep(time.Millisecond * 250) time.Sleep(time.Millisecond * 250)
_, err := net.Dial("tcp", "127.0.0.1:8125") _, err := net.Dial("tcp", "127.0.0.1:8125")
assert.NoError(t, err) require.NoError(t, err)
// Connection over the limit: // Connection over the limit:
conn, err := net.Dial("tcp", "127.0.0.1:8125") conn, err := net.Dial("tcp", "127.0.0.1:8125")
assert.NoError(t, err) require.NoError(t, err)
_, err = net.Dial("tcp", "127.0.0.1:8125") _, err = net.Dial("tcp", "127.0.0.1:8125")
assert.NoError(t, err) require.NoError(t, err)
_, err = conn.Write([]byte(testMsg)) _, err = conn.Write([]byte(testMsg))
assert.NoError(t, err) require.NoError(t, err)
time.Sleep(time.Millisecond * 100) time.Sleep(time.Millisecond * 100)
assert.Zero(t, acc.NFields()) require.Zero(t, acc.NFields())
} }
// Test that MaxTCPConnections is respected // Test that MaxTCPConnections is respected
@ -112,9 +111,9 @@ func TestCloseConcurrentConns(t *testing.T) {
time.Sleep(time.Millisecond * 250) time.Sleep(time.Millisecond * 250)
_, err := net.Dial("tcp", "127.0.0.1:8125") _, err := net.Dial("tcp", "127.0.0.1:8125")
assert.NoError(t, err) require.NoError(t, err)
_, err = net.Dial("tcp", "127.0.0.1:8125") _, err = net.Dial("tcp", "127.0.0.1:8125")
assert.NoError(t, err) require.NoError(t, err)
listener.Stop() listener.Stop()
} }
@ -156,7 +155,7 @@ func sendRequests(conn net.Conn, wg *sync.WaitGroup) {
defer wg.Done() defer wg.Done()
for i := 0; i < 25000; i++ { for i := 0; i < 25000; i++ {
//nolint:errcheck,revive //nolint:errcheck,revive
fmt.Fprintf(conn, testMsg) fmt.Fprint(conn, testMsg)
} }
} }
@ -476,7 +475,7 @@ func TestParse_Distributions(t *testing.T) {
parseMetrics() parseMetrics()
for key, value := range validMeasurementMap { for key, value := range validMeasurementMap {
field := map[string]interface{}{ field := map[string]interface{}{
"value": float64(value), "value": value,
} }
acc.AssertContainsFields(t, key, field) acc.AssertContainsFields(t, key, field)
} }
@ -1570,7 +1569,7 @@ func testValidateGauge(
} }
if valueExpected != valueActual { if valueExpected != valueActual {
return fmt.Errorf("Measurement: %s, expected %f, actual %f", name, valueExpected, valueActual) return fmt.Errorf("measurement: %s, expected %f, actual %f", name, valueExpected, valueActual)
} }
return nil return nil
} }
@ -1590,6 +1589,8 @@ func TestTCP(t *testing.T) {
addr := statsd.TCPlistener.Addr().String() addr := statsd.TCPlistener.Addr().String()
conn, err := net.Dial("tcp", addr) conn, err := net.Dial("tcp", addr)
require.NoError(t, err)
_, err = conn.Write([]byte("cpu.time_idle:42|c\n")) _, err = conn.Write([]byte("cpu.time_idle:42|c\n"))
require.NoError(t, err) require.NoError(t, err)
require.NoError(t, conn.Close()) require.NoError(t, conn.Close())

7
plugins/inputs/suricata/suricata_test.go
View File

@ -11,9 +11,10 @@ import (
"testing" "testing"
"time" "time"
"github.com/stretchr/testify/require"
"github.com/influxdata/telegraf" "github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/testutil" "github.com/influxdata/telegraf/testutil"
"github.com/stretchr/testify/require"
) )
var ex2 = `{"timestamp":"2017-03-06T07:43:39.000397+0000","event_type":"stats","stats":{"capture":{"kernel_packets":905344474,"kernel_drops":78355440,"kernel_packets_delta":2376742,"kernel_drops_delta":82049}}}` var ex2 = `{"timestamp":"2017-03-06T07:43:39.000397+0000","event_type":"stats","stats":{"capture":{"kernel_packets":905344474,"kernel_drops":78355440,"kernel_packets_delta":2376742,"kernel_drops_delta":82049}}}`
@ -388,11 +389,13 @@ func TestSuricataParse(t *testing.T) {
for _, tc := range tests { for _, tc := range tests {
data, err := os.ReadFile("testdata/" + tc.filename) data, err := os.ReadFile("testdata/" + tc.filename)
require.NoError(t, err) require.NoError(t, err)
s := Suricata{ s := Suricata{
Delimiter: "_", Delimiter: "_",
} }
acc := testutil.Accumulator{} acc := testutil.Accumulator{}
s.parse(&acc, data) err = s.parse(&acc, data)
require.NoError(t, err)
testutil.RequireMetricsEqual(t, tc.expected, acc.GetTelegrafMetrics(), testutil.IgnoreTime()) testutil.RequireMetricsEqual(t, tc.expected, acc.GetTelegrafMetrics(), testutil.IgnoreTime())
} }

16
plugins/inputs/synproxy/synproxy_test.go
View File

@ -7,9 +7,9 @@ import (
"os" "os"
"testing" "testing"
"github.com/influxdata/telegraf/testutil" "github.com/stretchr/testify/require"
"github.com/stretchr/testify/assert" "github.com/influxdata/telegraf/testutil"
) )
func TestSynproxyFileNormal(t *testing.T) { func TestSynproxyFileNormal(t *testing.T) {
@ -38,8 +38,8 @@ func TestSynproxyFileHeaderMismatch(t *testing.T) {
acc := testutil.Accumulator{} acc := testutil.Accumulator{}
err := k.Gather(&acc) err := k.Gather(&acc)
assert.Error(t, err) require.Error(t, err)
assert.Contains(t, err.Error(), "invalid number of columns in data") require.Contains(t, err.Error(), "invalid number of columns in data")
} }
func TestSynproxyFileInvalidHex(t *testing.T) { func TestSynproxyFileInvalidHex(t *testing.T) {
@ -52,8 +52,8 @@ func TestSynproxyFileInvalidHex(t *testing.T) {
acc := testutil.Accumulator{} acc := testutil.Accumulator{}
err := k.Gather(&acc) err := k.Gather(&acc)
assert.Error(t, err) require.Error(t, err)
assert.Contains(t, err.Error(), "invalid value") require.Contains(t, err.Error(), "invalid value")
} }
func TestNoSynproxyFile(t *testing.T) { func TestNoSynproxyFile(t *testing.T) {
@ -69,7 +69,7 @@ func TestNoSynproxyFile(t *testing.T) {
acc := testutil.Accumulator{} acc := testutil.Accumulator{}
err := k.Gather(&acc) err := k.Gather(&acc)
assert.Error(t, err) require.Error(t, err)
} }
// Valid Synproxy file // Valid Synproxy file
@ -149,7 +149,7 @@ func testSynproxyFileData(t *testing.T, fileData string, telegrafData map[string
acc := testutil.Accumulator{} acc := testutil.Accumulator{}
err := k.Gather(&acc) err := k.Gather(&acc)
assert.NoError(t, err) require.NoError(t, err)
acc.AssertContainsFields(t, "synproxy", telegrafData) acc.AssertContainsFields(t, "synproxy", telegrafData)
} }

11
plugins/inputs/syslog/syslog.go
View File

@ -13,7 +13,7 @@ import (
"time" "time"
"unicode" "unicode"
syslog "github.com/influxdata/go-syslog/v3" "github.com/influxdata/go-syslog/v3"
"github.com/influxdata/go-syslog/v3/nontransparent" "github.com/influxdata/go-syslog/v3/nontransparent"
"github.com/influxdata/go-syslog/v3/octetcounting" "github.com/influxdata/go-syslog/v3/octetcounting"
"github.com/influxdata/go-syslog/v3/rfc3164" "github.com/influxdata/go-syslog/v3/rfc3164"
@ -205,7 +205,7 @@ func (s *Syslog) Stop() {
// getAddressParts returns the address scheme and host // getAddressParts returns the address scheme and host
// it also sets defaults for them when missing // it also sets defaults for them when missing
// when the input address does not specify the protocol it returns an error // when the input address does not specify the protocol it returns an error
func getAddressParts(a string) (string, string, error) { func getAddressParts(a string) (scheme string, host string, err error) {
parts := strings.SplitN(a, "://", 2) parts := strings.SplitN(a, "://", 2)
if len(parts) != 2 { if len(parts) != 2 {
return "", "", fmt.Errorf("missing protocol within address '%s'", a) return "", "", fmt.Errorf("missing protocol within address '%s'", a)
@ -220,7 +220,6 @@ func getAddressParts(a string) (string, string, error) {
return parts[0], parts[1], nil return parts[0], parts[1], nil
} }
var host string
if u.Hostname() != "" { if u.Hostname() != "" {
host = u.Hostname() host = u.Hostname()
} }
@ -259,7 +258,7 @@ func (s *Syslog) listenPacket(acc telegraf.Accumulator) {
message, err := p.Parse(b[:n]) message, err := p.Parse(b[:n])
if message != nil { if message != nil {
acc.AddFields("syslog", fields(message, s), tags(message), s.time()) acc.AddFields("syslog", fields(message, s), tags(message), s.currentTime())
} }
if err != nil { if err != nil {
acc.AddError(err) acc.AddError(err)
@ -383,7 +382,7 @@ func (s *Syslog) store(res syslog.Result, acc telegraf.Accumulator) {
acc.AddError(res.Error) acc.AddError(res.Error)
} }
if res.Message != nil { if res.Message != nil {
acc.AddFields("syslog", fields(res.Message, s), tags(res.Message), s.time()) acc.AddFields("syslog", fields(res.Message, s), tags(res.Message), s.currentTime())
} }
} }
@ -473,7 +472,7 @@ func (uc unixCloser) Close() error {
return err return err
} }
func (s *Syslog) time() time.Time { func (s *Syslog) currentTime() time.Time {
t := s.now() t := s.now()
if t == s.lastTime { if t == s.lastTime {
t = t.Add(time.Nanosecond) t = t.Add(time.Nanosecond)

8
plugins/inputs/sysstat/sysstat.go
View File

@ -241,9 +241,9 @@ func (s *Sysstat) parse(acc telegraf.Accumulator, option string, ts time.Time) e
} }
r := bufio.NewReader(stdout) r := bufio.NewReader(stdout)
csv := csv.NewReader(r) csvReader := csv.NewReader(r)
csv.Comma = '\t' csvReader.Comma = '\t'
csv.FieldsPerRecord = 6 csvReader.FieldsPerRecord = 6
var measurement string var measurement string
// groupData to accumulate data when Group=true // groupData to accumulate data when Group=true
type groupData struct { type groupData struct {
@ -252,7 +252,7 @@ func (s *Sysstat) parse(acc telegraf.Accumulator, option string, ts time.Time) e
} }
m := make(map[string]groupData) m := make(map[string]groupData)
for { for {
record, err := csv.Read() record, err := csvReader.Read()
if err == io.EOF { if err == io.EOF {
break break
} }

4
plugins/inputs/sysstat/sysstat_test.go
View File

@ -10,8 +10,9 @@ import (
"path" "path"
"testing" "testing"
"github.com/influxdata/telegraf/testutil"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
"github.com/influxdata/telegraf/testutil"
) )
var s = Sysstat{ var s = Sysstat{
@ -310,5 +311,6 @@ dell-xps 5 2016-03-25 16:18:10 UTC sdb %util 0.30
default: default:
} }
// some code here to check arguments perhaps? // some code here to check arguments perhaps?
//nolint:revive // error code is important for this "test"
os.Exit(0) os.Exit(0)
} }

16
plugins/inputs/system/ps.go
View File

@ -5,13 +5,13 @@ import (
"path/filepath" "path/filepath"
"strings" "strings"
"github.com/influxdata/telegraf/internal"
"github.com/shirou/gopsutil/cpu" "github.com/shirou/gopsutil/cpu"
"github.com/shirou/gopsutil/disk" "github.com/shirou/gopsutil/disk"
"github.com/shirou/gopsutil/host" "github.com/shirou/gopsutil/host"
"github.com/shirou/gopsutil/mem" "github.com/shirou/gopsutil/mem"
"github.com/shirou/gopsutil/net" "github.com/shirou/gopsutil/net"
"github.com/influxdata/telegraf/internal"
) )
type PS interface { type PS interface {
@ -46,18 +46,18 @@ type SystemPSDisk struct{}
func (s *SystemPS) CPUTimes(perCPU, totalCPU bool) ([]cpu.TimesStat, error) { func (s *SystemPS) CPUTimes(perCPU, totalCPU bool) ([]cpu.TimesStat, error) {
var cpuTimes []cpu.TimesStat var cpuTimes []cpu.TimesStat
if perCPU { if perCPU {
if perCPUTimes, err := cpu.Times(true); err == nil { perCPUTimes, err := cpu.Times(true)
cpuTimes = append(cpuTimes, perCPUTimes...) if err != nil {
} else {
return nil, err return nil, err
} }
cpuTimes = append(cpuTimes, perCPUTimes...)
} }
if totalCPU { if totalCPU {
if totalCPUTimes, err := cpu.Times(false); err == nil { totalCPUTimes, err := cpu.Times(false)
cpuTimes = append(cpuTimes, totalCPUTimes...) if err != nil {
} else {
return nil, err return nil, err
} }
cpuTimes = append(cpuTimes, totalCPUTimes...)
} }
return cpuTimes, nil return cpuTimes, nil
} }