Merge pull request #4241 from gchq/gh-4240-multiple-shards

PR for #4240 - Stroom creating too many shards per partition

Author: at055612

stroom-core-shared/src/main/java/stroom/index/shared/IndexShard.java

@@ -313,6 +313,10 @@ public String toString() {
         return sb.toString();
     }
 
+
+    // --------------------------------------------------------------------------------
+
+
     /**
      * The status of this indexUuid shard
      */

stroom-index/stroom-index-impl-db/src/main/java/stroom/index/impl/db/IndexShardDaoImpl.java

@@ -30,6 +30,8 @@
 import stroom.query.language.functions.ValNull;
 import stroom.query.language.functions.ValString;
 import stroom.query.language.functions.ValuesConsumer;
+import stroom.util.logging.LambdaLogger;
+import stroom.util.logging.LambdaLoggerFactory;
 import stroom.util.shared.PageRequest;
 import stroom.util.shared.Range;
 import stroom.util.shared.ResultPage;
@@ -63,6 +65,8 @@
 @Singleton // holding all the volume selectors
 class IndexShardDaoImpl implements IndexShardDao {
 
+    private static final LambdaLogger LOGGER = LambdaLoggerFactory.getLogger(IndexShardDaoImpl.class);
+
     private static final Function<Record, IndexShard> RECORD_TO_INDEX_SHARD_MAPPER = record -> {
         final IndexShard indexShard = new IndexShard();
         indexShard.setId(record.get(INDEX_SHARD.ID));
@@ -301,12 +305,16 @@ public boolean setStatus(final Long id,
             case CORRUPT -> INDEX_SHARD.STATUS.ne(IndexShardStatus.DELETED.getPrimitiveValue());
         };
 
-        return JooqUtil.contextResult(indexDbConnProvider, context -> context
+        final boolean didUpdate = JooqUtil.contextResult(indexDbConnProvider, context -> context
                 .update(INDEX_SHARD)
                 .set(INDEX_SHARD.STATUS, status.getPrimitiveValue())
                 .where(INDEX_SHARD.ID.eq(id))
                 .and(currentStateCondition)
                 .execute()) > 0;
+
+        LOGGER.debug("Setting shard status to {} for shard id {}, didUpdate: {}", status, id, didUpdate);
+
+        return didUpdate;
     }
 
     @Override

stroom-index/stroom-index-impl/src/main/java/stroom/index/impl/ActiveShardsCacheImpl.java

@@ -11,8 +11,10 @@
 import stroom.index.shared.LuceneIndexDoc;
 import stroom.node.api.NodeInfo;
 import stroom.security.api.SecurityContext;
+import stroom.util.NullSafe;
 import stroom.util.logging.LambdaLogger;
 import stroom.util.logging.LambdaLoggerFactory;
+import stroom.util.logging.LogUtil;
 import stroom.util.shared.ResultPage;
 
 import jakarta.inject.Inject;
@@ -21,6 +23,7 @@
 
 import java.io.UncheckedIOException;
 import java.util.ArrayList;
+import java.util.EnumSet;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.locks.ReentrantLock;
@@ -78,6 +81,7 @@ private ActiveShards create(final IndexShardKey indexShardKey) {
                 throw new IndexException("Unable to find index with UUID: " + indexShardKey.getIndexUuid());
             }
 
+            LOGGER.debug("Creating ActiveShards for node: {}, indexShardKey: {}", nodeInfo, indexShardKey);
             return new ActiveShards(
                     nodeInfo,
                     indexShardWriterCache,
@@ -89,9 +93,24 @@ private ActiveShards create(final IndexShardKey indexShardKey) {
         });
     }
 
+
+    // --------------------------------------------------------------------------------
+
+
     public static class ActiveShards {
 
-        private static final int MAX_ATTEMPTS = 10000;
+        private static final LambdaLogger LOGGER = LambdaLoggerFactory.getLogger(ActiveShards.class);
+
+        // All shards are CLOSED on boot, but if this cache is cleared or items age off
+        // then we may shards in other states.
+        private static final EnumSet<IndexShardStatus> REQUIRED_SHARD_STATES = EnumSet.of(
+                IndexShardStatus.NEW,
+                IndexShardStatus.OPEN,
+                IndexShardStatus.OPENING,
+                IndexShardStatus.CLOSED,
+                IndexShardStatus.CLOSING);
+
+        private static final int MAX_ATTEMPTS = 10_000;
 
         private final NodeInfo nodeInfo;
         private final IndexShardWriterCache indexShardWriterCache;
@@ -130,6 +149,7 @@ public void addDocument(final IndexDocument document) {
 
             // Attempt under lock if we failed to add.
             if (!success) {
+                LOGGER.debug("Trying again under lock");
                 // If we failed then try under lock to make sure we get a new writer.
                 addDocumentUnderLock(document);
             }
@@ -220,13 +240,16 @@ private boolean addDocument(final IndexDocument document,
 
         private List<IndexShard> getIndexShards() {
             List<IndexShard> indexShards = this.indexShards;
+
             if (indexShards.size() < shardsPerPartition) {
                 indexShards = ensureShards();
             }
             return indexShards;
         }
 
         private synchronized List<IndexShard> ensureShards() {
+            LOGGER.debug(() -> LogUtil.message(
+                    "ensureShards, indexShards size before: {}", NullSafe.size(indexShards)));
             List<IndexShard> list = indexShards;
             if (list.size() < shardsPerPartition) {
                 list = new ArrayList<>(list);
@@ -235,10 +258,13 @@ private synchronized List<IndexShard> ensureShards() {
                 }
             }
             indexShards = list;
+            LOGGER.debug(() -> LogUtil.message(
+                    "ensureShards, indexShards size after: {}", NullSafe.size(indexShards)));
             return list;
         }
 
         private synchronized void addActiveShard(final IndexShardKey indexShardKey) {
+            LOGGER.debug("Adding shard for key {}", indexShardKey);
             final IndexShard indexShard = createNewShard(indexShardKey);
             final List<IndexShard> list = new ArrayList<>(indexShards);
             list.add(indexShard);
@@ -263,21 +289,34 @@ private List<IndexShard> getExistingShards(final IndexShardKey indexShardKey) {
 
             final List<IndexShard> indexShards = new ArrayList<>();
             final ResultPage<IndexShard> indexShardResultPage = indexShardDao.find(criteria);
+            LOGGER.debug(() -> LogUtil.message(
+                    "getExistingShards(), found {} un-filtered shards, maxDocsPerShard: {}",
+                    NullSafe.getOrElse(indexShardResultPage, ResultPage::size, 0),
+                    maxDocsPerShard));
             for (final IndexShard indexShard : indexShardResultPage.getValues()) {
                 // Look for non deleted, non-full, non-corrupt index shards.
-                if (IndexShardStatus.CLOSED.equals(indexShard.getStatus()) &&
-                        indexShard.getDocumentCount() < maxDocsPerShard) {
+                final IndexShardStatus status = indexShard.getStatus();
+                if (status != null
+                        && REQUIRED_SHARD_STATES.contains(status)
+                        && indexShard.getDocumentCount() < maxDocsPerShard) {
                     indexShards.add(indexShard);
+                } else {
+                    LOGGER.debug(() -> LogUtil.message("Ignoring shard {} with status: {}, docCount: {}",
+                            indexShard.getId(), status, indexShard.getDocumentCount()));
                 }
             }
+            LOGGER.debug(() -> LogUtil.message(
+                    "getExistingShards(), indexShards size: {}", NullSafe.size(indexShards)));
             return indexShards;
         }
 
         /**
          * Creates a new index shard writer for the specified key and opens a writer for it.
          */
         private IndexShard createNewShard(final IndexShardKey indexShardKey) {
-            return indexShardCreator.createIndexShard(indexShardKey, nodeInfo.getThisNodeName());
+            final String thisNodeName = nodeInfo.getThisNodeName();
+            LOGGER.debug("Creating shard for key {} on {}", indexShardKey, thisNodeName);
+            return indexShardCreator.createIndexShard(indexShardKey, thisNodeName);
         }
     }
 }

unreleased_changes/20240429_155759_127__4240.md

@@ -0,0 +1,24 @@
+* Issue **#4240** : Fix creation of new shards when non-full shards already exist for a partition.
+
+
+```sh
+# ********************************************************************************
+# Issue title: Stroom creating too many shards per partition
+# Issue link:  https://github.com/gchq/stroom/issues/4240
+# ********************************************************************************
+
+# ONLY the top line will be included as a change entry in the CHANGELOG.
+# The entry should be in GitHub flavour markdown and should be written on a SINGLE
+# line with no hard breaks. You can have multiple change files for a single GitHub issue.
+# The  entry should be written in the imperative mood, i.e. 'Fix nasty bug' rather than
+# 'Fixed nasty bug'.
+#
+# Examples of acceptable entries are:
+#
+#
+# * Issue **123** : Fix bug with an associated GitHub issue in this repository
+#
+# * Issue **namespace/other-repo#456** : Fix bug with an associated GitHub issue in another repository
+#
+# * Fix bug with no associated GitHub issue.
+```