Add new OTel pipeline for scraping Kueue metrics and publishing to CloudWatch

.github/workflows/PR-build.yml

@@ -68,72 +68,3 @@ jobs:
       - name: Check license and imports
         if: needs.changes.outputs.lint == 'true'
         run: make simple-lint
-
-  build:
-    needs: [lint, changes]
-    name: Build ${{ matrix.os }}
-    runs-on: ${{ matrix.os }}
-    strategy:
-      fail-fast: false
-      matrix:
-        os: [ ubuntu-latest, windows-2019, windows-latest, macos-12]
-        include:
-          - os: ubuntu-latest
-            family: linux
-            cache-path: |
-              ~/.cache/go-build
-              ~/go/pkg/mod
-          - os: macos-12
-            family: darwin
-            cache-path: |
-              ~/Library/Caches/go-build
-              ~/go/pkg/mod
-          - os: windows-2019
-            family: windows
-            cache-path: |
-              ~\AppData\Local\go-build
-              ~\go\pkg\mod
-          - os: windows-latest
-            family: windows
-            cache-path: |
-              ~\AppData\Local\go-build
-              ~\go\pkg\mod
-    steps:
-      - name: Set up Go 1.x
-        if: needs.changes.outputs.build == 'true'
-        uses: actions/setup-go@v4
-        with:
-          go-version: ~1.22.2
-          cache: false
-
-      - name: Check out code
-        if: needs.changes.outputs.build == 'true'
-        uses: actions/checkout@v3
-
-      - name: Cache binaries
-        id: cached_binaries
-        if: needs.changes.outputs.build == 'true'
-        uses: actions/cache@v3
-        with:
-          key: "cached-binaries-${{ matrix.os }}-${{ github.sha }}"
-          path: go.mod
-
-      - name: Cache build output
-        if: steps.cached_binaries.outputs.cache-hit != 'true' && needs.changes.outputs.build == 'true'
-        uses: actions/cache@v3
-        with:
-          path: ${{ matrix.cache-path }}
-          key: v1-go-pkg-mod-${{ matrix.os }}-${{ hashFiles('**/go.sum') }}
-
-      - name: Install make
-        if: matrix.family == 'windows' && steps.cached_binaries.outputs.cache-hit != 'true' && needs.changes.outputs.build == 'true'
-        run: choco install make
-
-      - name: Unit Test
-        if:  steps.cached_binaries.outputs.cache-hit != 'true' && needs.changes.outputs.build == 'true'
-        run: make test
-
-      - name: Build
-        if: steps.cached_binaries.outputs.cache-hit != 'true' && needs.changes.outputs.build == 'true'
-        run: make amazon-cloudwatch-agent-${{ matrix.family }}
-

go.mod

@@ -38,6 +38,7 @@ replace github.com/open-telemetry/opentelemetry-collector-contrib/processor/reso
 
 replace (
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awscontainerinsightreceiver => github.com/amazon-contributing/opentelemetry-collector-contrib/receiver/awscontainerinsightreceiver v0.0.0-20241104203805-20919412150d
+	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awscontainerinsightskueuereceiver => github.com/amazon-contributing/opentelemetry-collector-contrib/receiver/awscontainerinsightskueuereceiver v0.0.0-20241112181539-bc8278558352
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsxrayreceiver => github.com/amazon-contributing/opentelemetry-collector-contrib/receiver/awsxrayreceiver v0.0.0-20241104203805-20919412150d
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/jmxreceiver => github.com/amazon-contributing/opentelemetry-collector-contrib/receiver/jmxreceiver v0.0.0-20241104203805-20919412150d
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/prometheusreceiver => github.com/amazon-contributing/opentelemetry-collector-contrib/receiver/prometheusreceiver v0.0.0-20241104203805-20919412150d
@@ -208,6 +209,8 @@ require (
 	k8s.io/klog/v2 v2.120.1
 )
 
+require github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awscontainerinsightskueuereceiver v0.0.0-20241112181539-bc8278558352
+
 require (
 	cloud.google.com/go v0.112.1 // indirect
 	cloud.google.com/go/compute/metadata v0.3.0 // indirect

go.sum

@@ -220,6 +220,8 @@ github.com/amazon-contributing/opentelemetry-collector-contrib/processor/resourc
 github.com/amazon-contributing/opentelemetry-collector-contrib/processor/resourcedetectionprocessor v0.0.0-20241104203805-20919412150d/go.mod h1:uzpU7Y6+oL6RdOv8IWi6fjT8LNV6FYX6CN6NATLJOiQ=
 github.com/amazon-contributing/opentelemetry-collector-contrib/receiver/awscontainerinsightreceiver v0.0.0-20241104203805-20919412150d h1:LvHBMj+2Kh+SAESWTVoLbJ8bD4Xq3toB4SU66hYjg0M=
 github.com/amazon-contributing/opentelemetry-collector-contrib/receiver/awscontainerinsightreceiver v0.0.0-20241104203805-20919412150d/go.mod h1:StgsMi0cNUydO2N/7WbLYPUBXzvp9wIMcWp9P8x/Vck=
+github.com/amazon-contributing/opentelemetry-collector-contrib/receiver/awscontainerinsightskueuereceiver v0.0.0-20241112181539-bc8278558352 h1:WqkB8yOj8YCSxWzVUVCeVx263UWnmBUpMquetzwmntQ=
+github.com/amazon-contributing/opentelemetry-collector-contrib/receiver/awscontainerinsightskueuereceiver v0.0.0-20241112181539-bc8278558352/go.mod h1://9Xy+KG8K9KvujBh6sZXIYPDvbu8xsiU1l8StFHjMA=
 github.com/amazon-contributing/opentelemetry-collector-contrib/receiver/awsxrayreceiver v0.0.0-20241104203805-20919412150d h1:eQsOVRzXk32sVDkwJALfN0A4kZEMC2NNG8jBjZAZ7UM=
 github.com/amazon-contributing/opentelemetry-collector-contrib/receiver/awsxrayreceiver v0.0.0-20241104203805-20919412150d/go.mod h1:igQaQJt7eA/y3dZ2VLXVql+6k/ZXBgrAa2y9FrMMIKQ=
 github.com/amazon-contributing/opentelemetry-collector-contrib/receiver/jmxreceiver v0.0.0-20241104203805-20919412150d h1:Cld9lc7zzU/EV70Lv8EXylsx4ATjz9jHfsObEYfjyzQ=

internal/containerinsightscommon/const.go

@@ -18,6 +18,12 @@ const (
 	SourcesKey              = "Sources"
 	GpuDeviceKey            = "GpuDevice"
 
+	ClusterQueueNameKey     = "ClusterQueue"
+	ClusterQueueStatusKey   = "Status"
+	ClusterQueueReasonKey   = "Reason"
+	ClusterQueueResourceKey = "Resource"
+	Flavor                  = "Flavor"
+
 	// metric collected
 	CpuTotal                   = "cpu_usage_total"
 	CpuUser                    = "cpu_usage_user"
@@ -102,10 +108,17 @@ const (
 	NeuronHardware                              = "neuron_hardware"
 	NeuronExecutionLatency                      = "neuron_execution_latency"
 
+	KueuePendingWorkloads          = "kueue_pending_workloads"
+	KueueEvictedWorkloadsTotal     = "kueue_evicted_workloads_total"
+	KueueAdmittedActiveWorkloads   = "kueue_admitted_active_workloads"
+	KueueClusterQueueResourceUsage = "kueue_cluster_queue_resource_usage"
+	KueueClusterQueueNominalUsage  = "kueue_cluster_queue_nominal_quota"
+
 	TypeCluster          = "Cluster"
 	TypeClusterService   = "ClusterService"
 	TypeClusterNamespace = "ClusterNamespace"
 	TypeService          = "Service"
+	TypeClusterQueue     = "ClusterQueue"
 
 	// Both TypeInstance and TypeNode mean EC2 Instance, they are used in ECS and EKS separately
 	TypeInstance       = "Instance"

internal/containerinsightscommon/util.go

@@ -50,7 +50,7 @@ func MetricName(mType string, name string) string {
 		prefix = containerPrefix
 	case TypeService:
 		prefix = service
-	case TypeCluster, TypeGpuCluster:
+	case TypeCluster, TypeGpuCluster, TypeClusterQueue:
 		prefix = cluster
 	case K8sNamespace:
 		prefix = namespace

plugins/processors/kueueattributes/config.go

@@ -0,0 +1,19 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package kueueattributes
+
+import (
+	"go.opentelemetry.io/collector/component"
+)
+
+type Config struct{}
+
+// Verify Config implements Processor interface.
+var _ component.Config = (*Config)(nil)
+
+// Validate does not check for unsupported dimension key-value pairs, because those
+// get silently dropped and ignored during translation.
+func (cfg *Config) Validate() error {
+	return nil
+}

plugins/processors/kueueattributes/config_test.go

@@ -0,0 +1,18 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package kueueattributes
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"go.opentelemetry.io/collector/confmap"
+)
+
+func TestUnmarshalDefaultConfig(t *testing.T) {
+	factory := NewFactory()
+	cfg := factory.CreateDefaultConfig()
+	assert.NoError(t, confmap.New().Unmarshal(cfg))
+	assert.Equal(t, factory.CreateDefaultConfig(), cfg)
+}

plugins/processors/kueueattributes/factory.go

@@ -0,0 +1,56 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package kueueattributes
+
+import (
+	"context"
+	"fmt"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/processor"
+	"go.opentelemetry.io/collector/processor/processorhelper"
+)
+
+const (
+	stability = component.StabilityLevelBeta
+)
+
+var (
+	TypeStr, _            = component.NewType("kueueattributes")
+	processorCapabilities = consumer.Capabilities{MutatesData: true}
+)
+
+func NewFactory() processor.Factory {
+	return processor.NewFactory(
+		TypeStr,
+		createDefaultConfig,
+		processor.WithMetrics(createMetricsProcessor, stability))
+}
+
+func createDefaultConfig() component.Config {
+	return &Config{}
+}
+
+func createMetricsProcessor(
+	ctx context.Context,
+	set processor.CreateSettings,
+	cfg component.Config,
+	nextConsumer consumer.Metrics,
+) (processor.Metrics, error) {
+	processorConfig, ok := cfg.(*Config)
+	if !ok {
+		return nil, fmt.Errorf("configuration parsing error")
+	}
+
+	metricsProcessor := newKueueAttributesProcessor(processorConfig, set.Logger)
+
+	return processorhelper.NewMetricsProcessor(
+		ctx,
+		set,
+		cfg,
+		nextConsumer,
+		metricsProcessor.processMetrics,
+		processorhelper.WithCapabilities(processorCapabilities))
+}

plugins/processors/kueueattributes/factory_test.go

@@ -0,0 +1,45 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package kueueattributes
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/processor/processortest"
+)
+
+func TestCreateDefaultConfig(t *testing.T) {
+	factory := NewFactory()
+	require.NotNil(t, factory)
+
+	cfg := factory.CreateDefaultConfig()
+	assert.NotNil(t, cfg, "failed to create default config")
+	assert.NoError(t, componenttest.CheckConfigStruct(cfg))
+}
+
+func TestCreateProcessor(t *testing.T) {
+	factory := NewFactory()
+	require.NotNil(t, factory)
+
+	cfg := factory.CreateDefaultConfig()
+	setting := processortest.NewNopCreateSettings()
+
+	tProcessor, err := factory.CreateTracesProcessor(context.Background(), setting, cfg, consumertest.NewNop())
+	assert.Equal(t, err, component.ErrDataTypeIsNotSupported)
+	assert.Nil(t, tProcessor)
+
+	mProcessor, err := factory.CreateMetricsProcessor(context.Background(), setting, cfg, consumertest.NewNop())
+	assert.NoError(t, err)
+	assert.NotNil(t, mProcessor)
+
+	lProcessor, err := factory.CreateLogsProcessor(context.Background(), setting, cfg, consumertest.NewNop())
+	assert.Equal(t, err, component.ErrDataTypeIsNotSupported)
+	assert.Nil(t, lProcessor)
+}

plugins/processors/kueueattributes/processor.go

@@ -0,0 +1,106 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package kueueattributes
+
+import (
+	"context"
+	"strings"
+
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.uber.org/zap"
+
+	"github.com/aws/amazon-cloudwatch-agent/internal/containerinsightscommon"
+)
+
+const (
+	kueueMetricsIdentifier = "kueue"
+)
+
+var kueueLabelFilter = map[string]interface{}{
+	containerinsightscommon.ClusterNameKey:          nil,
+	containerinsightscommon.ClusterQueueNameKey:     nil,
+	containerinsightscommon.ClusterQueueStatusKey:   nil,
+	containerinsightscommon.ClusterQueueReasonKey:   nil,
+	containerinsightscommon.ClusterQueueResourceKey: nil,
+	containerinsightscommon.Flavor:                  nil,
+	containerinsightscommon.NodeNameKey:             nil,
+}
+
+type kueueAttributesProcessor struct {
+	*Config
+	logger      *zap.Logger
+	labelFilter map[string]interface{}
+}
+
+func newKueueAttributesProcessor(config *Config, logger *zap.Logger) *kueueAttributesProcessor {
+	d := &kueueAttributesProcessor{
+		Config:      config,
+		logger:      logger,
+		labelFilter: kueueLabelFilter,
+	}
+	return d
+}
+
+func (d *kueueAttributesProcessor) processMetrics(_ context.Context, md pmetric.Metrics) (pmetric.Metrics, error) {
+	rms := md.ResourceMetrics()
+	for i := 0; i < rms.Len(); i++ {
+		rm := rms.At(i)
+		sms := rm.ScopeMetrics()
+		for j := 0; j < sms.Len(); j++ {
+			metrics := sms.At(j).Metrics()
+			for k := 0; k < metrics.Len(); k++ {
+				m := metrics.At(k)
+				d.processMetricAttributes(m)
+			}
+		}
+		d.dropResourceMetricAttributes(rm)
+	}
+	return md, nil
+}
+
+func (d *kueueAttributesProcessor) processMetricAttributes(m pmetric.Metric) {
+	// only decorate kueue metrics
+	if !strings.HasPrefix(m.Name(), kueueMetricsIdentifier) {
+		return
+	}
+
+	var dps pmetric.NumberDataPointSlice
+	switch m.Type() {
+	case pmetric.MetricTypeGauge:
+		dps = m.Gauge().DataPoints()
+	case pmetric.MetricTypeSum:
+		dps = m.Sum().DataPoints()
+	default:
+		d.logger.Debug("Ignore unknown metric type", zap.String(containerinsightscommon.MetricType, m.Type().String()))
+	}
+
+	for i := 0; i < dps.Len(); i++ {
+		d.filterAttributes(dps.At(i).Attributes())
+	}
+}
+
+func (d *kueueAttributesProcessor) filterAttributes(attributes pcommon.Map) {
+	labels := d.labelFilter
+	if len(labels) == 0 {
+		return
+	}
+	// remove labels that are not in the keep list
+	attributes.RemoveIf(func(k string, _ pcommon.Value) bool {
+		if _, ok := labels[k]; ok {
+			return false
+		}
+		return true
+	})
+}
+
+func (d *kueueAttributesProcessor) dropResourceMetricAttributes(resourceMetric pmetric.ResourceMetrics) {
+	serviceNameKey := "service.name"
+	attributes := resourceMetric.Resource().Attributes()
+	serviceName, exists := attributes.Get(serviceNameKey)
+
+	if exists && (serviceName.Str() == "containerInsightsKueueMetricsScraper") {
+		resourceMetric.Resource().Attributes().Clear()
+	}
+}