[extension][ecsobserver] Init ECS service discovery using docker label

extension/observer/ecsobserver/Makefile

@@ -0,0 +1 @@
+include ../../../Makefile.Common

extension/observer/ecsobserver/config.go

@@ -0,0 +1,105 @@
+// Copyright  OpenTelemetry 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 ecsobserver
+
+import (
+	"os"
+	"time"
+
+	"go.opentelemetry.io/collector/config/configmodels"
+)
+
+const (
+	DefaultRefreshInterval             = 30 * time.Second
+	DefaultJobLabelName                = "prometheus_job"
+	AWSRegionEnvKey                    = "AWS_REGION"
+	DefaultDockerLabelMatcherPortLabel = "ECS_PROMETHEUS_EXPORTER_PORT"
+)
+
+type Config struct {
+	configmodels.ExtensionSettings `mapstructure:",squash" yaml:",inline"`
+
+	// ClusterName is the target ECS cluster name for service discovery.
+	ClusterName string `mapstructure:"cluster_name" yaml:"cluster_name"`
+	// ClusterRegion is the target ECS cluster's AWS region.
+	ClusterRegion string `mapstructure:"cluster_region" yaml:"cluster_region"`
+	// RefreshInterval determines how frequency at which the observer
+	// needs to poll for collecting information about new processes.
+	RefreshInterval time.Duration `mapstructure:"refresh_interval" yaml:"refresh_interval"`
+	// ResultFile is the output path of the discovered targets YAML file (optional).
+	// This is mainly used in conjunction with the Prometheus receiver.
+	ResultFile string `mapstructure:"result_file" yaml:"result_file"`
+	// JobLabelName is the override for prometheus job label, using `job` literal will cause error
+	// in otel prometheus receiver. See https://github.com/open-telemetry/opentelemetry-collector/issues/575
+	JobLabelName string `mapstructure:"job_label_name" yaml:"job_label_name"`
+	// Services is a list of service name patterns for filtering tasks.
+	Services []ServiceConfig `mapstructure:"services" yaml:"services"`
+	// TaskDefinitions is a list of task definition arn patterns for filtering tasks.
+	TaskDefinitions []TaskDefinitionConfig `mapstructure:"task_definitions" yaml:"task_definitions"`
+	// DockerLabels is a list of docker labels for filtering containers within tasks.
+	DockerLabels []DockerLabelConfig `mapstructure:"docker_labels" yaml:"docker_labels"`
+}
+
+// DefaultConfig only applies docker label
+func DefaultConfig() Config {
+	return Config{
+		ExtensionSettings: configmodels.ExtensionSettings{
+			TypeVal: typeStr,
+			NameVal: string(typeStr),
+		},
+		ClusterName:     "default",
+		ClusterRegion:   os.Getenv(AWSRegionEnvKey),
+		ResultFile:      "/etc/ecs_sd_targets.yaml",
+		RefreshInterval: DefaultRefreshInterval,
+		JobLabelName:    DefaultJobLabelName,
+		DockerLabels: []DockerLabelConfig{
+			{
+				PortLabel: DefaultDockerLabelMatcherPortLabel,
+			},
+		},
+	}
+}
+
+// ExampleConfig returns an example instance that matches testdata/config_example.yaml.
+// It can be used to validate if the struct tags like mapstructure, yaml are working properly.
+func ExampleConfig() Config {
+	return Config{
+		ClusterName:     "ecs-sd-test-1",
+		ClusterRegion:   "us-west-2",
+		ResultFile:      "/etc/ecs_sd_targets.yaml",
+		RefreshInterval: 15 * time.Second,
+		JobLabelName:    DefaultJobLabelName,
+		Services: []ServiceConfig{
+			{
+				NamePattern: "^retail-.*$",
+			},
+		},
+		TaskDefinitions: []TaskDefinitionConfig{
+			{
+				CommonExporterConfig: CommonExporterConfig{
+					JobName:      "task_def_1",
+					MetricsPath:  "/not/metrics",
+					MetricsPorts: []int{9113, 9090},
+				},
+				ArnPattern: ".*:task-definition/nginx:[0-9]+",
+			},
+		},
+		DockerLabels: []DockerLabelConfig{
+			{
+				PortLabel: "ECS_PROMETHEUS_EXPORTER_PORT",
+			},
+		},
+	}
+}

extension/observer/ecsobserver/config_test.go

@@ -0,0 +1,73 @@
+// Copyright  OpenTelemetry 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 ecsobserver
+
+import (
+	"path"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/config/configmodels"
+	"go.opentelemetry.io/collector/config/configtest"
+)
+
+func TestLoadConfig(t *testing.T) {
+	factories, err := componenttest.NopFactories()
+	assert.NoError(t, err)
+
+	factory := NewFactory()
+	factories.Extensions[typeStr] = factory
+	cfg, err := configtest.LoadConfigFile(t, path.Join(".", "testdata", "config.yaml"), factories)
+
+	require.Nil(t, err)
+	require.NotNil(t, cfg)
+
+	require.Len(t, cfg.Extensions, 4)
+
+	// Default
+	ext0 := cfg.Extensions["ecs_observer"]
+	assert.Equal(t, factory.CreateDefaultConfig(), ext0)
+
+	// Merge w/ Default
+	ext1 := cfg.Extensions["ecs_observer/1"]
+	assert.Equal(t, DefaultConfig().ClusterName, ext1.(*Config).ClusterName)
+	assert.NotEqual(t, DefaultConfig().ClusterRegion, ext1.(*Config).ClusterRegion)
+	assert.Equal(t, "my_prometheus_job", ext1.(*Config).JobLabelName)
+
+	// Example Config
+	ext2 := cfg.Extensions["ecs_observer/2"]
+	ext2Expected := ExampleConfig()
+	ext2Expected.ExtensionSettings = configmodels.ExtensionSettings{
+		TypeVal: "ecs_observer",
+		NameVal: "ecs_observer/2",
+	}
+	assert.Equal(t, &ext2Expected, ext2)
+
+	// Override docker label from default
+	ext3 := cfg.Extensions["ecs_observer/3"]
+	ext3Expected := DefaultConfig()
+	ext3Expected.ExtensionSettings = configmodels.ExtensionSettings{
+		TypeVal: "ecs_observer",
+		NameVal: "ecs_observer/3",
+	}
+	ext3Expected.DockerLabels = []DockerLabelConfig{
+		{
+			PortLabel: "IS_NOT_DEFAULT",
+		},
+	}
+	assert.Equal(t, &ext3Expected, ext3)
+}

extension/observer/ecsobserver/docker_label.go

@@ -0,0 +1,111 @@
+// Copyright  OpenTelemetry 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 ecsobserver
+
+import (
+	"fmt"
+	"strconv"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ecs"
+	"go.uber.org/zap"
+)
+
+// DockerLabelConfig matches all tasks based on their docker label.
+//
+// NOTE: it's possible to make DockerLabelConfig part of CommonExporterConfig
+// and use it both ServiceConfig and TaskDefinitionConfig.
+// However, based on existing users, few people mix different types of filters.
+// If that usecase arises in the future, we can rewrite the top level docker lable filter
+// using a task definition filter with arn_pattern:*.
+type DockerLabelConfig struct {
+	CommonExporterConfig `mapstructure:",squash" yaml:",inline"`
+
+	// PortLabel is mandetory, empty string means docker label based match is skipped.
+	PortLabel        string `mapstructure:"port_label" yaml:"port_label"`
+	JobNameLabel     string `mapstructure:"job_name_label" yaml:"job_name_label"`
+	MetricsPathLabel string `mapstructure:"metrics_path_label" yaml:"metrics_path_label"`
+}
+
+func (d *DockerLabelConfig) Init() error {
+	// It's possible to support it in the future, but for now just fail at config,
+	// so user don't need to wonder which port is used in the exported target.
+	if len(d.MetricsPorts) != 0 {
+		return fmt.Errorf("metrics_ports is not supported in docker_labels, got %v", d.MetricsPorts)
+	}
+	if d.PortLabel == "" {
+		return fmt.Errorf("port_label is empty")
+	}
+	return nil
+}
+
+func (d *DockerLabelConfig) NewMatcher(options MatcherOptions) (Matcher, error) {
+	return &DockerLabelMatcher{
+		logger: options.Logger,
+		cfg:    *d,
+	}, nil
+}
+
+func dockerLabelConfigToMatchers(cfgs []DockerLabelConfig) []MatcherConfig {
+	var matchers []MatcherConfig
+	for _, cfg := range cfgs {
+		// NOTE: &cfg points to the temp var, whose value would end up be the last one in the slice.
+		copied := cfg
+		matchers = append(matchers, &copied)
+	}
+	return matchers
+}
+
+type DockerLabelMatcher struct {
+	logger *zap.Logger
+	cfg    DockerLabelConfig
+}
+
+func (d *DockerLabelMatcher) Type() MatcherType {
+	return MatcherTypeDockerLabel
+}
+
+func (d *DockerLabelMatcher) MatchTargets(t *Task, c *ecs.ContainerDefinition) ([]MatchedTarget, error) {
+	portLabel := d.cfg.PortLabel
+
+	// Only check port label
+	ps, ok := c.DockerLabels[portLabel]
+	if !ok {
+		return nil, errNotMatched
+	}
+
+	// Convert port
+	s := aws.StringValue(ps)
+	port, err := strconv.Atoi(s)
+	if err != nil {
+		return nil, fmt.Errorf("invalid port_label value, container=%s labelKey=%s labelValue=%s: %w",
+			aws.StringValue(c.Name), d.cfg.PortLabel, s, err)
+	}
+	// Export only one target based on docker port label.
+	target := MatchedTarget{
+		Port: port,
+	}
+	if v, ok := c.DockerLabels[d.cfg.MetricsPathLabel]; ok {
+		target.MetricsPath = aws.StringValue(v)
+	}
+	if v, ok := c.DockerLabels[d.cfg.JobNameLabel]; ok {
+		target.Job = aws.StringValue(v)
+	}
+	// NOTE: we only override job name but keep port and metrics from docker label instead of using common export config.
+	if d.cfg.JobName != "" {
+		target.Job = d.cfg.JobName
+	}
+	return []MatchedTarget{target}, nil
+}