Add resource type and resource label support to stackdriver output (#5391)

plugins/outputs/stackdriver/README.md

@@ -7,6 +7,10 @@ Requires `project` to specify where Stackdriver metrics will be delivered to.
 
 Metrics are grouped by the `namespace` variable and metric key - eg: `custom.googleapis.com/telegraf/system/load5`
 
+[Resource type](https://cloud.google.com/monitoring/api/resources) is configured by the `resource_type` variable (default `global`).
+
+Additional resource labels can be configured by `resource_labels`. By default the required `project_id` label is always set to the `project` variable.
+
 ### Configuration
 
 ```toml
@@ -16,6 +20,15 @@ Metrics are grouped by the `namespace` variable and metric key - eg: `custom.goo
 
   # The namespace for the metric descriptor
   namespace = "telegraf"
+
+  # Custom resource type
+  resource_type = "generic_node"
+
+# Additonal resource labels
+[outputs.stackdriver.resource_labels]
+  node_id = "$HOSTNAME"
+  namespace = "myapp"
+  location = "eu-north0"
 ```
 
 ### Restrictions

plugins/outputs/stackdriver/stackdriver.go

@@ -21,6 +21,8 @@ import (
 type Stackdriver struct {
 	Project        string
 	Namespace      string
+	ResourceType   string            `toml:"resource_type"`
+	ResourceLabels map[string]string `toml:"resource_labels"`
 
 	client *monitoring.MetricClient
 }
@@ -47,11 +49,21 @@ const (
 )
 
 var sampleConfig = `
+  [[outputs.stackdriver]]
     # GCP Project
     project = "erudite-bloom-151019"
 
     # The namespace for the metric descriptor
     namespace = "telegraf"
+
+    # Custom resource type
+    resource_type = "generic_node"
+
+  # Additonal resource labels
+  [outputs.stackdriver.resource_labels]
+    node_id = "$HOSTNAME"
+    namespace = "myapp"
+    location = "eu-north0"
 `
 
 // Connect initiates the primary connection to the GCP project.
@@ -64,6 +76,16 @@ func (s *Stackdriver) Connect() error {
 		return fmt.Errorf("Namespace is a required field for stackdriver output")
 	}
 
+	if s.ResourceType == "" {
+		s.ResourceType = "global"
+	}
+
+	if s.ResourceLabels == nil {
+		s.ResourceLabels = make(map[string]string, 1)
+	}
+
+	s.ResourceLabels["project_id"] = s.Project
+
 	if s.client == nil {
 		ctx := context.Background()
 		client, err := monitoring.NewMetricClient(ctx)
@@ -137,10 +159,8 @@ func (s *Stackdriver) Write(metrics []telegraf.Metric) error {
 					},
 					MetricKind: metricKind,
 					Resource: &monitoredrespb.MonitoredResource{
-						Type: "global",
-						Labels: map[string]string{
-							"project_id": s.Project,
-						},
+						Type:   s.ResourceType,
+						Labels: s.ResourceLabels,
 					},
 					Points: []*monitoringpb.Point{
 						dataPoint,

plugins/outputs/stackdriver/stackdriver_test.go

@@ -97,6 +97,42 @@ func TestWrite(t *testing.T) {
 	require.NoError(t, err)
 	err = s.Write(testutil.MockMetrics())
 	require.NoError(t, err)
+
+	request := mockMetric.reqs[0].(*monitoringpb.CreateTimeSeriesRequest)
+	require.Equal(t, request.TimeSeries[0].Resource.Type, "global")
+	require.Equal(t, request.TimeSeries[0].Resource.Labels["project_id"], "projects/[PROJECT]")
+}
+
+func TestWriteResourceTypeAndLabels(t *testing.T) {
+	expectedResponse := &emptypb.Empty{}
+	mockMetric.err = nil
+	mockMetric.reqs = nil
+	mockMetric.resps = append(mockMetric.resps[:0], expectedResponse)
+
+	c, err := monitoring.NewMetricClient(context.Background(), clientOpt)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	s := &Stackdriver{
+		Project:      fmt.Sprintf("projects/%s", "[PROJECT]"),
+		Namespace:    "test",
+		ResourceType: "foo",
+		ResourceLabels: map[string]string{
+			"mylabel": "myvalue",
+		},
+		client: c,
+	}
+
+	err = s.Connect()
+	require.NoError(t, err)
+	err = s.Write(testutil.MockMetrics())
+	require.NoError(t, err)
+
+	request := mockMetric.reqs[0].(*monitoringpb.CreateTimeSeriesRequest)
+	require.Equal(t, request.TimeSeries[0].Resource.Type, "foo")
+	require.Equal(t, request.TimeSeries[0].Resource.Labels["project_id"], "projects/[PROJECT]")
+	require.Equal(t, request.TimeSeries[0].Resource.Labels["mylabel"], "myvalue")
 }
 
 func TestWriteAscendingTime(t *testing.T) {