fix(elasticsearch): fix regression system-update elasticsearch-setup (#15107)

Author: david-leifker

datahub-upgrade/build.gradle

@@ -72,6 +72,7 @@ dependencies {
   implementation externalDependency.slf4jApi
   compileOnly externalDependency.lombok
   implementation externalDependency.picocli
+  implementation externalDependency.resilience4j
   implementation externalDependency.parquet
   implementation externalDependency.protobuf
   implementation externalDependency.springBeans

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

@@ -4,10 +4,12 @@
 import com.linkedin.datahub.upgrade.UpgradeStep;
 import com.linkedin.datahub.upgrade.UpgradeStepResult;
 import com.linkedin.datahub.upgrade.impl.DefaultUpgradeStepResult;
+import com.linkedin.datahub.upgrade.system.elasticsearch.util.IndexUtils;
 import com.linkedin.datahub.upgrade.system.elasticsearch.util.UsageEventIndexUtils;
 import com.linkedin.gms.factory.config.ConfigurationProvider;
 import com.linkedin.gms.factory.search.BaseElasticSearchComponentsFactory;
 import com.linkedin.upgrade.DataHubUpgradeState;
+import io.datahubproject.metadata.context.OperationContext;
 import java.util.function.Function;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -28,28 +30,32 @@ public int retryCount() {
     return 3;
   }
 
+  @Override
+  public boolean skip(UpgradeContext context) {
+    boolean analyticsEnabled = configurationProvider.getPlatformAnalytics().isEnabled();
+    if (!analyticsEnabled) {
+      log.info("DataHub analytics is disabled, skipping usage event index setup");
+    }
+    return !analyticsEnabled;
+  }
+
   @Override
   public Function<UpgradeContext, UpgradeStepResult> executable() {
     return (context) -> {
       try {
-        boolean analyticsEnabled = configurationProvider.getPlatformAnalytics().isEnabled();
+
         String indexPrefix = configurationProvider.getElasticSearch().getIndex().getPrefix();
         // Handle null prefix by converting to empty string
         if (indexPrefix == null) {
           indexPrefix = "";
         }
 
-        if (!analyticsEnabled) {
-          log.info("DataHub analytics is disabled, skipping usage event index setup");
-          return new DefaultUpgradeStepResult(id(), DataHubUpgradeState.SUCCEEDED);
-        }
-
         boolean useOpenSearch = esComponents.getSearchClient().getEngineType().isOpenSearch();
         int numShards = esComponents.getIndexBuilder().getNumShards();
         int numReplicas = esComponents.getIndexBuilder().getNumReplicas();
 
         if (useOpenSearch) {
-          setupOpenSearchUsageEvents(indexPrefix, numShards, numReplicas);
+          setupOpenSearchUsageEvents(indexPrefix, numShards, numReplicas, context.opContext());
         } else {
           setupElasticsearchUsageEvents(indexPrefix, numShards, numReplicas);
         }
@@ -64,10 +70,10 @@ public Function<UpgradeContext, UpgradeStepResult> executable() {
 
   private void setupElasticsearchUsageEvents(String prefix, int numShards, int numReplicas)
       throws Exception {
-    String separator = prefix.isEmpty() ? "" : "_";
-    String prefixedPolicy = prefix + separator + "datahub_usage_event_policy";
-    String prefixedTemplate = prefix + separator + "datahub_usage_event_index_template";
-    String prefixedDataStream = prefix + separator + "datahub_usage_event";
+    String prefixedPolicy = IndexUtils.createPrefixedName(prefix, "datahub_usage_event_policy");
+    String prefixedTemplate =
+        IndexUtils.createPrefixedName(prefix, "datahub_usage_event_index_template");
+    String prefixedDataStream = IndexUtils.createPrefixedName(prefix, "datahub_usage_event");
 
     // Create ILM policy
     UsageEventIndexUtils.createIlmPolicy(esComponents, prefixedPolicy);
@@ -80,21 +86,36 @@ private void setupElasticsearchUsageEvents(String prefix, int numShards, int num
     UsageEventIndexUtils.createDataStream(esComponents, prefixedDataStream);
   }
 
-  private void setupOpenSearchUsageEvents(String prefix, int numShards, int numReplicas)
+  private void setupOpenSearchUsageEvents(
+      String prefix, int numShards, int numReplicas, OperationContext operationContext)
       throws Exception {
-    String separator = prefix.isEmpty() ? "" : "_";
-    String prefixedPolicy = prefix + separator + "datahub_usage_event_policy";
-    String prefixedTemplate = prefix + separator + "datahub_usage_event_index_template";
-    String prefixedIndex = prefix + separator + "datahub_usage_event-000001";
-
-    // Create ISM policy
-    UsageEventIndexUtils.createIsmPolicy(esComponents, prefixedPolicy, prefix);
-
-    // Create index template
-    UsageEventIndexUtils.createOpenSearchIndexTemplate(
-        esComponents, prefixedTemplate, numShards, numReplicas, prefix);
-
-    // Create initial index
-    UsageEventIndexUtils.createOpenSearchIndex(esComponents, prefixedIndex, prefix);
+    String prefixedPolicy = IndexUtils.createPrefixedName(prefix, "datahub_usage_event_policy");
+    String prefixedTemplate =
+        IndexUtils.createPrefixedName(prefix, "datahub_usage_event_index_template");
+    String prefixedIndex = IndexUtils.createPrefixedName(prefix, "datahub_usage_event-000001");
+
+    // Create ISM policy (both AWS and self-hosted OpenSearch use the same format)
+    boolean policyCreated =
+        UsageEventIndexUtils.createIsmPolicy(
+            esComponents, prefixedPolicy, prefix, operationContext);
+    log.info("ISM policy creation result: {}", policyCreated);
+
+    if (policyCreated) {
+      log.info("ISM policy created successfully, proceeding with template and index creation");
+
+      // Create index template (both AWS and self-hosted OpenSearch use the same format and
+      // endpoint)
+      log.info("Creating index template: {}", prefixedTemplate);
+      UsageEventIndexUtils.createOpenSearchIndexTemplate(
+          esComponents, prefixedTemplate, numShards, numReplicas, prefix);
+
+      // Create initial numbered index (both AWS and self-hosted OpenSearch use the same approach)
+      log.info("Creating initial index: {}", prefixedIndex);
+      UsageEventIndexUtils.createOpenSearchIndex(esComponents, prefixedIndex, prefix);
+    } else {
+      log.warn(
+          "ISM policy creation failed or is not supported. Skipping template and index creation to avoid configuration issues.");
+      log.info("Usage event tracking will not be available without proper policy configuration.");
+    }
   }
 }