Sanitize SQL in Apache Camel instrumentation

instrumentation/apache-camel-2.20/javaagent-unit-tests/build.gradle.kts

@@ -4,6 +4,8 @@ plugins {
 
 dependencies {
   testImplementation(project(":instrumentation:apache-camel-2.20:javaagent"))
+  testImplementation(project(":instrumentation-api"))
+
   testImplementation("org.apache.camel:camel-core:2.20.1")
   testImplementation("org.apache.camel:camel-aws:2.20.1")
   testImplementation("org.apache.camel:camel-http:2.20.1")

instrumentation/apache-camel-2.20/javaagent-unit-tests/src/test/groovy/io/opentelemetry/javaagent/instrumentation/apachecamel/decorators/SanitizationTest.groovy

@@ -0,0 +1,58 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.apachecamel.decorators
+
+import org.apache.camel.Exchange
+import org.apache.camel.Message
+import spock.lang.Specification
+
+class SanitizationTest extends Specification {
+
+  def "sanitize jdbc #originalSql"() {
+
+    setup:
+    def decorator = new DbSpanDecorator("jdbc", "")
+    def exchange = Mock(Exchange) {
+      getIn() >> Mock(Message) {
+        getBody() >> originalSql
+      }
+    }
+    def actualSanitized = decorator.getStatement(exchange, null)
+
+    expect:
+    actualSanitized == sanitizedSql
+
+    where:
+    originalSql                                       | sanitizedSql
+    "SELECT 3"                                        | "SELECT ?"
+    "SELECT * FROM TABLE WHERE FIELD = 1234"          | "SELECT * FROM TABLE WHERE FIELD = ?"
+    "SELECT * FROM TABLE WHERE FIELD<-1234"           | "SELECT * FROM TABLE WHERE FIELD<?"
+    "SELECT col1 AS col2 FROM users WHERE field=1234" | "SELECT col1 AS col2 FROM users WHERE field=?"
+  }
+
+  def "sanitize sql #originalSql"() {
+
+    setup:
+    def decorator = new DbSpanDecorator("sql", "")
+    def exchange = Mock(Exchange) {
+      getIn() >> Mock(Message) {
+        getHeader("CamelSqlQuery") >> originalSql
+      }
+    }
+    def actualSanitized = decorator.getStatement(exchange, null)
+
+    expect:
+    actualSanitized == sanitizedSql
+
+    where:
+    originalSql                                      | sanitizedSql
+    "SELECT * FROM table WHERE col1=1234 AND col2>3" | "SELECT * FROM table WHERE col1=? AND col2>?"
+    "UPDATE table SET col=12"                        | "UPDATE table SET col=?"
+    'insert into table where col=321'                | 'insert into table where col=?'
+  }
+
+
+}

instrumentation/apache-camel-2.20/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/apachecamel/decorators/DbSpanDecorator.java

@@ -24,6 +24,7 @@
 package io.opentelemetry.javaagent.instrumentation.apachecamel.decorators;
 
 import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.instrumentation.api.db.SqlStatementSanitizer;
 import io.opentelemetry.javaagent.instrumentation.apachecamel.CamelDirection;
 import io.opentelemetry.semconv.trace.attributes.SemanticAttributes;
 import java.net.URI;
@@ -58,11 +59,10 @@ public String getOperationName(
     }
   }
 
-  private String getStatement(Exchange exchange, Endpoint endpoint) {
+  // visible for testing
+  String getStatement(Exchange exchange, Endpoint endpoint) {
+    // TODO: sanitize cql
     switch (component) {
-      case "mongodb":
-        Map<String, String> mongoParameters = toQueryParameters(endpoint.getEndpointUri());
-        return mongoParameters.toString();
       case "cql":
         Object cqlObj = exchange.getIn().getHeader("CamelCqlQuery");
         if (cqlObj != null) {
@@ -76,13 +76,13 @@ private String getStatement(Exchange exchange, Endpoint endpoint) {
       case "jdbc":
         Object body = exchange.getIn().getBody();
         if (body instanceof String) {
-          return (String) body;
+          return SqlStatementSanitizer.sanitize((String) body).getFullStatement();
         }
         return null;
       case "sql":
         Object sqlquery = exchange.getIn().getHeader("CamelSqlQuery");
         if (sqlquery instanceof String) {
-          return (String) sqlquery;
+          return SqlStatementSanitizer.sanitize((String) sqlquery).getFullStatement();
         }
         return null;
       default: