metrics/lib/promscrape/discovery/marathon/marathon.go

package marathon

import (
	"flag"
	"fmt"
	"net"
	"strconv"
	"time"

	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutil"
	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutil"
	"github.com/VictoriaMetrics/VictoriaMetrics/lib/proxy"
)

// SDCheckInterval defines interval for targets refresh.
var SDCheckInterval = flag.Duration("promscrape.marathonSDCheckInterval", 30*time.Second, "Interval for checking for changes in Marathon REST API. "+
	"This works only if marathon_sd_configs is configured in '-promscrape.config' file. "+
	"See https://docs.victoriametrics.com/victoriametrics/sd_configs/#marathon_sd_configs for details")

// SDConfig is the configuration for Marathon service discovery.
type SDConfig struct {
	Servers []string `yaml:"servers"`

	HTTPClientConfig  promauth.HTTPClientConfig  `yaml:",inline"`
	ProxyURL          *proxy.URL                 `yaml:"proxy_url,omitempty"`
	ProxyClientConfig promauth.ProxyClientConfig `yaml:",inline"`
}

var configMap = discoveryutil.NewConfigMap()

// GetLabels returns Marathon labels according to sdc.
func (sdc *SDConfig) GetLabels(baseDir string) ([]*promutil.Labels, error) {
	ac, err := getAPIConfig(sdc, baseDir)
	if err != nil {
		return nil, fmt.Errorf("cannot get API config: %w", err)
	}

	apps, err := GetAppsList(ac)
	if err != nil {
		return nil, err
	}
	return getAppsLabels(apps), nil
}

// MustStop stops further usage for sdc.
func (sdc *SDConfig) MustStop() {
	_ = configMap.Delete(sdc)
}

// getAppsLabels takes an array of Marathon apps and converts them into labels.
func getAppsLabels(apps *AppList) []*promutil.Labels {
	ms := make([]*promutil.Labels, 0, len(apps.Apps))
	for _, a := range apps.Apps {
		ms = append(ms, getAppLabels(&a)...)
	}
	return ms
}

func getAppLabels(app *app) []*promutil.Labels {
	m := promutil.NewLabels(5)

	m.Add("__meta_marathon_app", app.ID)
	m.Add("__meta_marathon_image", app.Container.Docker.Image)

	var ports []uint32
	var labels []map[string]string
	var prefix string

	switch {
	case len(app.Container.PortMappings) != 0:
		// In Marathon 1.5.x the "container.docker.portMappings" object was moved
		// to "container.portMappings".
		ports, labels = extractPortMapping(app.Container.PortMappings, app.isContainerNet())
		prefix = "__meta_marathon_port_mapping_label_"

	case len(app.Container.Docker.PortMappings) != 0:
		// Prior to Marathon 1.5 the port mappings could be found at the path
		// "container.docker.portMappings".
		ports, labels = extractPortMapping(app.Container.Docker.PortMappings, app.isContainerNet())
		prefix = "__meta_marathon_port_mapping_label_"

	case len(app.PortDefinitions) != 0:
		// PortDefinitions deprecates the "ports" array and can be used to specify
		// a list of ports with metadata in case a mapping is not required.
		ports = make([]uint32, len(app.PortDefinitions))
		labels = make([]map[string]string, len(app.PortDefinitions))

		for i := range app.PortDefinitions {
			labels[i] = app.PortDefinitions[i].Labels
			// When requirePorts is false, this port becomes the 'servicePort', not the listen port.
			// In this case, the port needs to be taken from the task instead of the app.
			if app.RequirePorts {
				ports[i] = app.PortDefinitions[i].Port
			}
		}

		prefix = "__meta_marathon_port_definition_label_"
	}

	for ln, lv := range app.Labels {
		m.Add("__meta_marathon_app_label_"+discoveryutil.SanitizeLabelName(ln), lv)
	}

	labelss := make([]*promutil.Labels, 0, len(app.Tasks))

	// Gather info about the app's 'tasks'. Each instance (container) is considered a task
	// and can be reachable at one or more host:port endpoints.
	for _, t := range app.Tasks {
		mm := m.Clone()

		// There are no labels to gather if only Ports is defined. (eg. with host networking)
		// Ports can only be gathered from the Task (not from the app) and are guaranteed
		// to be the same across all tasks. If we haven't gathered any ports by now,
		// use the task's ports as the port list.
		if len(ports) == 0 && len(t.Ports) != 0 {
			ports = t.Ports
		}

		// Iterate over the ports we gathered using one of the methods above.
		for i, port := range ports {
			// A zero port here means that either the portMapping has a zero port defined,
			// or there is a portDefinition with requirePorts set to false. This means the port
			// is auto-generated by Mesos and needs to be looked up in the task.
			if port == 0 && len(t.Ports) == len(ports) {
				port = t.Ports[i]
			}

			// Each port represents a possible Prometheus target.
			targetAddress := targetEndpoint(&t, port, app.isContainerNet())
			mm.Add("__address__", targetAddress)
			mm.Add("__meta_marathon_task", t.ID)
			mm.Add("__meta_marathon_port_index", strconv.Itoa(i))

			// Gather all port labels and set them on the current target, skip if the port has no Marathon labels.
			// This will happen in the host networking case with only `ports` defined, where
			// it is inefficient to allocate a list of possibly hundreds of empty label maps per host port.
			if len(labels) > 0 {
				for ln, lv := range labels[i] {
					mm.Add(prefix+discoveryutil.SanitizeLabelName(ln), lv)
				}
			}
		}
		labelss = append(labelss, mm)
	}

	return labelss
}

// targetEndpoint Generate a target endpoint string in host:port format.
func targetEndpoint(task *task, port uint32, containerNet bool) string {
	var host string

	// Use the task's ipAddress field when it's in a container network
	if containerNet && len(task.IPAddresses) > 0 {
		host = task.IPAddresses[0].IPAddress
	} else {
		host = task.Host
	}

	return net.JoinHostPort(host, strconv.Itoa(int(port)))
}

// extractPortMapping Get a list of ports and a list of labels from a PortMapping.
func extractPortMapping(portMappings []portMapping, containerNet bool) ([]uint32, []map[string]string) {
	ports := make([]uint32, len(portMappings))
	labels := make([]map[string]string, len(portMappings))

	for i := range portMappings {
		labels[i] = portMappings[i].Labels

		if containerNet {
			// If the app is in a container network, connect directly to the container port.
			ports[i] = portMappings[i].ContainerPort
		} else {
			// Otherwise, connect to the allocated host port for the container.
			// Note that this host port is likely set to 0 in the app definition, which means it is
			// automatically generated and needs to be extracted from the task's 'ports' array at a later stage.
			ports[i] = portMappings[i].HostPort
		}
	}

	return ports, labels
}