Rollover max docs should only count primaries (#24977)

max_doc condition for index rollover should use document count only from primary shards 

Fixes #24217

core/src/main/java/org/elasticsearch/action/admin/indices/rollover/TransportRolloverAction.java

@@ -119,7 +119,7 @@ public class TransportRolloverAction extends TransportMasterNodeAction<RolloverR
                 @Override
                 public void onResponse(IndicesStatsResponse statsResponse) {
                     final Set<Condition.Result> conditionResults = evaluateConditions(rolloverRequest.getConditions(),
-                        statsResponse.getTotal().getDocs(), metaData.index(sourceIndexName));
+                        metaData.index(sourceIndexName), statsResponse);
 
                     if (rolloverRequest.isDryRun()) {
                         listener.onResponse(
@@ -201,6 +201,11 @@ public class TransportRolloverAction extends TransportMasterNodeAction<RolloverR
             .collect(Collectors.toSet());
     }
 
+    static Set<Condition.Result> evaluateConditions(final Set<Condition> conditions, final IndexMetaData metaData,
+                                                    final IndicesStatsResponse statsResponse) {
+        return evaluateConditions(conditions, statsResponse.getPrimaries().getDocs(), metaData);
+    }
+
     static void validate(MetaData metaData, RolloverRequest request) {
         final AliasOrIndex aliasOrIndex = metaData.getAliasAndIndexLookup().get(request.getAlias());
         if (aliasOrIndex == null) {

core/src/test/java/org/elasticsearch/action/admin/indices/rollover/TransportRolloverActionTests.java

@@ -22,6 +22,8 @@ package org.elasticsearch.action.admin.indices.rollover;
 import org.elasticsearch.Version;
 import org.elasticsearch.action.admin.indices.alias.IndicesAliasesClusterStateUpdateRequest;
 import org.elasticsearch.action.admin.indices.create.CreateIndexClusterStateUpdateRequest;
+import org.elasticsearch.action.admin.indices.stats.CommonStats;
+import org.elasticsearch.action.admin.indices.stats.IndicesStatsResponse;
 import org.elasticsearch.action.support.ActiveShardCount;
 import org.elasticsearch.cluster.metadata.AliasAction;
 import org.elasticsearch.cluster.metadata.AliasMetaData;
@@ -40,12 +42,30 @@ import java.util.List;
 import java.util.Locale;
 import java.util.Set;
 
+import org.mockito.ArgumentCaptor;
 import static org.elasticsearch.action.admin.indices.rollover.TransportRolloverAction.evaluateConditions;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
 
 public class TransportRolloverActionTests extends ESTestCase {
 
+    public void testDocStatsSelectionFromPrimariesOnly() throws Exception {
+        long docsInPrimaryShards = 100;
+        long docsInShards = 200;
+
+        final Condition condition = createTestCondition();
+        evaluateConditions(Sets.newHashSet(condition), createMetaData(), createIndecesStatResponse(docsInShards, docsInPrimaryShards));
+        final ArgumentCaptor<Condition.Stats> argument = ArgumentCaptor.forClass(Condition.Stats.class);
+        verify(condition).evaluate(argument.capture());
+
+        assertEquals(docsInPrimaryShards, argument.getValue().numDocs);
+    }
+
     public void testEvaluateConditions() throws Exception {
         MaxDocsCondition maxDocsCondition = new MaxDocsCondition(100L);
         MaxAgeCondition maxAgeCondition = new MaxAgeCondition(TimeValue.timeValueHours(2));
@@ -190,4 +210,37 @@ public class TransportRolloverActionTests extends ESTestCase {
         assertThat(createIndexRequest.index(), equalTo(rolloverIndex));
         assertThat(createIndexRequest.cause(), equalTo("rollover_index"));
     }
+
+    private IndicesStatsResponse createIndecesStatResponse(long totalDocs, long primaryDocs) {
+        final CommonStats primaryStats = mock(CommonStats.class);
+        when(primaryStats.getDocs()).thenReturn(new DocsStats(primaryDocs, 0));
+
+        final CommonStats totalStats = mock(CommonStats.class);
+        when(totalStats.getDocs()).thenReturn(new DocsStats(totalDocs, 0));
+
+        final IndicesStatsResponse response = mock(IndicesStatsResponse.class);
+        when(response.getPrimaries()).thenReturn(primaryStats);
+        when(response.getTotal()).thenReturn(totalStats);
+
+        return response;
+    }
+
+    private IndexMetaData createMetaData() {
+        final Settings settings = Settings.builder()
+            .put(IndexMetaData.SETTING_VERSION_CREATED, Version.CURRENT)
+            .put(IndexMetaData.SETTING_INDEX_UUID, UUIDs.randomBase64UUID())
+            .put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1)
+            .put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, 0)
+            .build();
+        return IndexMetaData.builder(randomAlphaOfLength(10))
+            .creationDate(System.currentTimeMillis() - TimeValue.timeValueHours(3).getMillis())
+            .settings(settings)
+            .build();
+    }
+
+    private Condition createTestCondition() {
+        final Condition condition = mock(Condition.class);
+        when(condition.evaluate(any())).thenReturn(new Condition.Result(condition, true));
+        return condition;
+    }
 }

rest-api-spec/src/main/resources/rest-api-spec/test/indices.rollover/20_max_doc_condition.yml

@@ -0,0 +1,57 @@
+---
+"Max docs rollover conditions matches only primary shards":
+  - skip:
+        version: "- 5.6.1"
+        reason:  "matching docs changed from all shards to primary shards"
+
+  # create index with alias and replica
+  - do:
+      indices.create:
+        index: logs-1
+        wait_for_active_shards: 1
+        body:
+          aliases:
+            logs_search: {}
+
+  # index first document and wait for refresh
+  - do:
+      index:
+        index: logs-1
+        type:  test
+        id:    "1"
+        body:  { "foo": "hello world" }
+        refresh: true
+
+  # perform alias rollover with no result
+  - do:
+      indices.rollover:
+        alias: "logs_search"
+        wait_for_active_shards: 1
+        body:
+          conditions:
+            max_docs: 2
+
+  - match: { conditions: { "[max_docs: 2]": false } }
+  - match: { rolled_over: false }
+
+  # index second document and wait for refresh
+  - do:
+      index:
+        index: logs-1
+        type:  test
+        id:    "2"
+        body:  { "foo": "hello world" }
+        refresh: true
+
+  # perform alias rollover
+  - do:
+      indices.rollover:
+        alias: "logs_search"
+        wait_for_active_shards: 1
+        body:
+          conditions:
+            max_docs: 2
+
+  - match: { conditions: { "[max_docs: 2]": true } }
+  - match: { rolled_over: true }
+