Parse db.statement attribute for database instrumentation (#755)

* Parse db.statement attribute for database instrumentation

* reverted file change

* fixed tests

* fixed formating

* applied comments

* fixed ordering

Author: AsakerMohd

awsagentprovider/build.gradle.kts

@@ -34,6 +34,8 @@ dependencies {
   implementation("io.opentelemetry.contrib:opentelemetry-aws-xray")
   // AWS Resource Detectors
   implementation("io.opentelemetry.contrib:opentelemetry-aws-resources")
+  // Json file reader
+  implementation("com.fasterxml.jackson.core:jackson-databind:2.16.1")
   // Export configuration
   compileOnly("io.opentelemetry:opentelemetry-exporter-otlp")
 

awsagentprovider/src/main/java/software/amazon/opentelemetry/javaagent/providers/AwsMetricAttributeGenerator.java

@@ -17,6 +17,7 @@
 
 import static io.opentelemetry.semconv.ResourceAttributes.SERVICE_NAME;
 import static io.opentelemetry.semconv.SemanticAttributes.DB_OPERATION;
+import static io.opentelemetry.semconv.SemanticAttributes.DB_STATEMENT;
 import static io.opentelemetry.semconv.SemanticAttributes.DB_SYSTEM;
 import static io.opentelemetry.semconv.SemanticAttributes.FAAS_INVOKED_NAME;
 import static io.opentelemetry.semconv.SemanticAttributes.FAAS_TRIGGER;
@@ -44,6 +45,8 @@
 import static software.amazon.opentelemetry.javaagent.providers.AwsAttributeKeys.AWS_SPAN_KIND;
 import static software.amazon.opentelemetry.javaagent.providers.AwsAttributeKeys.AWS_STREAM_NAME;
 import static software.amazon.opentelemetry.javaagent.providers.AwsAttributeKeys.AWS_TABLE_NAME;
+import static software.amazon.opentelemetry.javaagent.providers.AwsSpanProcessingUtil.MAX_KEYWORD_LENGTH;
+import static software.amazon.opentelemetry.javaagent.providers.AwsSpanProcessingUtil.SQL_DIALECT_PATTERN;
 import static software.amazon.opentelemetry.javaagent.providers.AwsSpanProcessingUtil.UNKNOWN_OPERATION;
 import static software.amazon.opentelemetry.javaagent.providers.AwsSpanProcessingUtil.UNKNOWN_REMOTE_OPERATION;
 import static software.amazon.opentelemetry.javaagent.providers.AwsSpanProcessingUtil.UNKNOWN_REMOTE_SERVICE;
@@ -68,6 +71,7 @@
 import java.util.Optional;
 import java.util.logging.Level;
 import java.util.logging.Logger;
+import java.util.regex.Matcher;
 import javax.annotation.Nullable;
 
 /**
@@ -266,9 +270,15 @@ private static void setRemoteServiceAndOperation(SpanData span, AttributesBuilde
     } else if (isKeyPresent(span, RPC_SERVICE) || isKeyPresent(span, RPC_METHOD)) {
       remoteService = normalizeServiceName(span, getRemoteService(span, RPC_SERVICE));
       remoteOperation = getRemoteOperation(span, RPC_METHOD);
-    } else if (isKeyPresent(span, DB_SYSTEM) || isKeyPresent(span, DB_OPERATION)) {
+    } else if (isKeyPresent(span, DB_SYSTEM)
+        || isKeyPresent(span, DB_OPERATION)
+        || isKeyPresent(span, DB_STATEMENT)) {
       remoteService = getRemoteService(span, DB_SYSTEM);
-      remoteOperation = getRemoteOperation(span, DB_OPERATION);
+      if (isKeyPresent(span, DB_OPERATION)) {
+        remoteOperation = getRemoteOperation(span, DB_OPERATION);
+      } else {
+        remoteOperation = getDBStatementRemoteOperation(span, DB_STATEMENT);
+      }
     } else if (isKeyPresent(span, FAAS_INVOKED_NAME) || isKeyPresent(span, FAAS_TRIGGER)) {
       remoteService = getRemoteService(span, FAAS_INVOKED_NAME);
       remoteOperation = getRemoteOperation(span, FAAS_TRIGGER);
@@ -449,6 +459,36 @@ private static String getRemoteOperation(SpanData span, AttributeKey<String> rem
     return remoteOperation;
   }
 
+  /**
+   * If no db.operation attribute provided in the span, we use db.statement to compute a valid
+   * remote operation in a best-effort manner. To do this, we take the first substring of the
+   * statement and compare to a regex list of known SQL keywords. The substring length is determined
+   * by the longest known SQL keywords.
+   */
+  private static String getDBStatementRemoteOperation(
+      SpanData span, AttributeKey<String> remoteOperationKey) {
+    String remoteOperation = span.getAttributes().get(remoteOperationKey);
+    if (remoteOperation == null) {
+      remoteOperation = UNKNOWN_REMOTE_OPERATION;
+    }
+
+    // Remove all whitespace and newline characters from the beginning of remote_operation
+    // and retrieve the first MAX_KEYWORD_LENGTH characters
+    remoteOperation = remoteOperation.stripLeading();
+    if (remoteOperation.length() > MAX_KEYWORD_LENGTH) {
+      remoteOperation = remoteOperation.substring(0, MAX_KEYWORD_LENGTH);
+    }
+
+    Matcher matcher = SQL_DIALECT_PATTERN.matcher(remoteOperation.toUpperCase());
+    if (matcher.find() && !matcher.group(0).isEmpty()) {
+      remoteOperation = matcher.group(0);
+    } else {
+      remoteOperation = UNKNOWN_REMOTE_OPERATION;
+    }
+
+    return remoteOperation;
+  }
+
   private static void logUnknownAttribute(AttributeKey<String> attributeKey, SpanData span) {
     String[] params = {
       attributeKey.getKey(), span.getKind().name(), span.getSpanContext().getSpanId()

awsagentprovider/src/main/java/software/amazon/opentelemetry/javaagent/providers/AwsSpanProcessingUtil.java

@@ -22,11 +22,20 @@
 import static io.opentelemetry.semconv.SemanticAttributes.RPC_SYSTEM;
 import static software.amazon.opentelemetry.javaagent.providers.AwsAttributeKeys.AWS_LOCAL_OPERATION;
 
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectReader;
 import io.opentelemetry.api.common.AttributeKey;
 import io.opentelemetry.api.trace.SpanContext;
 import io.opentelemetry.api.trace.SpanKind;
 import io.opentelemetry.sdk.common.InstrumentationScopeInfo;
 import io.opentelemetry.sdk.trace.data.SpanData;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.regex.Pattern;
 
 /** Utility class designed to support shared logic across AWS Span Processors. */
 final class AwsSpanProcessingUtil {
@@ -40,6 +49,29 @@ final class AwsSpanProcessingUtil {
   static final String LOCAL_ROOT = "LOCAL_ROOT";
   static final String SQS_RECEIVE_MESSAGE_SPAN_NAME = "Sqs.ReceiveMessage";
   static final String AWS_SDK_INSTRUMENTATION_SCOPE_PREFIX = "io.opentelemetry.aws-sdk-";
+  // Max keyword length supported by parsing into remote_operation from DB_STATEMENT.
+  // The current longest command word is DATETIME_INTERVAL_PRECISION at 27 characters.
+  // If we add a longer keyword to the sql dialect keyword list, need to update the constant below.
+  static final int MAX_KEYWORD_LENGTH = 27;
+  static final Pattern SQL_DIALECT_PATTERN =
+      Pattern.compile("^(?:" + String.join("|", getDialectKeywords()) + ")\\b");
+
+  private static final String SQL_DIALECT_KEYWORDS_JSON = "configuration/sql_dialect_keywords.json";
+
+  static List<String> getDialectKeywords() {
+    try (InputStream jsonFile =
+        AwsSpanProcessingUtil.class
+            .getClassLoader()
+            .getResourceAsStream(SQL_DIALECT_KEYWORDS_JSON)) {
+      ObjectMapper mapper = new ObjectMapper();
+      JsonNode jsonNode = mapper.readValue(jsonFile, JsonNode.class);
+      JsonNode arrayNode = jsonNode.get("keywords");
+      ObjectReader reader = mapper.readerFor(new TypeReference<List<String>>() {});
+      return reader.readValue(arrayNode);
+    } catch (IOException e) {
+      return new ArrayList<>();
+    }
+  }
 
   /**
    * Ingress operation (i.e. operation for Server and Consumer spans) will be generated from