dcos_metadata_test.go

package dcos_metadata

import (
	"testing"
	"time"

	"github.com/influxdata/telegraf"
	"github.com/influxdata/telegraf/internal"
	"github.com/influxdata/telegraf/metric"
	"github.com/stretchr/testify/assert"
)

type testCase struct {
	fixture                    string
	whitelist, whitelistPrefix []string
	inputs                     []telegraf.Metric
	expected                   []telegraf.Metric
	// cachedContainers prepopulates the plugin with container info
	cachedContainers map[string]containerInfo
	// containers is how the dm.containers map should look after
	// metrics are retrieved
	containers map[string]containerInfo
}

var (
	TEST_CASES = []testCase{
		// No metrics, no state; nothing to do
		{
			fixture:  "empty",
			inputs:   []telegraf.Metric{},
			expected: []telegraf.Metric{},
		},
		// One metric, cached state; tags are added
		{
			fixture:         "normal",
			whitelistPrefix: []string{"DCOS_METRICS_"},
			inputs: []telegraf.Metric{
				newMetric("test",
					map[string]string{"container_id": "abc123"},
					map[string]interface{}{"value": int64(1)},
					time.Now(),
				),
			},
			expected: []telegraf.Metric{
				newMetric("test",
					map[string]string{
						"container_id":  "abc123",
						"service_name":  "framework",
						"executor_name": "executor",
						"task_name":     "task",
						// Generated by mesos task labels:
						"FOO": "bar",
						"BAZ": "qux",
					},
					map[string]interface{}{"value": int64(1)},
					time.Now(),
				),
			},
			cachedContainers: map[string]containerInfo{
				"abc123": {"abc123", "task", "executor", "framework",
					map[string]string{"FOO": "bar", "BAZ": "qux"}},
			},
			containers: map[string]containerInfo{
				"abc123": {"abc123", "task", "executor", "framework",
					map[string]string{"FOO": "bar", "BAZ": "qux"}},
			},
		},
		// One metric, no cached state; no tags are added but state is updated (no additional whitelisted tags)
		{
			fixture: "fresh",
			inputs: []telegraf.Metric{
				newMetric("test",
					map[string]string{"container_id": "abc123"},
					map[string]interface{}{"value": int64(1)},
					time.Now(),
				),
			},
			expected: []telegraf.Metric{
				newMetric("test",
					// We don't expect tags, since no cache exists
					map[string]string{"container_id": "abc123"},
					map[string]interface{}{"value": int64(1)},
					time.Now(),
				),
			},
			cachedContainers: map[string]containerInfo{},
			// We do expect the cache to be updated when apply is done
			containers: map[string]containerInfo{
				"abc123": {"abc123", "task", "executor", "framework",
					// No whitelist/whitelisted prefixes configured
					map[string]string{}},
			},
		},
		// One metric, no cached state; no tags are added but state is updated (with prefix-whitelisted tags)
		{
			fixture:         "fresh",
			whitelistPrefix: []string{"DCOS_METRICS_"},
			inputs: []telegraf.Metric{
				newMetric("test",
					map[string]string{"container_id": "abc123"},
					map[string]interface{}{"value": int64(1)},
					time.Now(),
				),
			},
			expected: []telegraf.Metric{
				newMetric("test",
					// We don't expect tags, since no cache exists
					map[string]string{"container_id": "abc123"},
					map[string]interface{}{"value": int64(1)},
					time.Now(),
				),
			},
			cachedContainers: map[string]containerInfo{},
			// We do expect the cache to be updated when apply is done
			containers: map[string]containerInfo{
				"abc123": {"abc123", "task", "executor", "framework",
					// Ensure that the tags are picked up from state, including whitelisted DCOS_METRICS_-prefixed ones
					map[string]string{"FOO": "bar", "BAZ": "qux"}},
			},
		},
		// One metric, no cached state; no tags are added but state is updated (with a whitelisted tag,
		// no prefix-whitelisted tags)
		{
			fixture:   "fresh",
			whitelist: []string{"WHITELISTED_METRIC"},
			inputs: []telegraf.Metric{
				newMetric("test",
					map[string]string{"container_id": "abc123"},
					map[string]interface{}{"value": int64(1)},
					time.Now(),
				),
			},
			expected: []telegraf.Metric{
				newMetric("test",
					// We don't expect tags, since no cache exists
					map[string]string{"container_id": "abc123"},
					map[string]interface{}{"value": int64(1)},
					time.Now(),
				),
			},
			cachedContainers: map[string]containerInfo{},
			// We do expect the cache to be updated when apply is done
			containers: map[string]containerInfo{
				"abc123": {"abc123", "task", "executor", "framework",
					// Ensure that the tags are picked up from state, including whitelisted "WHITELISTED_METRIC" tag
					map[string]string{"WHITELISTED_METRIC": "foobar"}},
			},
		},
		// One metric, no cached state; no tags are added but state is updated (
		// with both whitelisted and prefix-whitelisted tags)
		{
			fixture:         "fresh",
			whitelist:       []string{"WHITELISTED_METRIC"},
			whitelistPrefix: []string{"DCOS_METRICS_"},
			inputs: []telegraf.Metric{
				newMetric("test",
					map[string]string{"container_id": "abc123"},
					map[string]interface{}{"value": int64(1)},
					time.Now(),
				),
			},
			expected: []telegraf.Metric{
				newMetric("test",
					// We don't expect tags, since no cache exists
					map[string]string{"container_id": "abc123"},
					map[string]interface{}{"value": int64(1)},
					time.Now(),
				),
			},
			cachedContainers: map[string]containerInfo{},
			// We do expect the cache to be updated when apply is done
			containers: map[string]containerInfo{
				"abc123": {"abc123", "task", "executor", "framework",
					// Ensure that the tags are picked up from state, including all whitelisted ones
					map[string]string{"FOO": "bar", "BAZ": "qux", "WHITELISTED_METRIC": "foobar"}},
			},
		},
		// One metric without a container ID; nothing to do
		{
			fixture:         "unrelated",
			whitelistPrefix: []string{"DCOS_METRICS_"},
			inputs: []telegraf.Metric{
				newMetric("test",
					map[string]string{}, // no container_id tag
					map[string]interface{}{"value": int64(1)},
					time.Now(),
				),
			},
			expected: []telegraf.Metric{
				newMetric("test",
					map[string]string{},
					map[string]interface{}{"value": int64(1)},
					time.Now(),
				),
			},
			cachedContainers: map[string]containerInfo{},
			// We do not expect the cache to be updated
			containers: map[string]containerInfo{},
		},
		// Fetching a nested container ID; not cached
		{
			fixture:         "nested",
			whitelistPrefix: []string{"DCOS_METRICS_"},
			inputs: []telegraf.Metric{
				newMetric("test",
					map[string]string{"container_id": "abc123"},
					map[string]interface{}{"value": int64(1)},
					time.Now(),
				),
			},
			expected: []telegraf.Metric{
				newMetric("test",
					map[string]string{"container_id": "abc123"},
					map[string]interface{}{"value": int64(1)},
					time.Now(),
				),
			},
			cachedContainers: map[string]containerInfo{},
			// We do expect the cache to be updated when apply is done
			// Parent container (executor) is fetched along with task
			containers: map[string]containerInfo{
				"abc123": {"abc123", "task", "executor", "framework",
					map[string]string{}},
				"xyz123": {"xyz123", "", "executor", "framework",
					nil},
			},
		},
		// Fetching a nested container ID; cached
		{
			fixture:         "nested",
			whitelistPrefix: []string{"DCOS_METRICS_"},
			inputs: []telegraf.Metric{
				newMetric("test",
					map[string]string{"container_id": "abc123"},
					map[string]interface{}{"value": int64(1)},
					time.Now(),
				),
				newMetric("test",
					map[string]string{"container_id": "xyz123"},
					map[string]interface{}{"value": int64(1)},
					time.Now(),
				),
			},
			expected: []telegraf.Metric{
				newMetric("test",
					map[string]string{
						"container_id":  "abc123",
						"service_name":  "framework",
						"executor_name": "executor",
						"task_name":     "task",
					},
					map[string]interface{}{"value": int64(1)},
					time.Now(),
				),
				newMetric("test",
					map[string]string{
						"container_id":  "xyz123",
						"service_name":  "framework",
						"executor_name": "executor",
						"task_name":     "",
					},
					map[string]interface{}{"value": int64(1)},
					time.Now(),
				),
			},
			cachedContainers: map[string]containerInfo{
				"abc123": {"abc123", "task", "executor", "framework",
					map[string]string{}},
				"xyz123": {"xyz123", "", "executor", "framework",
					nil},
			},
			// We do not expect the cache to be updated
			containers: map[string]containerInfo{
				"abc123": {"abc123", "task", "executor", "framework",
					map[string]string{}},
				"xyz123": {"xyz123", "", "executor", "framework",
					nil},
			},
		},
		// No executor;
		{
			fixture:         "noexecutor",
			whitelistPrefix: []string{"DCOS_METRICS_"},
			inputs: []telegraf.Metric{
				newMetric("test",
					map[string]string{"container_id": "abc123"},
					map[string]interface{}{"value": int64(1)},
					time.Now(),
				),
			},
			expected: []telegraf.Metric{
				newMetric("test",
					map[string]string{
						"container_id": "abc123",
						"service_name": "framework",
						// no executor tag at all
						"task_name": "task",
					},
					map[string]interface{}{"value": int64(1)},
					time.Now(),
				),
			},
			cachedContainers: map[string]containerInfo{
				"abc123": {"abc123", "task", "", "framework",
					map[string]string{}},
			},
			containers: map[string]containerInfo{
				"abc123": {"abc123", "task", "", "framework",
					map[string]string{}},
			},
		},
	}
)

func TestApply(t *testing.T) {
	for _, tc := range TEST_CASES {
		t.Run(tc.fixture, func(t *testing.T) {
			server := startTestServer(t, tc.fixture)
			defer server.Close()

			dm := DCOSMetadata{
				MesosAgentUrl:   server.URL,
				Timeout:         internal.Duration{Duration: 500 * time.Millisecond},
				RateLimit:       internal.Duration{Duration: 50 * time.Millisecond},
				Whitelist:       tc.whitelist,
				WhitelistPrefix: tc.whitelistPrefix,
				containers:      tc.cachedContainers,
			}

			outputs := dm.Apply(tc.inputs...)

			// No metrics were dropped
			assert.Equal(t, len(tc.expected), len(outputs))
			// Tags were added as expected
			for i, actual := range outputs {
				expected := tc.expected[i]
				assert.Equal(t, expected.Name(), actual.Name())
				assert.Equal(t, expected.Tags(), actual.Tags())
			}

			waitForContainersToEqual(t, &dm, tc.containers, 2*time.Second)
		})
	}
}

func TestGetClient(t *testing.T) {
	dm := DCOSMetadata{}
	client1, err1 := dm.getClient()
	client2, err2 := dm.getClient()
	assert.Nil(t, err1)
	assert.Nil(t, err2)
	assert.Equal(t, client1, client2)
}

// newMetric is a convenience method which allows us to define test cases at
// package level without doing error handling
func newMetric(name string, tags map[string]string, fields map[string]interface{}, tm time.Time) telegraf.Metric {
	m, err := metric.New(name, tags, fields, tm)
	if err != nil {
		panic(err)
	}
	return m
}

// waitForContainersToEqual waits for the length of the container cache to
// change and asserts that they equal the expected, or times out
func waitForContainersToEqual(t *testing.T, dm *DCOSMetadata, expected map[string]containerInfo, timeout time.Duration) {
	done := make(chan bool)
	go func() {
		for {
			// acquiring the lock here avoids triggering the go race detector
			dm.mu.Lock()
			if len(dm.containers) == len(expected) {
				done <- true
				break
			}
			dm.mu.Unlock()
			time.Sleep(10 * time.Millisecond)
		}
	}()

	select {
	case <-done:
		assert.Equal(t, expected, dm.containers)
		return
	case <-time.After(timeout):
		assert.Fail(t, "Timed out waiting for a container update")
		return
	}
}