[e2e] parse check run (#15362)

* [e2e] move parsing common code

* [e2e] ♻️ rename metricPayload to metricAggregator

* [e2e] ♻️ move common code out of metric aggregator

* [e2e] ✨ add check run aggregator

* [e2e] more generic aggregator

test/new-e2e/utils/payloadtesthelper/checkRunAggregator.go

@@ -0,0 +1,47 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016-present Datadog, Inc.
+
+package payloadtesthelper
+
+import (
+	"encoding/json"
+)
+
+type checkRun struct {
+	Check     string   `json:"check"`
+	HostName  string   `json:"host_name"`
+	Timestamp int      `json:"timestamp"`
+	Status    int      `json:"status"`
+	Message   string   `json:"message"`
+	Tags      []string `json:"tags"`
+}
+
+func (cr *checkRun) name() string {
+	return cr.Check
+}
+
+func (cr *checkRun) tags() []string {
+	return cr.Tags
+}
+
+func parseCheckRunPayload(data []byte) (checks []*checkRun, err error) {
+	enflated, err := enflate(data)
+	if err != nil {
+		return nil, err
+	}
+	checks = []*checkRun{}
+	err = json.Unmarshal(enflated, &checks)
+	return checks, err
+}
+
+type CheckRunAggregator struct {
+	Aggregator[*checkRun]
+}
+
+func NewCheckRunAggregator() CheckRunAggregator {
+	return CheckRunAggregator{
+		Aggregator: newAggregator(parseCheckRunPayload),
+	}
+}

test/new-e2e/utils/payloadtesthelper/checkRunAggregator_test.go

@@ -0,0 +1,41 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016-present Datadog, Inc.
+
+package payloadtesthelper
+
+import (
+	_ "embed"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+//go:embed fixtures/api_v1_check_run.txt
+var checkRunBody []byte
+
+func TestCheckRun(t *testing.T) {
+	t.Run("UnmarshallPayloads", func(t *testing.T) {
+		agg := NewCheckRunAggregator()
+		err := agg.UnmarshallPayloads(checkRunBody)
+		assert.NoError(t, err)
+		assert.Equal(t, 4, len(agg.payloadsByName))
+	})
+
+	t.Run("ContainsCheckName", func(t *testing.T) {
+		agg := NewCheckRunAggregator()
+		err := agg.UnmarshallPayloads(checkRunBody)
+		assert.NoError(t, err)
+		assert.True(t, agg.ContainsPayloadName("datadog.agent.up"))
+		assert.False(t, agg.ContainsPayloadName("invalid.check.name"))
+	})
+
+	t.Run("ContainsCheckNameAndTags", func(t *testing.T) {
+		agg := NewCheckRunAggregator()
+		err := agg.UnmarshallPayloads(checkRunBody)
+		assert.NoError(t, err)
+		assert.True(t, agg.ContainsPayloadNameAndTags("snmp.can_check", []string{"snmp_device:192.168.0.3", "snmp_host:41ba948911b9"}))
+		assert.False(t, agg.ContainsPayloadNameAndTags("snmp.can_check", []string{"invalid:tag"}))
+	})
+}

test/new-e2e/utils/payloadtesthelper/common.go

@@ -0,0 +1,113 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016-present Datadog, Inc.
+
+package payloadtesthelper
+
+import (
+	"bytes"
+	"compress/zlib"
+	"encoding/json"
+	"io/ioutil"
+)
+
+type GetPayloadResponse struct {
+	Payloads [][]byte `json:"payloads"`
+}
+
+type PayloadItem interface {
+	name() string
+	tags() []string
+}
+
+type parseFunc[P PayloadItem] func(data []byte) (items []P, err error)
+
+type Aggregator[P PayloadItem] struct {
+	payloadsByName map[string][]P
+	parse          parseFunc[P]
+}
+
+func newAggregator[P PayloadItem](parse parseFunc[P]) Aggregator[P] {
+	return Aggregator[P]{
+		payloadsByName: map[string][]P{},
+		parse:          parse,
+	}
+}
+
+func (agg *Aggregator[P]) UnmarshallPayloads(body []byte) error {
+	response := GetPayloadResponse{}
+
+	err := json.Unmarshal(body, &response)
+	if err != nil {
+		return err
+	}
+
+	// reset map
+	agg.payloadsByName = map[string][]P{}
+	// build map
+	for _, data := range response.Payloads {
+		payloads, err := agg.parse(data)
+		if err != nil {
+			return err
+		}
+		for _, item := range payloads {
+			if _, found := agg.payloadsByName[item.name()]; !found {
+				agg.payloadsByName[item.name()] = []P{}
+			}
+			agg.payloadsByName[item.name()] = append(agg.payloadsByName[item.name()], item)
+		}
+	}
+
+	return nil
+}
+
+func (agg *Aggregator[P]) ContainsPayloadName(name string) bool {
+	_, found := agg.payloadsByName[name]
+	return found
+}
+
+func (agg *Aggregator[P]) ContainsPayloadNameAndTags(name string, tags []string) bool {
+	payloads, found := agg.payloadsByName[name]
+	if !found {
+		return false
+	}
+
+	for _, payloadItem := range payloads {
+		if areTagsSubsetOfOtherTags(tags, payloadItem.tags()) {
+			return true
+		}
+	}
+
+	return false
+}
+
+func areTagsSubsetOfOtherTags(tags, otherTags []string) bool {
+	otherTagsSet := tagsToSet(otherTags)
+	for _, tag := range tags {
+		if _, found := otherTagsSet[tag]; !found {
+			return false
+		}
+	}
+	return true
+}
+
+func tagsToSet(tags []string) map[string]struct{} {
+	tagsSet := map[string]struct{}{}
+	for _, tag := range tags {
+		tagsSet[tag] = struct{}{}
+	}
+	return tagsSet
+}
+
+func enflate(payload []byte) (enflated []byte, err error) {
+	re, err := zlib.NewReader(bytes.NewReader(payload))
+	if err != nil {
+		return nil, err
+	}
+	enflated, err = ioutil.ReadAll(re)
+	if err != nil {
+		return nil, err
+	}
+	return enflated, nil
+}

test/new-e2e/utils/payloadtesthelper/fixtures/api_v1_check_run.txt

@@ -0,0 +1 @@
+{"payloads":["eJy01M+L4jAUB/D/5Z1raKpWk+te9qLrXSTE9G0tNj9o0l1kmP99aEYZHGEMQq5J+76f98rr/g3UCdUZOHijHVHSiM+DAk7WB2GkRuDQ4L9OIf/1Z7ObbZfVrt4ytvnNKasIrdekJHMoIHQafZDaAaf1almtypqVBfggw+iBlwVo9F62U8Hpcdl64HuQLZogprTv9aG4BkeGd/KRAEWEiyvwHhRvYuEFPUq2WDNKj+x24Qb7t+uRt2hw6NRssGPAAQ7vxddQGhlkY1sSjSSeims/9wN6ZL04jZjBJ+C9xARHOiP8xaiXoqtn0Wbs+4ytP82/tW6Cy/oN5skQDP/tcM6KWaRieiubrJJlqmTaGXGSpukxK6hOBSk3ZoSsyuSlHd30elYLTbVo1Ha4ZLUkb3Rnf3KM7qX0p2sc/2eHjwAAAP//oyssnw=="]}

test/new-e2e/utils/payloadtesthelper/metricAggregator.go

@@ -0,0 +1,52 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016-present Datadog, Inc.
+
+package payloadtesthelper
+
+import (
+	metricspb "github.com/DataDog/agent-payload/v5/gogen"
+)
+
+type MetricSeries struct {
+	// embed proto Metric Series struct
+	metricspb.MetricPayload_MetricSeries
+}
+
+func (mp *MetricSeries) name() string {
+	return mp.Metric
+}
+
+func (mp *MetricSeries) tags() []string {
+	return mp.Tags
+}
+
+func parseMetricSeries(data []byte) (metrics []*MetricSeries, err error) {
+	enflated, err := enflate(data)
+	if err != nil {
+		return nil, err
+	}
+	metricsPayload := new(metricspb.MetricPayload)
+	err = metricsPayload.Unmarshal(enflated)
+	if err != nil {
+		return nil, err
+	}
+
+	metrics = []*MetricSeries{}
+	for _, serie := range metricsPayload.Series {
+		metrics = append(metrics, &MetricSeries{MetricPayload_MetricSeries: *serie})
+	}
+
+	return metrics, err
+}
+
+type MetricAggregator struct {
+	Aggregator[*MetricSeries]
+}
+
+func NewMetricAggregator() MetricAggregator {
+	return MetricAggregator{
+		Aggregator: newAggregator(parseMetricSeries),
+	}
+}

test/new-e2e/utils/payloadtesthelper/metricPayloads_test.go → test/new-e2e/utils/payloadtesthelper/metricAggregator_test.go

@@ -17,25 +17,25 @@ var metricsBody []byte
 
 func TestNewMetricPayloads(t *testing.T) {
 	t.Run("UnmarshallPayloads", func(t *testing.T) {
-		mp := NewMetricPayloads()
-		err := mp.UnmarshallPayloads(metricsBody)
+		agg := NewMetricAggregator()
+		err := agg.UnmarshallPayloads(metricsBody)
 		assert.NoError(t, err)
-		assert.Equal(t, 74, len(mp.metricsByName))
+		assert.Equal(t, 74, len(agg.payloadsByName))
 	})
 
 	t.Run("ContainsMetricName", func(t *testing.T) {
-		mp := NewMetricPayloads()
-		err := mp.UnmarshallPayloads(metricsBody)
+		agg := NewMetricAggregator()
+		err := agg.UnmarshallPayloads(metricsBody)
 		assert.NoError(t, err)
-		assert.True(t, mp.ContainsMetricName("system.uptime"))
-		assert.False(t, mp.ContainsMetricName("invalid.name"))
+		assert.True(t, agg.ContainsPayloadName("system.uptime"))
+		assert.False(t, agg.ContainsPayloadName("invalid.name"))
 	})
 
 	t.Run("ContainsMetricName", func(t *testing.T) {
-		mp := NewMetricPayloads()
-		err := mp.UnmarshallPayloads(metricsBody)
+		agg := NewMetricAggregator()
+		err := agg.UnmarshallPayloads(metricsBody)
 		assert.NoError(t, err)
-		assert.True(t, mp.ContainsMetricNameAndTags("datadog.agent.python.version", []string{"python_version:3", "agent_version_major:7"}))
-		assert.False(t, mp.ContainsMetricNameAndTags("datadog.agent.python.version", []string{"python_version:3", "invalid:tag"}))
+		assert.True(t, agg.ContainsPayloadNameAndTags("datadog.agent.python.version", []string{"python_version:3", "agent_version_major:7"}))
+		assert.False(t, agg.ContainsPayloadNameAndTags("datadog.agent.python.version", []string{"python_version:3", "invalid:tag"}))
 	})
 }