fix(inputs.cisco_telemetry_mdt): Fix MDT source field overwrite (#13693)

plugins/inputs/cisco_telemetry_mdt/README.md

@@ -62,6 +62,9 @@ See the [CONFIGURATION.md][CONFIGURATION.md] for more details.
  ## Include the delete field in every telemetry message.
  # include_delete_field = false
 
+ ## Specify custom name for incoming MDT source field.
+ # source_field_name = "mdt_source"
+
  ## Define aliases to map telemetry encoding paths to simple measurement names
  [inputs.cisco_telemetry_mdt.aliases]
    ifstats = "ietf-interfaces:interfaces-state/interface/statistics"

plugins/inputs/cisco_telemetry_mdt/cisco_telemetry_mdt.go

@@ -60,6 +60,7 @@ type CiscoTelemetryMDT struct {
 	EmbeddedTags       []string              `toml:"embedded_tags"`
 	EnforcementPolicy  GRPCEnforcementPolicy `toml:"grpc_enforcement_policy"`
 	IncludeDeleteField bool                  `toml:"include_delete_field"`
+	SourceFieldName    string                `toml:"source_field_name"`
 
 	Log telegraf.Logger
 
@@ -397,6 +398,10 @@ func (c *CiscoTelemetryMDT) handleTelemetry(data []byte) {
 			for _, subfield := range keys.Fields {
 				c.parseKeyField(tags, subfield, "")
 			}
+			// If incoming MDT contains source key, copy to mdt_src
+			if _, ok := tags["source"]; ok {
+				tags[c.SourceFieldName] = tags["source"]
+			}
 		} else {
 			tags = make(map[string]string, 3)
 		}
@@ -746,6 +751,7 @@ func init() {
 		return &CiscoTelemetryMDT{
 			Transport:       "grpc",
 			ServiceAddress:  "127.0.0.1:57000",
+			SourceFieldName: "mdt_source",
 		}
 	})
 }

plugins/inputs/cisco_telemetry_mdt/cisco_telemetry_mdt_test.go

@@ -983,3 +983,52 @@ func TestGRPCDialoutKeepalive(t *testing.T) {
 	c.Stop()
 	require.NoError(t, conn.Close())
 }
+
+func TestSourceFieldRewrite(t *testing.T) {
+	c := &CiscoTelemetryMDT{Log: testutil.Logger{}, Transport: "dummy", Aliases: map[string]string{"alias": "type:model/some/path"}}
+	c.SourceFieldName = "mdt_source"
+	acc := &testutil.Accumulator{}
+	err := c.Start(acc)
+	// error is expected since we are passing in dummy transport
+	require.Error(t, err)
+
+	telemetry := &telemetryBis.Telemetry{
+		MsgTimestamp: 1543236572000,
+		EncodingPath: "type:model/some/path",
+		NodeId:       &telemetryBis.Telemetry_NodeIdStr{NodeIdStr: "hostname"},
+		Subscription: &telemetryBis.Telemetry_SubscriptionIdStr{SubscriptionIdStr: "subscription"},
+		DataGpbkv: []*telemetryBis.TelemetryField{
+			{
+				Fields: []*telemetryBis.TelemetryField{
+					{
+						Name: "keys",
+						Fields: []*telemetryBis.TelemetryField{
+							{
+								Name:        "source",
+								ValueByType: &telemetryBis.TelemetryField_StringValue{StringValue: "str"},
+							},
+						},
+					},
+					{
+						Name: "content",
+						Fields: []*telemetryBis.TelemetryField{
+							{
+								Name:        "bool",
+								ValueByType: &telemetryBis.TelemetryField_BoolValue{BoolValue: false},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+	data, err := proto.Marshal(telemetry)
+	require.NoError(t, err)
+
+	c.handleTelemetry(data)
+	require.Empty(t, acc.Errors)
+
+	tags := map[string]string{"path": "type:model/some/path", "mdt_source": "str", "source": "hostname", "subscription": "subscription"}
+	fields := map[string]interface{}{"bool": false}
+	acc.AssertContainsTaggedFields(t, "alias", fields, tags)
+}

plugins/inputs/cisco_telemetry_mdt/sample.conf

@@ -25,6 +25,9 @@
  ## Include the delete field in every telemetry message.
  # include_delete_field = false
 
+ ## Specify custom name for incoming MDT source field.
+ # source_field_name = "mdt_source"
+
  ## Define aliases to map telemetry encoding paths to simple measurement names
  [inputs.cisco_telemetry_mdt.aliases]
    ifstats = "ietf-interfaces:interfaces-state/interface/statistics"