Skip to content

Commit

Permalink
Separate coordinating and primary bytes in stats (elastic#59487)
Browse files Browse the repository at this point in the history 
Currently we combine coordinating and primary bytes into a single bucket
for indexing pressure stats. This makes sense for rejection logic.
However, for metrics it would be useful to separate them.
  • Loading branch information
@Tim-Brooks
Tim-Brooks committed Jul 14, 2020
1 parent 70fe553 commit 408a07f
Show file tree
Hide file tree
Showing 11 changed files with 275 additions and 109 deletions.
45 changes: 31 additions & 14 deletions qa/smoke-test-http/src/test/java/org/elasticsearch/http/IndexingPressureRestIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -76,32 +76,44 @@ public void testIndexingPressureStats() throws IOException {
ArrayList<Object> values = new ArrayList<>(((Map<Object, Object>) nodeStatsMap.get("nodes")).values());
assertThat(values.size(), equalTo(2));
XContentTestUtils.JsonMapView node1 = new XContentTestUtils.JsonMapView((Map<String, Object>) values.get(0));
Integer node1IndexingBytes = node1.get("indexing_pressure.total.coordinating_and_primary_bytes");
Integer node1CombinedBytes = node1.get("indexing_pressure.total.combined_coordinating_and_primary_bytes");
Integer node1PrimaryBytes = node1.get("indexing_pressure.total.primary_bytes");
Integer node1ReplicaBytes = node1.get("indexing_pressure.total.replica_bytes");
Integer node1Rejections = node1.get("indexing_pressure.total.coordinating_and_primary_memory_limit_rejections");
Integer node1CoordinatingRejections = node1.get("indexing_pressure.total.coordinating_rejections");
Integer node1PrimaryRejections = node1.get("indexing_pressure.total.primary_rejections");
XContentTestUtils.JsonMapView node2 = new XContentTestUtils.JsonMapView((Map<String, Object>) values.get(1));
Integer node2IndexingBytes = node2.get("indexing_pressure.total.coordinating_and_primary_bytes");
Integer node2IndexingBytes = node2.get("indexing_pressure.total.combined_coordinating_and_primary_bytes");
Integer node2PrimaryBytes = node2.get("indexing_pressure.total.primary_bytes");
Integer node2ReplicaBytes = node2.get("indexing_pressure.total.replica_bytes");
Integer node2Rejections = node2.get("indexing_pressure.total.coordinating_and_primary_memory_limit_rejections");
Integer node2CoordinatingRejections = node2.get("indexing_pressure.total.coordinating_rejections");
Integer node2PrimaryRejections = node2.get("indexing_pressure.total.primary_rejections");

if (node1IndexingBytes == 0) {
if (node1CombinedBytes == 0) {
assertThat(node2IndexingBytes, greaterThan(0));
assertThat(node2IndexingBytes, lessThan(1024));
} else {
assertThat(node1IndexingBytes, greaterThan(0));
assertThat(node1IndexingBytes, lessThan(1024));
assertThat(node1CombinedBytes, greaterThan(0));
assertThat(node1CombinedBytes, lessThan(1024));
}

if (node1ReplicaBytes == 0) {
assertThat(node1PrimaryBytes, greaterThan(0));
assertThat(node1PrimaryBytes, lessThan(1024));

assertThat(node2ReplicaBytes, greaterThan(0));
assertThat(node2ReplicaBytes, lessThan(1024));
} else {
assertThat(node2PrimaryBytes, greaterThan(0));
assertThat(node2PrimaryBytes, lessThan(1024));

assertThat(node2ReplicaBytes, equalTo(0));
assertThat(node1ReplicaBytes, lessThan(1024));
}

assertThat(node1Rejections, equalTo(0));
assertThat(node2Rejections, equalTo(0));
assertThat(node1CoordinatingRejections, equalTo(0));
assertThat(node1PrimaryRejections, equalTo(0));
assertThat(node2CoordinatingRejections, equalTo(0));
assertThat(node2PrimaryRejections, equalTo(0));

Request failedIndexingRequest = new Request("POST", "/index_name/_doc/");
String largeString = randomAlphaOfLength(10000);
Expand All @@ -116,14 +128,19 @@ public void testIndexingPressureStats() throws IOException {
ArrayList<Object> values2 = new ArrayList<>(((Map<Object, Object>) nodeStatsMap2.get("nodes")).values());
assertThat(values2.size(), equalTo(2));
XContentTestUtils.JsonMapView node1AfterRejection = new XContentTestUtils.JsonMapView((Map<String, Object>) values2.get(0));
node1Rejections = node1AfterRejection.get("indexing_pressure.total.coordinating_and_primary_memory_limit_rejections");
node1CoordinatingRejections = node1AfterRejection.get("indexing_pressure.total.coordinating_rejections");
node1PrimaryRejections = node1.get("indexing_pressure.total.primary_rejections");
XContentTestUtils.JsonMapView node2AfterRejection = new XContentTestUtils.JsonMapView((Map<String, Object>) values2.get(1));
node2Rejections = node2AfterRejection.get("indexing_pressure.total.coordinating_and_primary_memory_limit_rejections");
node2CoordinatingRejections = node2AfterRejection.get("indexing_pressure.total.coordinating_rejections");
node2PrimaryRejections = node2AfterRejection.get("indexing_pressure.total.primary_rejections");

if (node1Rejections == 0) {
assertThat(node2Rejections, equalTo(1));
if (node1CoordinatingRejections == 0) {
assertThat(node2CoordinatingRejections, equalTo(1));
} else {
assertThat(node1Rejections, equalTo(1));
assertThat(node1CoordinatingRejections, equalTo(1));
}

assertThat(node1PrimaryRejections, equalTo(0));
assertThat(node2PrimaryRejections, equalTo(0));
}
}
15 changes: 11 additions & 4 deletions rest-api-spec/src/main/resources/rest-api-spec/test/nodes.stats/50_indexing_pressure.yml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,12 +14,19 @@
nodes.stats:
metric: [ indexing_pressure ]

- gte: { nodes.$node_id.indexing_pressure.total.coordinating_and_primary_bytes: 0 }
- gte: { nodes.$node_id.indexing_pressure.total.combined_coordinating_and_primary_bytes: 0 }
- gte: { nodes.$node_id.indexing_pressure.total.coordinating_bytes: 0 }
- gte: { nodes.$node_id.indexing_pressure.total.primary_bytes: 0 }
- gte: { nodes.$node_id.indexing_pressure.total.replica_bytes: 0 }
- gte: { nodes.$node_id.indexing_pressure.total.all_bytes: 0 }
- gte: { nodes.$node_id.indexing_pressure.total.coordinating_and_primary_memory_limit_rejections: 0 }
- gte: { nodes.$node_id.indexing_pressure.total.replica_memory_limit_rejections: 0 }
- gte: { nodes.$node_id.indexing_pressure.current.coordinating_and_primary_bytes: 0 }

- gte: { nodes.$node_id.indexing_pressure.total.coordinating_rejections: 0 }
- gte: { nodes.$node_id.indexing_pressure.total.primary_rejections: 0 }
- gte: { nodes.$node_id.indexing_pressure.total.replica_rejections: 0 }

- gte: { nodes.$node_id.indexing_pressure.current.combined_coordinating_and_primary_bytes: 0 }
- gte: { nodes.$node_id.indexing_pressure.current.coordinating_bytes: 0 }
- gte: { nodes.$node_id.indexing_pressure.current.primary_bytes: 0 }
- gte: { nodes.$node_id.indexing_pressure.current.replica_bytes: 0 }
- gte: { nodes.$node_id.indexing_pressure.current.all_bytes: 0 }

Expand Down
77 changes: 51 additions & 26 deletions server/src/internalClusterTest/java/org/elasticsearch/index/IndexingPressureIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -56,8 +56,6 @@
@ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST, numDataNodes = 2, numClientNodes = 1, transportClientRatio = 0.0D)
public class IndexingPressureIT extends ESIntegTestCase {

// TODO: Add additional REST tests when metrics are exposed

public static final String INDEX_NAME = "test";

private static final Settings unboundedWriteQueue = Settings.builder().put("thread_pool.write.queue_size", -1).build();
Expand Down Expand Up @@ -140,11 +138,19 @@ public void testWriteBytesAreIncremented() throws Exception {
IndexingPressure replicaWriteLimits = internalCluster().getInstance(IndexingPressure.class, replicaName);
IndexingPressure coordinatingWriteLimits = internalCluster().getInstance(IndexingPressure.class, coordinatingOnlyNode);

assertThat(primaryWriteLimits.getCurrentCoordinatingAndPrimaryBytes(), greaterThan(bulkShardRequestSize));
assertThat(primaryWriteLimits.getCurrentCombinedCoordinatingAndPrimaryBytes(), greaterThan(bulkShardRequestSize));
assertThat(primaryWriteLimits.getCurrentPrimaryBytes(), greaterThan(bulkShardRequestSize));
assertEquals(0, primaryWriteLimits.getCurrentCoordinatingBytes());
assertEquals(0, primaryWriteLimits.getCurrentReplicaBytes());
assertEquals(0, replicaWriteLimits.getCurrentCoordinatingAndPrimaryBytes());

assertEquals(0, replicaWriteLimits.getCurrentCombinedCoordinatingAndPrimaryBytes());
assertEquals(0, replicaWriteLimits.getCurrentCoordinatingBytes());
assertEquals(0, replicaWriteLimits.getCurrentPrimaryBytes());
assertEquals(0, replicaWriteLimits.getCurrentReplicaBytes());
assertEquals(bulkRequestSize, coordinatingWriteLimits.getCurrentCoordinatingAndPrimaryBytes());

assertEquals(bulkRequestSize, coordinatingWriteLimits.getCurrentCombinedCoordinatingAndPrimaryBytes());
assertEquals(bulkRequestSize, coordinatingWriteLimits.getCurrentCoordinatingBytes());
assertEquals(0, coordinatingWriteLimits.getCurrentPrimaryBytes());
assertEquals(0, coordinatingWriteLimits.getCurrentReplicaBytes());

latchBlockingReplicationSend.countDown();
Expand All @@ -167,14 +173,25 @@ public void testWriteBytesAreIncremented() throws Exception {
final long secondBulkShardRequestSize = request.ramBytesUsed();

if (usePrimaryAsCoordinatingNode) {
assertThat(primaryWriteLimits.getCurrentCoordinatingAndPrimaryBytes(),
greaterThan(bulkShardRequestSize + secondBulkRequestSize));
assertEquals(0, replicaWriteLimits.getCurrentCoordinatingAndPrimaryBytes());
assertBusy(() -> {
assertThat(primaryWriteLimits.getCurrentCombinedCoordinatingAndPrimaryBytes(),
greaterThan(bulkShardRequestSize + secondBulkRequestSize));
assertEquals(secondBulkRequestSize, primaryWriteLimits.getCurrentCoordinatingBytes());
assertThat(primaryWriteLimits.getCurrentPrimaryBytes(),
greaterThan(bulkShardRequestSize + secondBulkRequestSize));

assertEquals(0, replicaWriteLimits.getCurrentCombinedCoordinatingAndPrimaryBytes());
assertEquals(0, replicaWriteLimits.getCurrentCoordinatingBytes());
assertEquals(0, replicaWriteLimits.getCurrentPrimaryBytes());
});
} else {
assertThat(primaryWriteLimits.getCurrentCoordinatingAndPrimaryBytes(), greaterThan(bulkShardRequestSize));
assertEquals(secondBulkRequestSize, replicaWriteLimits.getCurrentCoordinatingAndPrimaryBytes());
assertThat(primaryWriteLimits.getCurrentCombinedCoordinatingAndPrimaryBytes(), greaterThan(bulkShardRequestSize));

assertEquals(secondBulkRequestSize, replicaWriteLimits.getCurrentCombinedCoordinatingAndPrimaryBytes());
assertEquals(secondBulkRequestSize, replicaWriteLimits.getCurrentCoordinatingBytes());
assertEquals(0, replicaWriteLimits.getCurrentPrimaryBytes());
}
assertEquals(bulkRequestSize, coordinatingWriteLimits.getCurrentCoordinatingAndPrimaryBytes());
assertEquals(bulkRequestSize, coordinatingWriteLimits.getCurrentCombinedCoordinatingAndPrimaryBytes());
assertBusy(() -> assertThat(replicaWriteLimits.getCurrentReplicaBytes(),
greaterThan(bulkShardRequestSize + secondBulkShardRequestSize)));

Expand All @@ -183,11 +200,19 @@ public void testWriteBytesAreIncremented() throws Exception {
successFuture.actionGet();
secondFuture.actionGet();

assertEquals(0, primaryWriteLimits.getCurrentCoordinatingAndPrimaryBytes());
assertEquals(0, primaryWriteLimits.getCurrentCombinedCoordinatingAndPrimaryBytes());
assertEquals(0, primaryWriteLimits.getCurrentCoordinatingBytes());
assertEquals(0, primaryWriteLimits.getCurrentPrimaryBytes());
assertEquals(0, primaryWriteLimits.getCurrentReplicaBytes());
assertEquals(0, replicaWriteLimits.getCurrentCoordinatingAndPrimaryBytes());

assertEquals(0, replicaWriteLimits.getCurrentCombinedCoordinatingAndPrimaryBytes());
assertEquals(0, replicaWriteLimits.getCurrentCoordinatingBytes());
assertEquals(0, replicaWriteLimits.getCurrentPrimaryBytes());
assertEquals(0, replicaWriteLimits.getCurrentReplicaBytes());
assertEquals(0, coordinatingWriteLimits.getCurrentCoordinatingAndPrimaryBytes());

assertEquals(0, coordinatingWriteLimits.getCurrentCombinedCoordinatingAndPrimaryBytes());
assertEquals(0, coordinatingWriteLimits.getCurrentCoordinatingBytes());
assertEquals(0, coordinatingWriteLimits.getCurrentPrimaryBytes());
assertEquals(0, coordinatingWriteLimits.getCurrentReplicaBytes());
} finally {
if (replicationSendPointReached.getCount() > 0) {
Expand Down Expand Up @@ -237,11 +262,11 @@ public void testWriteCanBeRejectedAtCoordinatingLevel() throws Exception {
IndexingPressure coordinatingWriteLimits = internalCluster().getInstance(IndexingPressure.class, coordinatingOnlyNode);

assertBusy(() -> {
assertThat(primaryWriteLimits.getCurrentCoordinatingAndPrimaryBytes(), greaterThan(bulkShardRequestSize));
assertThat(primaryWriteLimits.getCurrentCombinedCoordinatingAndPrimaryBytes(), greaterThan(bulkShardRequestSize));
assertEquals(0, primaryWriteLimits.getCurrentReplicaBytes());
assertEquals(0, replicaWriteLimits.getCurrentCoordinatingAndPrimaryBytes());
assertEquals(0, replicaWriteLimits.getCurrentCombinedCoordinatingAndPrimaryBytes());
assertThat(replicaWriteLimits.getCurrentReplicaBytes(), greaterThan(bulkShardRequestSize));
assertEquals(bulkRequestSize, coordinatingWriteLimits.getCurrentCoordinatingAndPrimaryBytes());
assertEquals(bulkRequestSize, coordinatingWriteLimits.getCurrentCombinedCoordinatingAndPrimaryBytes());
assertEquals(0, coordinatingWriteLimits.getCurrentReplicaBytes());
});

Expand All @@ -259,11 +284,11 @@ public void testWriteCanBeRejectedAtCoordinatingLevel() throws Exception {

successFuture.actionGet();

assertEquals(0, primaryWriteLimits.getCurrentCoordinatingAndPrimaryBytes());
assertEquals(0, primaryWriteLimits.getCurrentCombinedCoordinatingAndPrimaryBytes());
assertEquals(0, primaryWriteLimits.getCurrentReplicaBytes());
assertEquals(0, replicaWriteLimits.getCurrentCoordinatingAndPrimaryBytes());
assertEquals(0, replicaWriteLimits.getCurrentCombinedCoordinatingAndPrimaryBytes());
assertEquals(0, replicaWriteLimits.getCurrentReplicaBytes());
assertEquals(0, coordinatingWriteLimits.getCurrentCoordinatingAndPrimaryBytes());
assertEquals(0, coordinatingWriteLimits.getCurrentCombinedCoordinatingAndPrimaryBytes());
assertEquals(0, coordinatingWriteLimits.getCurrentReplicaBytes());
}
}
Expand Down Expand Up @@ -301,11 +326,11 @@ public void testWriteCanBeRejectedAtPrimaryLevel() throws Exception {
IndexingPressure coordinatingWriteLimits = internalCluster().getInstance(IndexingPressure.class, coordinatingOnlyNode);

assertBusy(() -> {
assertThat(primaryWriteLimits.getCurrentCoordinatingAndPrimaryBytes(), greaterThan(bulkShardRequestSize));
assertThat(primaryWriteLimits.getCurrentCombinedCoordinatingAndPrimaryBytes(), greaterThan(bulkShardRequestSize));
assertEquals(0, primaryWriteLimits.getCurrentReplicaBytes());
assertEquals(0, replicaWriteLimits.getCurrentCoordinatingAndPrimaryBytes());
assertEquals(0, replicaWriteLimits.getCurrentCombinedCoordinatingAndPrimaryBytes());
assertThat(replicaWriteLimits.getCurrentReplicaBytes(), greaterThan(bulkShardRequestSize));
assertEquals(0, coordinatingWriteLimits.getCurrentCoordinatingAndPrimaryBytes());
assertEquals(0, coordinatingWriteLimits.getCurrentCombinedCoordinatingAndPrimaryBytes());
assertEquals(0, coordinatingWriteLimits.getCurrentReplicaBytes());
});

Expand All @@ -317,11 +342,11 @@ public void testWriteCanBeRejectedAtPrimaryLevel() throws Exception {

successFuture.actionGet();

assertEquals(0, primaryWriteLimits.getCurrentCoordinatingAndPrimaryBytes());
assertEquals(0, primaryWriteLimits.getCurrentCombinedCoordinatingAndPrimaryBytes());
assertEquals(0, primaryWriteLimits.getCurrentReplicaBytes());
assertEquals(0, replicaWriteLimits.getCurrentCoordinatingAndPrimaryBytes());
assertEquals(0, replicaWriteLimits.getCurrentCombinedCoordinatingAndPrimaryBytes());
assertEquals(0, replicaWriteLimits.getCurrentReplicaBytes());
assertEquals(0, coordinatingWriteLimits.getCurrentCoordinatingAndPrimaryBytes());
assertEquals(0, coordinatingWriteLimits.getCurrentCombinedCoordinatingAndPrimaryBytes());
assertEquals(0, coordinatingWriteLimits.getCurrentReplicaBytes());
}
}
Expand Down
2 changes: 1 addition & 1 deletion server/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -167,7 +167,7 @@ public static IndexRequest getIndexWriteRequest(DocWriteRequest<?> docWriteReque
@Override
protected void doExecute(Task task, BulkRequest bulkRequest, ActionListener<BulkResponse> listener) {
long indexingBytes = bulkRequest.ramBytesUsed();
final Releasable releasable = indexingPressure.markIndexingOperationStarted(indexingBytes);
final Releasable releasable = indexingPressure.markCoordinatingOperationStarted(indexingBytes);
final ActionListener<BulkResponse> releasingListener = ActionListener.runBefore(listener, releasable::close);
try {
doInternalExecute(task, bulkRequest, releasingListener);
Expand Down
Loading

0 comments on commit 408a07f

Please sign in to comment.