feat: Add RedisTimeSeries plugin (#11054)

etc/telegraf.conf

@@ -1677,6 +1677,23 @@
 #   ## Export metric collection time.
 #   # export_timestamp = false
 
+# # Configuration for the RedisTimeSeries server to send metrics to
+# [[outputs.redistimeseries]]
+#   ## The address of the RedisTimeSeries server.
+#   address = "127.0.0.1:6379"
+#   ## password to login Redis
+#   password = ""
+#
+#   ## username (optional)
+#   username = ""
+#   # redis database number (optional, must be an integer)
+#   database = 0
+#
+#   ## optional TLS configurations
+#   tls_ca = "/etc/telegraf/ca.pem
+#   tls_cert = "/etc/telegraf/cert.pem"
+#   tls_key = "/etc/telegraf/key.pem"
+#   insecure_skip_verify = false
 
 # # Configuration for Riemann to send metrics to
 # [[outputs.riemann]]
@@ -6216,9 +6233,9 @@
 
 
 # # A plugin to collect stats from Opensmtpd - a validating, recursive, and caching DNS resolver
-#  [[inputs.opensmtpd]]
-#    ## If running as a restricted user you can prepend sudo for additional access:
-#    #use_sudo = false
+# [[inputs.opensmtpd]]
+#   ## If running as a restricted user you can prepend sudo for additional access:
+#   #use_sudo = false
 #
 #    ## The default location of the smtpctl binary can be overridden with:
 #    binary = "/usr/sbin/smtpctl"
@@ -8403,35 +8420,35 @@
 
 # # Read Intel RDT metrics
 # [[inputs.intel_rdt]]
-#   ## Optionally set sampling interval to Nx100ms.
-#   ## This value is propagated to pqos tool. Interval format is defined by pqos itself.
-#   ## If not provided or provided 0, will be set to 10 = 10x100ms = 1s.
-#   # sampling_interval = "10"
+# 	## Optionally set sampling interval to Nx100ms.
+# 	## This value is propagated to pqos tool. Interval format is defined by pqos itself.
+# 	## If not provided or provided 0, will be set to 10 = 10x100ms = 1s.
+# 	# sampling_interval = "10"
 #
-#   ## Optionally specify the path to pqos executable.
-#   ## If not provided, auto discovery will be performed.
-#   # pqos_path = "/usr/local/bin/pqos"
+# 	## Optionally specify the path to pqos executable.
+# 	## If not provided, auto discovery will be performed.
+# 	# pqos_path = "/usr/local/bin/pqos"
 #
-#   ## Optionally specify if IPC and LLC_Misses metrics shouldn't be propagated.
-#   ## If not provided, default value is false.
-#   # shortened_metrics = false
+# 	## Optionally specify if IPC and LLC_Misses metrics shouldn't be propagated.
+# 	## If not provided, default value is false.
+# 	# shortened_metrics = false
 #
-#   ## Specify the list of groups of CPU core(s) to be provided as pqos input.
-#   ## Mandatory if processes aren't set and forbidden if processes are specified.
-#   ## e.g. ["0-3", "4,5,6"] or ["1-3,4"]
-#   # cores = ["0-3"]
+# 	## Specify the list of groups of CPU core(s) to be provided as pqos input.
+# 	## Mandatory if processes aren't set and forbidden if processes are specified.
+# 	## e.g. ["0-3", "4,5,6"] or ["1-3,4"]
+# 	# cores = ["0-3"]
 #
-#   ## Specify the list of processes for which Metrics will be collected.
-#   ## Mandatory if cores aren't set and forbidden if cores are specified.
-#   ## e.g. ["qemu", "pmd"]
-#   # processes = ["process"]
+# 	## Specify the list of processes for which Metrics will be collected.
+# 	## Mandatory if cores aren't set and forbidden if cores are specified.
+# 	## e.g. ["qemu", "pmd"]
+# 	# processes = ["process"]
 #
-#   ## Specify if the pqos process should be called with sudo.
-#   ## Mandatory if the telegraf process does not run as root.
-#   # use_sudo = false
+# 	## Specify if the pqos process should be called with sudo.
+# 	## Mandatory if the telegraf process does not run as root.
+# 	# use_sudo = false
 
 
-# # Subscribe and receive OpenConfig Telemetry data using JTI
+# # Read JTI OpenConfig Telemetry from listed sensors
 # [[inputs.jti_openconfig_telemetry]]
 #   ## List of device addresses to collect telemetry from
 #   servers = ["localhost:1883"]

go.mod

@@ -60,7 +60,6 @@ require (
 	github.com/go-ldap/ldap/v3 v3.4.1
 	github.com/go-logfmt/logfmt v0.5.1
 	github.com/go-ping/ping v0.0.0-20210201095549-52eed920f98c
-	github.com/go-redis/redis v6.15.9+incompatible
 	github.com/go-redis/redis/v8 v8.11.5
 	github.com/go-sql-driver/mysql v1.6.0
 	github.com/go-stomp/stomp v2.1.4+incompatible
@@ -174,6 +173,8 @@ require (
 	modernc.org/sqlite v1.17.3
 )
 
+require github.com/go-redis/redis/v7 v7.4.1
+
 require (
 	cloud.google.com/go v0.102.1 // indirect
 	cloud.google.com/go/compute v1.6.1 // indirect

go.sum

@@ -918,6 +918,8 @@ github.com/go-playground/locales v0.12.1/go.mod h1:IUMDtCfWo/w/mtMfIE/IG2K+Ey3yg
 github.com/go-playground/universal-translator v0.16.0/go.mod h1:1AnU7NaIRDWWzGEKwgtJRd2xk99HeFyHw3yid4rvQIY=
 github.com/go-redis/redis v6.15.9+incompatible h1:K0pv1D7EQUjfyoMql+r/jZqCLizCGKFlFgcHWWmHQjg=
 github.com/go-redis/redis v6.15.9+incompatible/go.mod h1:NAIEuMOZ/fxfXJIrKDQDz8wamY7mA7PouImQ2Jvg6kA=
+github.com/go-redis/redis/v7 v7.4.1 h1:PASvf36gyUpr2zdOUS/9Zqc80GbM+9BDyiJSJDDOrTI=
+github.com/go-redis/redis/v7 v7.4.1/go.mod h1:JDNMw23GTyLNC4GZu9njt15ctBQVn7xjRfnwdHj/Dcg=
 github.com/go-redis/redis/v8 v8.0.0-beta.6/go.mod h1:g79Vpae8JMzg5qjk8BiwU9tK+HmU3iDVyS4UAJLFycI=
 github.com/go-redis/redis/v8 v8.11.5 h1:AcZZR7igkdvfVmQTPnu9WE37LRrO/YrBH5zWyjDC0oI=
 github.com/go-redis/redis/v8 v8.11.5/go.mod h1:gREzHqY1hg6oD9ngVRbLStwAWKhA0FEgq8Jd4h5lpwo=

plugins/inputs/redis/README.md

@@ -28,7 +28,8 @@ The Redis input plugin gathers metrics from one or many Redis servers.
   #   # Can be "string", "integer", or "float"
   #   type = "string"
 
-  ## specify server password
+  ## specify server connection parameters
+  # username = "myuser"
   # password = "s#cr@t%"
 
   ## specify username for ACL auth (Redis 6.0+)

plugins/inputs/redis_sentinel/redis_sentinel.go

@@ -11,7 +11,7 @@ import (
 	"strings"
 	"sync"
 
-	"github.com/go-redis/redis"
+	"github.com/go-redis/redis/v7"
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/plugins/common/tls"

plugins/outputs/all/all.go

@@ -41,6 +41,7 @@ import (
 	_ "github.com/influxdata/telegraf/plugins/outputs/opentelemetry"
 	_ "github.com/influxdata/telegraf/plugins/outputs/opentsdb"
 	_ "github.com/influxdata/telegraf/plugins/outputs/prometheus_client"
+	_ "github.com/influxdata/telegraf/plugins/outputs/redistimeseries"
 	_ "github.com/influxdata/telegraf/plugins/outputs/riemann"
 	_ "github.com/influxdata/telegraf/plugins/outputs/riemann_legacy"
 	_ "github.com/influxdata/telegraf/plugins/outputs/sensu"

plugins/outputs/redistimeseries/README.md

@@ -0,0 +1,24 @@
+# RedisTimeSeries Producer Output Plugin
+
+The RedisTimeSeries output plugin writes metrics to the RedisTimeSeries server.
+
+## Configuration
+
+```toml
+[[outputs.redistimeseries]]
+  ## The address of the RedisTimeSeries server.
+  address = "127.0.0.1:6379"
+  ## password to login Redis
+  password = ""
+
+  ## username (optional)
+  # username = ""
+  # redis database number (optional, must be an integer)
+  # database = 0
+
+  ## optional TLS configurations
+  # tls_ca = "/etc/telegraf/ca.pem"
+  # tls_cert = "/etc/telegraf/cert.pem"
+  # tls_key = "/etc/telegraf/key.pem"
+  # insecure_skip_verify = false
+```

plugins/outputs/redistimeseries/redistimeseries.go

@@ -0,0 +1,91 @@
+package redistimeseries
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/go-redis/redis/v7"
+	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/plugins/common/tls"
+	"github.com/influxdata/telegraf/plugins/outputs"
+)
+
+const sampleConfig = `
+  ## The address of the RedisTimeSeries server.
+  address = "127.0.0.1:6379"
+
+  ## Redis ACL credentials
+  # username = ""
+  # password = ""
+  # database = 0
+
+  ## Optional TLS Config
+  # tls_ca = "/etc/telegraf/ca.pem"
+  # tls_cert = "/etc/telegraf/cert.pem"
+  # tls_key = "/etc/telegraf/key.pem"
+  # insecure_skip_verify = false
+`
+
+type RedisTimeSeries struct {
+	Address  string          `toml:"address"`
+	Username string          `toml:"username"`
+	Password string          `toml:"password"`
+	Database int             `toml:"database"`
+	Log      telegraf.Logger `toml:"-"`
+	tls.ClientConfig
+	client *redis.Client
+}
+
+func (r *RedisTimeSeries) Connect() error {
+	if r.Address == "" {
+		return errors.New("redis address must be specified")
+	}
+	r.client = redis.NewClient(&redis.Options{
+		Addr:     r.Address,
+		Password: r.Password,
+		Username: r.Username,
+		DB:       r.Database,
+	})
+	return r.client.Ping().Err()
+}
+
+func (r *RedisTimeSeries) Close() error {
+	return r.client.Close()
+}
+
+func (r *RedisTimeSeries) Description() string {
+	return "Plugin for sending metrics to RedisTimeSeries"
+}
+
+func (r *RedisTimeSeries) SampleConfig() string {
+	return sampleConfig
+}
+func (r *RedisTimeSeries) Write(metrics []telegraf.Metric) error {
+	for _, m := range metrics {
+		now := m.Time().UnixNano() / 1000000 // in milliseconds
+		name := m.Name()
+
+		var tags []interface{}
+		for k, v := range m.Tags() {
+			tags = append(tags, k, v)
+		}
+
+		for fieldName, value := range m.Fields() {
+			key := name + "_" + fieldName
+
+			addSlice := []interface{}{"TS.ADD", key, now, value}
+			addSlice = append(addSlice, tags...)
+
+			if err := r.client.Do(addSlice...).Err(); err != nil {
+				return fmt.Errorf("adding sample failed: %v", err)
+			}
+		}
+	}
+	return nil
+}
+
+func init() {
+	outputs.Add("redistimeseries", func() telegraf.Output {
+		return &RedisTimeSeries{}
+	})
+}

plugins/outputs/redistimeseries/redistimeseries_test.go

@@ -0,0 +1,26 @@
+package redistimeseries
+
+import (
+	"github.com/influxdata/telegraf/testutil"
+	"github.com/stretchr/testify/require"
+	"testing"
+)
+
+func TestConnectAndWrite(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	address := testutil.GetLocalHost() + ":6379"
+	redis := &RedisTimeSeries{
+		Address: address,
+	}
+
+	// Verify that we can connect to the RedisTimeSeries server
+	err := redis.Connect()
+	require.NoError(t, err)
+
+	// Verify that we can successfully write data to the RedisTimeSeries server
+	err = redis.Write(testutil.MockMetrics())
+	require.NoError(t, err)
+}