NuOpenLineageLog - Adding or setting log level to info for debugging the metrics degradation issue. This version should not be merged to main/master

Author: jrosend

client/java/src/main/java/io/openlineage/client/OpenLineageClient.java

@@ -77,7 +77,7 @@ public void emit(@NonNull OpenLineage.RunEvent runEvent) {
     }
     if (circuitBreaker.isPresent() && circuitBreaker.get().currentState().isClosed()) {
       engagedCircuitBreaker.set(1);
-      log.warn("OpenLineageClient disabled with circuit breaker");
+      log.warn("NuOpenLineageLog: OpenLineageClient: emit: OpenLineageClient disabled with circuit breaker");
       return;
     } else {
       engagedCircuitBreaker.set(0);

integration/spark/app/src/main/java/io/openlineage/spark/agent/NuEventEmitter.java

@@ -1,6 +1,7 @@
 package io.openlineage.spark.agent;
 
 import io.openlineage.client.OpenLineage;
+import io.openlineage.client.OpenLineageClientUtils;
 import lombok.extern.slf4j.Slf4j;
 
 import java.lang.reflect.Field;
@@ -19,6 +20,19 @@
 @Slf4j
 public class NuEventEmitter {
 
+    public static long estimateShallowSize() {
+        return 24;
+    }
+
+    public static long estimateDeepSize(String str) {
+        long shallowSize = estimateShallowSize();
+        boolean isLatin1 = str.chars().allMatch(ch -> ch <= 0xFF);
+        long dataSize = isLatin1 ? str.length() : str.length() * 2L;
+        long arraySize = 12 + 4 + dataSize;
+        arraySize = (arraySize + 7) & ~7;
+        return shallowSize + arraySize;
+    }
+
     private static final Set<String> WANTED_JOB_TYPES = new HashSet<>(
             Collections.singletonList(
                     "SQL_JOB" // as defined in SparkSQLExecutionContext.SPARK_JOB_TYPE
@@ -35,15 +49,15 @@ public class NuEventEmitter {
     private static Boolean isPermittedJobType(RunEvent event) {
         String jobType = event.getJob().getFacets().getJobType().getJobType();
         if (WANTED_JOB_TYPES.stream().noneMatch(jobType::equals)) {
-            log.info("NuOpenLineageLog: OpenLineage event with job type {} has no lineage value and should not be emitted", jobType);
+            log.info("NuOpenLineageLog: NuEventEmitter: isPermittedJobType: OpenLineage event with job type {} has no lineage value and should not be emitted", jobType);
             return false;
         }
         return true;
     }
 
-    private static Boolean isPermitedEventType(RunEvent event) {
+    private static Boolean isPermittedEventType(RunEvent event) {
         if (RUNNING.equals(event.getEventType())) {
-            log.info("NuOpenLineageLog: OpenLineage event is {} and should not be emitted", RUNNING);
+            log.info("NuOpenLineageLog: NuEventEmitter: isPermittedEventType: OpenLineage event is {} and should not be emitted", RUNNING);
             return false;
         }
         return true;
@@ -52,11 +66,11 @@ private static Boolean isPermitedEventType(RunEvent event) {
     private static Boolean isPermittedJobName(RunEvent event) {
         String jobName = event.getJob().getName();
         if (isNull(jobName)) {
-            log.info("NuOpenLineageLog: OpenLineage event has no job name and should not be emitted");
+            log.info("NuOpenLineageLog: NuEventEmitter: isPermittedJobName: OpenLineage event has no job name and should not be emitted");
             return false;
         }
         if (WANTED_EVENT_NAME_SUBSTRINGS.stream().noneMatch(jobName::contains)) {
-            log.info("NuOpenLineageLog: OpenLineage event job name {} has no permitted substring and should not be emitted", jobName);
+            log.info("NuOpenLineageLog: NuEventEmitter: isPermittedJobName: OpenLineage event job name {} has no permitted substring and should not be emitted", jobName);
             return false;
         }
         return true;
@@ -65,7 +79,7 @@ private static Boolean isPermittedJobName(RunEvent event) {
     private static Boolean shouldEmit(RunEvent event) {
         return Stream.of(
                 isPermittedJobType(event),
-                isPermitedEventType(event),
+                isPermittedEventType(event),
                 isPermittedJobName(event)
         ).noneMatch(Boolean.FALSE::equals);
     }
@@ -83,34 +97,38 @@ private static void discardColumnLineageFacet(RunEvent event) {
                     .collect(Collectors.toList())
                     .forEach(dataset -> {
                         try {
-                            log.info("NuOpenLineageLog: Discarding column lineage facet for dataset {} {} {}",
+                            log.info("NuOpenLineageLog: NuEventEmitter: discardColumnLineageFacet: Discarding column lineage facet for dataset {} {} {}",
                                     dataset.getClass().getSimpleName(), dataset.getNamespace(), dataset.getName());
                             columnLineageFacetField.set(dataset.getFacets(), null);
                         } catch (IllegalAccessException e) {
-                            log.info("NuOpenLineageLog: Failed to discard column lineage facet", e);
+                            log.info("NuOpenLineageLog: NuEventEmitter: discardColumnLineageFacet:  Failed to discard column lineage facet", e);
                         }
                     });
         } catch (NoSuchFieldException e) {
-            log.info("NuOpenLineageLog: Failed to discard column lineage facet: columnLineage field not found at OpenLineage.DatasetFacets", e);
+            log.info("NuOpenLineageLog: NuEventEmitter: discardColumnLineageFacet: Failed to discard column lineage facet: columnLineage field not found at OpenLineage.DatasetFacets", e);
         }
     }
 
     public static void emit(RunEvent event, EventEmitter eventEmitter) {
-        log.info("NuOpenLineageLog: Begin: Emitting OpenLineage event {} with job name {} and job type {}",
-                event.getEventType(), event.getJob().getName(), event.getJob().getFacets().getJobType().getJobType());
+        String jsonEvent = OpenLineageClientUtils.toJson(event);
+        double estimatedEventSize = estimateDeepSize(jsonEvent) / 1024.0;
+
+        log.info("NuOpenLineageLog: NuEventEmitter: emit:  Begin: Emitting OpenLineage event {} with job name {} and job type {} and size {} KB",
+                event.getEventType(), event.getJob().getName(), event.getJob().getFacets().getJobType().getJobType(), estimatedEventSize);
         if (!shouldEmit(event)) {
-            log.info("NuOpenLineageLog: OpenLineage event {} has no lineage value and should not be emitted", event.getEventType());
+            log.info("NuOpenLineageLog: NuEventEmitter: emit: OpenLineage event {} has no lineage value and should not be emitted", event.getEventType());
             return;
         }
 
         if (shouldDiscardColumnLineageFacet(event.getEventType())) {
-            log.info("NuOpenLineageLog: Discarding column lineage facet for event {}", event.getEventType());
+            log.info("NuOpenLineageLog: NuEventEmitter: emit: Discarding column lineage facet for event {}", event.getEventType());
             discardColumnLineageFacet(event);
         }
 
         eventEmitter.emit(event);
 
-        log.info("NuOpenLineageLog: End: Emitting OpenLineage event {} with job name {} and job type {}",
-                event.getEventType(), event.getJob().getName(), event.getJob().getFacets().getJobType().getJobType());
+        log.info("NuOpenLineageLog: NuEventEmitter: emit: End: Emitting OpenLineage event {} with job name {} and job type {} and size {} KB",
+                event.getEventType(), event.getJob().getName(), event.getJob().getFacets().getJobType().getJobType(), estimatedEventSize);
     }
+
 }