WIP: prometheus service discovery dependency inversion

config.go

// Copyright 2016 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 discovery

import (
	"context"
	"errors"
	"reflect"
	"sort"
	"strings"

	"github.com/go-kit/kit/log"
	"github.com/prometheus/prometheus/discovery/targetgroup"
)

const configFieldPrefix = "AUTO_"

var (
	configFieldNames = make(map[reflect.Type]string)
	configFields     []reflect.StructField
	configType       reflect.Type
)

func getConfigType() reflect.Type {
	if configType == nil {
		sort.Slice(configFields, func(i, k int) bool {
			return configFields[i].Name < configFields[k].Name
		})
		configType = reflect.StructOf(configFields)
	}
	return configType
}

// Discoverer provides information about target groups. It maintains a set
// of sources from which TargetGroups can originate. Whenever a discovery provider
// detects a potential change, it sends the TargetGroup through its channel.
//
// Discoverer does not know if an actual change happened.
// It does guarantee that it sends the new TargetGroup whenever a change happens.
//
// Discoverers should initially send a full set of all discoverable TargetGroups.
type Discoverer interface {
	// Run hands a channel to the discovery provider (Consul, DNS, etc.) through which
	// it can send updated target groups. It must return when the context is canceled.
	// It should not close the update channel on returning.
	Run(ctx context.Context, up chan<- []*targetgroup.Group)
}

// DiscovererOptions provides options for a Discoverer.
type DiscovererOptions struct {
	Logger log.Logger
}

// ConfigOptions provides options for a Config.
type ConfigOptions struct {
	// WorkingDirectory may be used to resolve relative file paths
	// in the config (e.g. TLS certificates).
	WorkingDirectory string

	// TODO: are others options required...?
}

// A Config provides the configuration and constructor for a Discoverer.
type Config interface {
	Name() string
	NewDiscoverer(DiscovererOptions) (Discoverer, error)
	SetOptions(ConfigOptions)
}

// RegisterConfig registers the given Config type along with the YAML key for the
// list of its type in the Configs object.
func RegisterConfig(yamlListKey string, config Config) {
	typ := reflect.TypeOf(config)
	fieldName := configFieldPrefix + yamlListKey // must be exported
	configFieldNames[typ] = fieldName
	configFields = append(configFields, reflect.StructField{
		Name: fieldName,
		Type: reflect.SliceOf(typ),
		Tag:  reflect.StructTag(`yaml:"` + yamlListKey + `,omitempty"`),
	})
}

// Configs contains configuration for registered service discovery mechanisms.
type Configs []Config

// UnmarshalYAML implements the yaml.Unmarshaler interface.
func (c *Configs) UnmarshalYAML(unmarshal func(interface{}) error) error {
	ptr := reflect.New(getConfigType())
	if err := unmarshal(ptr.Interface()); err != nil {
		return err
	}
	obj := ptr.Elem()

	var cfgs []Config
	for i := 0; i < obj.NumField(); i++ {
		list := obj.Field(i)
		for k := 0; k < list.Len(); k++ {
			val := list.Index(k)
			if val.IsZero() || (val.Kind() == reflect.Ptr && val.Elem().IsZero()) {
				key := configFieldNames[list.Type().Elem()]
				key = strings.TrimPrefix(key, configFieldPrefix)
				return errors.New("empty or null section in " + key)
			}
			cfgs = append(cfgs, val.Interface().(Config))
		}
	}
	*c = cfgs
	return nil
}

// MarshalYAML implements the yaml.Marshaler interface.
func (c *Configs) MarshalYAML() (interface{}, error) {
	typ := getConfigType()
	if c == nil {
		return reflect.Zero(reflect.PtrTo(typ)).Interface(), nil
	}
	ptr := reflect.New(typ)
	obj := ptr.Elem()

	for _, cfg := range *c {
		name := configFieldNames[reflect.TypeOf(cfg)]
		field := obj.FieldByName(name)
		field.Set(
			reflect.Append(field, reflect.ValueOf(cfg)),
		)
	}

	return ptr.Interface(), nil
}

func init() {
	// N.B.: static_configs is the only Config type implemented by default.
	// It must be implemented here to prevent a cyclic dependency in targetgroup.
	// All other types are registered at init by their implementing packages.
	RegisterConfig("static_configs", &staticConfig{})
}

type staticConfig targetgroup.Group

func (g *staticConfig) targetGroup() *targetgroup.Group { return (*targetgroup.Group)(g) }

func (g *staticConfig) UnmarshalYAML(unmarshal func(interface{}) error) error {
	return g.targetGroup().UnmarshalYAML(unmarshal)
}
func (g *staticConfig) MarshalYAML() (interface{}, error) {
	return g.targetGroup().MarshalYAML()
}

func (g *staticConfig) Name() string { return "static" }
func (g *staticConfig) NewDiscoverer(_ DiscovererOptions) (Discoverer, error) {
	return g, nil
}
func (g *staticConfig) SetOptions(_ ConfigOptions) {}

func (g *staticConfig) Run(ctx context.Context, up chan<- []*targetgroup.Group) {
	// Apparently, sending an empty update is sometimes useful.
	var groups []*targetgroup.Group
	if g != nil {
		groups = []*targetgroup.Group{g.targetGroup()}
	}
	select {
	case <-ctx.Done():
		return
	case up <- groups:
	}
	// TODO: existing implementation closes up chan, but documentation explicitly forbids it...?
}

config_test.go

// Copyright 2016 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 discovery

import (
	"errors"
	"testing"

	"github.com/prometheus/common/model"
	"github.com/prometheus/prometheus/util/testutil"
	"gopkg.in/yaml.v2"
)

func TestConfig(t *testing.T) {
	tests := []struct {
		name string
		yaml string
		cfgs Configs
		err  error
	}{
		{
			name: "empty",
		},
		{
			name: "empty static_configs list",
			yaml: `{static_configs: []}`,
		},
		{
			name: "empty static_configs item",
			yaml: `{static_configs: [{}]}`,
			err:  errors.New("empty or null section in static_configs"),
		},
		{
			name: "static_configs",
			yaml: `
static_configs:
  - targets:
      - foo
      - bar
    labels:
      node: foobar
  - targets:
      - baz
      - qux
    labels:
      node: bazqux
`,
			cfgs: Configs{
				&staticConfig{
					Targets: []model.LabelSet{
						{model.AddressLabel: model.LabelValue("foo")},
						{model.AddressLabel: model.LabelValue("bar")},
					},
					Labels: model.LabelSet{
						"node": "foobar",
					},
				},
				&staticConfig{
					Targets: []model.LabelSet{
						{model.AddressLabel: model.LabelValue("baz")},
						{model.AddressLabel: model.LabelValue("qux")},
					},
					Labels: model.LabelSet{
						"node": "bazqux",
					},
				},
			},
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			var cfgs Configs
			err := yaml.Unmarshal([]byte(tt.yaml), &cfgs)
			testutil.ErrorEqual(t, tt.err, err, "unmarshal error")
			testutil.Equals(t, tt.cfgs, cfgs, "unmarshal result")
		})
	}
}