feat(outputs.redistimeseries): Add option to expire values

plugins/outputs/redistimeseries/README.md

@@ -41,6 +41,13 @@ to use them.
   ## Timeout for operations such as ping or sending metrics
   # timeout = "10s"
 
+  ## Set a time-to-live (TTL) on each Redis key
+  ## If set, Redis will expire the key after the specified duration
+  ## The TTL is refreshed on every write, so the key only expires
+  ## if no new data arrives within the configured period
+  ## Disabled by default (no expiry)
+  # expire = ""
+
   ## Enable attempt to convert string fields to numeric values
   ## If "false" or in case the string value cannot be converted the string
   ## field will be dropped.

plugins/outputs/redistimeseries/redistimeseries.go

@@ -22,17 +22,25 @@ import (
 var sampleConfig string
 
 type RedisTimeSeries struct {
-	Address             string          `toml:"address"`
-	Username            config.Secret   `toml:"username"`
-	Password            config.Secret   `toml:"password"`
-	Database            int             `toml:"database"`
-	ConvertStringFields bool            `toml:"convert_string_fields"`
-	Timeout             config.Duration `toml:"timeout"`
-	Log                 telegraf.Logger `toml:"-"`
+	Address             string           `toml:"address"`
+	Username            config.Secret    `toml:"username"`
+	Password            config.Secret    `toml:"password"`
+	Database            int              `toml:"database"`
+	ConvertStringFields bool             `toml:"convert_string_fields"`
+	Timeout             config.Duration  `toml:"timeout"`
+	Expire              *config.Duration `toml:"expire"`
+	Log                 telegraf.Logger  `toml:"-"`
 	tls.ClientConfig
 	client *redis.Client
 }
 
+func (r *RedisTimeSeries) Init() error {
+	if r.Expire != nil && time.Duration(*r.Expire) <= 0 {
+		return errors.New("expire must be a positive duration")
+	}
+	return nil
+}
+
 func (r *RedisTimeSeries) Connect() error {
 	if r.Address == "" {
 		return errors.New("redis address must be specified")
@@ -104,9 +112,18 @@ func (r *RedisTimeSeries) Write(metrics []telegraf.Metric) error {
 				}
 			}
 
-			resp := r.client.TSAddWithArgs(ctx, key, m.Time().UnixMilli(), value, &redis.TSOptions{Labels: m.Tags()})
-			if err := resp.Err(); err != nil {
-				return fmt.Errorf("adding sample %q failed: %w", key, err)
+			if r.Expire != nil {
+				pipe := r.client.Pipeline()
+				pipe.TSAddWithArgs(ctx, key, m.Time().UnixMilli(), value, &redis.TSOptions{Labels: m.Tags()})
+				pipe.Expire(ctx, key, time.Duration(*r.Expire))
+				if _, err := pipe.Exec(ctx); err != nil {
+					return fmt.Errorf("writing sample %q with expiry failed: %w", key, err)
+				}
+			} else {
+				resp := r.client.TSAddWithArgs(ctx, key, m.Time().UnixMilli(), value, &redis.TSOptions{Labels: m.Tags()})
+				if err := resp.Err(); err != nil {
+					return fmt.Errorf("adding sample %q failed: %w", key, err)
+				}
 			}
 		}
 	}

plugins/outputs/redistimeseries/redistimeseries_test.go

@@ -161,8 +161,12 @@ func getAllRecords(testContext context.Context, address string) []string {
 		}
 
 		result := client.TSRange(ctx, key, 0, int(time.Now().UnixMilli()))
+		var expires string
+		if client.TTL(ctx, key).Val() > 0 {
+			expires = "; expires"
+		}
 		for _, point := range result.Val() {
-			records = append(records, fmt.Sprintf("%s: %f %d%s", result.Args()[1], point.Value, point.Timestamp, labels))
+			records = append(records, fmt.Sprintf("%s: %f %d%s%s", result.Args()[1], point.Value, point.Timestamp, labels, expires))
 		}
 	}
 

plugins/outputs/redistimeseries/sample.conf

@@ -11,6 +11,13 @@
   ## Timeout for operations such as ping or sending metrics
   # timeout = "10s"
 
+  ## Set a time-to-live (TTL) on each Redis key
+  ## If set, Redis will expire the key after the specified duration
+  ## The TTL is refreshed on every write, so the key only expires
+  ## if no new data arrives within the configured period
+  ## Disabled by default (no expiry)
+  # expire = ""
+
   ## Enable attempt to convert string fields to numeric values
   ## If "false" or in case the string value cannot be converted the string
   ## field will be dropped.

plugins/outputs/redistimeseries/testcases/expire/expected.out

@@ -0,0 +1,6 @@
+weather_temperature: 23.100000 1696489223000 location=somewhere; expires
+weather_humidity: 52.300000 1696489223000 location=somewhere; expires
+weather_windspeed: 3.200000 1696489223000 location=somewhere; expires
+weather_temperature: 23.200000 1696489223100 location=somewhere; expires
+weather_humidity: 52.100000 1696489223100 location=somewhere; expires
+weather_windspeed: 13.200000 1696489223100 location=somewhere; expires

plugins/outputs/redistimeseries/testcases/expire/input.influx

@@ -0,0 +1,2 @@
+weather,location=somewhere temperature=23.1,humidity=52.3,windspeed=3.2 1696489223000000000
+weather,location=somewhere temperature=23.2,humidity=52.1,windspeed=13.2 1696489223100000000

plugins/outputs/redistimeseries/testcases/expire/telegraf.conf

@@ -0,0 +1,3 @@
+[[outputs.redistimeseries]]
+  address = "127.0.0.1:6379"
+  expire = "60s"