Ci neuron

.chloggen-aws/nvidia-gpu.yaml

@@ -0,0 +1,24 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: 'enhancement'
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: containerinsightsreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Adds DCGM scraper to collect NVIDIA GPU metrics"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [160]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext: Supports NVIDIA GPU metrics by adding a new prometheus data scraper in a k8s environment. The new scraper |
+  relabels the default DCGM labels into existing Container Insights labels.
+
+# e.g. '[aws]'
+# Include 'aws' if the change is done by cwa
+# Default: '[user]'
+change_logs: [aws]

exporter/awsemfexporter/metric_translator.go

@@ -131,7 +131,9 @@ func (mt metricTranslator) translateOTelToGroupedMetric(rm pmetric.ResourceMetri
 	}
 
 	if serviceName, ok := rm.Resource().Attributes().Get("service.name"); ok {
-		if strings.HasPrefix(serviceName.Str(), "containerInsightsKubeAPIServerScraper") {
+		if strings.HasPrefix(serviceName.Str(), "containerInsightsKubeAPIServerScraper") ||
+			strings.HasPrefix(serviceName.Str(), "containerInsightsDCGMExporterScraper") ||
+			strings.HasPrefix(serviceName.Str(), "containerInsightsNeuronMonitorScraper") {
 			// the prometheus metrics that come from the container insight receiver need to be clearly tagged as coming from container insights
 			metricReceiver = containerInsightsReceiver
 		}

exporter/awsemfexporter/metric_translator_test.go

@@ -280,6 +280,10 @@ func TestTranslateOtToGroupedMetric(t *testing.T) {
 	// need to have 1 more metric than the default because the first is not going to be retained because it is a delta metric
 	containerInsightMetric := createTestResourceMetricsHelper(defaultNumberOfTestMetrics + 1)
 	containerInsightMetric.Resource().Attributes().PutStr(conventions.AttributeServiceName, "containerInsightsKubeAPIServerScraper")
+	gpuMetric := createTestResourceMetricsHelper(defaultNumberOfTestMetrics + 1)
+	gpuMetric.Resource().Attributes().PutStr(conventions.AttributeServiceName, "containerInsightsDCGMExporterScraper")
+	neuronMetric := createTestResourceMetricsHelper(defaultNumberOfTestMetrics + 1)
+	neuronMetric.Resource().Attributes().PutStr(conventions.AttributeServiceName, "containerInsightsNeuronMonitorScraper")
 
 	counterSumMetrics := map[string]*metricInfo{
 		"spanCounter": {
@@ -368,12 +372,40 @@ func TestTranslateOtToGroupedMetric(t *testing.T) {
 				"spanName":    "testSpan",
 			},
 			map[string]string{
-				(oTellibDimensionKey): "cloudwatch-lib",
-				"spanName":            "testSpan",
+				oTellibDimensionKey: "cloudwatch-lib",
+				"spanName":          "testSpan",
 			},
 			"myServiceNS/containerInsightsKubeAPIServerScraper",
 			containerInsightsReceiver,
 		},
+		{
+			"dcgm receiver",
+			gpuMetric,
+			map[string]string{
+				"isItAnError": "false",
+				"spanName":    "testSpan",
+			},
+			map[string]string{
+				oTellibDimensionKey: "cloudwatch-lib",
+				"spanName":          "testSpan",
+			},
+			"myServiceNS/containerInsightsDCGMExporterScraper",
+			containerInsightsReceiver,
+		},
+		{
+			"neuron monitor receiver",
+			neuronMetric,
+			map[string]string{
+				"isItAnError": "false",
+				"spanName":    "testSpan",
+			},
+			map[string]string{
+				oTellibDimensionKey: "cloudwatch-lib",
+				"spanName":          "testSpan",
+			},
+			"myServiceNS/containerInsightsNeuronMonitorScraper",
+			containerInsightsReceiver,
+		},
 	}
 
 	for _, tc := range testCases {

receiver/awscontainerinsightreceiver/config.go

@@ -57,4 +57,11 @@ type Config struct {
 	// EnableControlPlaneMetrics enables additional metrics sourced from the Kubernetes API server /metrics prometheus endpoint
 	// The default value is false.
 	EnableControlPlaneMetrics bool `mapstructure:"enable_control_plane_metrics"`
+
+	// EnableGpuMetric toggles GPU monitoring where metrics are scraped from vendor specific sources
+	EnableGpuMetric bool `mapstructure:"gpu_metrics"`
+
+	// EnableNeuronMetric toggles Neuron monitoring where metrics are scraped from neuron monitor
+	// The default value is false.
+	EnableNeuronMetric bool `mapstructure:"neuron_metrics"`
 }

receiver/awscontainerinsightreceiver/factory.go

@@ -42,6 +42,9 @@ const (
 
 	// Don't enable EKS control plane metrics by default
 	defaultEnableControlPlaneMetrics = false
+
+	// Don't enable Neuron metrics by default
+	defaultEnableNeuronMetrics = false
 )
 
 // NewFactory creates a factory for AWS container insight receiver
@@ -64,6 +67,7 @@ func createDefaultConfig() component.Config {
 		ClusterName:                 defaultClusterName,
 		LeaderLockName:              defaultLeaderLockName,
 		EnableControlPlaneMetrics:   defaultEnableControlPlaneMetrics,
+		EnableNeuronMetric:          defaultEnableNeuronMetrics,
 	}
 }
 

receiver/awscontainerinsightreceiver/internal/gpu/dcgmscraper.go

@@ -0,0 +1,219 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package gpu
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/prometheus/common/model"
+	"github.com/prometheus/prometheus/config"
+	"github.com/prometheus/prometheus/discovery"
+	"github.com/prometheus/prometheus/discovery/kubernetes"
+	"github.com/prometheus/prometheus/model/relabel"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/receiver"
+	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/prometheusreceiver"
+)
+
+const (
+	caFile             = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"
+	collectionInterval = 60 * time.Second
+	jobName            = "containerInsightsDCGMExporterScraper"
+)
+
+type DcgmScraper struct {
+	ctx                context.Context
+	settings           component.TelemetrySettings
+	host               component.Host
+	hostInfoProvider   hostInfoProvider
+	prometheusReceiver receiver.Metrics
+	running            bool
+}
+
+type DcgmScraperOpts struct {
+	Ctx               context.Context
+	TelemetrySettings component.TelemetrySettings
+	Consumer          consumer.Metrics
+	Host              component.Host
+	HostInfoProvider  hostInfoProvider
+}
+
+type hostInfoProvider interface {
+	GetClusterName() string
+	GetInstanceID() string
+}
+
+func NewDcgmScraper(opts DcgmScraperOpts) (*DcgmScraper, error) {
+	if opts.Consumer == nil {
+		return nil, errors.New("consumer cannot be nil")
+	}
+	if opts.Host == nil {
+		return nil, errors.New("host cannot be nil")
+	}
+	if opts.HostInfoProvider == nil {
+		return nil, errors.New("cluster name provider cannot be nil")
+	}
+
+	scrapeConfig := &config.ScrapeConfig{
+		ScrapeInterval: model.Duration(collectionInterval),
+		ScrapeTimeout:  model.Duration(collectionInterval),
+		JobName:        jobName,
+		Scheme:         "http",
+		MetricsPath:    "/metrics",
+		ServiceDiscoveryConfigs: discovery.Configs{
+			&kubernetes.SDConfig{
+				Role: kubernetes.RoleService,
+				NamespaceDiscovery: kubernetes.NamespaceDiscovery{
+					IncludeOwnNamespace: true,
+				},
+				Selectors: []kubernetes.SelectorConfig{
+					{
+						Role:  kubernetes.RoleService,
+						Label: "k8s-app=dcgm-exporter-service",
+					},
+				},
+				AttachMetadata: kubernetes.AttachMetadataConfig{
+					Node: true,
+				},
+			},
+		},
+		RelabelConfigs: []*relabel.Config{
+			{
+				SourceLabels: model.LabelNames{"__address__"},
+				Regex:        relabel.MustNewRegexp("([^:]+)(?::\\d+)?"),
+				Replacement:  "${1}:9400",
+				TargetLabel:  "__address__",
+				Action:       relabel.Replace,
+			},
+		},
+		MetricRelabelConfigs: []*relabel.Config{
+			{
+				SourceLabels: model.LabelNames{"__name__"},
+				Regex:        relabel.MustNewRegexp("DCGM_.*"),
+				Action:       relabel.Keep,
+			},
+			{
+				SourceLabels: model.LabelNames{"Hostname"},
+				TargetLabel:  "NodeName",
+				Regex:        relabel.MustNewRegexp("(.*)"),
+				Replacement:  "${1}",
+				Action:       relabel.Replace,
+			},
+			{
+				SourceLabels: model.LabelNames{"namespace"},
+				TargetLabel:  "Namespace",
+				Regex:        relabel.MustNewRegexp("(.*)"),
+				Replacement:  "${1}",
+				Action:       relabel.Replace,
+			},
+			// hacky way to inject static values (clusterName & instanceId) to label set without additional processor
+			// relabel looks up an existing label then creates another label with given key (TargetLabel) and value (static)
+			{
+				SourceLabels: model.LabelNames{"namespace"},
+				TargetLabel:  "ClusterName",
+				Regex:        relabel.MustNewRegexp(".*"),
+				Replacement:  opts.HostInfoProvider.GetClusterName(),
+				Action:       relabel.Replace,
+			},
+			{
+				SourceLabels: model.LabelNames{"namespace"},
+				TargetLabel:  "InstanceId",
+				Regex:        relabel.MustNewRegexp(".*"),
+				Replacement:  opts.HostInfoProvider.GetInstanceID(),
+				Action:       relabel.Replace,
+			},
+			{
+				SourceLabels: model.LabelNames{"pod"},
+				TargetLabel:  "FullPodName",
+				Regex:        relabel.MustNewRegexp("(.*)"),
+				Replacement:  "${1}",
+				Action:       relabel.Replace,
+			},
+			{
+				SourceLabels: model.LabelNames{"pod"},
+				TargetLabel:  "PodName",
+				Regex:        relabel.MustNewRegexp("(.+)-(.+)"),
+				Replacement:  "${1}",
+				Action:       relabel.Replace,
+			},
+			// additional k8s podname for service name decoration
+			{
+				SourceLabels: model.LabelNames{"pod"},
+				TargetLabel:  "K8sPodName",
+				Regex:        relabel.MustNewRegexp("(.*)"),
+				Replacement:  "${1}",
+				Action:       relabel.Replace,
+			},
+			{
+				SourceLabels: model.LabelNames{"container"},
+				TargetLabel:  "ContainerName",
+				Regex:        relabel.MustNewRegexp("(.*)"),
+				Replacement:  "${1}",
+				Action:       relabel.Replace,
+			},
+			{
+				SourceLabels: model.LabelNames{"device"},
+				TargetLabel:  "GpuDevice",
+				Regex:        relabel.MustNewRegexp("(.*)"),
+				Replacement:  "${1}",
+				Action:       relabel.Replace,
+			},
+		},
+	}
+
+	promConfig := prometheusreceiver.Config{
+		PrometheusConfig: &config.Config{
+			ScrapeConfigs: []*config.ScrapeConfig{scrapeConfig},
+		},
+	}
+
+	params := receiver.CreateSettings{
+		TelemetrySettings: opts.TelemetrySettings,
+	}
+
+	promFactory := prometheusreceiver.NewFactory()
+	promReceiver, err := promFactory.CreateMetricsReceiver(opts.Ctx, params, &promConfig, opts.Consumer)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create prometheus receiver: %w", err)
+	}
+
+	return &DcgmScraper{
+		ctx:                opts.Ctx,
+		settings:           opts.TelemetrySettings,
+		host:               opts.Host,
+		hostInfoProvider:   opts.HostInfoProvider,
+		prometheusReceiver: promReceiver,
+	}, nil
+}
+
+func (ds *DcgmScraper) GetMetrics() []pmetric.Metrics {
+	// This method will never return metrics because the metrics are collected by the scraper.
+	// This method will ensure the scraper is running
+	if !ds.running {
+		ds.settings.Logger.Info("The scraper is not running, starting up the scraper")
+		err := ds.prometheusReceiver.Start(ds.ctx, ds.host)
+		if err != nil {
+			ds.settings.Logger.Error("Unable to start PrometheusReceiver", zap.Error(err))
+		}
+		ds.running = err == nil
+	}
+	return nil
+}
+
+func (ds *DcgmScraper) Shutdown() {
+	if ds.running {
+		err := ds.prometheusReceiver.Shutdown(ds.ctx)
+		if err != nil {
+			ds.settings.Logger.Error("Unable to shutdown PrometheusReceiver", zap.Error(err))
+		}
+		ds.running = false
+	}
+}