Merge branch 'main' of github.com:aws/amazon-cloudwatch-agent-test in…

…to apm

test/cloudwatchlogs/publish_logs_test.go

@@ -13,6 +13,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs/types"
 	"github.com/stretchr/testify/assert"
 
 	"github.com/aws/amazon-cloudwatch-agent-test/environment"
@@ -99,11 +100,15 @@ func TestWriteLogsToCloudWatch(t *testing.T) {
 			end := time.Now()
 
 			// check CWL to ensure we got the expected number of logs in the log stream
-			ok, err := awsservice.ValidateLogs(instanceId, instanceId, &start, &end, func(logs []string) bool {
-				return param.numExpectedLogs == len(logs)
-			})
+			err = awsservice.ValidateLogs(
+				instanceId,
+				instanceId,
+				&start,
+				&end,
+				awsservice.AssertLogsCount(param.numExpectedLogs),
+				awsservice.AssertNoDuplicateLogs(),
+			)
 			assert.NoError(t, err)
-			assert.True(t, ok)
 		})
 	}
 }
@@ -151,23 +156,24 @@ func TestRotatingLogsDoesNotSkipLines(t *testing.T) {
 
 	end := time.Now()
 
-	ok, err := awsservice.ValidateLogs(logGroup, logStream, &start, &end, func(logs []string) bool {
-		if len(logs) != len(lines) {
-			return false
-		}
-
-		for i := 0; i < len(logs); i++ {
-			expected := strings.ReplaceAll(lines[i], "'", "\"")
-			actual := strings.ReplaceAll(logs[i], "'", "\"")
-			if expected != actual {
-				return false
+	err := awsservice.ValidateLogs(
+		logGroup,
+		logStream,
+		&start,
+		&end,
+		awsservice.AssertLogsCount(len(lines)),
+		func(events []types.OutputLogEvent) error {
+			for i := 0; i < len(events); i++ {
+				expected := strings.ReplaceAll(lines[i], "'", "\"")
+				actual := strings.ReplaceAll(*events[i].Message, "'", "\"")
+				if expected != actual {
+					return fmt.Errorf("actual log event %q does not match the expected %q", actual, expected)
+				}
 			}
-		}
-
-		return true
-	})
+			return nil
+		},
+	)
 	assert.NoError(t, err)
-	assert.True(t, ok)
 }
 
 func writeLogs(t *testing.T, f *os.File, iterations int) {

test/ecs/ecs_metadata/ecs_metadata_test.go

@@ -7,12 +7,12 @@ import (
 	_ "embed"
 	"flag"
 	"fmt"
+	"github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs/types"
 	"log"
 	"strings"
 	"testing"
 	"time"
 
-	"github.com/qri-io/jsonschema"
 	"github.com/stretchr/testify/assert"
 
 	"github.com/aws/amazon-cloudwatch-agent-test/util/awsservice"
@@ -36,8 +36,6 @@ var clusterName = flag.String("clusterName", "", "Please provide the os preferen
 var schema string
 
 func TestValidatingCloudWatchLogs(t *testing.T) {
-	rs := jsonschema.Must(schema)
-
 	start := time.Now()
 
 	logGroupName := fmt.Sprintf(ECSLogGroupNameFormat, *clusterName)
@@ -57,25 +55,25 @@ func TestValidatingCloudWatchLogs(t *testing.T) {
 
 		end := time.Now()
 
-		ok, err := awsservice.ValidateLogs(logGroupName, LogStreamName, &start, &end, func(logs []string) bool {
-			if len(logs) < 1 {
-				return false
-			}
-			for _, l := range logs {
-				if !awsservice.MatchEMFLogWithSchema(l, rs, func(s string) bool {
-					ok := true
-					if strings.Contains(l, "CloudWatchMetrics") {
-						ok = ok && strings.Contains(l, "\"Namespace\":\"ECS/ContainerInsights/Prometheus\"")
+		err := awsservice.ValidateLogs(
+			logGroupName,
+			LogStreamName,
+			&start,
+			&end,
+			awsservice.AssertLogsNotEmpty(),
+			awsservice.AssertPerLog(
+				awsservice.AssertLogSchema(awsservice.WithSchema(schema)),
+				func(event types.OutputLogEvent) error {
+					if strings.Contains(*event.Message, "CloudWatchMetrics") &&
+						!strings.Contains(*event.Message, "\"Namespace\":\"ECS/ContainerInsights/Prometheus\"") {
+						return fmt.Errorf("emf log found for non ECS/ContainerInsights/Prometheus namespace: %s", *event.Message)
 					}
-					return ok && strings.Contains(l, "\"job\":\"prometheus-redis\"")
-				}) {
-					return false
-				}
-			}
-			return true
-		})
+					return nil
+				},
+				awsservice.AssertLogContainsSubstring("\"job\":\"prometheus-redis\""),
+			),
+		)
 		assert.NoError(t, err)
-		assert.True(t, ok)
 
 		break
 	}

test/fluent/fluent_test.go

@@ -11,6 +11,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs/types"
+
 	"github.com/aws/amazon-cloudwatch-agent-test/environment"
 	"github.com/aws/amazon-cloudwatch-agent-test/util/awsservice"
 )
@@ -43,41 +45,47 @@ func TestFluentLogs(t *testing.T) {
 
 	now := time.Now()
 	for group, fieldsArr := range logGroupToKey {
-		group := fmt.Sprintf("/aws/containerinsights/%s/%s", env.EKSClusterName, group)
+		group = fmt.Sprintf("/aws/containerinsights/%s/%s", env.EKSClusterName, group)
 		if !awsservice.IsLogGroupExists(group) {
 			t.Fatalf("fluent log group doesn't exsit: %s", group)
 		}
 
 		streams := awsservice.GetLogStreams(group)
-		if len(streams) < 1 {
+		if len(streams) == 0 {
 			t.Fatalf("fluent log streams are empty for log group: %s", group)
 		}
 
-		ok, err := awsservice.ValidateLogs(group, *(streams[0].LogStreamName), nil, &now, func(logs []string) bool {
-			if len(logs) < 1 {
-				return false
-			}
-
-			// only 1 log message gets validate
-			// log message must include expected fields, and there could be more than 1 set of expected fields  per log group
-			var found = false
-			for _, fields := range fieldsArr {
-				match := 0
-				for _, field := range fields {
-					if strings.Contains(logs[0], "\""+field+"\"") {
-						match += 1
+		err := awsservice.ValidateLogs(
+			group,
+			*(streams[0].LogStreamName),
+			nil,
+			&now,
+			awsservice.AssertLogsNotEmpty(),
+			func(events []types.OutputLogEvent) error {
+				// only 1 log message gets validated
+				// log message must include expected fields, and there could be more than 1 set of expected fields per log group
+				var found bool
+				for _, fields := range fieldsArr {
+					var match int
+					for _, field := range fields {
+						if strings.Contains(*events[0].Message, "\""+field+"\"") {
+							match += 1
+						}
+					}
+					if match == len(fields) {
+						found = true
+						break
 					}
 				}
-				if match == len(fields) {
-					found = true
-					break
+				if !found {
+					return fmt.Errorf("fluent log entry doesn't include expected message fields: %s", *events[0].Message)
 				}
-			}
-			return found
-		})
+				return nil
+			},
+		)
 
-		if err != nil || !ok {
-			t.Fatalf("fluent log entry doesn't include expected message fields for logGroup: %s", group)
+		if err != nil {
+			t.Fatalf("failed validation for log group %s: %v", group, err)
 		}
 	}
 

test/metric_value_benchmark/container_insights_test.go

@@ -8,11 +8,9 @@ package metric_value_benchmark
 import (
 	_ "embed"
 	"fmt"
-	"strings"
+	"log"
 	"time"
 
-	"github.com/qri-io/jsonschema"
-
 	"github.com/aws/amazon-cloudwatch-agent-test/environment"
 	"github.com/aws/amazon-cloudwatch-agent-test/test/metric"
 	"github.com/aws/amazon-cloudwatch-agent-test/test/metric/dimension"
@@ -113,8 +111,6 @@ func validateLogsForContainerInsights(e *environment.MetaData) status.TestResult
 		Status: status.FAILED,
 	}
 
-	rs := jsonschema.Must(emfContainerInsightsSchema)
-
 	now := time.Now()
 	group := fmt.Sprintf("/aws/ecs/containerinsights/%s/performance", e.EcsClusterName)
 
@@ -125,26 +121,21 @@ func validateLogsForContainerInsights(e *environment.MetaData) status.TestResult
 	}
 
 	for _, container := range containers {
-		validateLogContents := func(s string) bool {
-			return strings.Contains(s, fmt.Sprintf("\"ContainerInstanceId\":\"%s\"", container.ContainerInstanceId))
-		}
-
-		var ok bool
 		stream := fmt.Sprintf("NodeTelemetry-%s", container.ContainerInstanceId)
-		ok, err = awsservice.ValidateLogs(group, stream, nil, &now, func(logs []string) bool {
-			if len(logs) < 1 {
-				return false
-			}
-
-			for _, l := range logs {
-				if !awsservice.MatchEMFLogWithSchema(l, rs, validateLogContents) {
-					return false
-				}
-			}
-			return true
-		})
-
-		if err != nil || !ok {
+		err = awsservice.ValidateLogs(
+			group,
+			stream,
+			nil,
+			&now,
+			awsservice.AssertLogsNotEmpty(),
+			awsservice.AssertPerLog(
+				awsservice.AssertLogSchema(awsservice.WithSchema(emfContainerInsightsSchema)),
+				awsservice.AssertLogContainsSubstring(fmt.Sprintf("\"ContainerInstanceId\":\"%s\"", container.ContainerInstanceId)),
+			),
+		)
+
+		if err != nil {
+			log.Printf("log validation (%s/%s) for container (%s) failed: %v", group, stream, container.ContainerInstanceId, err)
 			return testResult
 		}
 	}

test/metric_value_benchmark/eks_daemonset_test.go

@@ -7,13 +7,12 @@ package metric_value_benchmark
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"log"
-	"strings"
 	"time"
 
 	"github.com/aws/aws-sdk-go-v2/service/cloudwatch/types"
-	"github.com/qri-io/jsonschema"
 	"golang.org/x/exp/slices"
 
 	"github.com/aws/amazon-cloudwatch-agent-test/environment"
@@ -127,43 +126,34 @@ func (e *EKSDaemonTestRunner) validateLogs(env *environment.MetaData) status.Tes
 	}
 
 	for _, instance := range eKSInstances {
-		validateLogContents := func(s string) bool {
-			return strings.Contains(s, fmt.Sprintf("\"ClusterName\":\"%s\"", env.EKSClusterName))
-		}
-
-		var ok bool
 		stream := *instance.InstanceName
-		ok, err = awsservice.ValidateLogs(group, stream, nil, &now, func(logs []string) bool {
-			if len(logs) < 1 {
-				log.Println(fmt.Sprintf("failed to get logs for instance: %s", stream))
-				return false
-			}
+		err = awsservice.ValidateLogs(
+			group,
+			stream,
+			nil,
+			&now,
+			awsservice.AssertLogsNotEmpty(),
+			awsservice.AssertPerLog(
+				awsservice.AssertLogSchema(func(message string) (string, error) {
+					var eksClusterType awsservice.EKSClusterType
+					innerErr := json.Unmarshal([]byte(message), &eksClusterType)
+					if innerErr != nil {
+						return "", fmt.Errorf("failed to unmarshal log file: %w", innerErr)
+					}
+
+					log.Printf("eksClusterType is: %s", eksClusterType.Type)
+					jsonSchema, ok := eks_resources.EksClusterValidationMap[eksClusterType.Type]
+					if !ok {
+						return "", errors.New("invalid cluster type provided")
+					}
+					return jsonSchema, nil
+				}),
+				awsservice.AssertLogContainsSubstring(fmt.Sprintf("\"ClusterName\":\"%s\"", env.EKSClusterName)),
+			),
+		)
 
-			for _, l := range logs {
-				var eksClusterType awsservice.EKSClusterType
-				err := json.Unmarshal([]byte(l), &eksClusterType)
-				if err != nil {
-					log.Println("failed to unmarshal log file")
-				}
-
-				log.Println(fmt.Sprintf("eksClusterType is: %s", eksClusterType.Type))
-				jsonSchema, ok := eks_resources.EksClusterValidationMap[eksClusterType.Type]
-				if !ok {
-					log.Println("invalid cluster type provided")
-					return false
-				}
-				rs := jsonschema.Must(jsonSchema)
-
-				if !awsservice.MatchEMFLogWithSchema(l, rs, validateLogContents) {
-					log.Println("failed to match log with schema")
-					log.Printf("log entry %s json schema %s", l, jsonSchema)
-					return false
-				}
-			}
-			return true
-		})
-
-		if err != nil || !ok {
+		if err != nil {
+			log.Printf("log validation (%s/%s) failed: %v", group, stream, err)
 			return testResult
 		}
 	}