new failures metadata for segment tracking

failure_reasons: array of all failures (as json objects) for a job
- for general analytics on failures
main_failure_reason: main failure reason (as json object) for this job
- for operational usage (for Intercom)
- currently this is just the first failure reason chronologically
    - we'll probably to change this when we have more data on how to
determine failure reasons more intelligently

- added an attempt_id to failures so we can group failures by attempt
- removed stacktrace from failures since it's not clear how we'd use
these in an analytics use case (and because segment has a 32kb size
limit for events)

airbyte-commons/src/main/java/io/airbyte/commons/json/Jsons.java

@@ -14,6 +14,7 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.ObjectWriter;
+import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.google.common.base.Charsets;
 import com.google.common.base.Preconditions;
@@ -91,6 +92,10 @@ public static JsonNode emptyObject() {
     return jsonNode(Collections.emptyMap());
   }
 
+  public static ArrayNode arrayNode() {
+    return OBJECT_MAPPER.createArrayNode();
+  }
+
   public static <T> T object(final JsonNode jsonNode, final Class<T> klass) {
     return OBJECT_MAPPER.convertValue(jsonNode, klass);
   }

airbyte-commons/src/test/java/io/airbyte/commons/json/JsonsTest.java

@@ -134,6 +134,11 @@ void testEmptyObject() {
     assertEquals(Jsons.deserialize("{}"), Jsons.emptyObject());
   }
 
+  @Test
+  void testArrayNode() {
+    assertEquals(Jsons.deserialize("[]"), Jsons.arrayNode());
+  }
+
   @Test
   void testToObject() {
     final ToClass expected = new ToClass("abc", 999, 888L);

airbyte-scheduler/persistence/src/main/java/io/airbyte/scheduler/persistence/job_tracker/TrackingMetadata.java

@@ -4,8 +4,12 @@
 
 package io.airbyte.scheduler.persistence.job_tracker;
 
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableMap.Builder;
+import io.airbyte.commons.json.Jsons;
 import io.airbyte.config.JobOutput;
 import io.airbyte.config.ResourceRequirements;
 import io.airbyte.config.StandardDestinationDefinition;
@@ -16,7 +20,9 @@
 import io.airbyte.scheduler.models.Attempt;
 import io.airbyte.scheduler.models.Job;
 import java.util.List;
+import java.util.Optional;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
 import org.apache.logging.log4j.util.Strings;
 
 public class TrackingMetadata {
@@ -102,9 +108,37 @@ public static ImmutableMap<String, Object> generateJobAttemptMetadata(final Job
             metadata.put("volume_rows", syncSummary.getRecordsSynced());
           }
         }
+
+        final ArrayNode failuresAsJson = failureReasonsAsJsonArray(attempts);
+        if (!failuresAsJson.isEmpty()) {
+          metadata.put("failure_reasons", failuresAsJson.toString());
+          metadata.put("main_failure_reason", failuresAsJson.get(0).toString());
+        }
       }
     }
     return metadata.build();
   }
 
+  private static ArrayNode failureReasonsAsJsonArray(final List<Attempt> attempts) {
+    final ArrayNode failureJsonsArray = Jsons.arrayNode();
+
+    final IntStream attemptIndices = IntStream.range(0, attempts.size());
+    attemptIndices.mapToObj(attemptIndex -> attempts.get(attemptIndex)
+        .getFailureSummary()
+        .map(failureSummary -> failureSummary.getFailures()
+            .stream()
+            .map(failureReason -> {
+              final JsonNode jsonNode = Jsons.jsonNode(failureReason);
+              ((ObjectNode) jsonNode).remove("stacktrace");
+              ((ObjectNode) jsonNode).put("attempt_id", attemptIndex + 1);
+              return jsonNode;
+            })
+            .toList()))
+        .filter(Optional::isPresent)
+        .map(Optional::get)
+        .forEach(failureJsonsArray::addAll);
+
+    return failureJsonsArray;
+  }
+
 }