Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Initialize all shards on index creation to avoid mapping conflicts #799

Closed
wants to merge 1 commit into from
Closed

Initialize all shards on index creation to avoid mapping conflicts #799

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,8 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.1.0/)
- Register system index descriptors through SystemIndexPlugin.getSystemIndexDescriptors ([#750](https://github.com/opensearch-project/flow-framework/pull/750))

### Bug Fixes
- Initialize all shards on index creation to avoid mapping conflicts ([#799](https://github.com/opensearch-project/flow-framework/pull/799))


### Infrastructure
### Documentation
Expand Down
5 changes: 4 additions & 1 deletion src/main/java/org/opensearch/flowframework/indices/FlowFrameworkIndicesHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import org.opensearch.action.get.GetRequest;
import org.opensearch.action.index.IndexRequest;
import org.opensearch.action.index.IndexResponse;
import org.opensearch.action.support.ActiveShardCount;
import org.opensearch.action.support.WriteRequest;
import org.opensearch.action.update.UpdateRequest;
import org.opensearch.action.update.UpdateResponse;
Expand Down Expand Up @@ -193,7 +194,9 @@ public void initFlowFrameworkIndexIfAbsent(FlowFrameworkIndex index, ActionListe
logger.error(errorMessage, e);
internalListener.onFailure(new FlowFrameworkException(errorMessage, ExceptionsHelper.status(e)));
});
CreateIndexRequest request = new CreateIndexRequest(indexName).mapping(mapping).settings(indexSettings);
CreateIndexRequest request = new CreateIndexRequest(indexName).mapping(mapping)
.settings(indexSettings)
.waitForActiveShards(ActiveShardCount.ALL);
Copy link
Member

@owaiskazi19 owaiskazi19 Jul 25, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This would wait for all shards be it primary and replicas to become active and cause a performance hit just for creating an index especially in large clusters or when dealing with indices with a high number of shards.

One option I can think of here:

  1. We can register a cluster state listener to monitor a cluster state that would tell us if the index has been created

(Pseudo code below)

ClusterStateListener listener = new ClusterStateListener() {
    @Override
    public void clusterChanged(ClusterChangedEvent event) {
        if (event.state().metadata().hasIndex(indexName)) {
            // Index is created, proceed with mapping update
            updateIndexMapping(indexName, mapping);
            clusterService.removeListener(this);
        }
    }
};

clusterService.addListener(listener);
  1. We can separate out updating the mapping to the index
void updateIndexMapping(String indexName, String mapping) {
    PutMappingRequest request = new PutMappingRequest(indexName)
        .source(mapping);

    client.admin().indices().putMapping(request, ActionListener.wrap(........
  1. [Optional] We can also add retries creating index if required

Copy link
Member Author

@dbwiddis dbwiddis Jul 25, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This would wait for all shards be it primary and replicas to become active and cause a performance hit just for creating an index especially in large clusters or when dealing with indices with a high number of shards.

Good concern, but we control this index and we are creating the settings at the same time as we are creating the mappings. We have either zero replicas in a single-node cluster (in which case this PR is no change) or exactly one replica, which is the very replica creating this race condition.

flow-framework/src/main/java/org/opensearch/flowframework/indices/FlowFrameworkIndicesHandler.java

Line 80 in f3a9e99

private static final Map<String, Object> indexSettings = Map.of("index.auto_expand_replicas", "0-1");

  1. We can register a cluster state listener to monitor a cluster state that would tell us if the index has been created

This is unneeded, the existing method doesn't return until the index has been created and assigned to the primary shard. And we already check the cluster state before indexing the document.

  1. We can separate out updating the mapping to the index

Fair enough but assuming you're doing the immediate refresh of this, it's no different than waiting for the one replica to have the mapping created.

  1. [Optional] We can also add retries creating index if required

Unfortunately in this case the retries will always fail. If you create an index with a text field mapping, it is impossible to change it to keyword without deleting the index.

The only other approach I can see that could possibly work is doing a GetMapping call before the first index request. This seems reasonable to do for the config index as we know we'll only initialize it once. But for creating a template we would need to check the mapping with every subsequent template request, not just the first one.

Note the performance/latency is a one-time event for the very first template creation.

Other possibilities for performance improvements that are well beyond the scope of a quick bug fix past code freeze:

  • we could pre-create these indices on startup, somewhat like ml commons does with their config index on a cron job,
  • we could create all the system indices in parallel

Copy link
Member

@owaiskazi19 owaiskazi19 Jul 25, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we could pre-create these indices on startup, somewhat like ml commons does with their config index on a cron job

This can be a good solution to this problem. I am still not aligned with using ActiveShardCount.ALL even as a quick fix. Probably need a second opinion @amitgalitz @joshpalis thoughts?

For now we can at least separate out update mapping, that way we will be sure that the index is created before we update the mapping.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For now we can at least separate out update mapping, that way we will be sure that the index is created before we update the mapping.

Wouldn't separating the update mapping still lead to same possible problem if we create the index and then we update the document before we update mapping?

I am still not aligned with using ActiveShardCount.ALL even as a quick fix.

I think perf hit shouldn't be too big here because we have a max replica of 0-1 and maximum of 5 shards for our system indices so we aren't waiting for hundreds of shards across dozens of nodes ever.

I just want to confirm problem again though, it looks like ActiveShardCount waits for 1 primary shard, we are saying we think we can have a replica shard on node 2 not initialized yet but it gets a document inserted before shard creation which leads to the wrong mapping?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I just want to confirm problem again though

I believe the sequence is:

  • one node gets the CreateIndex request with the mapping, updates that node, returns acknolwedged response
  • workflow sees index created and moves to create first document
  • that checks that the index exists (it does) but not whether the mapping has been updated across all nodes, and then inserts a document
  • the inserted document doesn't have a mapping and uses dynamic mapping of text
  • the initial index creation fails because it can't update/overwrite

I may be wrong on the diagnosis here, but I see no other way for this to fail

Copy link
Member

@owaiskazi19 owaiskazi19 Jul 25, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@reta @andrross @kaituo @sohami @saratvemulapalli any idea here

client.admin().indices().create(request, actionListener);
} else {
logger.debug("index: {} is already created", indexName);
Expand Down
Loading