Add OTLP Logs Testcase

terraform/templates/defaults/ecs_taskdef.tpl

@@ -42,16 +42,28 @@
             "value": "${sample_app_listen_address}"
         },
         {
-        "name": "JAEGER_RECEIVER_ENDPOINT",
-        "value": "127.0.0.1:${http_port}"
+            "name": "JAEGER_RECEIVER_ENDPOINT",
+            "value": "127.0.0.1:${http_port}"
         },
         {
-        "name": "ZIPKIN_RECEIVER_ENDPOINT",
-        "value": "127.0.0.1:${http_port}"
+            "name": "ZIPKIN_RECEIVER_ENDPOINT",
+            "value": "127.0.0.1:${http_port}"
         },
         {
-        "name": "OTEL_METRICS_EXPORTER",
-        "value": "otlp"
+            "name": "OTEL_METRICS_EXPORTER",
+            "value": "none"
+        },
+        {
+            "name": "OTEL_TRACES_EXPORTER",
+            "value": "otlp"
+        },
+        {
+            "name": "OTEL_LOGS_EXPORTER",
+            "value": "otlp"
+        },
+        {
+            "name": "SAMPLE_APP_LOG_LEVEL",
+            "value": "INFO"
         }
       ],
       "dependsOn": [

terraform/testcases/otlp_logs/otconfig.tpl

@@ -0,0 +1,30 @@
+extensions:
+  pprof:
+    endpoint: 0.0.0.0:1777
+receivers:
+  otlp:
+    protocols:
+      grpc:
+        endpoint: 0.0.0.0:${grpc_port}
+
+processors:
+  batch:
+
+exporters:
+  logging:
+    verbosity: detailed
+  awscloudwatchlogs:
+    log_group_name: "/aws/ecs/otlp/${testing_id}/logs"
+    log_stream_name: "otlp-logs"
+    region: ${region}
+
+service:
+  pipelines:
+    logs:
+      receivers: [otlp]
+      processors: [batch]
+      exporters: [logging, awscloudwatchlogs]
+  extensions: [pprof]
+  telemetry:
+    logs:
+      level: ${log_level}

terraform/testcases/otlp_logs/parameters.tfvars

@@ -0,0 +1,5 @@
+validation_config = "spark-otel-log-validation.yml"
+
+sample_app = "spark"
+
+sample_app_image = "public.ecr.aws/aws-otel-test/aws-otel-java-spark:latest"

validator/src/main/java/com/amazon/aoc/App.java

@@ -153,10 +153,8 @@ public Integer call() throws Exception {
     // load config
     List<ValidationConfig> validationConfigList =
         new ConfigLoadHelper().loadConfigFromFile(configPath);
-
     // run validation
     validate(context, validationConfigList);
-
     Instant endTime = Instant.now();
     Duration duration = Duration.between(startTime, endTime);
     log.info("Validation has completed in {} minutes.", duration.toMinutes());

validator/src/main/java/com/amazon/aoc/fileconfigs/PredefinedExpectedTemplate.java

@@ -77,6 +77,7 @@ public enum PredefinedExpectedTemplate implements FileConfig {
   CONTAINER_INSIGHT_ECS_LOG("/expected-data-template/container-insight/ecs/ecs-instance"),
   CONTAINER_INSIGHT_ECS_PROMETHEUS_LOG("/expected-data-template/container-insight/ecs/prometheus"),
   CONTAINER_INSIGHT_FARGATE_EKS_LOG("/expected-data-template/container-insight/eks/fargate"),
+  DEFAULT_EXPECTED_LOG("/expected-data-template/otlpExpectedLog.mustache"),
   ;
 
   private String path;

validator/src/main/java/com/amazon/aoc/validators/CWLogValidator.java

@@ -0,0 +1,114 @@
+package com.amazon.aoc.validators;
+
+import com.amazon.aoc.callers.ICaller;
+import com.amazon.aoc.exception.BaseException;
+import com.amazon.aoc.exception.ExceptionCode;
+import com.amazon.aoc.fileconfigs.FileConfig;
+import com.amazon.aoc.helpers.MustacheHelper;
+import com.amazon.aoc.helpers.RetryHelper;
+import com.amazon.aoc.models.Context;
+import com.amazon.aoc.models.ValidationConfig;
+import com.amazon.aoc.services.CloudWatchService;
+import com.amazonaws.services.logs.model.OutputLogEvent;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.github.fge.jsonschema.main.JsonSchema;
+import com.github.fge.jsonschema.main.JsonSchemaFactory;
+import com.github.fge.jsonschema.report.ListReportProvider;
+import com.github.fge.jsonschema.report.LogLevel;
+import com.github.fge.jsonschema.report.ProcessingReport;
+import com.github.fge.jsonschema.util.JsonLoader;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.*;
+import lombok.extern.log4j.Log4j2;
+
+@Log4j2
+public class CWLogValidator implements IValidator {
+
+  protected String logStreamName = "otlp-logs";
+
+  private static final String LOGGROUPPATH = "/aws/ecs/otlp/%s/logs";
+
+  protected CloudWatchService cloudWatchService;
+  private static final int CHECK_INTERVAL_IN_MILLI = 30 * 1000;
+  private static final int CHECK_DURATION_IN_SECONDS = 2 * 60;
+  private static final int MAX_RETRY_COUNT = 12;
+  private static final int QUERY_LIMIT = 100;
+  private JsonSchema schema;
+  protected String logGroupName;
+
+  private Context context;
+
+  protected final ObjectMapper mapper = new ObjectMapper();
+
+  @Override
+  public void init(
+      Context context,
+      ValidationConfig validationConfig,
+      ICaller caller,
+      FileConfig expectedDataTemplate)
+      throws Exception {
+    this.context = context;
+    cloudWatchService = new CloudWatchService(context.getRegion());
+    logGroupName = String.format(LOGGROUPPATH, context.getTestingId());
+    MustacheHelper mustacheHelper = new MustacheHelper();
+    String templateInput = mustacheHelper.render(expectedDataTemplate, context);
+    JsonNode jsonNode = JsonLoader.fromString(templateInput);
+    JsonSchemaFactory jsonSchemaFactory =
+        JsonSchemaFactory.newBuilder()
+            .setReportProvider(new ListReportProvider(LogLevel.INFO, LogLevel.FATAL))
+            .freeze();
+    JsonSchema jsonSchema = jsonSchemaFactory.getJsonSchema(jsonNode);
+    this.schema = jsonSchema;
+    caller.callSampleApp();
+  }
+
+  @Override
+  public void validate() throws Exception {
+    RetryHelper.retry(
+        getMaxRetryCount(),
+        CHECK_INTERVAL_IN_MILLI,
+        true,
+        () -> {
+          Instant startTime =
+              Instant.now().minusSeconds(CHECK_DURATION_IN_SECONDS).truncatedTo(ChronoUnit.MINUTES);
+          fetchAndValidateLogs(startTime);
+        });
+  }
+
+  protected void fetchAndValidateLogs(Instant startTime) throws Exception {
+    List<OutputLogEvent> logEvents =
+        cloudWatchService.getLogs(
+            logGroupName, logStreamName, startTime.toEpochMilli(), QUERY_LIMIT);
+    if (logEvents.isEmpty()) {
+      throw new BaseException(
+          ExceptionCode.LOG_FORMAT_NOT_MATCHED,
+          String.format(
+              "[StructuredLogValidator] no logs found under log stream %s" + " in log group %s",
+              logStreamName, logGroupName));
+    }
+    for (OutputLogEvent logEvent : logEvents) {
+      if (logEvent.getMessage().contains("Executing outgoing-http-call")) {
+        validateJsonSchema(logEvent.getMessage());
+      }
+    }
+  }
+
+  protected void validateJsonSchema(String logEventMsg) throws Exception {
+    JsonNode logEventNode = mapper.readTree(logEventMsg);
+    if (schema != null) {
+      ProcessingReport report = schema.validate(JsonLoader.fromString(logEventNode.toString()));
+      if (report.isSuccess()) {
+        log.info("Report was a success");
+      } else {
+        log.info("[StructuredLogValidator] failed to validate schema \n");
+        log.info(report.toString() + "\n");
+      }
+    }
+  }
+
+  protected int getMaxRetryCount() {
+    return MAX_RETRY_COUNT;
+  }
+}

validator/src/main/java/com/amazon/aoc/validators/ValidatorFactory.java

@@ -54,6 +54,10 @@ public IValidator launchValidator(ValidationConfig validationConfig) throws Exce
         validator = new CWMetricValidator();
         expectedData = validationConfig.getExpectedMetricTemplate();
         break;
+      case "cw-logs":
+        validator = new CWLogValidator();
+        expectedData = validationConfig.getExpectedLogStructureTemplate();
+        break;
       case "ecs-describe-task":
         validator = new ECSHealthCheckValidator(new TaskService(), 10);
         expectedData = validationConfig.getExpectedMetricTemplate();

validator/src/main/resources/expected-data-template/otlpExpectedLog.mustache

@@ -0,0 +1,10 @@
+{
+    "body": "Executing outgoing-http-call",
+    "severity_number": "{{severity_number}}",
+    "severity_text": "INFO",
+    "trace_id": "{{trace_id}}",
+    "span_id": "{{span_id}}",
+    "resource": {
+        "service.name": "{{resource.service.name}}"
+    }
+}

validator/src/main/resources/validations/spark-otel-log-validation.yml

@@ -0,0 +1,6 @@
+-
+  validationType: "cw-logs"
+  httpPath: "/outgoing-http-call"
+  httpMethod: "get"
+  callingType: "http"
+  expectedLogStructureTemplate: "DEFAULT_EXPECTED_LOG"