Gate Spark Plan parsing with flag

Author: charlesmyu

dd-java-agent/instrumentation/spark/spark_2.12/src/main/java/datadog/trace/instrumentation/spark/Spark212Instrumentation.java

@@ -109,7 +109,7 @@ public static class SparkPlanInfoAdvice {
     public static void exit(
         @Advice.Return(readOnly = false) SparkPlanInfo planInfo,
         @Advice.Argument(0) SparkPlan plan) {
-      if (planInfo.metadata().size() == 0) {
+      if (planInfo.metadata().size() == 0 && Config.get().isDataJobsParseSparkPlanEnabled()) {
         Spark212PlanSerializer planUtils = new Spark212PlanSerializer();
         HashMap<String, String> args = new HashMap<>();
         planInfo =

dd-java-agent/instrumentation/spark/spark_2.13/src/main/java/datadog/trace/instrumentation/spark/Spark213Instrumentation.java

@@ -110,7 +110,7 @@ public static class SparkPlanInfoAdvice {
     public static void exit(
         @Advice.Return(readOnly = false) SparkPlanInfo planInfo,
         @Advice.Argument(0) SparkPlan plan) {
-      if (planInfo.metadata().size() == 0) {
+      if (planInfo.metadata().size() == 0 && Config.get().isDataJobsParseSparkPlanEnabled()) {
         Spark213PlanSerializer planUtils = new Spark213PlanSerializer();
         planInfo =
             new SparkPlanInfo(

dd-java-agent/instrumentation/spark/src/testFixtures/groovy/datadog/trace/instrumentation/spark/AbstractSpark24SqlTest.groovy

@@ -16,6 +16,7 @@ abstract class AbstractSpark24SqlTest extends InstrumentationSpecification {
     super.configurePreAgent()
     injectSysConfig("dd.integration.spark.enabled", "true")
     injectSysConfig("dd.integration.openlineage-spark.enabled", "true")
+    injectSysConfig("dd.data.jobs.parse_spark_plan.enabled", "true")
   }
 
   static Dataset<Row> generateSampleDataframe(SparkSession spark) {

dd-java-agent/instrumentation/spark/src/testFixtures/groovy/datadog/trace/instrumentation/spark/AbstractSpark32SqlTest.groovy

@@ -11,6 +11,7 @@ abstract class AbstractSpark32SqlTest extends InstrumentationSpecification {
     super.configurePreAgent()
     injectSysConfig("dd.integration.spark.enabled", "true")
     injectSysConfig("dd.integration.spark-openlineage.enabled", "true")
+    injectSysConfig("dd.data.jobs.parse_spark_plan.enabled", "true")
   }
 
   def "compute a GROUP BY sql query plan"() {

dd-trace-api/src/main/java/datadog/trace/api/ConfigDefaults.java

@@ -231,6 +231,7 @@ public final class ConfigDefaults {
   static final boolean DEFAULT_DATA_JOBS_ENABLED = false;
   static final boolean DEFAULT_DATA_JOBS_OPENLINEAGE_ENABLED = false;
   static final boolean DEFAULT_DATA_JOBS_OPENLINEAGE_TIMEOUT_ENABLED = true;
+  static final boolean DEFAULT_DATA_JOBS_PARSE_SPARK_PLAN_ENABLED = false;
 
   static final boolean DEFAULT_DATA_STREAMS_ENABLED = false;
   static final int DEFAULT_DATA_STREAMS_BUCKET_DURATION = 10; // seconds

dd-trace-api/src/main/java/datadog/trace/api/config/GeneralConfig.java

@@ -80,6 +80,8 @@ public final class GeneralConfig {
   public static final String DATA_JOBS_OPENLINEAGE_ENABLED = "data.jobs.openlineage.enabled";
   public static final String DATA_JOBS_OPENLINEAGE_TIMEOUT_ENABLED =
       "data.jobs.openlineage.timeout.enabled";
+  public static final String DATA_JOBS_PARSE_SPARK_PLAN_ENABLED =
+      "data.jobs.parse_spark_plan.enabled";
 
   public static final String DATA_STREAMS_ENABLED = "data.streams.enabled";
   public static final String DATA_STREAMS_BUCKET_DURATION_SECONDS =

internal-api/src/main/java/datadog/trace/api/Config.java

@@ -50,6 +50,7 @@
 import static datadog.trace.api.ConfigDefaults.DEFAULT_DATA_JOBS_ENABLED;
 import static datadog.trace.api.ConfigDefaults.DEFAULT_DATA_JOBS_OPENLINEAGE_ENABLED;
 import static datadog.trace.api.ConfigDefaults.DEFAULT_DATA_JOBS_OPENLINEAGE_TIMEOUT_ENABLED;
+import static datadog.trace.api.ConfigDefaults.DEFAULT_DATA_JOBS_PARSE_SPARK_PLAN_ENABLED;
 import static datadog.trace.api.ConfigDefaults.DEFAULT_DATA_STREAMS_BUCKET_DURATION;
 import static datadog.trace.api.ConfigDefaults.DEFAULT_DATA_STREAMS_ENABLED;
 import static datadog.trace.api.ConfigDefaults.DEFAULT_DB_CLIENT_HOST_SPLIT_BY_HOST;
@@ -344,6 +345,7 @@
 import static datadog.trace.api.config.GeneralConfig.DATA_JOBS_ENABLED;
 import static datadog.trace.api.config.GeneralConfig.DATA_JOBS_OPENLINEAGE_ENABLED;
 import static datadog.trace.api.config.GeneralConfig.DATA_JOBS_OPENLINEAGE_TIMEOUT_ENABLED;
+import static datadog.trace.api.config.GeneralConfig.DATA_JOBS_PARSE_SPARK_PLAN_ENABLED;
 import static datadog.trace.api.config.GeneralConfig.DATA_STREAMS_BUCKET_DURATION_SECONDS;
 import static datadog.trace.api.config.GeneralConfig.DATA_STREAMS_ENABLED;
 import static datadog.trace.api.config.GeneralConfig.DOGSTATSD_ARGS;
@@ -1189,6 +1191,7 @@ public static String getHostName() {
   private final boolean dataJobsEnabled;
   private final boolean dataJobsOpenLineageEnabled;
   private final boolean dataJobsOpenLineageTimeoutEnabled;
+  private final boolean dataJobsParseSparkPlanEnabled;
 
   private final boolean dataStreamsEnabled;
   private final float dataStreamsBucketDurationSeconds;
@@ -2627,6 +2630,9 @@ PROFILING_DATADOG_PROFILER_ENABLED, isDatadogProfilerSafeInCurrentEnvironment())
     dataJobsOpenLineageTimeoutEnabled =
         configProvider.getBoolean(
             DATA_JOBS_OPENLINEAGE_TIMEOUT_ENABLED, DEFAULT_DATA_JOBS_OPENLINEAGE_TIMEOUT_ENABLED);
+    dataJobsParseSparkPlanEnabled =
+        configProvider.getBoolean(
+            DATA_JOBS_PARSE_SPARK_PLAN_ENABLED, DEFAULT_DATA_JOBS_PARSE_SPARK_PLAN_ENABLED);
 
     dataStreamsEnabled =
         configProvider.getBoolean(DATA_STREAMS_ENABLED, DEFAULT_DATA_STREAMS_ENABLED);
@@ -4556,6 +4562,10 @@ public boolean isDataJobsOpenLineageTimeoutEnabled() {
     return dataJobsOpenLineageTimeoutEnabled;
   }
 
+  public boolean isDataJobsParseSparkPlanEnabled() {
+    return dataJobsParseSparkPlanEnabled;
+  }
+
   public boolean isApmTracingEnabled() {
     return apmTracingEnabled;
   }
@@ -5898,6 +5908,8 @@ public String toString() {
         + dataJobsOpenLineageEnabled
         + ", dataJobsOpenLineageTimeoutEnabled="
         + dataJobsOpenLineageTimeoutEnabled
+        + ", dataJobsParseSparkPlanEnabled="
+        + dataJobsParseSparkPlanEnabled
         + ", apmTracingEnabled="
         + apmTracingEnabled
         + ", jdkSocketEnabled="