[corechecks] Support custom tags in every corechecks

docs/agent/config.md

@@ -75,6 +75,7 @@ advanced options in the `instance` section:
 that should run less frequently than the default 15 seconds interval
 * `empty_default_hostname`: submit metrics, events and service checks with no
 hostname when set to `true`
+* `tags`: send custom tags in addition to the tags sent by the check.
 
 ## Removed options
 

pkg/aggregator/mocksender/mocked_methods.go

@@ -64,6 +64,11 @@ func (m *MockSender) Commit() {
 	m.Called()
 }
 
+//SetCheckCustomTags enables the set of check custom tags mock call.
+func (m *MockSender) SetCheckCustomTags(tags []string) {
+	m.Called(tags)
+}
+
 //GetMetricStats enables the get metric stats mock call.
 func (m *MockSender) GetMetricStats() map[string]int64 {
 	m.Called()

pkg/aggregator/mocksender/mocksender.go

@@ -51,6 +51,7 @@ func (m *MockSender) SetupAcceptAll() {
 	m.On("Event", mock.AnythingOfType("metrics.Event")).Return()
 	m.On("GetMetricStats", mock.AnythingOfType("map[string]int64")).Return()
 	m.On("DisableDefaultHostname", mock.AnythingOfType("bool")).Return()
+	m.On("SetCheckCustomTags", mock.AnythingOfType("[]string")).Return()
 	m.On("Commit").Return()
 }
 

pkg/aggregator/sender.go

@@ -35,6 +35,7 @@ type Sender interface {
 	Event(e metrics.Event)
 	GetMetricStats() map[string]int64
 	DisableDefaultHostname(disable bool)
+	SetCheckCustomTags(tags []string)
 }
 
 type metricStats struct {
@@ -61,6 +62,7 @@ type checkSender struct {
 	smsOut                  chan<- senderMetricSample
 	serviceCheckOut         chan<- metrics.ServiceCheck
 	eventOut                chan<- metrics.Event
+	checkTags               []string
 }
 
 type senderMetricSample struct {
@@ -152,6 +154,12 @@ func (s *checkSender) DisableDefaultHostname(disable bool) {
 	s.defaultHostnameDisabled = disable
 }
 
+// SetCheckCustomTags stores the tags set in the check configuration file.
+// They will be appended to each send (metric, event and service)
+func (s *checkSender) SetCheckCustomTags(tags []string) {
+	s.checkTags = tags
+}
+
 // Commit commits the metric samples that were added during a check run
 // Should be called at the end of every check run
 func (s *checkSender) Commit() {
@@ -191,6 +199,8 @@ func (s *checkSender) SendRawMetricSample(sample *metrics.MetricSample) {
 }
 
 func (s *checkSender) sendMetricSample(metric string, value float64, hostname string, tags []string, mType metrics.MetricType) {
+	tags = append(tags, s.checkTags...)
+
 	log.Trace(mType.String(), " sample: ", metric, ": ", value, " for hostname: ", hostname, " tags: ", tags)
 
 	metricSample := &metrics.MetricSample{
@@ -267,7 +277,7 @@ func (s *checkSender) ServiceCheck(checkName string, status metrics.ServiceCheck
 		Status:    status,
 		Host:      hostname,
 		Ts:        time.Now().Unix(),
-		Tags:      tags,
+		Tags:      append(tags, s.checkTags...),
 		Message:   message,
 	}
 
@@ -284,6 +294,8 @@ func (s *checkSender) ServiceCheck(checkName string, status metrics.ServiceCheck
 
 // Event submits an event
 func (s *checkSender) Event(e metrics.Event) {
+	e.Tags = append(e.Tags, s.checkTags...)
+
 	log.Trace("Event submitted: ", e.Title, " for hostname: ", e.Host, " tags: ", e.Tags)
 
 	if e.Host == "" && !s.defaultHostnameDisabled {

pkg/aggregator/sender_test.go

@@ -10,6 +10,7 @@ import (
 	"fmt"
 	"sync"
 	"testing"
+	"time"
 
 	// 3p
 
@@ -133,6 +134,125 @@ func TestGetSenderDefaultHostname(t *testing.T) {
 	assert.Equal(t, false, checksender.defaultHostnameDisabled)
 }
 
+func TestGetSenderAddCheckCustomTagsMetrics(t *testing.T) {
+	resetAggregator()
+	InitAggregator(nil, "testhostname", "")
+
+	senderMetricSampleChan := make(chan senderMetricSample, 10)
+	serviceCheckChan := make(chan metrics.ServiceCheck, 10)
+	eventChan := make(chan metrics.Event, 10)
+	checkSender := newCheckSender(checkID1, "", senderMetricSampleChan, serviceCheckChan, eventChan)
+
+	// no custom tags
+	checkSender.sendMetricSample("metric.test", 42.0, "testhostname", nil, metrics.CounterType)
+	sms := <-senderMetricSampleChan
+	assert.Nil(t, sms.metricSample.Tags)
+
+	// only tags added by the check
+	checkTags := []string{"check:tag1", "check:tag2"}
+	checkSender.sendMetricSample("metric.test", 42.0, "testhostname", checkTags, metrics.CounterType)
+	sms = <-senderMetricSampleChan
+	assert.Equal(t, checkTags, sms.metricSample.Tags)
+
+	// simulate tags in the configuration file
+	customTags := []string{"custom:tag1", "custom:tag2"}
+	checkSender.SetCheckCustomTags(customTags)
+	assert.Len(t, checkSender.checkTags, 2)
+
+	// only tags coming from the configuration file
+	checkSender.sendMetricSample("metric.test", 42.0, "testhostname", nil, metrics.CounterType)
+	sms = <-senderMetricSampleChan
+	assert.Equal(t, customTags, sms.metricSample.Tags)
+
+	// tags added by the check + tags coming from the configuration file
+	checkSender.sendMetricSample("metric.test", 42.0, "testhostname", checkTags, metrics.CounterType)
+	sms = <-senderMetricSampleChan
+	assert.Equal(t, append(checkTags, customTags...), sms.metricSample.Tags)
+}
+
+func TestGetSenderAddCheckCustomTagsService(t *testing.T) {
+	resetAggregator()
+	InitAggregator(nil, "testhostname", "")
+
+	senderMetricSampleChan := make(chan senderMetricSample, 10)
+	serviceCheckChan := make(chan metrics.ServiceCheck, 10)
+	eventChan := make(chan metrics.Event, 10)
+	checkSender := newCheckSender(checkID1, "", senderMetricSampleChan, serviceCheckChan, eventChan)
+
+	// no custom tags
+	checkSender.ServiceCheck("test", metrics.ServiceCheckOK, "testhostname", nil, "test message")
+	sc := <-serviceCheckChan
+	assert.Nil(t, sc.Tags)
+
+	// only tags added by the check
+	checkTags := []string{"check:tag1", "check:tag2"}
+	checkSender.ServiceCheck("test", metrics.ServiceCheckOK, "testhostname", checkTags, "test message")
+	sc = <-serviceCheckChan
+	assert.Equal(t, checkTags, sc.Tags)
+
+	// simulate tags in the configuration file
+	customTags := []string{"custom:tag1", "custom:tag2"}
+	checkSender.SetCheckCustomTags(customTags)
+	assert.Len(t, checkSender.checkTags, 2)
+
+	// only tags coming from the configuration file
+	checkSender.ServiceCheck("test", metrics.ServiceCheckOK, "testhostname", nil, "test message")
+	sc = <-serviceCheckChan
+	assert.Equal(t, customTags, sc.Tags)
+
+	// tags added by the check + tags coming from the configuration file
+	checkSender.ServiceCheck("test", metrics.ServiceCheckOK, "testhostname", checkTags, "test message")
+	sc = <-serviceCheckChan
+	assert.Equal(t, append(checkTags, customTags...), sc.Tags)
+}
+
+func TestGetSenderAddCheckCustomTagsEvent(t *testing.T) {
+	resetAggregator()
+	InitAggregator(nil, "testhostname", "")
+
+	senderMetricSampleChan := make(chan senderMetricSample, 10)
+	serviceCheckChan := make(chan metrics.ServiceCheck, 10)
+	eventChan := make(chan metrics.Event, 10)
+	checkSender := newCheckSender(checkID1, "", senderMetricSampleChan, serviceCheckChan, eventChan)
+
+	event := metrics.Event{
+		Title: "title",
+		Host:  "testhostname",
+		Ts:    time.Now().Unix(),
+		Text:  "text",
+		Tags:  nil,
+	}
+
+	// no custom tags
+	checkSender.Event(event)
+	e := <-eventChan
+	assert.Nil(t, e.Tags)
+
+	// only tags added by the check
+	checkTags := []string{"check:tag1", "check:tag2"}
+	event.Tags = checkTags
+	checkSender.Event(event)
+	e = <-eventChan
+	assert.Equal(t, checkTags, e.Tags)
+
+	// simulate tags in the configuration file
+	customTags := []string{"custom:tag1", "custom:tag2"}
+	checkSender.SetCheckCustomTags(customTags)
+	assert.Len(t, checkSender.checkTags, 2)
+
+	// only tags coming from the configuration file
+	event.Tags = nil
+	checkSender.Event(event)
+	e = <-eventChan
+	assert.Equal(t, customTags, e.Tags)
+
+	// tags added by the check + tags coming from the configuration file
+	event.Tags = checkTags
+	checkSender.Event(event)
+	e = <-eventChan
+	assert.Equal(t, append(checkTags, customTags...), e.Tags)
+}
+
 func TestCheckSenderInterface(t *testing.T) {
 	senderMetricSampleChan := make(chan senderMetricSample, 10)
 	serviceCheckChan := make(chan metrics.ServiceCheck, 10)

pkg/autodiscovery/integration/config.go

@@ -52,8 +52,9 @@ type Config struct {
 
 // CommonInstanceConfig holds the reserved fields for the yaml instance data
 type CommonInstanceConfig struct {
-	MinCollectionInterval int  `yaml:"min_collection_interval"`
-	EmptyDefaultHostname  bool `yaml:"empty_default_hostname"`
+	MinCollectionInterval int      `yaml:"min_collection_interval"`
+	EmptyDefaultHostname  bool     `yaml:"empty_default_hostname"`
+	Tags                  []string `yaml:"tags"`
 }
 
 // Equal determines whether the passed config is the same

pkg/collector/corechecks/checkbase.go

@@ -32,6 +32,10 @@ import (
 // Integration warnings are handled via the Warn and Warnf methods
 // that forward the warning to the logger and send the warning to
 // the collector for display in the status page and the web UI.
+//
+// If custom tags are set in the check configuration file,
+// they will be automatically appended to each send done by this
+// check.
 type CheckBase struct {
 	checkName      string
 	checkID        check.ID
@@ -84,6 +88,17 @@ func (c *CheckBase) CommonConfigure(instance integration.Data) error {
 		}
 		s.DisableDefaultHostname(true)
 	}
+
+	// Set custom tags configured for this check
+	if commonOptions.Tags != nil && len(commonOptions.Tags) > 0 {
+		s, err := aggregator.GetSender(c.checkID)
+		if err != nil {
+			log.Errorf("failed to retrieve a sender for check %s: %s", string(c.ID()), err)
+			return err
+		}
+		s.SetCheckCustomTags(commonOptions.Tags)
+	}
+
 	return nil
 }
 

pkg/collector/corechecks/cluster/kubernetes_apiserver.go

@@ -37,7 +37,6 @@ const (
 
 // KubeASConfig is the config of the API server.
 type KubeASConfig struct {
-	Tags                     []string `yaml:"tags"`
 	CollectEvent             bool     `yaml:"collect_events"`
 	CollectOShiftQuotas      bool     `yaml:"collect_openshift_clusterquotas"`
 	FilteredEventType        []string `yaml:"filtered_event_types"`
@@ -279,7 +278,7 @@ func (k *KubeASCheck) parseComponentStatus(sender aggregator.Sender, componentsS
 			log.Debug("API Server component's structure is not expected")
 			continue
 		}
-		tagComp := append(k.instance.Tags, fmt.Sprintf("component:%s", component.Name))
+		tagComp := []string{fmt.Sprintf("component:%s", component.Name)}
 		for _, condition := range component.Conditions {
 			status_check := metrics.ServiceCheckUnknown
 
@@ -341,7 +340,6 @@ ITER_EVENTS:
 			k.Warnf("Error while formatting bundled events, %s. Not submitting", err.Error())
 			continue
 		}
-		datadogEv.Tags = append(datadogEv.Tags, k.instance.Tags...)
 		sender.Event(datadogEv)
 	}
 	return nil

pkg/collector/corechecks/cluster/kubernetes_apiserver_test.go

@@ -89,33 +89,31 @@ func TestParseComponentStatus(t *testing.T) {
 
 	// FIXME: use the factory instead
 	kubeASCheck := &KubeASCheck{
-		instance: &KubeASConfig{
-			Tags: []string{"test"},
-		},
+		instance:              &KubeASConfig{},
 		CheckBase:             core.NewCheckBase(kubernetesAPIServerCheckName),
 		KubeAPIServerHostname: "hostname",
 	}
 
 	mocked := mocksender.NewMockSender(kubeASCheck.ID())
-	mocked.On("ServiceCheck", "kube_apiserver_controlplane.up", metrics.ServiceCheckOK, "hostname", []string{"test", "component:Zookeeper"}, "")
+	mocked.On("ServiceCheck", "kube_apiserver_controlplane.up", metrics.ServiceCheckOK, "hostname", []string{"component:Zookeeper"}, "")
 	kubeASCheck.parseComponentStatus(mocked, expected)
 
 	mocked.AssertNumberOfCalls(t, "ServiceCheck", 1)
-	mocked.AssertServiceCheck(t, "kube_apiserver_controlplane.up", metrics.ServiceCheckOK, "hostname", []string{"test", "component:Zookeeper"}, "")
+	mocked.AssertServiceCheck(t, "kube_apiserver_controlplane.up", metrics.ServiceCheckOK, "hostname", []string{"component:Zookeeper"}, "")
 
 	err := kubeASCheck.parseComponentStatus(mocked, unExpected)
 	assert.EqualError(t, err, "metadata structure has changed. Not collecting API Server's Components status")
 	mocked.AssertNotCalled(t, "ServiceCheck", "kube_apiserver_controlplane.up")
 
-	mocked.On("ServiceCheck", "kube_apiserver_controlplane.up", metrics.ServiceCheckCritical, "hostname", []string{"test", "component:ETCD"}, "")
+	mocked.On("ServiceCheck", "kube_apiserver_controlplane.up", metrics.ServiceCheckCritical, "hostname", []string{"component:ETCD"}, "")
 	kubeASCheck.parseComponentStatus(mocked, unHealthy)
 	mocked.AssertNumberOfCalls(t, "ServiceCheck", 2)
-	mocked.AssertServiceCheck(t, "kube_apiserver_controlplane.up", metrics.ServiceCheckCritical, "hostname", []string{"test", "component:ETCD"}, "")
+	mocked.AssertServiceCheck(t, "kube_apiserver_controlplane.up", metrics.ServiceCheckCritical, "hostname", []string{"component:ETCD"}, "")
 
-	mocked.On("ServiceCheck", "kube_apiserver_controlplane.up", metrics.ServiceCheckUnknown, "hostname", []string{"test", "component:DCA"}, "")
+	mocked.On("ServiceCheck", "kube_apiserver_controlplane.up", metrics.ServiceCheckUnknown, "hostname", []string{"component:DCA"}, "")
 	kubeASCheck.parseComponentStatus(mocked, unknown)
 	mocked.AssertNumberOfCalls(t, "ServiceCheck", 3)
-	mocked.AssertServiceCheck(t, "kube_apiserver_controlplane.up", metrics.ServiceCheckUnknown, "hostname", []string{"test", "component:DCA"}, "")
+	mocked.AssertServiceCheck(t, "kube_apiserver_controlplane.up", metrics.ServiceCheckUnknown, "hostname", []string{"component:DCA"}, "")
 
 	empty_resp := kubeASCheck.parseComponentStatus(mocked, empty)
 	assert.Nil(t, empty_resp, "metadata structure has changed. Not collecting API Server's Components status")
@@ -159,7 +157,6 @@ func TestProcessBundledEvents(t *testing.T) {
 
 	kubeASCheck := &KubeASCheck{
 		instance: &KubeASConfig{
-			Tags:              []string{"test"},
 			FilteredEventType: []string{"ignored"},
 		},
 		CheckBase:             core.NewCheckBase(kubernetesAPIServerCheckName),
@@ -195,7 +192,7 @@ func TestProcessBundledEvents(t *testing.T) {
 		Title:          "Events from the machine-blue Node",
 		Text:           "%%% \n30 **MissingClusterDNS**: MountVolume.SetUp succeeded\n \n _Events emitted by the kubelet seen at " + time.Unix(709675200, 0).String() + "_ \n\n %%%",
 		Priority:       "normal",
-		Tags:           []string{"test", "namespace:default", "source_component:kubelet"},
+		Tags:           []string{"namespace:default", "source_component:kubelet"},
 		AggregationKey: "kubernetes_apiserver:e63e74fa-f566-11e7-9749-0e4863e1cbf4",
 		SourceTypeName: "kubernetes",
 		Ts:             709675200,
@@ -222,7 +219,7 @@ func TestProcessBundledEvents(t *testing.T) {
 		Title:          "Events from the machine-blue Node",
 		Text:           "%%% \n30 **MissingClusterDNS**: MountVolume.SetUp succeeded\n \n _Events emitted by the kubelet seen at " + time.Unix(709675200, 0).String() + "_ \n\n %%%",
 		Priority:       "normal",
-		Tags:           []string{"test", "namespace:default", "source_component:kubelet"},
+		Tags:           []string{"namespace:default", "source_component:kubelet"},
 		AggregationKey: "kubernetes_apiserver:e63e74fa-f566-11e7-9749-0e4863e1cbf4",
 		SourceTypeName: "kubernetes",
 		Ts:             709675200,
@@ -247,7 +244,6 @@ func TestProcessEvent(t *testing.T) {
 
 	kubeASCheck := &KubeASCheck{
 		instance: &KubeASConfig{
-			Tags:              []string{"test"},
 			FilteredEventType: []string{"ignored"},
 		},
 		CheckBase:             core.NewCheckBase(kubernetesAPIServerCheckName),
@@ -263,7 +259,7 @@ func TestProcessEvent(t *testing.T) {
 		Title:          "Events from the dca-789976f5d7-2ljx6 ReplicaSet",
 		Text:           "%%% \n2 **Scheduled**: Successfully assigned dca-789976f5d7-2ljx6 to ip-10-0-0-54\n \n _New events emitted by the default-scheduler seen at " + time.Unix(709662600000, 0).String() + "_ \n\n %%%",
 		Priority:       "normal",
-		Tags:           []string{"test", "source_component:default-scheduler", "namespace:default"},
+		Tags:           []string{"source_component:default-scheduler", "namespace:default"},
 		AggregationKey: "kubernetes_apiserver:e6417a7f-f566-11e7-9749-0e4863e1cbf4",
 		SourceTypeName: "kubernetes",
 		Ts:             709662600,

pkg/collector/corechecks/cluster/kubernetes_openshift.go

@@ -51,7 +51,7 @@ func (k *KubeASCheck) retrieveOShiftClusterQuotas() ([]osq.ClusterResourceQuota,
 // reportClusterQuotas reports metrics on OpenShift ClusterResourceQuota objects
 func (k *KubeASCheck) reportClusterQuotas(quotas []osq.ClusterResourceQuota, sender aggregator.Sender) {
 	for _, quota := range quotas {
-		quotaTags := append(k.instance.Tags, fmt.Sprintf("clusterquota:%s", quota.Name))
+		quotaTags := []string{fmt.Sprintf("clusterquota:%s", quota.Name)}
 		remaining := computeQuotaRemaining(quota.Status.Total.Used, quota.Status.Total.Hard)
 
 		k.reportQuota(quota.Status.Total.Hard, "openshift.clusterquota", "limit", quotaTags, sender)

pkg/collector/corechecks/cluster/kubernetes_openshift_test.go

@@ -26,7 +26,7 @@ func TestReportClusterQuotas(t *testing.T) {
 	json.Unmarshal(raw, &list)
 	require.Len(t, list.Items, 1)
 
-	var instanceCfg = []byte("tags: [customtag]")
+	var instanceCfg = []byte("")
 	var initCfg = []byte("")
 	kubeASCheck := KubernetesASFactory().(*KubeASCheck)
 	err = kubeASCheck.Configure(instanceCfg, initCfg)
@@ -38,7 +38,7 @@ func TestReportClusterQuotas(t *testing.T) {
 	mocked.AssertNumberOfCalls(t, "Gauge", 9*3)
 
 	// Total
-	expectedTags := []string{"customtag", "clusterquota:multiproj-test"}
+	expectedTags := []string{"clusterquota:multiproj-test"}
 
 	mocked.AssertMetric(t, "Gauge", "openshift.clusterquota.cpu.limit", 3.0, "", expectedTags)
 	mocked.AssertMetric(t, "Gauge", "openshift.clusterquota.cpu.used", 0.6, "", expectedTags)