config.go

package cloudwatch_exporter

import (
	"crypto/md5"
	"encoding/hex"
	"fmt"
	"time"

	"github.com/go-kit/log"
	yaceConf "github.com/nerdswords/yet-another-cloudwatch-exporter/pkg/config"
	yaceModel "github.com/nerdswords/yet-another-cloudwatch-exporter/pkg/model"
	"gopkg.in/yaml.v2"

	"github.com/grafana/agent/pkg/integrations"
	integrations_v2 "github.com/grafana/agent/pkg/integrations/v2"
	"github.com/grafana/agent/pkg/integrations/v2/metricsutils"
)

const (
	metricsPerQuery       = 500
	cloudWatchConcurrency = 5
	tagConcurrency        = 5
	labelsSnakeCase       = false
)

// Since we are gathering metrics from CloudWatch and writing them in prometheus during each scrape, the timestamp
// used should be the scrape one
var addCloudwatchTimestamp = false

// Avoid producing absence of values in metrics
var nilToZero = true

func init() {
	integrations.RegisterIntegration(&Config{})
	integrations_v2.RegisterLegacy(&Config{}, integrations_v2.TypeMultiplex, metricsutils.NewNamedShim("cloudwatch"))
}

// Config is the configuration for the CloudWatch metrics integration
type Config struct {
	STSRegion    string          `yaml:"sts_region"`
	FIPSDisabled bool            `yaml:"fips_disabled"`
	Discovery    DiscoveryConfig `yaml:"discovery"`
	Static       []StaticJob     `yaml:"static"`
}

// DiscoveryConfig configures scraping jobs that will auto-discover metrics dimensions for a given service.
type DiscoveryConfig struct {
	ExportedTags TagsPerNamespace `yaml:"exported_tags"`
	Jobs         []*DiscoveryJob  `yaml:"jobs"`
}

// TagsPerNamespace represents for each namespace, a list of tags that will be exported as labels in each metric.
type TagsPerNamespace map[string][]string

// DiscoveryJob configures a discovery job for a given service.
type DiscoveryJob struct {
	InlineRegionAndRoles `yaml:",inline"`
	InlineCustomTags     `yaml:",inline"`
	SearchTags           []Tag    `yaml:"search_tags"`
	Type                 string   `yaml:"type"`
	Metrics              []Metric `yaml:"metrics"`
}

// StaticJob will scrape metrics that match all defined dimensions.
type StaticJob struct {
	InlineRegionAndRoles `yaml:",inline"`
	InlineCustomTags     `yaml:",inline"`
	Name                 string      `yaml:"name"`
	Namespace            string      `yaml:"namespace"`
	Dimensions           []Dimension `yaml:"dimensions"`
	Metrics              []Metric    `yaml:"metrics"`
}

// InlineRegionAndRoles exposes for each supported job, the AWS regions and IAM roles in which the agent should perform the
// scrape.
type InlineRegionAndRoles struct {
	Regions []string `yaml:"regions"`
	Roles   []Role   `yaml:"roles"`
}

type InlineCustomTags struct {
	CustomTags []Tag `yaml:"custom_tags"`
}

type Role struct {
	RoleArn    string `yaml:"role_arn"`
	ExternalID string `yaml:"external_id"`
}

type Dimension struct {
	Name  string `yaml:"name"`
	Value string `yaml:"value"`
}

type Tag struct {
	Key   string `yaml:"key"`
	Value string `yaml:"value"`
}

type Metric struct {
	Name       string        `yaml:"name"`
	Statistics []string      `yaml:"statistics"`
	Period     time.Duration `yaml:"period"`
}

// Name returns the name of the integration this config is for.
func (c *Config) Name() string {
	return "cloudwatch_exporter"
}

func (c *Config) InstanceKey(agentKey string) (string, error) {
	return getHash(c)
}

// NewIntegration creates a new integration from the config.
func (c *Config) NewIntegration(l log.Logger) (integrations.Integration, error) {
	exporterConfig, fipsEnabled, err := ToYACEConfig(c)
	if err != nil {
		return nil, fmt.Errorf("invalid cloudwatch exporter configuration: %w", err)
	}
	return newCloudwatchExporter(c.Name(), l, exporterConfig, fipsEnabled), nil
}

// getHash calculates the MD5 hash of the yaml representation of the config
func getHash(c *Config) (string, error) {
	bytes, err := yaml.Marshal(c)
	if err != nil {
		return "", err
	}
	hash := md5.Sum(bytes)
	return hex.EncodeToString(hash[:]), nil
}

// ToYACEConfig converts a Config into YACE's config model. Note that the conversion is not direct, some values
// have been opinionated to simplify the config model the agent exposes for this integration.
// The returned boolean is whether or not AWS FIPS endpoints will be enabled.
func ToYACEConfig(c *Config) (yaceConf.ScrapeConf, bool, error) {
	discoveryJobs := []*yaceConf.Job{}
	for _, job := range c.Discovery.Jobs {
		discoveryJobs = append(discoveryJobs, toYACEDiscoveryJob(job))
	}
	staticJobs := []*yaceConf.Static{}
	for _, stat := range c.Static {
		staticJobs = append(staticJobs, toYACEStaticJob(stat))
	}
	conf := yaceConf.ScrapeConf{
		APIVersion: "v1alpha1",
		StsRegion:  c.STSRegion,
		Discovery: yaceConf.Discovery{
			ExportedTagsOnMetrics: yaceConf.ExportedTagsOnMetrics(c.Discovery.ExportedTags),
			Jobs:                  discoveryJobs,
		},
		Static: staticJobs,
	}

	// yaceSess expects a default value of True
	fipsEnabled := !c.FIPSDisabled

	// Run the exporter's config validation. Between other things, it will check that the service for which a discovery
	// job is instantiated, it's supported.
	if err := conf.Validate(); err != nil {
		return conf, fipsEnabled, err
	}
	patchYACEDefaults(&conf)

	return conf, fipsEnabled, nil
}

// patchYACEDefaults overrides some default values YACE applies after validation.
func patchYACEDefaults(yc *yaceConf.ScrapeConf) {
	// YACE doesn't allow during validation a zero-delay in each metrics scrape. Override this behaviour since it's taken
	// into account by the rounding period.
	// https://github.com/nerdswords/yet-another-cloudwatch-exporter/blob/7e5949124bb5f26353eeff298724a5897de2a2a4/pkg/config/config.go#L320
	for _, job := range yc.Discovery.Jobs {
		for _, metric := range job.Metrics {
			metric.Delay = 0
		}
	}
}

func toYACEStaticJob(job StaticJob) *yaceConf.Static {
	return &yaceConf.Static{
		Name:       job.Name,
		Regions:    job.Regions,
		Roles:      toYACERoles(job.Roles),
		Namespace:  job.Namespace,
		CustomTags: toYACETags(job.CustomTags),
		Dimensions: toYACEDimensions(job.Dimensions),
		Metrics:    toYACEMetrics(job.Metrics),
	}
}

func toYACEDimensions(dim []Dimension) []yaceConf.Dimension {
	yaceDims := []yaceConf.Dimension{}
	for _, d := range dim {
		yaceDims = append(yaceDims, yaceConf.Dimension{
			Name:  d.Name,
			Value: d.Value,
		})
	}
	return yaceDims
}

func toYACEDiscoveryJob(job *DiscoveryJob) *yaceConf.Job {
	roles := toYACERoles(job.Roles)
	yaceJob := yaceConf.Job{
		Regions:    job.Regions,
		Roles:      roles,
		CustomTags: toYACETags(job.CustomTags),
		Type:       job.Type,
		Metrics:    toYACEMetrics(job.Metrics),
		SearchTags: toYACETags(job.SearchTags),

		// By setting RoundingPeriod to nil, the exporter will align the start and end times for retrieving CloudWatch
		// metrics, with the smallest period in the retrieved batch.
		RoundingPeriod: nil,

		JobLevelMetricFields: yaceConf.JobLevelMetricFields{
			// Set to zero job-wide scraping time settings. This should be configured at the metric level to make the data
			// being fetched more explicit.
			Period:                 0,
			Length:                 0,
			Delay:                  0,
			NilToZero:              &nilToZero,
			AddCloudwatchTimestamp: &addCloudwatchTimestamp,
		},
	}
	return &yaceJob
}

func toYACEMetrics(metrics []Metric) []*yaceConf.Metric {
	yaceMetrics := []*yaceConf.Metric{}
	for _, metric := range metrics {
		periodSeconds := int64(metric.Period.Seconds())
		lengthSeconds := periodSeconds
		yaceMetrics = append(yaceMetrics, &yaceConf.Metric{
			Name:       metric.Name,
			Statistics: metric.Statistics,

			// Length dictates the size of the window for whom we request metrics, that is, endTime - startTime. Period
			// dictates the size of the buckets in which we aggregate data, inside that window. Since data will be scraped
			// by the agent every so often, dictated by the scrapedInterval, CloudWatch should return a single datapoint
			// for each requested metric. That is if Period >= Length, but is Period > Length, we will be getting not enough
			// data to fill the whole aggregation bucket. Therefore, Period == Length.
			Period: periodSeconds,
			Length: lengthSeconds,

			// Delay moves back the time window for whom CloudWatch is requested data. Since we are already adjusting
			// this with RoundingPeriod (see toYACEDiscoveryJob), we should omit this setting.
			Delay: 0,

			NilToZero:              &nilToZero,
			AddCloudwatchTimestamp: &addCloudwatchTimestamp,
		})
	}
	return yaceMetrics
}

func toYACERoles(roles []Role) []yaceConf.Role {
	yaceRoles := []yaceConf.Role{}
	// YACE defaults to an empty role, which means the environment configured role is used
	// https://github.com/nerdswords/yet-another-cloudwatch-exporter/blob/30aeceb2324763cdd024a1311045f83a09c1df36/pkg/config/config.go#L111
	if len(roles) == 0 {
		yaceRoles = append(yaceRoles, yaceConf.Role{})
	}
	for _, role := range roles {
		yaceRoles = append(yaceRoles, yaceConf.Role{
			RoleArn:    role.RoleArn,
			ExternalID: role.ExternalID,
		})
	}
	return yaceRoles
}

func toYACETags(tags []Tag) []yaceModel.Tag {
	outTags := []yaceModel.Tag{}
	for _, t := range tags {
		outTags = append(outTags, yaceModel.Tag{
			Key:   t.Key,
			Value: t.Value,
		})
	}
	return outTags
}