fix(prefix): improve/fix es prefix calculation (#15110)

This commit is contained in:
david-leifker 2025-10-25 15:11:47 -05:00 committed by GitHub
parent bc25b82da5
commit ee20dbbcb3
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 116 additions and 121 deletions

View File

@ -4,7 +4,6 @@ import com.linkedin.datahub.upgrade.UpgradeContext;
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;
@ -44,11 +43,8 @@ public class CreateUsageEventIndicesStep implements UpgradeStep {
return (context) -> {
try {
String indexPrefix = configurationProvider.getElasticSearch().getIndex().getPrefix();
// Handle null prefix by converting to empty string
if (indexPrefix == null) {
indexPrefix = "";
}
final String indexPrefix =
configurationProvider.getElasticSearch().getIndex().getFinalPrefix();
boolean useOpenSearch = esComponents.getSearchClient().getEngineType().isOpenSearch();
int numShards = esComponents.getIndexBuilder().getNumShards();
@ -70,10 +66,9 @@ public class CreateUsageEventIndicesStep implements UpgradeStep {
private void setupElasticsearchUsageEvents(String prefix, int numShards, int numReplicas)
throws Exception {
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");
String prefixedPolicy = prefix + "datahub_usage_event_policy";
String prefixedTemplate = prefix + "datahub_usage_event_index_template";
String prefixedDataStream = prefix + "datahub_usage_event";
// Create ILM policy
UsageEventIndexUtils.createIlmPolicy(esComponents, prefixedPolicy);
@ -89,10 +84,9 @@ public class CreateUsageEventIndicesStep implements UpgradeStep {
private void setupOpenSearchUsageEvents(
String prefix, int numShards, int numReplicas, OperationContext operationContext)
throws Exception {
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");
String prefixedPolicy = prefix + "datahub_usage_event_policy";
String prefixedTemplate = prefix + "datahub_usage_event_index_template";
String prefixedIndex = prefix + "datahub_usage_event-000001";
// Create ISM policy (both AWS and self-hosted OpenSearch use the same format)
boolean policyCreated =

View File

@ -145,25 +145,6 @@ public class IndexUtils {
return -1;
}
/**
* Creates a prefixed name for usage event resources.
*
* <p>This method handles the logic for adding prefixes to usage event resource names, including
* the proper separator handling. If the prefix is empty, no separator is added. If the prefix is
* not empty, an underscore separator is added between the prefix and the resource name.
*
* @param prefix the index prefix (e.g., "prod", "dev", or empty string)
* @param resourceName the base resource name (e.g., "datahub_usage_event_policy")
* @return the prefixed resource name (e.g., "prod_datahub_usage_event_policy" or
* "datahub_usage_event_policy")
*/
public static String createPrefixedName(String prefix, String resourceName) {
if (prefix == null || prefix.isEmpty()) {
return resourceName;
}
return prefix + "_" + resourceName;
}
/**
* Loads a resource file as a UTF-8 encoded string.
*

View File

@ -1,5 +1,7 @@
package com.linkedin.datahub.upgrade.system.elasticsearch.steps;
import static org.mockito.Mockito.atLeastOnce;
import com.linkedin.datahub.upgrade.UpgradeContext;
import com.linkedin.datahub.upgrade.UpgradeStepResult;
import com.linkedin.gms.factory.config.ConfigurationProvider;
@ -60,7 +62,7 @@ public class CreateUsageEventIndicesStepTest {
Mockito.when(configurationProvider.getPlatformAnalytics()).thenReturn(platformAnalytics);
Mockito.when(configurationProvider.getElasticSearch()).thenReturn(elasticSearch);
Mockito.when(elasticSearch.getIndex()).thenReturn(index);
Mockito.when(index.getPrefix()).thenReturn("test_");
Mockito.when(index.getFinalPrefix()).thenReturn("test_");
Mockito.when(upgradeContext.opContext()).thenReturn(opContext);
@ -279,7 +281,7 @@ public class CreateUsageEventIndicesStepTest {
// Arrange
Mockito.when(platformAnalytics.isEnabled()).thenReturn(true);
Mockito.when(searchEngineType.isOpenSearch()).thenReturn(false);
Mockito.when(index.getPrefix()).thenReturn(""); // Empty prefix
Mockito.when(index.getFinalPrefix()).thenReturn(""); // Empty prefix
// Act
Function<UpgradeContext, UpgradeStepResult> executable = step.executable();
@ -291,44 +293,7 @@ public class CreateUsageEventIndicesStepTest {
Assert.assertEquals(result.result(), DataHubUpgradeState.SUCCEEDED);
// Verify empty prefix was used and no underscore separator was added
Mockito.verify(index).getPrefix();
// Verify that the low-level requests were made with correct names (no underscore prefix)
Mockito.verify(searchClient, Mockito.atLeast(2)).performLowLevelRequest(Mockito.any());
// Verify specific endpoint calls were made
Mockito.verify(searchClient)
.performLowLevelRequest(
Mockito.argThat(
request ->
request.getEndpoint().equals("/_ilm/policy/datahub_usage_event_policy")));
Mockito.verify(searchClient)
.performLowLevelRequest(
Mockito.argThat(
request ->
request
.getEndpoint()
.equals("/_index_template/datahub_usage_event_index_template")));
}
@Test
public void testExecutable_WithNullPrefix() throws Exception {
// Arrange
Mockito.when(platformAnalytics.isEnabled()).thenReturn(true);
Mockito.when(searchEngineType.isOpenSearch()).thenReturn(false);
Mockito.when(index.getPrefix()).thenReturn(null); // Null prefix
// Act
Function<UpgradeContext, UpgradeStepResult> executable = step.executable();
UpgradeStepResult result = executable.apply(upgradeContext);
// Assert
Assert.assertNotNull(result);
Assert.assertEquals(result.stepId(), "CreateUsageEventIndicesStep");
Assert.assertEquals(result.result(), DataHubUpgradeState.SUCCEEDED);
// Verify null prefix was handled and no underscore separator was added
Mockito.verify(index).getPrefix();
Mockito.verify(index, atLeastOnce()).getFinalPrefix();
// Verify that the low-level requests were made with correct names (no underscore prefix)
Mockito.verify(searchClient, Mockito.atLeast(2)).performLowLevelRequest(Mockito.any());
@ -353,7 +318,7 @@ public class CreateUsageEventIndicesStepTest {
// Arrange
Mockito.when(platformAnalytics.isEnabled()).thenReturn(true);
Mockito.when(searchEngineType.isOpenSearch()).thenReturn(false);
Mockito.when(index.getPrefix()).thenReturn("prod"); // Non-empty prefix
Mockito.when(index.getFinalPrefix()).thenReturn("prod_"); // Non-empty prefix
// Act
Function<UpgradeContext, UpgradeStepResult> executable = step.executable();
@ -365,7 +330,7 @@ public class CreateUsageEventIndicesStepTest {
Assert.assertEquals(result.result(), DataHubUpgradeState.SUCCEEDED);
// Verify non-empty prefix was used and underscore separator was added
Mockito.verify(index).getPrefix();
Mockito.verify(index, atLeastOnce()).getFinalPrefix();
// Verify that the low-level requests were made with correct names (with underscore prefix)
Mockito.verify(searchClient)
@ -387,8 +352,8 @@ public class CreateUsageEventIndicesStepTest {
// Arrange
Mockito.when(platformAnalytics.isEnabled()).thenReturn(true);
Mockito.when(searchEngineType.isOpenSearch()).thenReturn(false);
Mockito.when(index.getPrefix())
.thenReturn("kbcpyv7ss3-staging-test"); // Specific prefix from issue
Mockito.when(index.getFinalPrefix())
.thenReturn("kbcpyv7ss3-staging-test_"); // Specific prefix from issue
// Act
Function<UpgradeContext, UpgradeStepResult> executable = step.executable();
@ -400,7 +365,7 @@ public class CreateUsageEventIndicesStepTest {
Assert.assertEquals(result.result(), DataHubUpgradeState.SUCCEEDED);
// Verify specific prefix was used and underscore separator was added
Mockito.verify(index).getPrefix();
Mockito.verify(index, atLeastOnce()).getFinalPrefix();
// Verify that the low-level requests were made with correct names (with underscore prefix)
Mockito.verify(searchClient)
@ -426,7 +391,7 @@ public class CreateUsageEventIndicesStepTest {
// Arrange
Mockito.when(platformAnalytics.isEnabled()).thenReturn(true);
Mockito.when(searchEngineType.isOpenSearch()).thenReturn(true);
Mockito.when(index.getPrefix()).thenReturn(""); // Empty prefix
Mockito.when(index.getFinalPrefix()).thenReturn(""); // Empty prefix
// Act
Function<UpgradeContext, UpgradeStepResult> executable = step.executable();
@ -439,7 +404,7 @@ public class CreateUsageEventIndicesStepTest {
// Verify OpenSearch path was taken and empty prefix was used
Mockito.verify(searchEngineType).isOpenSearch();
Mockito.verify(index).getPrefix();
Mockito.verify(index, atLeastOnce()).getFinalPrefix();
// Verify that the low-level requests were made with correct names (no underscore prefix)
// Note: createIsmPolicy makes 2 calls - one for creation and one for update attempt
@ -464,7 +429,7 @@ public class CreateUsageEventIndicesStepTest {
// Arrange
Mockito.when(platformAnalytics.isEnabled()).thenReturn(true);
Mockito.when(searchEngineType.isOpenSearch()).thenReturn(true);
Mockito.when(index.getPrefix()).thenReturn("prod"); // Non-empty prefix
Mockito.when(index.getFinalPrefix()).thenReturn("prod_"); // Non-empty prefix
// Act
Function<UpgradeContext, UpgradeStepResult> executable = step.executable();
@ -477,7 +442,7 @@ public class CreateUsageEventIndicesStepTest {
// Verify OpenSearch path was taken and non-empty prefix was used
Mockito.verify(searchEngineType).isOpenSearch();
Mockito.verify(index).getPrefix();
Mockito.verify(index, atLeastOnce()).getFinalPrefix();
// Verify that the low-level requests were made with correct names (with underscore prefix)
// Note: createIsmPolicy makes 2 calls - one for creation and one for update attempt

View File

@ -31,45 +31,6 @@ public class IndexUtilsTest {
Mockito.when(operationContext.getObjectMapper()).thenReturn(objectMapper);
}
@Test
public void testCreatePrefixedName_WithPrefix() {
// Arrange
String prefix = "test";
String resourceName = "index";
// Act
String result = IndexUtils.createPrefixedName(prefix, resourceName);
// Assert
Assert.assertEquals(result, "test_index");
}
@Test
public void testCreatePrefixedName_EmptyPrefix() {
// Arrange
String prefix = "";
String resourceName = "index";
// Act
String result = IndexUtils.createPrefixedName(prefix, resourceName);
// Assert
Assert.assertEquals(result, "index");
}
@Test
public void testCreatePrefixedName_NullPrefix() {
// Arrange
String prefix = null;
String resourceName = "index";
// Act
String result = IndexUtils.createPrefixedName(prefix, resourceName);
// Assert
Assert.assertEquals(result, "index");
}
@Test
public void testIsAwsOpenSearchService_AwsHost() {
// Arrange

View File

@ -7,4 +7,12 @@ public class IndexConfiguration {
private String prefix;
private DocIdsConfiguration docIds;
private Integer minSearchFilterLength;
public String getFinalPrefix() {
if (prefix == null || prefix.isEmpty()) {
return "";
} else {
return prefix + "_";
}
}
}

View File

@ -0,0 +1,86 @@
package com.linkedin.metadata.config.search;
import org.testng.Assert;
import org.testng.annotations.Test;
public class IndexConfigurationTest {
@Test
public void testGetFinalPrefix_WithPrefix() {
// Arrange
IndexConfiguration config = new IndexConfiguration();
config.setPrefix("prod");
// Act
String result = config.getFinalPrefix();
// Assert
Assert.assertEquals(result, "prod_");
}
@Test
public void testGetFinalPrefix_EmptyPrefix() {
// Arrange
IndexConfiguration config = new IndexConfiguration();
config.setPrefix("");
// Act
String result = config.getFinalPrefix();
// Assert
Assert.assertEquals(result, "");
}
@Test
public void testGetFinalPrefix_NullPrefix() {
// Arrange
IndexConfiguration config = new IndexConfiguration();
config.setPrefix(null);
// Act
String result = config.getFinalPrefix();
// Assert
Assert.assertEquals(result, "");
}
@Test
public void testGetFinalPrefix_DefaultConstructor() {
// Arrange
IndexConfiguration config = new IndexConfiguration();
// prefix is null by default
// Act
String result = config.getFinalPrefix();
// Assert
Assert.assertEquals(result, "");
}
@Test
public void testGetFinalPrefix_WithComplexPrefix() {
// Arrange
IndexConfiguration config = new IndexConfiguration();
config.setPrefix("kbcpyv7ss3-staging-test");
// Act
String result = config.getFinalPrefix();
// Assert
Assert.assertEquals(result, "kbcpyv7ss3-staging-test_");
}
@Test
public void testGetFinalPrefix_WithWhitespacePrefix() {
// Arrange
IndexConfiguration config = new IndexConfiguration();
config.setPrefix(" ");
// Act
String result = config.getFinalPrefix();
// Assert
// Whitespace-only string is not considered empty by isEmpty()
Assert.assertEquals(result, " _");
}
}