feat(search): implement multi-client search engine shim for ES8 support (#14904)

Co-authored-by: Esteban Gutierrez <[email protected]>

Author: RyanHolstien

.github/workflows/build-and-test.yml

@@ -64,8 +64,15 @@ jobs:
       - name: Free up disk space
         run: |
           sudo apt-get remove 'dotnet-*' azure-cli || true
+          sudo rm -rf /usr/local/.ghcup || true
+          sudo rm -rf /usr/share/dotnet || true
+          sudo rm -rf /usr/share/swift || true
+          sudo rm -rf /usr/local/julia* || true
+          sudo rm -rf /usr/local/share/powershell || true
+          sudo rm -rf /usr/share/miniconda || true
           sudo rm -rf /usr/local/lib/android/ || true
-          sudo docker image prune -a -f || true
+          sudo docker system prune -a -f || true
+          df -h
       - uses: szenius/[email protected]
         with:
           timezoneLinux: ${{ matrix.timezone }}
@@ -89,6 +96,13 @@ jobs:
           distribution: "zulu"
           java-version: 17
       - uses: gradle/actions/setup-gradle@v4
+      - name: Disk Space Analysis
+        run: |
+          echo "=== Disk Usage Overview ==="
+          df -h
+
+          echo -e "\n=== Docker Disk Usage ==="
+          docker system df -v
       - name: Gradle build (and test) for NOT metadata ingestion
         if: ${{  matrix.command == 'except_metadata_ingestion' && needs.setup.outputs.backend_change == 'true' }}
         # datahub-schematron:cli excluded due to dependency on metadata-ingestion
@@ -112,6 +126,13 @@ jobs:
             -x :metadata-integration:java:datahub-schematron:cli:test
         env:
           CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
+      - name: Disk Space Analysis
+        run: |
+          echo "=== Disk Usage Overview ==="
+          df -h
+
+          echo -e "\n=== Docker Disk Usage ==="
+          docker system df -v
       - name: Gradle build (and test) for frontend
         if: ${{  matrix.command == 'frontend' && needs.setup.outputs.frontend_change == 'true' }}
         run: |

.github/workflows/docker-unified.yml

@@ -422,11 +422,17 @@ jobs:
       MIXPANEL_PROJECT_ID: ${{ secrets.MIXPANEL_PROJECT_ID }}
     steps:
       - name: Free up disk space
-        if: ${{ !contains(needs.setup.outputs.test_runner_type, 'depot') }}
         run: |
           sudo apt-get remove 'dotnet-*' azure-cli || true
+          sudo rm -rf /usr/local/.ghcup || true
+          sudo rm -rf /usr/share/dotnet || true
+          sudo rm -rf /usr/share/swift || true
+          sudo rm -rf /usr/local/julia* || true
+          sudo rm -rf /usr/local/share/powershell || true
+          sudo rm -rf /usr/share/miniconda || true
           sudo rm -rf /usr/local/lib/android/ || true
-          sudo docker image prune -a -f || true
+          sudo docker system prune -a -f || true
+          df -h
 
       - uses: actions/cache/restore@v4
         with:
@@ -468,6 +474,14 @@ jobs:
           username: ${{ secrets.ACRYL_DOCKER_USERNAME }}
           password: ${{ secrets.ACRYL_DOCKER_PASSWORD }}
 
+      - name: Disk Space Analysis
+        run: |
+          echo "=== Disk Usage Overview ==="
+          df -h
+
+          echo -e "\n=== Docker Disk Usage ==="
+          docker system df -v
+
       - name: build images
         if: ${{ needs.setup.outputs.use_depot_cache != 'true' }}
         run: |
@@ -482,6 +496,13 @@ jobs:
           depot pull --project ${{ env.DEPOT_PROJECT_ID }} ${{ needs.base_build.outputs.build_id }}
           docker images
 
+      - name: Disk Space Analysis
+        run: |
+          echo "=== Disk Usage Overview ==="
+          df -h
+
+          echo -e "\n=== Docker Disk Usage ==="
+          docker system df -v
       - name: run quickstart
         env:
           DATAHUB_TELEMETRY_ENABLED: false

build.gradle

@@ -46,6 +46,7 @@ buildscript {
   ext.neo4jApocVersion = '5.20.0'
   ext.testContainersVersion = '1.21.1'
   ext.elasticsearchVersion = '2.11.1' // ES 7.10, Opensearch 1.x, 2.x
+  ext.elasticsearch8Version = '8.17.4' // ES 8.x Java client
   ext.jacksonVersion = '2.18.4'
   ext.jettyVersion = '12.0.21'
   // see also datahub-frontend/play.gradle
@@ -151,6 +152,8 @@ project.ext.externalDependency = [
     'ebeanDdl': 'io.ebean:ebean-ddl-generator:' + ebeanVersion,
     'ebeanQueryBean': 'io.ebean:querybean-generator:' + ebeanVersion,
     'elasticSearchRest': 'org.opensearch.client:opensearch-rest-high-level-client:' + elasticsearchVersion,
+    // Multi-client shim dependencies
+    'elasticsearch8Client': 'co.elastic.clients:elasticsearch-java:' + elasticsearch8Version,
     'findbugsAnnotations': 'com.google.code.findbugs:annotations:3.0.1',
     'graphqlJava': 'com.graphql-java:graphql-java:22.3',
     'graphqlJavaScalars': 'com.graphql-java:graphql-java-extended-scalars:22.0',

datahub-graphql-core/src/main/java/com/linkedin/datahub/graphql/analytics/service/AnalyticsService.java

@@ -13,6 +13,7 @@
 import com.linkedin.datahub.graphql.types.entitytype.EntityTypeMapper;
 import com.linkedin.metadata.datahubusage.DataHubUsageEventConstants;
 import com.linkedin.metadata.utils.elasticsearch.IndexConvention;
+import com.linkedin.metadata.utils.elasticsearch.SearchClientShim;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -24,7 +25,6 @@
 import org.opensearch.action.search.SearchRequest;
 import org.opensearch.action.search.SearchResponse;
 import org.opensearch.client.RequestOptions;
-import org.opensearch.client.RestHighLevelClient;
 import org.opensearch.index.query.BoolQueryBuilder;
 import org.opensearch.index.query.QueryBuilder;
 import org.opensearch.index.query.QueryBuilders;
@@ -45,7 +45,7 @@
 @RequiredArgsConstructor
 public class AnalyticsService {
 
-  private final RestHighLevelClient _elasticClient;
+  private final SearchClientShim<?> _elasticClient;
   private final IndexConvention _indexConvention;
 
   private static final String FILTERED = "filtered";

datahub-upgrade/src/main/java/com/linkedin/datahub/upgrade/config/NonBlockingConfigs.java

@@ -18,8 +18,8 @@
 import com.linkedin.metadata.search.SearchService;
 import com.linkedin.metadata.search.elasticsearch.ElasticSearchService;
 import com.linkedin.metadata.search.elasticsearch.update.ESWriteDAO;
+import com.linkedin.metadata.utils.elasticsearch.SearchClientShim;
 import io.datahubproject.metadata.context.OperationContext;
-import org.opensearch.client.RestHighLevelClient;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.beans.factory.annotation.Value;
@@ -75,7 +75,7 @@ public NonBlockingSystemUpgrade backfillProcessInstancesHasRunEvents(
       final OperationContext opContext,
       EntityService<?> entityService,
       ElasticSearchService elasticSearchService,
-      RestHighLevelClient restHighLevelClient,
+      SearchClientShim<?> restHighLevelClient,
       @Value("${systemUpdate.processInstanceHasRunEvents.enabled}") final boolean enabled,
       @Value("${systemUpdate.processInstanceHasRunEvents.reprocess.enabled}")
           boolean reprocessEnabled,

datahub-upgrade/src/main/java/com/linkedin/datahub/upgrade/system/dataprocessinstances/BackfillDataProcessInstances.java

@@ -5,9 +5,9 @@
 import com.linkedin.datahub.upgrade.system.NonBlockingSystemUpgrade;
 import com.linkedin.metadata.entity.EntityService;
 import com.linkedin.metadata.search.elasticsearch.ElasticSearchService;
+import com.linkedin.metadata.utils.elasticsearch.SearchClientShim;
 import io.datahubproject.metadata.context.OperationContext;
 import java.util.List;
-import org.opensearch.client.RestHighLevelClient;
 
 public class BackfillDataProcessInstances implements NonBlockingSystemUpgrade {
 
@@ -17,7 +17,7 @@ public BackfillDataProcessInstances(
       OperationContext opContext,
       EntityService<?> entityService,
       ElasticSearchService elasticSearchService,
-      RestHighLevelClient restHighLevelClient,
+      SearchClientShim<?> restHighLevelClient,
       boolean enabled,
       boolean reprocessEnabled,
       Integer batchSize,

datahub-upgrade/src/main/java/com/linkedin/datahub/upgrade/system/dataprocessinstances/BackfillDataProcessInstancesHasRunEventsStep.java

@@ -14,6 +14,7 @@
 import com.linkedin.metadata.entity.EntityService;
 import com.linkedin.metadata.search.elasticsearch.ElasticSearchService;
 import com.linkedin.metadata.utils.elasticsearch.IndexConvention;
+import com.linkedin.metadata.utils.elasticsearch.SearchClientShim;
 import com.linkedin.upgrade.DataHubUpgradeState;
 import io.datahubproject.metadata.context.OperationContext;
 import java.io.IOException;
@@ -28,7 +29,6 @@
 import org.opensearch.action.search.SearchRequest;
 import org.opensearch.action.search.SearchResponse;
 import org.opensearch.client.RequestOptions;
-import org.opensearch.client.RestHighLevelClient;
 import org.opensearch.index.query.QueryBuilder;
 import org.opensearch.index.query.QueryBuilders;
 import org.opensearch.search.aggregations.Aggregation;
@@ -47,7 +47,7 @@ public class BackfillDataProcessInstancesHasRunEventsStep implements UpgradeStep
   private final OperationContext opContext;
   private final EntityService<?> entityService;
   private final ElasticSearchService elasticSearchService;
-  private final RestHighLevelClient restHighLevelClient;
+  private final SearchClientShim<?> restHighLevelClient;
 
   private final boolean reprocessEnabled;
   private final Integer batchSize;
@@ -60,7 +60,7 @@ public BackfillDataProcessInstancesHasRunEventsStep(
       OperationContext opContext,
       EntityService<?> entityService,
       ElasticSearchService elasticSearchService,
-      RestHighLevelClient restHighLevelClient,
+      SearchClientShim<?> restHighLevelClient,
       boolean reprocessEnabled,
       Integer batchSize,
       Integer batchDelayMs,

datahub-upgrade/src/main/java/com/linkedin/datahub/upgrade/system/elasticsearch/steps/BuildIndicesPostStep.java

@@ -63,8 +63,7 @@ public Function<UpgradeContext, UpgradeStepResult> executable() {
           boolean ack =
               esComponents
                   .getSearchClient()
-                  .indices()
-                  .putSettings(request, RequestOptions.DEFAULT)
+                  .updateIndexSettings(request, RequestOptions.DEFAULT)
                   .isAcknowledged();
           log.info(
               "Updated index {} with new settings. Settings: {}, Acknowledged: {}",

datahub-upgrade/src/main/java/com/linkedin/datahub/upgrade/system/elasticsearch/steps/BuildIndicesPreStep.java

@@ -80,8 +80,7 @@ public Function<UpgradeContext, UpgradeStepResult> executable() {
             boolean cloneAck =
                 esComponents
                     .getSearchClient()
-                    .indices()
-                    .clone(resizeRequest, RequestOptions.DEFAULT)
+                    .cloneIndex(resizeRequest, RequestOptions.DEFAULT)
                     .isAcknowledged();
             log.info("Cloned index {} into {}, Acknowledged: {}", indexName, clonedName, cloneAck);
             if (!cloneAck) {
@@ -110,8 +109,7 @@ private boolean blockWrites(String indexName) throws InterruptedException, IOExc
       ack =
           esComponents
               .getSearchClient()
-              .indices()
-              .putSettings(request, RequestOptions.DEFAULT)
+              .updateIndexSettings(request, RequestOptions.DEFAULT)
               .isAcknowledged();
       log.info(
           "Updated index {} with new settings. Settings: {}, Acknowledged: {}",

datahub-upgrade/src/main/java/com/linkedin/datahub/upgrade/system/elasticsearch/steps/CleanIndicesStep.java

@@ -9,24 +9,24 @@
 import com.linkedin.metadata.config.search.ElasticSearchConfiguration;
 import com.linkedin.metadata.search.elasticsearch.indexbuilder.ESIndexBuilder;
 import com.linkedin.metadata.shared.ElasticSearchIndexed;
+import com.linkedin.metadata.utils.elasticsearch.SearchClientShim;
 import com.linkedin.structured.StructuredPropertyDefinition;
 import com.linkedin.upgrade.DataHubUpgradeState;
 import com.linkedin.util.Pair;
 import java.util.List;
 import java.util.Set;
 import java.util.function.Function;
 import lombok.extern.slf4j.Slf4j;
-import org.opensearch.client.RestHighLevelClient;
 
 @Slf4j
 public class CleanIndicesStep implements UpgradeStep {
-  private final RestHighLevelClient searchClient;
+  private final SearchClientShim<?> searchClient;
   private final ElasticSearchConfiguration esConfig;
   private final List<ElasticSearchIndexed> indexedServices;
   private final Set<Pair<Urn, StructuredPropertyDefinition>> structuredProperties;
 
   public CleanIndicesStep(
-      final RestHighLevelClient searchClient,
+      final SearchClientShim<?> searchClient,
       final ElasticSearchConfiguration esConfig,
       final List<ElasticSearchIndexed> indexedServices,
       final Set<Pair<Urn, StructuredPropertyDefinition>> structuredProperties) {