Add support for Apache Elasticjob

.fossa.yml

@@ -73,6 +73,9 @@ targets:
     - type: gradle
       path: ./
       target: ':instrumentation:apache-dubbo-2.7:library-autoconfigure'
+    - type: gradle
+      path: ./
+      target: ':instrumentation:apache-elasticjob-3.0:javaagent'
     - type: gradle
       path: ./
       target: ':instrumentation:apache-httpasyncclient-4.1:javaagent'

docs/supported-libraries.md

@@ -29,6 +29,7 @@ These are the supported libraries and frameworks:
 | [Apache CXF JAX-WS](https://cxf.apache.org/)                                                                                                | 3.0+ (not including 4.0+ yet)      | N/A                                                                                                                                                                                                                                                                                                                                                                                     | Provides `http.route` [2], Controller Spans [3]                                                                                                   |
 | [Apache DBCP](https://commons.apache.org/proper/commons-dbcp/)                                                                              | 2.0+                               | [opentelemetry-apache-dbcp-2.0](../instrumentation/apache-dbcp-2.0/library)                                                                                                                                                                                                                                                                                                             | [Database Pool Metrics]                                                                                                                           |
 | [Apache Dubbo](https://github.com/apache/dubbo/)                                                                                            | 2.7+                               | [opentelemetry-apache-dubbo-2.7](../instrumentation/apache-dubbo-2.7/library-autoconfigure)                                                                                                                                                                                                                                                                                             | [RPC Client Spans], [RPC Server Spans]                                                                                                            |
+| [Apache ElasticJob](https://shardingsphere.apache.org/elasticjob/)                                                                          | 3.0+                               | N/A                                                                                                                                                                                                                                                                                                                                                                                     | none                                                                                                                                              |
 | [Apache HttpAsyncClient](https://hc.apache.org/index.html)                                                                                  | 4.1+                               | N/A                                                                                                                                                                                                                                                                                                                                                                                     | [HTTP Client Spans], [HTTP Client Metrics]                                                                                                        |
 | [Apache HttpClient](https://hc.apache.org/index.html)                                                                                       | 2.0+                               | [opentelemetry-apache-httpclient-4.3](../instrumentation/apache-httpclient/apache-httpclient-4.3/library),<br>[opentelemetry-apache-httpclient-5.2](../instrumentation/apache-httpclient/apache-httpclient-5.2/library)                                                                                                                                                                 | [HTTP Client Spans], [HTTP Client Metrics]                                                                                                        |
 | [Apache ShenYu](https://shenyu.apache.org/)                                                                                                 | 2.4+                               | N/A                                                                                                                                                                                                                                                                                                                                                                                     | Provides `http.route` [2]                                                                                                                         |

instrumentation/apache-elasticjob-3.0/README.md

@@ -0,0 +1,5 @@
+# Settings for the Apache ElasticJob instrumentation
+
+| System property                                                       | Type    | Default | Description                                         |
+|-----------------------------------------------------------------------|---------|---------|-----------------------------------------------------|
+| `otel.instrumentation.apache-elasticjob.experimental-span-attributes` | Boolean | `false` | Enable the capture of experimental span attributes. |

instrumentation/apache-elasticjob-3.0/javaagent/build.gradle.kts

@@ -0,0 +1,25 @@
+plugins {
+  id("otel.javaagent-instrumentation")
+}
+
+muzzle {
+  pass {
+    group.set("org.apache.shardingsphere.elasticjob")
+    module.set("elasticjob-lite-core")
+    versions.set("[3.0.0,)")
+    assertInverse.set(true)
+  }
+}
+
+dependencies {
+  library("org.apache.shardingsphere.elasticjob:elasticjob-lite-core:3.0.0")
+
+  testImplementation("org.apache.curator:curator-test:5.1.0")
+}
+
+tasks.withType<Test>().configureEach {
+  // required on jdk17
+  jvmArgs("--add-opens=java.base/java.lang=ALL-UNNAMED")
+  jvmArgs("-XX:+IgnoreUnrecognizedVMOptions")
+  jvmArgs("-Dotel.instrumentation.apache-elasticjob.experimental-span-attributes=true")
+}

instrumentation/apache-elasticjob-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/apacheelasticjob/v3_0/DataflowJobExecutorInstrumentation.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.apacheelasticjob.v3_0;
+
+import static io.opentelemetry.javaagent.instrumentation.apacheelasticjob.v3_0.ElasticJobSingletons.helper;
+import static net.bytebuddy.matcher.ElementMatchers.isMethod;
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
+
+import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
+import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
+import javax.annotation.Nullable;
+import net.bytebuddy.asm.Advice;
+import net.bytebuddy.description.type.TypeDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+import org.apache.shardingsphere.elasticjob.api.ShardingContext;
+import org.apache.shardingsphere.elasticjob.dataflow.job.DataflowJob;
+
+public class DataflowJobExecutorInstrumentation implements TypeInstrumentation {
+
+  @Override
+  public ElementMatcher<TypeDescription> typeMatcher() {
+    return named("org.apache.shardingsphere.elasticjob.dataflow.executor.DataflowJobExecutor");
+  }
+
+  @Override
+  public void transform(TypeTransformer transformer) {
+    transformer.applyAdviceToMethod(
+        isMethod().and(named("process")).and(takesArguments(4)),
+        DataflowJobExecutorInstrumentation.class.getName() + "$ProcessAdvice");
+  }
+
+  @SuppressWarnings("unused")
+  public static class ProcessAdvice {
+
+    @Nullable
+    @Advice.OnMethodEnter(suppress = Throwable.class)
+    public static ElasticJobHelper.ElasticJobScope onEnter(
+        @Advice.Argument(0) DataflowJob<?> elasticJob,
+        @Advice.Argument(3) ShardingContext shardingContext) {
+
+      ElasticJobProcessRequest request =
+          ElasticJobProcessRequest.createWithUserJobInfo(
+              shardingContext.getJobName(),
+              shardingContext.getTaskId(),
+              shardingContext.getShardingItem(),
+              shardingContext.getShardingTotalCount(),
+              shardingContext.getShardingParameter() != null
+                  ? shardingContext.getShardingParameter()
+                  : "",
+              "DATAFLOW",
+              elasticJob.getClass(),
+              "processData");
+
+      return helper().startSpan(request);
+    }
+
+    @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
+    public static void onExit(
+        @Advice.Enter ElasticJobHelper.ElasticJobScope scope, @Advice.Thrown Throwable throwable) {
+      if (scope != null) {
+        helper().endSpan(scope, throwable);
+      }
+    }
+  }
+}

instrumentation/apache-elasticjob-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/apacheelasticjob/v3_0/ElasticJobCodeAttributesGetter.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.apacheelasticjob.v3_0;
+
+import io.opentelemetry.instrumentation.api.incubator.semconv.code.CodeAttributesGetter;
+import javax.annotation.Nullable;
+
+class ElasticJobCodeAttributesGetter implements CodeAttributesGetter<ElasticJobProcessRequest> {
+  @Nullable
+  @Override
+  public Class<?> getCodeClass(ElasticJobProcessRequest request) {
+    if (request.isScriptJob() || request.isHttpJob()) {
+      return null;
+    }
+    return request.getUserJobClass();
+  }
+
+  @Nullable
+  @Override
+  public String getMethodName(ElasticJobProcessRequest request) {
+    if (request.isScriptJob() || request.isHttpJob()) {
+      return null;
+    }
+    return request.getUserMethodName() != null ? request.getUserMethodName() : "process";
+  }
+}

instrumentation/apache-elasticjob-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/apacheelasticjob/v3_0/ElasticJobExecutorInstrumentation.java

@@ -0,0 +1,98 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.apacheelasticjob.v3_0;
+
+import static io.opentelemetry.javaagent.instrumentation.apacheelasticjob.v3_0.ElasticJobSingletons.helper;
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.takesArgument;
+
+import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
+import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
+import javax.annotation.Nullable;
+import net.bytebuddy.asm.Advice;
+import net.bytebuddy.description.type.TypeDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+import org.apache.shardingsphere.elasticjob.infra.listener.ShardingContexts;
+
+public class ElasticJobExecutorInstrumentation implements TypeInstrumentation {
+  @Override
+  public ElementMatcher<TypeDescription> typeMatcher() {
+    return named("org.apache.shardingsphere.elasticjob.executor.ElasticJobExecutor");
+  }
+
+  @Override
+  public void transform(TypeTransformer transformer) {
+    transformer.applyAdviceToMethod(
+        named("process")
+            .and(
+                takesArgument(
+                    0, named("org.apache.shardingsphere.elasticjob.api.JobConfiguration")))
+            .and(
+                takesArgument(
+                    1,
+                    named("org.apache.shardingsphere.elasticjob.infra.listener.ShardingContexts")))
+            .and(takesArgument(2, int.class))
+            .and(
+                takesArgument(
+                    3,
+                    named("org.apache.shardingsphere.elasticjob.tracing.event.JobExecutionEvent"))),
+        ElasticJobExecutorInstrumentation.class.getName() + "$ElasticJobExecutorAdvice");
+  }
+
+  @SuppressWarnings("unused")
+  public static class ElasticJobExecutorAdvice {
+
+    @Nullable
+    @Advice.OnMethodEnter(suppress = Throwable.class)
+    public static ElasticJobHelper.ElasticJobScope onEnter(
+        @Advice.FieldValue("jobItemExecutor") Object jobItemExecutor,
+        @Advice.Argument(1) ShardingContexts shardingContexts,
+        @Advice.Argument(2) int item) {
+
+      String jobType = determineJobTypeFromExecutor(jobItemExecutor);
+      if (!"SCRIPT".equals(jobType) && !"HTTP".equals(jobType)) {
+        return null;
+      }
+      ElasticJobProcessRequest request =
+          ElasticJobProcessRequest.create(
+              shardingContexts.getJobName(),
+              shardingContexts.getTaskId(),
+              item,
+              shardingContexts.getShardingTotalCount(),
+              shardingContexts.getShardingItemParameters() == null
+                  ? ""
+                  : shardingContexts.getShardingItemParameters().toString(),
+              jobType);
+      return helper().startSpan(request);
+    }
+
+    public static String determineJobTypeFromExecutor(Object jobItemExecutor) {
+      if (jobItemExecutor == null) {
+        return "UNKNOWN";
+      } else {
+        switch (jobItemExecutor.getClass().getSimpleName()) {
+          case "HttpJobExecutor":
+            return "HTTP";
+          case "ScriptJobExecutor":
+            return "SCRIPT";
+          case "SimpleJobExecutor":
+            return "SIMPLE";
+          case "DataflowJobExecutor":
+            return "DATAFLOW";
+          default:
+            return "UNKNOWN";
+        }
+      }
+    }
+
+    @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
+    public static void onExit(
+        @Advice.Enter @Nullable ElasticJobHelper.ElasticJobScope scope,
+        @Advice.Thrown @Nullable Throwable throwable) {
+      helper().endSpan(scope, throwable);
+    }
+  }
+}

instrumentation/apache-elasticjob-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/apacheelasticjob/v3_0/ElasticJobExperimentalAttributeExtractor.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.apacheelasticjob.v3_0;
+
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.common.AttributesBuilder;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor;
+import javax.annotation.Nullable;
+
+class ElasticJobExperimentalAttributeExtractor
+    implements AttributesExtractor<ElasticJobProcessRequest, Void> {
+
+  private static final AttributeKey<String> ELASTICJOB_JOB_NAME =
+      AttributeKey.stringKey("scheduling.apache-elasticjob.job.name");
+  private static final AttributeKey<String> ELASTICJOB_TASK_ID =
+      AttributeKey.stringKey("scheduling.apache-elasticjob.task.id");
+  private static final AttributeKey<Long> ELASTICJOB_ITEM =
+      AttributeKey.longKey("scheduling.apache-elasticjob.item");
+  private static final AttributeKey<Long> ELASTICJOB_SHARDING_TOTAL_COUNT =
+      AttributeKey.longKey("scheduling.apache-elasticjob.sharding.total.count");
+  private static final AttributeKey<String> ELASTICJOB_SHARDING_ITEM_PARAMETERS =
+      AttributeKey.stringKey("scheduling.apache-elasticjob.sharding.item.parameters");
+
+  @Override
+  public void onStart(
+      AttributesBuilder attributes,
+      Context parentContext,
+      ElasticJobProcessRequest elasticJobProcessRequest) {
+    attributes.put(ELASTICJOB_JOB_NAME, elasticJobProcessRequest.getJobName());
+    attributes.put(ELASTICJOB_TASK_ID, elasticJobProcessRequest.getTaskId());
+    attributes.put(ELASTICJOB_ITEM, elasticJobProcessRequest.getItem());
+    attributes.put(
+        ELASTICJOB_SHARDING_TOTAL_COUNT, elasticJobProcessRequest.getShardingTotalCount());
+    if (elasticJobProcessRequest.getShardingItemParameters() != null) {
+      attributes.put(
+          ELASTICJOB_SHARDING_ITEM_PARAMETERS,
+          elasticJobProcessRequest.getShardingItemParameters());
+    }
+  }
+
+  @Override
+  public void onEnd(
+      AttributesBuilder attributes,
+      Context context,
+      ElasticJobProcessRequest elasticJobProcessRequest,
+      @Nullable Void unused,
+      @Nullable Throwable error) {}
+}

instrumentation/apache-elasticjob-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/apacheelasticjob/v3_0/ElasticJobHelper.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.apacheelasticjob.v3_0;
+
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.Scope;
+import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
+import javax.annotation.Nullable;
+
+public final class ElasticJobHelper {
+  private final Instrumenter<ElasticJobProcessRequest, Void> instrumenter;
+
+  private ElasticJobHelper(Instrumenter<ElasticJobProcessRequest, Void> instrumenter) {
+    this.instrumenter = instrumenter;
+  }
+
+  public static ElasticJobHelper create(Instrumenter<ElasticJobProcessRequest, Void> instrumenter) {
+    return new ElasticJobHelper(instrumenter);
+  }
+
+  @Nullable
+  public ElasticJobScope startSpan(ElasticJobProcessRequest request) {
+    Context parentContext = Context.current();
+    if (!this.instrumenter.shouldStart(parentContext, request)) {
+      return null;
+    } else {
+      Context context = this.instrumenter.start(parentContext, request);
+      return new ElasticJobScope(request, context, context.makeCurrent());
+    }
+  }
+
+  public void endSpan(@Nullable ElasticJobScope scope, @Nullable Throwable throwable) {
+    if (scope != null) {
+      if (throwable != null) {
+        scope.request.setFailed();
+      }
+
+      scope.scope.close();
+      this.instrumenter.end(scope.context, scope.request, null, throwable);
+    }
+  }
+
+  public static class ElasticJobScope {
+    private final ElasticJobProcessRequest request;
+    private final Context context;
+    private final Scope scope;
+
+    private ElasticJobScope(ElasticJobProcessRequest request, Context context, Scope scope) {
+      this.request = request;
+      this.context = context;
+      this.scope = scope;
+    }
+  }
+}

instrumentation/apache-elasticjob-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/apacheelasticjob/v3_0/ElasticJobInstrumentationModule.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.apacheelasticjob.v3_0;
+
+import com.google.auto.service.AutoService;
+import io.opentelemetry.javaagent.extension.instrumentation.InstrumentationModule;
+import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
+import io.opentelemetry.javaagent.extension.instrumentation.internal.ExperimentalInstrumentationModule;
+import java.util.Arrays;
+import java.util.List;
+
+@AutoService(InstrumentationModule.class)
+public class ElasticJobInstrumentationModule extends InstrumentationModule
+    implements ExperimentalInstrumentationModule {
+
+  public ElasticJobInstrumentationModule() {
+    super("apache-elasticjob", "apache-elasticjob-3.0");
+  }
+
+  @Override
+  public List<TypeInstrumentation> typeInstrumentations() {
+    return Arrays.asList(
+        new ElasticJobExecutorInstrumentation(),
+        new SimpleJobExecutorInstrumentation(),
+        new DataflowJobExecutorInstrumentation());
+  }
+
+  @Override
+  public boolean isIndyReady() {
+    return true;
+  }
+}