Add base64decode operation to string processor (#6740)

plugins/processors/strings/README.md

@@ -12,6 +12,7 @@ Implemented functions are:
 - trim_suffix
 - replace
 - left
+- base64decode
 
 Please note that in this implementation these are processed in the order that they appear above.
 
@@ -68,6 +69,10 @@ If you'd like to apply multiple processings to the same `tag_key` or `field_key`
   # [[processors.strings.left]]
   #   field = "message"
   #   width = 10
+
+  ## Decode a base64 encoded utf-8 string
+  # [[processors.strings.base64decode]]
+  #   field = "message"
 ```
 
 #### Trim, TrimLeft, TrimRight

plugins/processors/strings/strings.go

@@ -1,23 +1,26 @@
 package strings
 
 import (
+	"encoding/base64"
 	"strings"
 	"unicode"
+	"unicode/utf8"
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/plugins/processors"
 )
 
 type Strings struct {
-	Lowercase  []converter `toml:"lowercase"`
+	Lowercase    []converter `toml:"lowercase"`
-	Uppercase  []converter `toml:"uppercase"`
+	Uppercase    []converter `toml:"uppercase"`
-	Trim       []converter `toml:"trim"`
+	Trim         []converter `toml:"trim"`
-	TrimLeft   []converter `toml:"trim_left"`
+	TrimLeft     []converter `toml:"trim_left"`
-	TrimRight  []converter `toml:"trim_right"`
+	TrimRight    []converter `toml:"trim_right"`
-	TrimPrefix []converter `toml:"trim_prefix"`
+	TrimPrefix   []converter `toml:"trim_prefix"`
-	TrimSuffix []converter `toml:"trim_suffix"`
+	TrimSuffix   []converter `toml:"trim_suffix"`
-	Replace    []converter `toml:"replace"`
+	Replace      []converter `toml:"replace"`
-	Left       []converter `toml:"left"`
+	Left         []converter `toml:"left"`
+	Base64Decode []converter `toml:"base64decode"`
 
 	converters []converter
 	init       bool
@@ -86,6 +89,10 @@ const sampleConfig = `
   # [[processors.strings.left]]
   #   field = "message"
   #   width = 10
+
+  ## Decode a base64 encoded utf-8 string
+  # [[processors.strings.base64decode]]
+  #   field = "message"
 `
 
 func (s *Strings) SampleConfig() string {
@@ -288,6 +295,20 @@ func (s *Strings) initOnce() {
 		}
 		s.converters = append(s.converters, c)
 	}
+	for _, c := range s.Base64Decode {
+		c := c
+		c.fn = func(s string) string {
+			data, err := base64.StdEncoding.DecodeString(s)
+			if err != nil {
+				return s
+			}
+			if utf8.Valid(data) {
+				return string(data)
+			}
+			return s
+		}
+		s.converters = append(s.converters, c)
+	}
 
 	s.init = true
 }

plugins/processors/strings/strings_test.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/metric"
+	"github.com/influxdata/telegraf/testutil"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -892,3 +893,110 @@ func TestMeasurementCharDeletion(t *testing.T) {
 	assert.Equal(t, "foofoofoo", results[1].Name(), "Should have refused to delete the whole string")
 	assert.Equal(t, "barbarbar", results[2].Name(), "Should not have changed the input")
 }
+
+func TestBase64Decode(t *testing.T) {
+	tests := []struct {
+		name     string
+		plugin   *Strings
+		metric   []telegraf.Metric
+		expected []telegraf.Metric
+	}{
+		{
+			name: "base64decode success",
+			plugin: &Strings{
+				Base64Decode: []converter{
+					{
+						Field: "message",
+					},
+				},
+			},
+			metric: []telegraf.Metric{
+				testutil.MustMetric(
+					"cpu",
+					map[string]string{},
+					map[string]interface{}{
+						"message": "aG93ZHk=",
+					},
+					time.Unix(0, 0),
+				),
+			},
+			expected: []telegraf.Metric{
+				testutil.MustMetric(
+					"cpu",
+					map[string]string{},
+					map[string]interface{}{
+						"message": "howdy",
+					},
+					time.Unix(0, 0),
+				),
+			},
+		},
+		{
+			name: "base64decode not valid base64 returns original string",
+			plugin: &Strings{
+				Base64Decode: []converter{
+					{
+						Field: "message",
+					},
+				},
+			},
+			metric: []telegraf.Metric{
+				testutil.MustMetric(
+					"cpu",
+					map[string]string{},
+					map[string]interface{}{
+						"message": "_not_base64_",
+					},
+					time.Unix(0, 0),
+				),
+			},
+			expected: []telegraf.Metric{
+				testutil.MustMetric(
+					"cpu",
+					map[string]string{},
+					map[string]interface{}{
+						"message": "_not_base64_",
+					},
+					time.Unix(0, 0),
+				),
+			},
+		},
+		{
+			name: "base64decode not valid utf-8 returns original string",
+			plugin: &Strings{
+				Base64Decode: []converter{
+					{
+						Field: "message",
+					},
+				},
+			},
+			metric: []telegraf.Metric{
+				testutil.MustMetric(
+					"cpu",
+					map[string]string{},
+					map[string]interface{}{
+						"message": "//5oAG8AdwBkAHkA",
+					},
+					time.Unix(0, 0),
+				),
+			},
+			expected: []telegraf.Metric{
+				testutil.MustMetric(
+					"cpu",
+					map[string]string{},
+					map[string]interface{}{
+						"message": "//5oAG8AdwBkAHkA",
+					},
+					time.Unix(0, 0),
+				),
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			actual := tt.plugin.Apply(tt.metric...)
+			testutil.RequireMetricsEqual(t, tt.expected, actual)
+		})
+	}
+}