feat(custom-monitored-resource): impl, tests

Author: zdunecki

.gitignore

@@ -0,0 +1 @@
+*.iml

stackdriver.go

@@ -55,6 +55,8 @@ type Sink struct {
 	prefix    string
 	taskInfo  *taskInfo
 
+	monitoredResource *monitoredrespb.MonitoredResource
+
 	mu        sync.Mutex
 	debugLogs bool
 }
@@ -104,6 +106,15 @@ type Config struct {
 	// will log additional information that is helpful when debugging errors.
 	// Optional. Defaults to false.
 	DebugLogs bool
+
+	//MonitoredResource identifies the machine/service/resource
+	//that is monitored.
+	//Different possible settings are defined here:
+	//https://cloud.google.com/monitoring/api/resources
+	//
+	//setting a nil MonitoredResource will run
+	//a defaultMonitoredResource function.
+	MonitoredResource *monitoredrespb.MonitoredResource
 }
 
 type taskInfo struct {
@@ -123,6 +134,20 @@ type BucketFn func([]string) []float64
 // writing to stackdriver.
 type ExtractLabelsFn func([]string, string) ([]string, []metrics.Label, error)
 
+// defaultMonitoredResource returns default monitored resource
+func defaultMonitoredResource(taskInfo *taskInfo) *monitoredrespb.MonitoredResource {
+	return &monitoredrespb.MonitoredResource{
+		Type: "generic_task",
+		Labels: map[string]string{
+			"project_id": taskInfo.ProjectID,
+			"location":   taskInfo.Location,
+			"namespace":  taskInfo.Namespace,
+			"job":        taskInfo.Job,
+			"task_id":    taskInfo.TaskID,
+		},
+	}
+}
+
 // DefaultBucketer is the default BucketFn used to determing bucketing values
 // for metrics.
 func DefaultBucketer(key []string) []float64 {
@@ -211,6 +236,12 @@ func NewSink(client *monitoring.MetricClient, config *Config) *Sink {
 		s.taskInfo.TaskID = "go-" + strconv.Itoa(os.Getpid()) + "@" + hostname
 	}
 
+	if config.MonitoredResource != nil {
+		s.monitoredResource = config.MonitoredResource
+	} else {
+		s.monitoredResource = defaultMonitoredResource(s.taskInfo)
+	}
+
 	s.reset()
 
 	// run cancelable goroutine that reports on interval
@@ -291,16 +322,7 @@ func (s *Sink) report(ctx context.Context) {
 	end, rGauges, rCounters, rHistograms := s.deep()
 
 	// https://cloud.google.com/monitoring/api/resources
-	resource := &monitoredrespb.MonitoredResource{
-		Type: "generic_task",
-		Labels: map[string]string{
-			"project_id": s.taskInfo.ProjectID,
-			"location":   s.taskInfo.Location,
-			"namespace":  s.taskInfo.Namespace,
-			"job":        s.taskInfo.Job,
-			"task_id":    s.taskInfo.TaskID,
-		},
-	}
+	resource := s.monitoredResource
 
 	ts := []*monitoringpb.TimeSeries{}
 

stackdriver_test.go

@@ -14,9 +14,12 @@
 package stackdriver
 
 import (
+	"bytes"
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
+	monitoredrespb "google.golang.org/genproto/googleapis/api/monitoredres"
 	"net"
 	"sync"
 	"testing"
@@ -1091,3 +1094,84 @@ func diffCreateMsg(want, got *monitoringpb.CreateTimeSeriesRequest) string {
 	}
 	return out
 }
+
+func TestCustomMonitorResource(t *testing.T) {
+	checkLabels := func(sink *Sink, labels map[string]string) error {
+		expectedLabels := sink.monitoredResource.GetLabels()
+
+		expectedLabelsBytes, _ := json.Marshal(expectedLabels)
+		labelsBytes, _ := json.Marshal(labels)
+
+		if expectedLabelsBytes == nil || !bytes.Equal(expectedLabelsBytes, labelsBytes) {
+			return errors.New("invalid labels")
+		}
+
+		return nil
+	}
+
+	{
+		labels := map[string]string{
+			"project_id":     "project",
+			"location":       "zone",
+			"cluster_name":   "cluster",
+			"container_name": "container_name",
+			"namespace_name": "namespace_name",
+			"pod_name":       "pod_name",
+		}
+
+		sink := NewSink(nil, &Config{
+			ProjectID: "example_project",
+			Prefix:    sPtr(""),
+			MonitoredResource: &monitoredrespb.MonitoredResource{
+				Labels: labels,
+				Type:   "k8s_container",
+			},
+		})
+
+		if err := checkLabels(sink, labels); err != nil {
+			t.Error(err)
+		}
+	}
+
+	{
+		sink := NewSink(nil, &Config{
+			ProjectID: "example_project",
+			Prefix:    sPtr(""),
+		})
+
+		labels := defaultMonitoredResource(sink.taskInfo).GetLabels()
+
+		if err := checkLabels(sink, labels); err != nil {
+			t.Error(err)
+		}
+	}
+
+	{
+		labels := map[string]string{
+			"project_id":     "project",
+			"location":       "zone",
+			"cluster_name":   "cluster",
+			"container_name": "container_name",
+			"namespace_name": "namespace_name",
+			"pod_name":       "pod_name",
+		}
+
+		invalidLabels := map[string]string{
+			"project_id": "project",
+		}
+
+		sink := NewSink(nil, &Config{
+			ProjectID: "example_project",
+			Prefix:    sPtr(""),
+			MonitoredResource: &monitoredrespb.MonitoredResource{
+				Labels: labels,
+				Type:   "k8s_container",
+			},
+		})
+
+		if err := checkLabels(sink, invalidLabels); err == nil {
+			t.Error("labels should be the same")
+		}
+	}
+
+}