Minor: Fix ingestion pipeline name search not working for observability alert

openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/searchIndex/SearchIndexApp.java

@@ -57,6 +57,7 @@ public class SearchIndexApp extends AbstractNativeApplication {
           "dashboard",
           "topic",
           "pipeline",
+          "ingestionPipeline",
           "searchIndex",
           "user",
           "team",

openmetadata-service/src/main/java/org/openmetadata/service/events/subscription/AlertsRuleEvaluator.java

@@ -350,7 +350,7 @@ public boolean matchIngestionPipelineState(List<String> pipelineState) {
     for (FieldChange fieldChange : changeEvent.getChangeDescription().getFieldsUpdated()) {
       if (fieldChange.getName().equals("pipelineStatus") && fieldChange.getNewValue() != null) {
         PipelineStatus pipelineStatus =
-            JsonUtils.convertValue(fieldChange.getNewValue(), PipelineStatus.class);
+            JsonUtils.readOrConvertValue(fieldChange.getNewValue(), PipelineStatus.class);
         PipelineStatusType status = pipelineStatus.getPipelineState();
         for (String givenStatus : pipelineState) {
           if (givenStatus.equalsIgnoreCase(status.value())) {

openmetadata-service/src/main/java/org/openmetadata/service/formatter/entity/IngestionPipelineFormatter.java

@@ -22,6 +22,7 @@
 import org.openmetadata.schema.type.FieldChange;
 import org.openmetadata.service.formatter.decorators.MessageDecorator;
 import org.openmetadata.service.formatter.util.FormatterUtil;
+import org.openmetadata.service.util.JsonUtils;
 
 public class IngestionPipelineFormatter implements EntityFormatter {
   private static final String PIPELINE_STATUS_FIELD = "pipelineStatus";
@@ -41,10 +42,12 @@ public String format(
   private String transformIngestionPipelineStatus(
       MessageDecorator<?> messageFormatter, FieldChange fieldChange, EntityInterface entity) {
     String ingestionPipelineName = entity.getName();
-    PipelineStatus status = (PipelineStatus) fieldChange.getNewValue();
+    PipelineStatus status =
+        JsonUtils.readOrConvertValue(fieldChange.getNewValue(), PipelineStatus.class);
     if (status != null) {
+      // In case of running
       String date =
-          new SimpleDateFormat("dd/MM/yyyy HH:mm:ss").format(new Date(status.getEndDate()));
+          new SimpleDateFormat("dd/MM/yyyy HH:mm:ss").format(new Date(status.getTimestamp()));
       String format =
           String.format(
               "Ingestion Pipeline %s %s at %s",

openmetadata-service/src/main/java/org/openmetadata/service/resources/events/subscription/EventSubscriptionResource.java

@@ -614,7 +614,7 @@ private List<SubscriptionDestination> getSubscriptions(
     List<SubscriptionDestination> result = new ArrayList<>();
     subscriptions.forEach(
         subscription -> {
-          if (subscription.getId() == null) {
+          if (nullOrEmpty(subscription.getId())) {
             subscription.withId(UUID.randomUUID());
           }
           result.add(subscription);

openmetadata-service/src/main/java/org/openmetadata/service/search/SearchIndexFactory.java

@@ -28,6 +28,7 @@
 import org.openmetadata.schema.entity.services.PipelineService;
 import org.openmetadata.schema.entity.services.SearchService;
 import org.openmetadata.schema.entity.services.StorageService;
+import org.openmetadata.schema.entity.services.ingestionPipelines.IngestionPipeline;
 import org.openmetadata.schema.entity.teams.Team;
 import org.openmetadata.schema.entity.teams.User;
 import org.openmetadata.schema.tests.TestCase;
@@ -49,6 +50,7 @@
 import org.openmetadata.service.search.indexes.EntityReportDataIndex;
 import org.openmetadata.service.search.indexes.GlossaryIndex;
 import org.openmetadata.service.search.indexes.GlossaryTermIndex;
+import org.openmetadata.service.search.indexes.IngestionPipelineIndex;
 import org.openmetadata.service.search.indexes.MessagingServiceIndex;
 import org.openmetadata.service.search.indexes.MetadataServiceIndex;
 import org.openmetadata.service.search.indexes.MlModelIndex;
@@ -82,6 +84,7 @@ public SearchIndex buildIndex(String entityType, Object entity) {
       case Entity.DASHBOARD -> new DashboardIndex((Dashboard) entity);
       case Entity.TOPIC -> new TopicIndex((Topic) entity);
       case Entity.PIPELINE -> new PipelineIndex((Pipeline) entity);
+      case Entity.INGESTION_PIPELINE -> new IngestionPipelineIndex((IngestionPipeline) entity);
       case Entity.USER -> new UserIndex((User) entity);
       case Entity.TEAM -> new TeamIndex((Team) entity);
       case Entity.GLOSSARY -> new GlossaryIndex((Glossary) entity);

openmetadata-service/src/main/java/org/openmetadata/service/search/indexes/IngestionPipelineIndex.java

@@ -0,0 +1,72 @@
+package org.openmetadata.service.search.indexes;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import org.openmetadata.common.utils.CommonUtil;
+import org.openmetadata.schema.entity.services.ingestionPipelines.IngestionPipeline;
+import org.openmetadata.service.Entity;
+import org.openmetadata.service.search.ParseTags;
+import org.openmetadata.service.search.SearchIndexUtils;
+import org.openmetadata.service.search.models.SearchSuggest;
+import org.openmetadata.service.util.JsonUtils;
+
+public class IngestionPipelineIndex implements SearchIndex {
+  final IngestionPipeline ingestionPipeline;
+  final List<String> excludeFields =
+      List.of("changeDescription", "sourceConfig", "openMetadataServerConnection", "airflowConfig");
+
+  public IngestionPipelineIndex(IngestionPipeline ingestionPipeline) {
+    this.ingestionPipeline = ingestionPipeline;
+  }
+
+  public Map<String, Object> buildESDoc() {
+    Map<String, Object> doc = JsonUtils.getMap(ingestionPipeline);
+    SearchIndexUtils.removeNonIndexableFields(doc, excludeFields);
+    List<SearchSuggest> suggest = new ArrayList<>();
+    List<SearchSuggest> serviceSuggest = new ArrayList<>();
+    suggest.add(
+        SearchSuggest.builder().input(ingestionPipeline.getFullyQualifiedName()).weight(5).build());
+    suggest.add(
+        SearchSuggest.builder().input(ingestionPipeline.getDisplayName()).weight(10).build());
+    serviceSuggest.add(
+        SearchSuggest.builder().input(ingestionPipeline.getService().getName()).weight(5).build());
+    ParseTags parseTags =
+        new ParseTags(Entity.getEntityTags(Entity.INGESTION_PIPELINE, ingestionPipeline));
+    doc.put(
+        "name",
+        ingestionPipeline.getName() != null
+            ? ingestionPipeline.getName()
+            : ingestionPipeline.getDisplayName());
+    doc.put(
+        "displayName",
+        ingestionPipeline.getDisplayName() != null
+            ? ingestionPipeline.getDisplayName()
+            : ingestionPipeline.getName());
+    doc.put("followers", SearchIndexUtils.parseFollowers(ingestionPipeline.getFollowers()));
+    doc.put("tags", parseTags.getTags());
+    doc.put("tier", parseTags.getTierTag());
+    doc.put("suggest", suggest);
+    doc.put("service_suggest", serviceSuggest);
+    doc.put("entityType", Entity.INGESTION_PIPELINE);
+    doc.put(
+        "totalVotes",
+        CommonUtil.nullOrEmpty(ingestionPipeline.getVotes())
+            ? 0
+            : ingestionPipeline.getVotes().getUpVotes()
+                - ingestionPipeline.getVotes().getDownVotes());
+    doc.put(
+        "fqnParts",
+        getFQNParts(
+            ingestionPipeline.getFullyQualifiedName(),
+            suggest.stream().map(SearchSuggest::getInput).toList()));
+    doc.put("owner", getEntityWithDisplayName(ingestionPipeline.getOwner()));
+    doc.put("service", getEntityWithDisplayName(ingestionPipeline.getService()));
+    doc.put("domain", getEntityWithDisplayName(ingestionPipeline.getDomain()));
+    return doc;
+  }
+
+  public static Map<String, Float> getFields() {
+    return SearchIndex.getDefaultFields();
+  }
+}