fix cloudwatchlogs flaky tests

test/cloudwatchlogs/publish_logs_test.go

@@ -24,11 +24,13 @@ import (
 )
 
 const (
-	configOutputPath              = "/opt/aws/amazon-cloudwatch-agent/bin/config.json"
-	logLineId1                    = "foo"
-	logLineId2                    = "bar"
-	logFilePath                   = "/tmp/cwagent_log_test.log" // TODO: not sure how well this will work on Windows
-	sleepForFlush                 = 20 * time.Second            // default flush interval is 5 seconds
+	configOutputPath = "/opt/aws/amazon-cloudwatch-agent/bin/config.json"
+	logLineId1       = "foo"
+	logLineId2       = "bar"
+	logFilePath      = "/tmp/cwagent_log_test.log" // TODO: not sure how well this will work on Windows
+	sleepForFlush    = 20 * time.Second            // default flush interval is 5 seconds
+	cwPropagationAttempts         = 6
+	cwPropagationInterval         = 15 * time.Second
 	configPathAutoRemoval         = "resources/config_auto_removal.json"
 	standardLogGroupClass         = "STANDARD"
 	infrequentAccessLogGroupClass = "INFREQUENT_ACCESS"
@@ -130,12 +132,15 @@ func TestWriteLogsToCloudWatch(t *testing.T) {
 			common.StopAgent()
 			end := time.Now()
 
-			// check CWL to ensure we got the expected number of logs in the log stream
-			err = awsservice.ValidateLogs(
+			// Retry to absorb CloudWatch Logs propagation lag: events can take a few
+			// seconds to become queryable after the agent flushes.
+			err = awsservice.ValidateLogsWithRetry(
 				instanceId,
 				instanceId,
 				&start,
 				&end,
+				cwPropagationAttempts,
+				cwPropagationInterval,
 				awsservice.AssertLogsCount(param.numExpectedLogs),
 				awsservice.AssertNoDuplicateLogs(),
 			)
@@ -287,7 +292,7 @@ func TestLogGroupClass(t *testing.T) {
 			}
 			t.Logf("Agent logs %s", string(agentLog))
 
-			assert.True(t, awsservice.IsLogGroupExists(logGroupName, param.logGroupClass))
+			assert.True(t, awsservice.IsLogGroupExistsWithRetry(logGroupName, cwPropagationAttempts, cwPropagationInterval, param.logGroupClass))
 		})
 	}
 }

util/awsservice/cloudwatchlogs.go

@@ -70,6 +70,30 @@ func ValidateLogs(logGroup, logStream string, since, until *time.Time, validator
 	return nil
 }
 
+// ValidateLogsWithRetry is ValidateLogs with bounded re-validation. GetLogsSince does not
+// retry a stream that exists but whose events CloudWatch has not surfaced yet, so a single
+// query right after the agent stops is racy; should re-query.
+func ValidateLogsWithRetry(logGroup, logStream string, since, until *time.Time, attempts int, interval time.Duration, validators ...LogEventsValidator) error {
+	if attempts < 1 {
+		attempts = 1
+	}
+	var err error
+	for attempt := 1; attempt <= attempts; attempt++ {
+		err = ValidateLogs(logGroup, logStream, since, until, validators...)
+		if err == nil {
+			if attempt > 1 {
+				log.Printf("Log validation for %s/%s passed on attempt %d/%d", logGroup, logStream, attempt, attempts)
+			}
+			return nil
+		}
+		if attempt < attempts {
+			log.Printf("Log validation for %s/%s failed on attempt %d/%d: %v; retrying in %s", logGroup, logStream, attempt, attempts, err, interval)
+			time.Sleep(interval)
+		}
+	}
+	return fmt.Errorf("log validation for %s/%s failed after %d attempts over %s: %w", logGroup, logStream, attempts, time.Duration(attempts-1)*interval, err)
+}
+
 // GetLogsSince makes GetLogEvents API calls, paginates through the results for the given time frame, and returns
 // the raw log strings
 func GetLogsSince(logGroup, logStream string, since, until *time.Time) ([]types.OutputLogEvent, error) {
@@ -152,6 +176,18 @@ func IsLogGroupExists(logGroupName string, logGroupClassArg ...types.LogGroupCla
 	return len(describeLogGroupOutput.LogGroups) > 0
 }
 
+func IsLogGroupExistsWithRetry(logGroupName string, attempts int, interval time.Duration, logGroupClassArg ...types.LogGroupClass) bool {
+	for i := 0; i < attempts; i++ {
+		if IsLogGroupExists(logGroupName, logGroupClassArg...) {
+			return true
+		}
+		if i < attempts-1 {
+			time.Sleep(interval)
+		}
+	}
+	return false
+}
+
 // GetLogQueryStats for the log group between start/end (in epoch seconds) for the
 // query string.
 func GetLogQueryStats(logGroupName string, startTime, endTime int64, queryString string) (*types.QueryStatistics, error) {