Add IntegTests to Cover changes from PR:1713, Aggregate NeuronCore Utilization Metrics

docs/resources/dummy-neuron-monitor/Dockerfile

@@ -27,6 +27,7 @@ RUN apt-get update \
  && rm -rf /tmp/tmp* \
  && apt-get clean
 
+COPY neuron-monitor-output.json /opt/aws/neuron/bin/neuron-monitor-output.json
 COPY dummy_neuron_monitor.py /opt/aws/neuron/bin/dummy_neuron_monitor.py
 RUN chmod 755 /opt/aws/neuron/bin/dummy_neuron_monitor.py
 RUN pip3 install prometheus_client boto3 requests

terraform/eks/daemon/awsneuron/main.tf

@@ -395,7 +395,7 @@ resource "kubernetes_daemonset" "neuron_monitor" {
         }
         container {
           name  = "neuron-monitor-prometheus"
-          image = "506463145083.dkr.ecr.us-west-2.amazonaws.com/mocked-neuron-monitor:v2"
+          image = "506463145083.dkr.ecr.us-west-2.amazonaws.com/mocked-neuron-monitor:v4"
           port {
             container_port = 8000
           }

test/awsneuron/neuron_metrics_test.go

@@ -40,6 +40,7 @@ func (t *AwsNeuronTestRunner) Validate() status.TestGroupResult {
 	testResults = append(testResults, metric.ValidateMetrics(t.env, awsNeuronMetricIndicator, expectedDimsToMetrics)...)
 	testResults = append(testResults, metric.ValidateLogs(t.env))
 	testResults = append(testResults, metric.ValidateLogsFrequency(t.env))
+	testResults = append(testResults, metric.ValidateNeuronCoreUtilizationValuesLogs(t.env))
 	return status.TestGroupResult{
 		Name:        t.GetTestName(),
 		TestResults: testResults,

test/metric/container_insights_util.go

@@ -10,8 +10,10 @@ import (
 	"errors"
 	"fmt"
 	"log"
+	"math"
 	"math/rand"
 	"sort"
+	"strconv"
 	"strings"
 	"time"
 
@@ -296,3 +298,63 @@ func ValidateLogsFrequency(env *environment.MetaData) status.TestResult {
 	testResult.Status = status.SUCCESSFUL
 	return testResult
 }
+
+func ValidateNeuronCoreUtilizationValuesLogs(env *environment.MetaData) status.TestResult {
+	const core = "core"
+	testResult := status.TestResult{
+		Name:   "emf-logs-neuron-core-utilization",
+		Status: status.FAILED,
+	}
+	var testFailed = false
+
+	end := time.Now().Add(-2 * time.Minute).Truncate(time.Minute)
+	start := end.Add(-1 * time.Minute)
+	group := fmt.Sprintf("/aws/containerinsights/%s/performance", env.EKSClusterName)
+
+	// need to get the instances used for the EKS cluster
+	eKSInstances, err := awsservice.GetEKSInstances(env.EKSClusterName)
+	if err != nil {
+		log.Println("failed to get EKS instances", err)
+		return testResult
+	}
+
+	for _, instance := range eKSInstances {
+		stream := *instance.InstanceName
+		coreMap, err := awsservice.GetNeuronCoreUtilizationPerCore(group, stream, &start, &end)
+
+		if err != nil {
+			log.Printf("log validation (%s/%s) failed: %v, start time : %s, error is : %s", group, stream, err, start, err)
+			return testResult
+		}
+
+		// We expect 32 Cores from the current test, anything less or more is a bug
+		if len(coreMap) != 32 {
+			log.Printf("32 Cores not found")
+			var coreMapStr strings.Builder
+			for k, v := range coreMap {
+				coreMapStr.WriteString(fmt.Sprintf("%s: %f, ", k, v))
+			}
+			log.Printf("coreMap: %s", coreMapStr.String())
+			testFailed = true
+		}
+
+		// Check if coreMap has the expected core utilization values
+		for coreKey, actualValue := range coreMap {
+			if strings.HasPrefix(coreKey, core) {
+				coreNumStr := strings.TrimPrefix(coreKey, core)
+				expectedValue, err := strconv.Atoi(coreNumStr)
+				if err != nil || math.Round(actualValue) != float64(expectedValue) {
+					log.Printf("Core utilization validation failed: expected %s:%d, got %v", 
+						coreKey, expectedValue, actualValue)
+					testFailed = true
+				}
+			}
+		}
+	}
+
+	testResult.Status = status.SUCCESSFUL
+	if testFailed {
+		testResult.Status = status.FAILED
+	}
+	return testResult
+}

util/awsservice/cloudwatchlogs.go

@@ -431,3 +431,38 @@ func CountMetricsInEMFLogs(logGroupName string) (int, error) {
 
 	return totalMetrics, nil
 }
+
+func GetNeuronCoreUtilizationPerCore(logGroup, logStream string, since, until *time.Time) (map[string]float64, error) {
+	var coreUtilization = make(map[string]float64)
+	var data map[string]interface{}
+
+	events, err := GetLogsSince(logGroup, logStream, since, until)
+
+	// if there is an error, return the empty map
+	if err != nil {
+		return coreUtilization, err
+	}
+
+	for _, event := range events {
+		message := *event.Message
+
+		var eksClusterType EKSClusterType
+		innerErr := json.Unmarshal([]byte(message), &eksClusterType)
+		if innerErr != nil || !strings.Contains(eksClusterType.Type, "NodeAWSNeuronCore") {
+			continue
+		}
+
+		err := json.Unmarshal([]byte(message), &data)
+		if err != nil {
+			continue
+		}
+
+		if core, ok := data["NeuronCore"].(string); ok {
+			if util, ok := data["node_neuroncore_utilization"].(float64); ok {
+				coreUtilization[core] = util
+			}
+		}
+	}
+
+	return coreUtilization, nil
+}