Merge pull request #1560 from mwringe/hawkular-label-tags

[Hawkular] Add labels as individual tags

Author: DirectXMan12

docs/sink-configuration.md

@@ -86,6 +86,7 @@ The following options are available:
   * `name` - The syntax is `name(regexp)` where MetricName is matched (such as `cpu/usage`) with a `regexp` filter
 * `batchSize`- How many metrics are sent in each request to Hawkular-Metrics (default is 1000)
 * `concurrencyLimit`- How many concurrent requests are used to send data to the Hawkular-Metrics (default is 5)
+* `labelTagPrefix` - A prefix to be placed in front of each label when stored as a tag for the metric (default is `labels.`)
 
 A combination of `insecure` / `caCert` / `auth` is not supported, only a single of these parameters is allowed at once. Also, combination of `useServiceAccount` and `user` + `pass` is not supported. To increase the performance of Hawkular sink in case of multiple instances of Hawkular-Metrics (such as scaled scenario in OpenShift) modify the parameters of batchSize and concurrencyLimit to balance the load on Hawkular-Metrics instances.
 

metrics/sinks/hawkular/client.go

@@ -174,6 +174,17 @@ func (h *hawkularSink) registerLabeledIfNecessary(ms *core.MetricSet, metric cor
 			// Set tag values
 			for k, v := range ms.Labels {
 				mdd.Tags[k] = v
+				if k == core.LabelLabels.Key {
+					labels := strings.Split(v, ",")
+					for _, label := range labels {
+						labelKeyValue := strings.Split(label, ":")
+						if len(labelKeyValue) != 2 {
+							glog.V(4).Infof("Could not split the label %v into its key and value pair. This label will not be added as a tag in Hawkular Metrics.", label)
+						} else {
+							mdd.Tags[h.labelTagPrefix+labelKeyValue[0]] = labelKeyValue[1]
+						}
+					}
+				}
 			}
 
 			// Set the labeled values

metrics/sinks/hawkular/driver.go

@@ -43,6 +43,9 @@ const (
 
 	nodeId string = "labelNodeId"
 
+	labelTagPrefixOpts    = "labelTagPrefix"
+	labelTagPrefixDefault = "labels."
+
 	defaultServiceAccountFile = "/var/run/secrets/kubernetes.io/serviceaccount/token"
 )
 
@@ -212,6 +215,12 @@ func (h *hawkularSink) init() error {
 		h.labelTenant = v[0]
 	}
 
+	if v, found := opts[labelTagPrefixOpts]; found {
+		h.labelTagPrefix = v[0]
+	} else {
+		h.labelTagPrefix = labelTagPrefixDefault
+	}
+
 	if v, found := opts[nodeId]; found {
 		h.labelNodeId = v[0]
 	}

metrics/sinks/hawkular/driver_test.go

@@ -457,6 +457,106 @@ func TestStoreTimeseries(t *testing.T) {
 	assert.NotEqual(t, ids[0], ids[1])
 }
 
+func TestTags(t *testing.T) {
+	m := &sync.Mutex{}
+	calls := make([]string, 0, 2)
+	// how many times tags have been updated
+	tagsUpdated := 0
+	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		m.Lock()
+		defer m.Unlock()
+		calls = append(calls, r.RequestURI)
+		w.Header().Set("Content-Type", "application/json")
+
+		defer r.Body.Close()
+		b, err := ioutil.ReadAll(r.Body)
+		assert.NoError(t, err)
+
+		if strings.HasSuffix(r.RequestURI, "/tags") {
+			tags := make(map[string]string)
+			err := json.Unmarshal(b, &tags)
+			assert.NoError(t, err)
+
+			assert.Equal(t, 10, len(tags))
+			assert.Equal(t, "test-label", tags["projectId"])
+			assert.Equal(t, "test-container", tags[core.LabelContainerName.Key])
+			assert.Equal(t, "test-podid", tags[core.LabelPodId.Key])
+			assert.Equal(t, "test-container/test/metric/A", tags["group_id"])
+			assert.Equal(t, "test/metric/A", tags["descriptor_name"])
+			assert.Equal(t, "XYZ", tags[core.LabelResourceID.Key])
+			assert.Equal(t, "bytes", tags["units"])
+
+			assert.Equal(t, "testLabelA:testValueA,testLabelB:testValueB", tags[core.LabelLabels.Key])
+			assert.Equal(t, "testValueA", tags["labels.testLabelA"])
+			assert.Equal(t, "testValueB", tags["labels.testLabelB"])
+
+			tagsUpdated++
+		}
+	}))
+	defer s.Close()
+
+	hSink, err := integSink(s.URL + "?tenant=test-heapster&labelToTenant=projectId")
+	assert.NoError(t, err)
+
+	l := make(map[string]string)
+	l["projectId"] = "test-label"
+	l[core.LabelContainerName.Key] = "test-container"
+	l[core.LabelPodId.Key] = "test-podid"
+	l[core.LabelLabels.Key] = "testLabelA:testValueA,testLabelB:testValueB"
+
+	labeledMetric := core.LabeledMetric{
+		Name: "test/metric/A",
+		Labels: map[string]string{
+			core.LabelResourceID.Key: "XYZ",
+		},
+		MetricValue: core.MetricValue{
+			MetricType: core.MetricGauge,
+			FloatValue: 124.456,
+		},
+	}
+
+	metricSet := core.MetricSet{
+		Labels:         l,
+		LabeledMetrics: []core.LabeledMetric{labeledMetric},
+		MetricValues: map[string]core.MetricValue{
+			"test/metric/A": {
+				ValueType:  core.ValueInt64,
+				MetricType: core.MetricCumulative,
+				IntValue:   123456,
+			},
+		},
+	}
+
+	smd := core.MetricDescriptor{
+		Name:      "test/metric/A",
+		Units:     core.UnitsBytes,
+		ValueType: core.ValueInt64,
+		Type:      core.MetricGauge,
+		Labels:    []core.LabelDescriptor{},
+	}
+
+	//register the metric definitions
+	hSink.Register([]core.MetricDescriptor{smd})
+	//register the metrics themselves
+	hSink.registerLabeledIfNecessary(&metricSet, labeledMetric)
+
+	assert.Equal(t, 1, tagsUpdated)
+
+	tags := hSink.reg["test-container/test-podid/test/metric/A/XYZ"].Tags
+	assert.Equal(t, 10, len(tags))
+	assert.Equal(t, "test-label", tags["projectId"])
+	assert.Equal(t, "test-container", tags[core.LabelContainerName.Key])
+	assert.Equal(t, "test-podid", tags[core.LabelPodId.Key])
+	assert.Equal(t, "test-container/test/metric/A", tags["group_id"])
+	assert.Equal(t, "test/metric/A", tags["descriptor_name"])
+	assert.Equal(t, "XYZ", tags[core.LabelResourceID.Key])
+	assert.Equal(t, "bytes", tags["units"])
+
+	assert.Equal(t, "testLabelA:testValueA,testLabelB:testValueB", tags[core.LabelLabels.Key])
+	assert.Equal(t, "testValueA", tags["labels.testLabelA"])
+	assert.Equal(t, "testValueB", tags["labels.testLabelB"])
+}
+
 func TestUserPass(t *testing.T) {
 	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		w.Header().Set("X-Authorization", r.Header.Get("Authorization"))

metrics/sinks/hawkular/types.go

@@ -53,10 +53,11 @@ type hawkularSink struct {
 
 	uri *url.URL
 
-	labelTenant string
-	labelNodeId string
-	modifiers   []metrics.Modifier
-	filters     []Filter
+	labelTenant    string
+	labelNodeId    string
+	labelTagPrefix string
+	modifiers      []metrics.Modifier
+	filters        []Filter
 
 	batchSize int
 }