forked from opensearch-project/opensearch-spark
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Separate metadata log entry data model and persistence (opensearch-pr…
…oject#406) * update metadata log entry model to be generic Signed-off-by: Sean Kao <[email protected]> * move OS specific stuff away from log entry Signed-off-by: Sean Kao <[email protected]> refactor: utils for OS log entry storage Signed-off-by: Sean Kao <[email protected]> * remove index name from log entry Signed-off-by: Sean Kao <[email protected]> * add log entry storage context Signed-off-by: Sean Kao <[email protected]> * fix error Signed-off-by: Sean Kao <[email protected]> * rename storage utils to OS converter and add test Signed-off-by: Sean Kao <[email protected]> * update comment for log entry for OpenSearch Signed-off-by: Sean Kao <[email protected]> * rename storageContext to properties Signed-off-by: Sean Kao <[email protected]> * remove unused failLogEntry Signed-off-by: Sean Kao <[email protected]> * use jackson log entry toJson Signed-off-by: Sean Kao <[email protected]> --------- Signed-off-by: Sean Kao <[email protected]>
- Loading branch information
1 parent
a4126b8
commit c3bbe63
Showing
14 changed files
with
401 additions
and
206 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
146 changes: 146 additions & 0 deletions
146
...ain/scala/org/opensearch/flint/core/storage/FlintMetadataLogEntryOpenSearchConverter.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,146 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.flint.core.storage; | ||
|
|
||
| import com.fasterxml.jackson.core.JsonProcessingException; | ||
| import com.fasterxml.jackson.databind.ObjectMapper; | ||
| import com.fasterxml.jackson.databind.node.ObjectNode; | ||
| import org.opensearch.flint.common.metadata.log.FlintMetadataLogEntry; | ||
|
|
||
| import java.util.Map; | ||
|
|
||
| /** | ||
| * Utility class for handling Flint metadata log entries in OpenSearch storage. | ||
| */ | ||
| public class FlintMetadataLogEntryOpenSearchConverter { | ||
|
|
||
| public static final String QUERY_EXECUTION_REQUEST_MAPPING = String.join("\n", | ||
| "{", | ||
| " \"dynamic\": false,", | ||
| " \"properties\": {", | ||
| " \"version\": {", | ||
| " \"type\": \"keyword\"", | ||
| " },", | ||
| " \"type\": {", | ||
| " \"type\": \"keyword\"", | ||
| " },", | ||
| " \"state\": {", | ||
| " \"type\": \"keyword\"", | ||
| " },", | ||
| " \"statementId\": {", | ||
| " \"type\": \"keyword\"", | ||
| " },", | ||
| " \"applicationId\": {", | ||
| " \"type\": \"keyword\"", | ||
| " },", | ||
| " \"sessionId\": {", | ||
| " \"type\": \"keyword\"", | ||
| " },", | ||
| " \"sessionType\": {", | ||
| " \"type\": \"keyword\"", | ||
| " },", | ||
| " \"error\": {", | ||
| " \"type\": \"text\"", | ||
| " },", | ||
| " \"lang\": {", | ||
| " \"type\": \"keyword\"", | ||
| " },", | ||
| " \"query\": {", | ||
| " \"type\": \"text\"", | ||
| " },", | ||
| " \"dataSourceName\": {", | ||
| " \"type\": \"keyword\"", | ||
| " },", | ||
| " \"submitTime\": {", | ||
| " \"type\": \"date\",", | ||
| " \"format\": \"strict_date_time||epoch_millis\"", | ||
| " },", | ||
| " \"jobId\": {", | ||
| " \"type\": \"keyword\"", | ||
| " },", | ||
| " \"lastUpdateTime\": {", | ||
| " \"type\": \"date\",", | ||
| " \"format\": \"strict_date_time||epoch_millis\"", | ||
| " },", | ||
| " \"queryId\": {", | ||
| " \"type\": \"keyword\"", | ||
| " },", | ||
| " \"excludeJobIds\": {", | ||
| " \"type\": \"keyword\"", | ||
| " }", | ||
| " }", | ||
| "}"); | ||
|
|
||
| public static final String QUERY_EXECUTION_REQUEST_SETTINGS = String.join("\n", | ||
| "{", | ||
| " \"index\": {", | ||
| " \"number_of_shards\": \"1\",", | ||
| " \"auto_expand_replicas\": \"0-2\",", | ||
| " \"number_of_replicas\": \"0\"", | ||
| " }", | ||
| "}"); | ||
|
|
||
| /** | ||
| * Convert a log entry to json string for persisting to OpenSearch. | ||
| * Expects the following field in properties: | ||
| * - dataSourceName: data source name | ||
| * | ||
| * @param logEntry | ||
| * log entry to convert | ||
| * @return | ||
| * json string representation of the log entry | ||
| */ | ||
| public static String toJson(FlintMetadataLogEntry logEntry) throws JsonProcessingException { | ||
| String applicationId = System.getenv().getOrDefault("SERVERLESS_EMR_VIRTUAL_CLUSTER_ID", "unknown"); | ||
| String jobId = System.getenv().getOrDefault("SERVERLESS_EMR_JOB_ID", "unknown"); | ||
| long lastUpdateTime = System.currentTimeMillis(); | ||
|
|
||
| ObjectMapper mapper = new ObjectMapper(); | ||
| ObjectNode json = mapper.createObjectNode(); | ||
|
|
||
| json.put("version", "1.0"); | ||
| json.put("latestId", logEntry.id()); | ||
| json.put("type", "flintindexstate"); | ||
| json.put("state", logEntry.state().toString()); | ||
| json.put("applicationId", applicationId); | ||
| json.put("jobId", jobId); | ||
| json.put("dataSourceName", logEntry.properties().get("dataSourceName").get().toString()); | ||
| json.put("jobStartTime", logEntry.createTime()); | ||
| json.put("lastUpdateTime", lastUpdateTime); | ||
| json.put("error", logEntry.error()); | ||
|
|
||
| return mapper.writerWithDefaultPrettyPrinter().writeValueAsString(json); | ||
| } | ||
|
|
||
| /** | ||
| * Construct a log entry from OpenSearch document fields. | ||
| * | ||
| * @param id | ||
| * OpenSearch document id | ||
| * @param seqNo | ||
| * OpenSearch document sequence number | ||
| * @param primaryTerm | ||
| * OpenSearch document primary term | ||
| * @param sourceMap | ||
| * OpenSearch document source as a map | ||
| * @return | ||
| * log entry | ||
| */ | ||
| public static FlintMetadataLogEntry constructLogEntry( | ||
| String id, | ||
| Long seqNo, | ||
| Long primaryTerm, | ||
| Map<String, Object> sourceMap) { | ||
| return new FlintMetadataLogEntry( | ||
| id, | ||
| /* sourceMap may use Integer or Long even though it's always long in index mapping */ | ||
| ((Number) sourceMap.get("jobStartTime")).longValue(), | ||
| FlintMetadataLogEntry.IndexState$.MODULE$.from((String) sourceMap.get("state")), | ||
| Map.of("seqNo", seqNo, "primaryTerm", primaryTerm), | ||
| (String) sourceMap.get("error"), | ||
| Map.of("dataSourceName", (String) sourceMap.get("dataSourceName"))); | ||
| } | ||
| } |
Oops, something went wrong.