POC for hollow shard

Author: kingherc

server/src/main/java/org/elasticsearch/index/CompositeIndexEventListener.java

@@ -349,4 +349,28 @@ public void afterFilesRestoredFromRepository(IndexShard indexShard) {
             }
         }
     }
+
+    @Override
+    public void onPrimaryPermitAcquire(IndexShard indexShard) {
+        for (IndexEventListener listener : listeners) {
+            try {
+                listener.onPrimaryPermitAcquire(indexShard);
+            } catch (Exception e) {
+                logger.warn(() -> "[" + indexShard.shardId() + "] failed to invoke on primary permit acquire", e);
+                throw e;
+            }
+        }
+    }
+
+    @Override
+    public void onPrimaryPermitReleased(IndexShard indexShard) {
+        for (IndexEventListener listener : listeners) {
+            try {
+                listener.onPrimaryPermitReleased(indexShard);
+            } catch (Exception e) {
+                logger.warn(() -> "[" + indexShard.shardId() + "] failed to invoke on primary permit released", e);
+                throw e;
+            }
+        }
+    }
 }

server/src/main/java/org/elasticsearch/index/shard/IndexEventListener.java

@@ -190,4 +190,18 @@ default void afterIndexShardRecovery(IndexShard indexShard, ActionListener<Void>
      * @param indexShard the shard that is recovering
      */
     default void afterFilesRestoredFromRepository(IndexShard indexShard) {}
+
+    /**
+     * Called when one or more primary permits are attempted to be acquired.
+     *
+     * @param indexShard    the shard
+     */
+    default void onPrimaryPermitAcquire(IndexShard indexShard) {}
+
+    /**
+     * Called when one or more primary permits are released.
+     *
+     * @param indexShard    the shard
+     */
+    default void onPrimaryPermitReleased(IndexShard indexShard) {}
 }

server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

@@ -776,6 +776,7 @@ public void relocated(
         final String targetNodeId,
         final String targetAllocationId,
         final BiConsumer<ReplicationTracker.PrimaryContext, ActionListener<Void>> consumer,
+        // TODO primary permits if already held to release after primary context
         final ActionListener<Void> listener
     ) throws IllegalIndexShardStateException, IllegalStateException {
         assert shardRouting.primary() : "only primaries can be marked as relocated: " + shardRouting;
@@ -1827,6 +1828,23 @@ public void postRecovery(String reason, ActionListener<Void> listener) throws In
         }
     }
 
+    /**
+     * Resets the engine, by closing the current engine reference and creating a new engine.
+     */
+    public void resetEngine() throws IOException {
+        assert indexShardOperationPermits.getActiveOperationsCount() == OPERATIONS_BLOCKED : "must hold permits to reset the engine";
+        final EngineConfig config = newEngineConfig(replicationTracker);
+        synchronized (engineMutex) {
+            IOUtils.close(currentEngineReference.get());
+            final Engine newEngine = createEngine(config);
+            currentEngineReference.set(newEngine);
+            onNewEngine(newEngine);
+            active.set(true);
+        }
+        onSettingsChanged();
+        checkAndCallWaitForEngineOrClosedShardListeners();
+    }
+
     /**
      * called before starting to copy index files over
      */
@@ -3553,6 +3571,7 @@ public void acquirePrimaryOperationPermit(
     ) {
         verifyNotClosed();
         assert shardRouting.primary() : "acquirePrimaryOperationPermit should only be called on primary shard: " + shardRouting;
+        indexEventListener.onPrimaryPermitAcquire(this);
         indexShardOperationPermits.acquire(wrapPrimaryOperationPermitListener(onPermitAcquired), executorOnDelay, forceExecution);
     }
 
@@ -3581,25 +3600,32 @@ public void acquireAllPrimaryOperationsPermits(final ActionListener<Releasable>
      */
     private ActionListener<Releasable> wrapPrimaryOperationPermitListener(final ActionListener<Releasable> listener) {
         return listener.delegateFailure((l, r) -> {
-            if (isPrimaryMode()) {
-                l.onResponse(r);
+            var wrappedReleasable = wrapPrimaryPermitReleasable(r);
+            if (isPrimaryMode() || state() == IndexShardState.POST_RECOVERY) {
+                l.onResponse(wrappedReleasable);
             } else {
-                r.close();
+                wrappedReleasable.close();
                 l.onFailure(new ShardNotInPrimaryModeException(shardId, state));
             }
         });
     }
 
+    private Releasable wrapPrimaryPermitReleasable(Releasable releasable) {
+        return Releasables.wrap(releasable, () -> { indexEventListener.onPrimaryPermitReleased(this); });
+    }
+
     private void asyncBlockOperations(ActionListener<Releasable> onPermitAcquired, long timeout, TimeUnit timeUnit) {
         final Releasable forceRefreshes = refreshListeners.forceRefreshes();
         final ActionListener<Releasable> wrappedListener = ActionListener.wrap(r -> {
+            var wrappedReleasable = wrapPrimaryPermitReleasable(r);
             forceRefreshes.close();
-            onPermitAcquired.onResponse(r);
+            onPermitAcquired.onResponse(wrappedReleasable);
         }, e -> {
             forceRefreshes.close();
             onPermitAcquired.onFailure(e);
         });
         try {
+            indexEventListener.onPrimaryPermitAcquire(this);
             indexShardOperationPermits.blockOperations(wrappedListener, timeout, timeUnit, threadPool.generic());
         } catch (Exception e) {
             forceRefreshes.close();
@@ -3637,6 +3663,7 @@ private <E extends Exception> void bumpPrimaryTerm(
         assert newPrimaryTerm > pendingPrimaryTerm || (newPrimaryTerm >= pendingPrimaryTerm && combineWithAction != null);
         assert getOperationPrimaryTerm() <= pendingPrimaryTerm;
         final CountDownLatch termUpdated = new CountDownLatch(1);
+        indexEventListener.onPrimaryPermitAcquire(this);
         asyncBlockOperations(new ActionListener<Releasable>() {
             @Override
             public void onFailure(final Exception e) {
@@ -3659,7 +3686,7 @@ private void innerFail(final Exception e) {
 
             @Override
             public void onResponse(final Releasable releasable) {
-                final Releasable releaseOnce = Releasables.releaseOnce(releasable);
+                final Releasable releaseOnce = Releasables.releaseOnce(wrapPrimaryPermitReleasable(releasable));
                 try {
                     assert getOperationPrimaryTerm() <= pendingPrimaryTerm;
                     termUpdated.await();

server/src/main/java/org/elasticsearch/node/NodeConstruction.java

@@ -221,6 +221,7 @@
 import java.io.Closeable;
 import java.io.IOException;
 import java.io.UncheckedIOException;
+import java.lang.management.ManagementFactory;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -538,15 +539,17 @@ private SettingsModule validateSettings(Settings envSettings, Settings settings,
         // after settings validation.
         nodeEnvironment = new NodeEnvironment(envSettings, environment);
         logger.info(
-            "node name [{}], node ID [{}], cluster name [{}], roles {}",
+            "node name [{}], node ID [{}], cluster name [{}], roles {}, myPID {}",
             Node.NODE_NAME_SETTING.get(envSettings),
             nodeEnvironment.nodeId(),
             ClusterName.CLUSTER_NAME_SETTING.get(envSettings).value(),
             DiscoveryNode.getRolesFromSettings(settings)
                 .stream()
                 .map(DiscoveryNodeRole::roleName)
-                .collect(Collectors.toCollection(LinkedHashSet::new))
+                .collect(Collectors.toCollection(LinkedHashSet::new)),
+            ManagementFactory.getRuntimeMXBean().getName()
         );
+
         resourcesToClose.add(nodeEnvironment);
         modules.bindToInstance(NodeEnvironment.class, nodeEnvironment);