feat(gms): store update events in a new index in ElasticSearch

docker/elasticsearch-setup/create-indices.sh

@@ -107,6 +107,10 @@ function create_datahub_usage_event_datastream() {
   create_if_not_exists "_index_template/${PREFIX}datahub_usage_event_index_template" index_template.json
   #   3. although indexing request creates the data stream, it's not queryable before creation, causing GMS to throw exceptions
   create_if_not_exists "_data_stream/${PREFIX}datahub_usage_event" "datahub_usage_event"
+
+  # Create index template for update_events
+  create_if_not_exists "_index_template/${PREFIX}datahub_update_event_index_template" update_event_template.json
+  create_if_not_exists "_data_stream/${PREFIX}datahub_update_event" "datahub_update_event"
 }
 
 # create indices for ES OSS (AWS)

metadata-io/src/main/java/com/linkedin/metadata/search/EntitySearchService.java

@@ -38,6 +38,13 @@ public interface EntitySearchService {
    */
   void upsertDocument(@Nonnull String entityName, @Nonnull String document, @Nonnull String docId);
 
+  /**
+   *
+   * @param document the document to update into update index
+   * @param docId the ID of the document
+   */
+  void createUpdateDocument(@Nonnull String document, @Nonnull String docId);
+
   /**
    * Deletes the document with the given document ID from the index.
    *

metadata-io/src/main/java/com/linkedin/metadata/search/elasticsearch/ElasticSearchService.java

@@ -70,6 +70,12 @@ public void upsertDocument(@Nonnull String entityName, @Nonnull String document,
     esWriteDAO.upsertDocument(entityName, document, docId);
   }
 
+  @Override
+  public void createUpdateDocument(@Nonnull String document, @Nonnull String docId) {
+    log.debug(String.format("Creating Update document document: %s, docId %s", document, docId));
+    esWriteDAO.createUpdateDocument(document, docId);
+  }
+
   @Override
   public void deleteDocument(@Nonnull String entityName, @Nonnull String docId) {
     log.debug(String.format("Deleting Search document entityName: %s, docId: %s", entityName, docId));

metadata-io/src/main/java/com/linkedin/metadata/search/elasticsearch/update/ESWriteDAO.java

@@ -6,7 +6,9 @@
 import javax.annotation.Nonnull;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.elasticsearch.action.DocWriteRequest;
 import org.elasticsearch.action.delete.DeleteRequest;
+import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.update.UpdateRequest;
 import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.RestHighLevelClient;
@@ -46,6 +48,23 @@ public void upsertDocument(@Nonnull String entityName, @Nonnull String document,
     bulkProcessor.add(updateRequest);
   }
 
+  /**
+   * Creates a request to insert new document into datahub_update_event index
+   *
+   * @param document the document to insert
+   * @param docId the ID of the document
+   */
+  public void createUpdateDocument(@Nonnull String document, @Nonnull String docId) {
+    final String indexName = indexConvention.getIndexName("datahub_update_event");
+    final IndexRequest indexRequest = new IndexRequest(
+        indexName)
+        .id(docId)
+        .source(document, XContentType.JSON)
+        .opType(DocWriteRequest.OpType.CREATE);
+
+    bulkProcessor.add(indexRequest);
+  }
+
   /**
    * Deletes the document with the given document ID from the index.
    *

metadata-io/src/main/java/com/linkedin/metadata/search/transformer/SearchDocumentTransformer.java

@@ -4,6 +4,7 @@
 import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.JsonNodeFactory;
 import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.linkedin.common.AuditStamp;
 import com.linkedin.common.urn.Urn;
 import com.linkedin.data.schema.DataSchema;
 import com.linkedin.data.template.RecordTemplate;
@@ -14,6 +15,7 @@
 import com.linkedin.metadata.models.annotation.SearchableAnnotation.FieldType;
 import com.linkedin.metadata.models.extractor.FieldExtractor;
 
+import com.linkedin.mxe.MetadataChangeLog;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -82,6 +84,23 @@ public Optional<String> transformAspect(
     return Optional.of(searchDocument.toString());
   }
 
+  public String transformEvent(MetadataChangeLog event) {
+    final ObjectNode searchDocument = JsonNodeFactory.instance.objectNode();
+    searchDocument.put("urn", event.getEntityUrn().toString());
+    searchDocument.put("changeType", event.getChangeType().toString());
+    searchDocument.put("entityType", event.getEntityType().toString());
+    searchDocument.put("aspectName", event.getAspectName().toString());
+
+    AuditStamp eventCreationInfo = event.getCreated();
+    String actor = eventCreationInfo.getActor().toString();
+    Long time = eventCreationInfo.getTime();
+
+    searchDocument.put("actorUrn", actor);
+    searchDocument.put("timestamp", time);
+    searchDocument.put("@timestamp", time);
+    return searchDocument.toString();
+  }
+
   public void setSearchableValue(final SearchableFieldSpec fieldSpec, final List<Object> fieldValues,
       final ObjectNode searchDocument, final Boolean forDelete) {
     DataSchema.Type valueType = fieldSpec.getPegasusSchema().getType();

metadata-io/src/main/java/com/linkedin/metadata/search/utils/SearchUtils.java

@@ -22,6 +22,10 @@
 import java.io.InputStream;
 import java.io.UnsupportedEncodingException;
 import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.Base64;
 import java.util.Collections;
 import java.util.Map;
 import java.util.Optional;
@@ -65,6 +69,16 @@ public static Optional<String> getDocId(@Nonnull Urn urn) {
     }
   }
 
+  public static String getDocHash(@Nonnull String aspectValue) {
+    try {
+      MessageDigest hashDigest = MessageDigest.getInstance("SHA-256");
+      String hashId = Base64.getEncoder().encodeToString(hashDigest.digest(aspectValue.getBytes(StandardCharsets.UTF_8)));
+      return hashId;
+    } catch (NoSuchAlgorithmException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
   /**
    * Validates the request params and create a request map out of it.
    *

metadata-jobs/mae-consumer/src/main/java/com/linkedin/metadata/kafka/hook/UpdateIndicesHook.java

@@ -194,6 +194,9 @@ private void handleUpdateChangeEvent(@Nonnull final MetadataChangeLog event) {
     } else {
       updateGraphService(urn, aspectSpec, aspect, event);
     }
+
+    // Populate update index with update event
+    updateUpdateIndex(event);
   }
 
   /**
@@ -489,6 +492,20 @@ private void updateTimeseriesFields(String entityType, String aspectName, Urn ur
     });
   }
 
+  /**
+   * Process event and update Update index in Elastic
+   */
+  private void updateUpdateIndex(@Nonnull final MetadataChangeLog event) {
+    String updateDocument = _searchDocumentTransformer.transformEvent(event);
+
+    // Generating a hash using event urn, event content and time, to be used as a unique id for ES documents
+    String stringToBeHashed = event.getEntityType().toString() + '_'
+        + event.getAspect().getValue().asAvroString()
+        + '_' + event.getCreated().getTime().toString();
+    String docHash = SearchUtils.getDocHash(stringToBeHashed);
+    _entitySearchService.createUpdateDocument(updateDocument, docHash);
+  }
+
   private void updateSystemMetadata(SystemMetadata systemMetadata, Urn urn, AspectSpec aspectSpec, RecordTemplate aspect) {
     _systemMetadataService.insert(systemMetadata, urn.toString(), aspectSpec.getName());
 

metadata-service/restli-servlet-impl/src/main/resources/index/usage-event/update_event_template.json

@@ -0,0 +1,23 @@
+{
+  "index_patterns": ["*PREFIXdatahub_update_event*"],
+  "data_stream": { },
+  "priority": 499,
+  "template": {
+    "mappings": {
+      "properties": {
+        "@timestamp": {
+          "type": "date"
+        },
+        "type": {
+          "type": "keyword"
+        },
+        "timestamp": {
+          "type": "date"
+        }
+      }
+    },
+    "settings": {
+      "index.lifecycle.name": "PREFIXdatahub_usage_event_policy"
+    }
+  }
+}