prometheus/discovery/xds/kuma_test.go

// Copyright 2021 The Prometheus Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package xds

import (
	"context"
	"errors"
	"fmt"
	"testing"
	"time"

	v3 "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/common/model"
	"github.com/stretchr/testify/require"
	"google.golang.org/protobuf/proto"
	"google.golang.org/protobuf/types/known/anypb"
	"gopkg.in/yaml.v2"

	"github.com/prometheus/prometheus/discovery"
	"github.com/prometheus/prometheus/discovery/targetgroup"
)

var (
	kumaConf = sdConf

	testKumaMadsV1Resources = []*MonitoringAssignment{
		{
			Mesh:    "metrics",
			Service: "prometheus",
			Targets: []*MonitoringAssignment_Target{
				{
					Name:        "prometheus-01",
					Scheme:      "http",
					Address:     "10.1.4.32:9090",
					MetricsPath: "/custom-metrics",
					Labels: map[string]string{
						"commit_hash": "620506a88",
					},
				},
				{
					Name:    "prometheus-02",
					Scheme:  "http",
					Address: "10.1.4.33:9090",
					Labels: map[string]string{
						"commit_hash": "3513bba00",
					},
				},
			},
			Labels: map[string]string{
				"kuma.io/zone": "us-east-1",
				"team":         "infra",
			},
		},
		{
			Mesh:    "metrics",
			Service: "grafana",
			Targets: []*MonitoringAssignment_Target{},
			Labels: map[string]string{
				"kuma.io/zone": "us-east-1",
				"team":         "infra",
			},
		},
		{
			Mesh:    "data",
			Service: "elasticsearch",
			Targets: []*MonitoringAssignment_Target{
				{
					Name:    "elasticsearch-01",
					Scheme:  "http",
					Address: "10.1.1.1",
					Labels: map[string]string{
						"role": "ml",
					},
				},
			},
		},
	}
)

func getKumaMadsV1DiscoveryResponse(resources ...*MonitoringAssignment) (*v3.DiscoveryResponse, error) {
	serialized := make([]*anypb.Any, len(resources))
	for i, res := range resources {
		data, err := proto.Marshal(res)
		if err != nil {
			return nil, err
		}

		serialized[i] = &anypb.Any{
			TypeUrl: KumaMadsV1ResourceTypeURL,
			Value:   data,
		}
	}
	return &v3.DiscoveryResponse{
		TypeUrl:   KumaMadsV1ResourceTypeURL,
		Resources: serialized,
	}, nil
}

func newKumaTestHTTPDiscovery(c KumaSDConfig) (*fetchDiscovery, error) {
	reg := prometheus.NewRegistry()
	refreshMetrics := discovery.NewRefreshMetrics(reg)
	// TODO(ptodev): Add the ability to unregister refresh metrics.
	metrics := c.NewDiscovererMetrics(reg, refreshMetrics)
	err := metrics.Register()
	if err != nil {
		return nil, err
	}

	kd, err := NewKumaHTTPDiscovery(&c, nopLogger, metrics)
	if err != nil {
		return nil, err
	}

	pd, ok := kd.(*fetchDiscovery)
	if !ok {
		return nil, errors.New("not a fetchDiscovery")
	}
	return pd, nil
}

func TestKumaMadsV1ResourceParserInvalidTypeURL(t *testing.T) {
	resources := make([]*anypb.Any, 0)
	groups, err := kumaMadsV1ResourceParser(resources, "type.googleapis.com/some.api.v1.Monitoring")
	require.Nil(t, groups)
	require.Error(t, err)
}

func TestKumaMadsV1ResourceParserEmptySlice(t *testing.T) {
	resources := make([]*anypb.Any, 0)
	groups, err := kumaMadsV1ResourceParser(resources, KumaMadsV1ResourceTypeURL)
	require.Empty(t, groups)
	require.NoError(t, err)
}

func TestKumaMadsV1ResourceParserValidResources(t *testing.T) {
	res, err := getKumaMadsV1DiscoveryResponse(testKumaMadsV1Resources...)
	require.NoError(t, err)

	targets, err := kumaMadsV1ResourceParser(res.Resources, KumaMadsV1ResourceTypeURL)
	require.NoError(t, err)
	require.Len(t, targets, 3)

	expectedTargets := []model.LabelSet{
		{
			"__address__":                    "10.1.4.32:9090",
			"__metrics_path__":               "/custom-metrics",
			"__scheme__":                     "http",
			"instance":                       "prometheus-01",
			"__meta_kuma_mesh":               "metrics",
			"__meta_kuma_service":            "prometheus",
			"__meta_kuma_label_team":         "infra",
			"__meta_kuma_label_kuma_io_zone": "us-east-1",
			"__meta_kuma_label_commit_hash":  "620506a88",
			"__meta_kuma_dataplane":          "prometheus-01",
		},
		{
			"__address__":                    "10.1.4.33:9090",
			"__metrics_path__":               "",
			"__scheme__":                     "http",
			"instance":                       "prometheus-02",
			"__meta_kuma_mesh":               "metrics",
			"__meta_kuma_service":            "prometheus",
			"__meta_kuma_label_team":         "infra",
			"__meta_kuma_label_kuma_io_zone": "us-east-1",
			"__meta_kuma_label_commit_hash":  "3513bba00",
			"__meta_kuma_dataplane":          "prometheus-02",
		},
		{
			"__address__":            "10.1.1.1",
			"__metrics_path__":       "",
			"__scheme__":             "http",
			"instance":               "elasticsearch-01",
			"__meta_kuma_mesh":       "data",
			"__meta_kuma_service":    "elasticsearch",
			"__meta_kuma_label_role": "ml",
			"__meta_kuma_dataplane":  "elasticsearch-01",
		},
	}
	require.Equal(t, expectedTargets, targets)
}

func TestKumaMadsV1ResourceParserInvalidResources(t *testing.T) {
	data, err := protoJSONMarshalOptions.Marshal(&MonitoringAssignment_Target{})
	require.NoError(t, err)

	resources := []*anypb.Any{{
		TypeUrl: KumaMadsV1ResourceTypeURL,
		Value:   data,
	}}
	groups, err := kumaMadsV1ResourceParser(resources, KumaMadsV1ResourceTypeURL)
	require.Nil(t, groups)

	require.ErrorContains(t, err, "cannot parse")
}

func TestNewKumaHTTPDiscovery(t *testing.T) {
	kd, err := newKumaTestHTTPDiscovery(kumaConf)
	require.NoError(t, err)
	require.NotNil(t, kd)

	resClient, ok := kd.client.(*HTTPResourceClient)
	require.True(t, ok)
	require.Equal(t, kumaConf.Server, resClient.Server())
	require.Equal(t, KumaMadsV1ResourceTypeURL, resClient.ResourceTypeURL())
	require.Equal(t, kumaConf.ClientID, resClient.ID())
	require.Equal(t, KumaMadsV1ResourceType, resClient.config.ResourceType)

	kd.metrics.Unregister()
}

func TestKumaHTTPDiscoveryRefresh(t *testing.T) {
	s := createTestHTTPServer(t, func(request *v3.DiscoveryRequest) (*v3.DiscoveryResponse, error) {
		if request.VersionInfo == "1" {
			return nil, nil
		}

		res, err := getKumaMadsV1DiscoveryResponse(testKumaMadsV1Resources...)
		require.NoError(t, err)

		res.VersionInfo = "1"
		res.Nonce = "abc"

		return res, nil
	})
	defer s.Close()

	cfgString := fmt.Sprintf(`
---
server: %s
refresh_interval: 10s
tls_config:
  insecure_skip_verify: true
`, s.URL)

	var cfg KumaSDConfig
	require.NoError(t, yaml.Unmarshal([]byte(cfgString), &cfg))

	kd, err := newKumaTestHTTPDiscovery(cfg)
	require.NoError(t, err)
	require.NotNil(t, kd)

	ch := make(chan []*targetgroup.Group, 1)
	kd.poll(context.Background(), ch)

	groups := <-ch
	require.Len(t, groups, 1)

	targets := groups[0].Targets
	require.Len(t, targets, 3)

	expectedTargets := []model.LabelSet{
		{
			"__address__":                    "10.1.4.32:9090",
			"__metrics_path__":               "/custom-metrics",
			"__scheme__":                     "http",
			"instance":                       "prometheus-01",
			"__meta_kuma_mesh":               "metrics",
			"__meta_kuma_service":            "prometheus",
			"__meta_kuma_label_team":         "infra",
			"__meta_kuma_label_kuma_io_zone": "us-east-1",
			"__meta_kuma_label_commit_hash":  "620506a88",
			"__meta_kuma_dataplane":          "prometheus-01",
		},
		{
			"__address__":                    "10.1.4.33:9090",
			"__metrics_path__":               "",
			"__scheme__":                     "http",
			"instance":                       "prometheus-02",
			"__meta_kuma_mesh":               "metrics",
			"__meta_kuma_service":            "prometheus",
			"__meta_kuma_label_team":         "infra",
			"__meta_kuma_label_kuma_io_zone": "us-east-1",
			"__meta_kuma_label_commit_hash":  "3513bba00",
			"__meta_kuma_dataplane":          "prometheus-02",
		},
		{
			"__address__":            "10.1.1.1",
			"__metrics_path__":       "",
			"__scheme__":             "http",
			"instance":               "elasticsearch-01",
			"__meta_kuma_mesh":       "data",
			"__meta_kuma_service":    "elasticsearch",
			"__meta_kuma_label_role": "ml",
			"__meta_kuma_dataplane":  "elasticsearch-01",
		},
	}
	require.Equal(t, expectedTargets, targets)

	// Should skip the next update.
	ctx, cancel := context.WithCancel(context.Background())
	go func() {
		time.Sleep(1 * time.Second)
		cancel()
	}()

	kd.poll(ctx, ch)
	select {
	case <-ctx.Done():
		return
	case <-ch:
		require.Fail(t, "no update expected")
	}

	kd.metrics.Unregister()
}