Merge pull request #1472 from Danielzolty/addotlplogs

Add OTLP Logs Testcase

terraform/templates/defaults/ecs_taskdef.tpl

@@ -42,16 +42,20 @@
             "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_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/exception/ExceptionCode.java

@@ -37,6 +37,7 @@ public enum ExceptionCode {
   NOT_ENOUGH_SPANS(50010, "not enough spans in the trace"),
   COLLECTOR_ID_NOT_MATCHED(50011, "span collector-id attribute does not match"),
   NULL_VAR(50012, "variable is null"),
+  EXPECTED_LOG_NOT_FOUND(50013, "expected log not found"),
 
   // build validator
   VALIDATION_TYPE_NOT_EXISTED(60001, "validation type not existed"),

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/otlp/otlp-log.json"),
   ;
 
   private String path;

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

@@ -0,0 +1,139 @@
+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 int MAX_RETRY_COUNT = 12;
+  private static final int QUERY_LIMIT = 100;
+  private JsonSchema schema;
+  protected String logGroupName;
+
+  private ICaller caller;
+
+  private Context context;
+  private String templateInput;
+
+  private ProcessingReport processingReport = null;
+
+  protected final ObjectMapper mapper = new ObjectMapper();
+
+  @Override
+  public void init(
+      Context context,
+      ValidationConfig validationConfig,
+      ICaller caller,
+      FileConfig expectedDataTemplate)
+      throws Exception {
+    this.context = context;
+    this.caller = caller;
+    cloudWatchService = new CloudWatchService(context.getRegion());
+    logGroupName = String.format(LOGGROUPPATH, context.getTestingId());
+    MustacheHelper mustacheHelper = new MustacheHelper();
+    this.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;
+  }
+
+  @Override
+  public void validate() throws Exception {
+    if (caller != null) {
+      caller.callSampleApp();
+    }
+    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());
+      }
+    }
+    if (processingReport == null || !processingReport.isSuccess()) {
+      throw new BaseException(ExceptionCode.EXPECTED_LOG_NOT_FOUND);
+    }
+  }
+
+  protected void validateJsonSchema(String logEventMsg) throws Exception {
+    JsonNode logEventNode = mapper.readTree(logEventMsg);
+    if (schema != null) {
+      processingReport = schema.validate(JsonLoader.fromString(logEventNode.toString()));
+      if (processingReport.isSuccess()) {
+        log.info("Report was a success");
+      } else {
+        log.info("[StructuredLogValidator] failed to validate schema \n");
+        log.info(processingReport.toString() + "\n");
+        log.info(("Actual Message: " + logEventMsg));
+        log.info("Expected Schema: " + templateInput);
+      }
+    }
+  }
+
+  protected int getMaxRetryCount() {
+    return MAX_RETRY_COUNT;
+  }
+
+  protected ProcessingReport getProcessingReport() {
+    return processingReport;
+  }
+
+  public void setCloudWatchService(CloudWatchService cloudWatchService) {
+    this.cloudWatchService = cloudWatchService;
+  }
+
+  public void setMaxRetryCount(int maxRetryCount) {
+    this.MAX_RETRY_COUNT = maxRetryCount;
+  }
+}

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/otlp/otlp-log.json

@@ -0,0 +1,32 @@
+{
+  "title": "structured log schema",
+  "description": "json schema for the container insights receiver ECS EC2 structured log",
+  "type": "object",
+
+  "properties": {
+    "body": {},
+    "severity_number": {},
+    "severity_text": {},
+    "flags": {},
+    "trace_id": {},
+    "span_id": {},
+    "resource": {
+      "properties": {
+        "service.name": {}
+      },
+      "required": [
+        "service.name"
+      ]
+    }
+  },
+  "required": [
+    "body",
+    "severity_number",
+    "severity_text",
+    "flags",
+    "trace_id",
+    "span_id",
+    "resource"
+  ],
+  "additionalProperties": true
+}

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"