Skip to content

Commit

Permalink
Add index_types for OTEL logs and metrics opensearch-project#3148
Browse files Browse the repository at this point in the history 
Signed-off-by: Jürgen Walter <juergen.walter@sap.com>
  • Loading branch information
@juergen-walter
juergen-walter committed Jan 12, 2024
1 parent 787064e commit 8b1fd94
Show file tree
Hide file tree
Showing 5 changed files with 152 additions and 1 deletion.
88 changes: 88 additions & 0 deletions ...grationTest/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSinkIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -216,6 +216,94 @@ public void testInstantiateSinkRawSpanDefault() throws IOException {
}
}

@Test
@DisabledIf(value = "isES6", disabledReason = LOG_INGESTION_TEST_DISABLED_REASON)
public void testInstantiateSinkMetricsDefault() throws IOException {
final PluginSetting pluginSetting = generatePluginSetting(IndexType.LOG_ANALYTICS.getValue(), null, null);
OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
final String indexAlias = IndexConstants.TYPE_TO_DEFAULT_ALIAS.get(IndexType.LOG_ANALYTICS);
Request request = new Request(HttpMethod.HEAD, indexAlias);
Response response = client.performRequest(request);
assertThat(response.getStatusLine().getStatusCode(), equalTo(SC_OK));
final String index = String.format("%s-000001", indexAlias);
final Map<String, Object> mappings = getIndexMappings(index);
assertThat(mappings, notNullValue());
assertThat((boolean) mappings.get("date_detection"), equalTo(false));
sink.shutdown();

if (isOSBundle()) {
// Check managed index
await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> {
assertThat(getIndexPolicyId(index), equalTo(IndexConstants.LOGS_ISM_POLICY));
}
);
}

// roll over initial index
request = new Request(HttpMethod.POST, String.format("%s/_rollover", indexAlias));
request.setJsonEntity("{ \"conditions\" : { } }\n");
response = client.performRequest(request);
assertThat(response.getStatusLine().getStatusCode(), equalTo(SC_OK));

// Instantiate sink again
sink = createObjectUnderTest(pluginSetting, true);
// Make sure no new write index *-000001 is created under alias
final String rolloverIndexName = String.format("%s-000002", indexAlias);
request = new Request(HttpMethod.GET, rolloverIndexName + "/_alias");
response = client.performRequest(request);
assertThat(checkIsWriteIndex(EntityUtils.toString(response.getEntity()), indexAlias, rolloverIndexName), equalTo(true));
sink.shutdown();

if (isOSBundle()) {
// Check managed index
assertThat(getIndexPolicyId(rolloverIndexName), equalTo(IndexConstants.LOGS_ISM_POLICY));
}
}

@Test
@DisabledIf(value = "isES6", disabledReason = METRIC_INGESTION_TEST_DISABLED_REASON)
public void testInstantiateSinkMetricsDefault() throws IOException {
final PluginSetting pluginSetting = generatePluginSetting(IndexType.METRIC_ANALYTICS.getValue(), null, null);
OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
final String indexAlias = IndexConstants.TYPE_TO_DEFAULT_ALIAS.get(IndexType.METRIC_ANALYTICS);
Request request = new Request(HttpMethod.HEAD, indexAlias);
Response response = client.performRequest(request);
assertThat(response.getStatusLine().getStatusCode(), equalTo(SC_OK));
final String index = String.format("%s-000001", indexAlias);
final Map<String, Object> mappings = getIndexMappings(index);
assertThat(mappings, notNullValue());
assertThat((boolean) mappings.get("date_detection"), equalTo(false));
sink.shutdown();

if (isOSBundle()) {
// Check managed index
await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> {
assertThat(getIndexPolicyId(index), equalTo(IndexConstants.METRRICS_ISM_POLICY));
}
);
}

// roll over initial index
request = new Request(HttpMethod.POST, String.format("%s/_rollover", indexAlias));
request.setJsonEntity("{ \"conditions\" : { } }\n");
response = client.performRequest(request);
assertThat(response.getStatusLine().getStatusCode(), equalTo(SC_OK));

// Instantiate sink again
sink = createObjectUnderTest(pluginSetting, true);
// Make sure no new write index *-000001 is created under alias
final String rolloverIndexName = String.format("%s-000002", indexAlias);
request = new Request(HttpMethod.GET, rolloverIndexName + "/_alias");
response = client.performRequest(request);
assertThat(checkIsWriteIndex(EntityUtils.toString(response.getEntity()), indexAlias, rolloverIndexName), equalTo(true));
sink.shutdown();

if (isOSBundle()) {
// Check managed index
assertThat(getIndexPolicyId(rolloverIndexName), equalTo(IndexConstants.METRICS_ISM_POLICY));
}
}

@Test
@DisabledIf(value = "isES6", disabledReason = TRACE_INGESTION_TEST_DISABLED_REASON)
public void testInstantiateSinkRawSpanReservedAliasAlreadyUsedAsIndex() throws IOException {
Expand Down
4 changes: 4 additions & 0 deletions ...ain/java/org/opensearch/dataprepper/plugins/sink/opensearch/index/IndexConfiguration.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -412,6 +412,10 @@ private Map<String, Object> readIndexTemplate(final String templateFile, final I
templateURL = loadExistingTemplate(templateType, IndexConstants.RAW_DEFAULT_TEMPLATE_FILE);
} else if (indexType.equals(IndexType.TRACE_ANALYTICS_SERVICE_MAP)) {
templateURL = loadExistingTemplate(templateType, IndexConstants.SERVICE_MAP_DEFAULT_TEMPLATE_FILE);
} else if (indexType.equals(IndexType.LOG_ANALYTICS)) {
templateURL = loadExistingTemplate(templateType, IndexConstants.LOGS_DEFAULT_TEMPLATE_FILE);
} else if (indexType.equals(IndexType.METRIC_ANALYTICS)) {
templateURL = loadExistingTemplate(templateType, IndexConstants.METRICS_DEFAULT_TEMPLATE_FILE);
} else if (templateFile != null) {
if (templateFile.toLowerCase().startsWith(S3_PREFIX)) {
FileReader s3FileReader = new S3FileReader(s3Client);
Expand Down
15 changes: 14 additions & 1 deletion ...rc/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/index/IndexConstants.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,17 @@ public class IndexConstants {
public static final String RAW_ISM_POLICY = "raw-span-policy";
public static final String RAW_ISM_FILE_NO_ISM_TEMPLATE = "raw-span-policy-no-ism-template.json";
public static final String RAW_ISM_FILE_WITH_ISM_TEMPLATE = "raw-span-policy-with-ism-template.json";

public static final String LOGS_DEFAULT_TEMPLATE_FILE = "logs-otel-v1-index-template.json";
public static final String LOGS_ISM_POLICY = "logs-policy";
public static final String LOGS_ISM_FILE_NO_ISM_TEMPLATE = "logs-policy-no-ism-template.json";
public static final String LOGS_ISM_FILE_WITH_ISM_TEMPLATE = "logs-policy-with-ism-template.json";

public static final String METRICS_DEFAULT_TEMPLATE_FILE = "metrics-otel-v1-index-template.json";
public static final String METRICS_ISM_POLICY = "metrics-policy";
public static final String METRICS_ISM_FILE_NO_ISM_TEMPLATE = "logs-policy-no-ism-template.json";
public static final String METRICS_ISM_FILE_WITH_ISM_TEMPLATE = "logs-policy-with-ism-template.json";

public static final String ISM_ENABLED_SETTING = "opendistro.index_state_management.enabled";
public static final String ISM_POLICY_ID_SETTING = "opendistro.index_state_management.policy_id";
public static final String ISM_ROLLOVER_ALIAS_SETTING = "opendistro.index_state_management.rollover_alias";
Expand All @@ -26,7 +37,9 @@ public class IndexConstants {

static {
// TODO: extract out version number into version enum
TYPE_TO_DEFAULT_ALIAS.put(IndexType.TRACE_ANALYTICS_RAW, "otel-v1-apm-span");
TYPE_TO_DEFAULT_ALIAS.put(IndexType.TRACE_ANALYTICS_SERVICE_MAP, "otel-v1-apm-service-map");
TYPE_TO_DEFAULT_ALIAS.put(IndexType.TRACE_ANALYTICS_RAW, "otel-v1-apm-span");
TYPE_TO_DEFAULT_ALIAS.put(IndexType.LOG_ANALYTICS, "logs-otel-v1");
TYPE_TO_DEFAULT_ALIAS.put(IndexType.METRIC_ANALYTICS, "metrics-otel-v1");
}
}
44 changes: 44 additions & 0 deletions ...in/java/org/opensearch/dataprepper/plugins/sink/opensearch/index/IndexManagerFactory.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,14 @@ public final IndexManager getIndexManager(final IndexType indexType,
indexManager = new TraceAnalyticsServiceMapIndexManager(
restHighLevelClient, openSearchClient, openSearchSinkConfiguration, clusterSettingsParser, templateStrategy, indexAlias);
break;
case LOG_ANALYTICS:
indexManager = new LogAnalyticsIndexManager(
restHighLevelClient, openSearchClient, openSearchSinkConfiguration, clusterSettingsParser, templateStrategy, indexAlias);
break;
case METRIC_ANALYTICS:
indexManager = new MetricAnalyticsIndexManager(
restHighLevelClient, openSearchClient, openSearchSinkConfiguration, clusterSettingsParser, templateStrategy, indexAlias);
break;
case MANAGEMENT_DISABLED:
indexManager = new ManagementDisabledIndexManager(
restHighLevelClient, openSearchClient, openSearchSinkConfiguration, clusterSettingsParser, templateStrategy, indexAlias);
Expand Down Expand Up @@ -141,6 +149,42 @@ public TraceAnalyticsServiceMapIndexManager(final RestHighLevelClient restHighLe
}
}

private static class LogAnalyticsIndexManager extends AbstractIndexManager {

public LogsIndexManager(final RestHighLevelClient restHighLevelClient,
final OpenSearchClient openSearchClient,
final OpenSearchSinkConfiguration openSearchSinkConfiguration,
final ClusterSettingsParser clusterSettingsParser,
final TemplateStrategy templateStrategy,
final String indexAlias) {
super(restHighLevelClient, openSearchClient, openSearchSinkConfiguration, clusterSettingsParser, templateStrategy, indexAlias);
this.ismPolicyManagementStrategy = new IsmPolicyManagement(
openSearchClient,
restHighLevelClient,
IndexConstants.LOGS_ISM_POLICY,
IndexConstants.LOGS_ISM_FILE_WITH_ISM_TEMPLATE,
IndexConstants.LOGS_ISM_FILE_NO_ISM_TEMPLATE);
}
}

private static class MetricAnalyticsIndexManager extends AbstractIndexManager {

public MetricsIndexManager(final RestHighLevelClient restHighLevelClient,
final OpenSearchClient openSearchClient,
final OpenSearchSinkConfiguration openSearchSinkConfiguration,
final ClusterSettingsParser clusterSettingsParser,
final TemplateStrategy templateStrategy,
final String indexAlias) {
super(restHighLevelClient, openSearchClient, openSearchSinkConfiguration, clusterSettingsParser, templateStrategy, indexAlias);
this.ismPolicyManagementStrategy = new IsmPolicyManagement(
openSearchClient,
restHighLevelClient,
IndexConstants.METRICS_ISM_POLICY,
IndexConstants.METRICS_ISM_FILE_WITH_ISM_TEMPLATE,
IndexConstants.METRICS_ISM_FILE_NO_ISM_TEMPLATE);
}
}

private class ManagementDisabledIndexManager extends AbstractIndexManager {
protected ManagementDisabledIndexManager(final RestHighLevelClient restHighLevelClient,
final OpenSearchClient openSearchClient,
Expand Down
2 changes: 2 additions & 0 deletions ...rch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/index/IndexType.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,8 @@
public enum IndexType {
TRACE_ANALYTICS_RAW("trace-analytics-raw"),
TRACE_ANALYTICS_SERVICE_MAP("trace-analytics-service-map"),
LOG_ANALYTICS("log-analytics"),
METRIC_ANALYTICS("metric-analytics"),
CUSTOM("custom"),
MANAGEMENT_DISABLED("management_disabled");

Expand Down

0 comments on commit 8b1fd94

Please sign in to comment.