diff --git a/rest-api-spec/src/main/resources/rest-api-spec/api/indices.stats.json b/rest-api-spec/src/main/resources/rest-api-spec/api/indices.stats.json index 322a43dcdd134..f1bfc33daf8f7 100644 --- a/rest-api-spec/src/main/resources/rest-api-spec/api/indices.stats.json +++ b/rest-api-spec/src/main/resources/rest-api-spec/api/indices.stats.json @@ -84,7 +84,8 @@ "segments", "store", "warmer", - "suggest" + "suggest", + "bulk" ], "description":"Limit the information returned the specific metrics." } diff --git a/rest-api-spec/src/main/resources/rest-api-spec/api/nodes.stats.json b/rest-api-spec/src/main/resources/rest-api-spec/api/nodes.stats.json index 09027ba3a3103..0dc5e159a4528 100644 --- a/rest-api-spec/src/main/resources/rest-api-spec/api/nodes.stats.json +++ b/rest-api-spec/src/main/resources/rest-api-spec/api/nodes.stats.json @@ -168,7 +168,8 @@ "segments", "store", "warmer", - "suggest" + "suggest", + "bulk" ], "description":"Limit the information returned for `indices` metric to the specific index metrics. Isn't used if `indices` (or `all`) metric isn't specified." }, diff --git a/rest-api-spec/src/main/resources/rest-api-spec/test/cat.shards/10_basic.yml b/rest-api-spec/src/main/resources/rest-api-spec/test/cat.shards/10_basic.yml index de5e632975752..f8409d7a31da3 100644 --- a/rest-api-spec/src/main/resources/rest-api-spec/test/cat.shards/10_basic.yml +++ b/rest-api-spec/src/main/resources/rest-api-spec/test/cat.shards/10_basic.yml @@ -78,6 +78,9 @@ warmer.current .+ \n warmer.total .+ \n warmer.total_time .+ \n + bulk.total_operations .+ \n + bulk.total_time .+ \n + bulk.total_size_in_bytes .+ \n $/ --- "Test cat shards output": diff --git a/server/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java b/server/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java index 828147e997343..3a28b4eba1779 100644 --- a/server/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java +++ b/server/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java @@ -168,6 +168,7 @@ protected void doRun() throws Exception { } // We're done, there's no more operations to execute so we resolve the wrapped listener finishRequest(); + primary.getBulkOperationListener().afterBulk(request.totalSizeInBytes(), System.nanoTime() - startBulkTime); } @Override @@ -189,7 +190,6 @@ private void finishRequest() { () -> new WritePrimaryResult<>( context.getBulkShardRequest(), context.buildShardResponse(), context.getLocationToSync(), null, context.getPrimary(), logger)); - primary.getBulkOperationListener().afterBulk(request.totalSizeInBytes(), System.nanoTime() - startBulkTime); } }.run(); } diff --git a/server/src/main/java/org/elasticsearch/index/bulk/stats/BulkStats.java b/server/src/main/java/org/elasticsearch/index/bulk/stats/BulkStats.java index 63dfd68d48820..dad40c418686b 100644 --- a/server/src/main/java/org/elasticsearch/index/bulk/stats/BulkStats.java +++ b/server/src/main/java/org/elasticsearch/index/bulk/stats/BulkStats.java @@ -28,10 +28,15 @@ import org.elasticsearch.common.xcontent.XContentBuilder; import java.io.IOException; +import java.util.Objects; +/** + * Bulk related statistics, including the time and size of shard bulk requests, + * starting at the shard level and allowing aggregation to indices and node level + */ public class BulkStats implements Writeable, ToXContentFragment { - private long total = 0; + private long totalOperations = 0; private long totalTimeInMillis = 0; private long totalSizeInBytes = 0; @@ -40,13 +45,13 @@ public BulkStats() { } public BulkStats(StreamInput in) throws IOException { - total = in.readVLong(); + totalOperations = in.readVLong(); totalTimeInMillis = in.readVLong(); totalSizeInBytes = in.readVLong(); } - public BulkStats(long total, long totalTimeInMillis, long totalSizeInBytes) { - this.total = total; + public BulkStats(long totalOperations, long totalTimeInMillis, long totalSizeInBytes) { + this.totalOperations = totalOperations; this.totalTimeInMillis = totalTimeInMillis; this.totalSizeInBytes = totalSizeInBytes; } @@ -59,7 +64,7 @@ public void addTotals(BulkStats bulkStats) { if (bulkStats == null) { return; } - this.total += bulkStats.total; + this.totalOperations += bulkStats.totalOperations; this.totalTimeInMillis += bulkStats.totalTimeInMillis; this.totalSizeInBytes += bulkStats.totalSizeInBytes; } @@ -68,8 +73,8 @@ public long getTotalSizeInBytes() { return totalSizeInBytes; } - public long getTotal() { - return total; + public long getTotalOperations() { + return totalOperations; } public TimeValue getTotalTime() { @@ -80,24 +85,46 @@ public long getTotalTimeInMillis() { return totalTimeInMillis; } - @Override public void writeTo(StreamOutput out) throws IOException { - out.writeVLong(total); + @Override + public void writeTo(StreamOutput out) throws IOException { + out.writeVLong(totalOperations); out.writeVLong(totalTimeInMillis); out.writeVLong(totalSizeInBytes); } - @Override public XContentBuilder toXContent(XContentBuilder builder, ToXContent.Params params) throws IOException { + @Override + public XContentBuilder toXContent(XContentBuilder builder, ToXContent.Params params) throws IOException { builder.startObject(Fields.BULK); - builder.field(Fields.TOTAL, total); + builder.field(Fields.TOTAL_OPERATIONS, totalOperations); builder.humanReadableField(Fields.TOTAL_TIME_IN_MILLIS, Fields.TOTAL_TIME, getTotalTime()); builder.field(Fields.TOTAL_SIZE_IN_BYTES, totalSizeInBytes); builder.endObject(); return builder; } + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + + final BulkStats that = (BulkStats) o; + return Objects.equals(this.totalOperations, that.totalOperations) + && Objects.equals(this.totalTimeInMillis, that.totalTimeInMillis) + && Objects.equals(this.totalSizeInBytes, that.totalSizeInBytes); + } + + @Override + public int hashCode() { + return Objects.hash(totalOperations, totalTimeInMillis, totalSizeInBytes); + } + static final class Fields { static final String BULK = "bulk"; - static final String TOTAL = "total"; + static final String TOTAL_OPERATIONS = "total_operations"; static final String TOTAL_TIME = "total_time"; static final String TOTAL_TIME_IN_MILLIS = "total_time_in_millis"; static final String TOTAL_SIZE_IN_BYTES = "total_size_in_bytes"; diff --git a/server/src/main/java/org/elasticsearch/index/bulk/stats/ShardBulkStats.java b/server/src/main/java/org/elasticsearch/index/bulk/stats/ShardBulkStats.java index ec3b1ded3c39a..8f1869c1ec289 100644 --- a/server/src/main/java/org/elasticsearch/index/bulk/stats/ShardBulkStats.java +++ b/server/src/main/java/org/elasticsearch/index/bulk/stats/ShardBulkStats.java @@ -21,9 +21,14 @@ import org.elasticsearch.common.metrics.CounterMetric; import org.elasticsearch.common.metrics.MeanMetric; +import org.elasticsearch.index.shard.IndexShard; import java.util.concurrent.TimeUnit; +/** + * Internal class that maintains relevant shard bulk statistics / metrics. + * @see IndexShard + */ public class ShardBulkStats implements BulkOperationListener { private final StatsHolder totalStats = new StatsHolder(); @@ -32,7 +37,8 @@ public BulkStats stats() { return totalStats.stats(); } - @Override public void afterBulk(long shardBulkSizeInBytes, long tookInNanos) { + @Override + public void afterBulk(long shardBulkSizeInBytes, long tookInNanos) { totalStats.totalSizeInBytes.inc(shardBulkSizeInBytes); totalStats.shardBulkMetric.inc(tookInNanos); } diff --git a/server/src/main/java/org/elasticsearch/rest/action/cat/RestIndicesAction.java b/server/src/main/java/org/elasticsearch/rest/action/cat/RestIndicesAction.java index d90736be20426..230de7c40254d 100644 --- a/server/src/main/java/org/elasticsearch/rest/action/cat/RestIndicesAction.java +++ b/server/src/main/java/org/elasticsearch/rest/action/cat/RestIndicesAction.java @@ -487,9 +487,9 @@ protected Table getTableWithHeader(final RestRequest request) { table.addCell("search.throttled", "alias:sth;default:false;desc:indicates if the index is search throttled"); - table.addCell("bulk.total", - "sibling:pri;alias:bto,bulkTotal;default:false;text-align:right;desc:number of bulk shard ops"); - table.addCell("pri.bulk.shard_bulk_time", "default:false;text-align:right;desc:number of bulk shard ops"); + table.addCell("bulk.total_operations", + "sibling:pri;alias:bto,bulkTotalOperation;default:false;text-align:right;desc:number of bulk shard ops"); + table.addCell("pri.bulk.total_operations", "default:false;text-align:right;desc:number of bulk shard ops"); table.addCell("bulk.total_time", "sibling:pri;alias:btti,bulkTotalTime;default:false;text-align:right;desc:time spend in shard bulk"); table.addCell("pri.bulk.shard_bulk_time", "default:false;text-align:right;desc:time spend in shard bulk"); @@ -756,8 +756,8 @@ Table buildTable(final RestRequest request, table.addCell(searchThrottled); - table.addCell(totalStats.getBulk() == null ? null : totalStats.getBulk().getTotal()); - table.addCell(primaryStats.getBulk() == null ? null : primaryStats.getBulk().getTotal()); + table.addCell(totalStats.getBulk() == null ? null : totalStats.getBulk().getTotalOperations()); + table.addCell(primaryStats.getBulk() == null ? null : primaryStats.getBulk().getTotalOperations()); table.addCell(totalStats.getBulk() == null ? null : totalStats.getBulk().getTotalTime()); table.addCell(primaryStats.getBulk() == null ? null : primaryStats.getBulk().getTotalTime()); diff --git a/server/src/main/java/org/elasticsearch/rest/action/cat/RestNodesAction.java b/server/src/main/java/org/elasticsearch/rest/action/cat/RestNodesAction.java index ccf2d664fafcb..607d14e4d2f01 100644 --- a/server/src/main/java/org/elasticsearch/rest/action/cat/RestNodesAction.java +++ b/server/src/main/java/org/elasticsearch/rest/action/cat/RestNodesAction.java @@ -243,7 +243,7 @@ protected Table getTableWithHeader(final RestRequest request) { table.addCell("suggest.time", "alias:suti,suggestTime;default:false;text-align:right;desc:time spend in suggest"); table.addCell("suggest.total", "alias:suto,suggestTotal;default:false;text-align:right;desc:number of suggest ops"); - table.addCell("bulk.total", "alias:bto,bulkTotal;default:false;text-align:right;desc:number of bulk shard ops"); + table.addCell("bulk.total_operations", "alias:bto,bulkTotalOperations;default:false;text-align:right;desc:number of bulk shard ops"); table.addCell("bulk.total_time", "alias:btti,bulkTotalTime;default:false;text-align:right;desc:time spend in shard bulk"); table.addCell("bulk.total_size_in_bytes", "alias:btsi,bulkTotalSizeInBytes;default:false;text-align:right;desc:total size in bytes of shard bulk"); @@ -422,7 +422,7 @@ Table buildTable(boolean fullId, RestRequest req, ClusterStateResponse state, No table.addCell(searchStats == null ? null : searchStats.getTotal().getSuggestCount()); BulkStats bulkStats = indicesStats == null ? null : indicesStats.getBulk(); - table.addCell(bulkStats == null ? null : bulkStats.getTotal()); + table.addCell(bulkStats == null ? null : bulkStats.getTotalOperations()); table.addCell(bulkStats == null ? null : bulkStats.getTotalTime()); table.addCell(bulkStats == null ? null : bulkStats.getTotalSizeInBytes()); diff --git a/server/src/main/java/org/elasticsearch/rest/action/cat/RestShardsAction.java b/server/src/main/java/org/elasticsearch/rest/action/cat/RestShardsAction.java index 427580735d9d9..6f9ab28b0c5ba 100644 --- a/server/src/main/java/org/elasticsearch/rest/action/cat/RestShardsAction.java +++ b/server/src/main/java/org/elasticsearch/rest/action/cat/RestShardsAction.java @@ -200,7 +200,7 @@ protected Table getTableWithHeader(final RestRequest request) { table.addCell("warmer.total", "alias:wto,warmerTotal;default:false;text-align:right;desc:total warmer ops"); table.addCell("warmer.total_time", "alias:wtt,warmerTotalTime;default:false;text-align:right;desc:time spent in warmers"); - table.addCell("bulk.total", "alias:bto,bulkTotal;default:false;text-align:right;desc:number of bulk shard ops"); + table.addCell("bulk.total_operations", "alias:bto,bulkTotalOperations;default:false;text-align:right;desc:number of bulk shard ops"); table.addCell("bulk.total_time", "alias:btti,bulkTotalTime;default:false;text-align:right;desc:time spend in shard bulk"); table.addCell("bulk.total_size_in_bytes", "alias:btsi,bulkTotalSizeInBytes;default:false;text-align:right;desc:total size in bytes of shard bulk"); @@ -355,7 +355,7 @@ private Table buildTable(RestRequest request, ClusterStateResponse state, Indice table.addCell(getOrNull(commonStats, CommonStats::getWarmer, WarmerStats::total)); table.addCell(getOrNull(commonStats, CommonStats::getWarmer, WarmerStats::totalTime)); - table.addCell(getOrNull(commonStats, CommonStats::getBulk, BulkStats::getTotal)); + table.addCell(getOrNull(commonStats, CommonStats::getBulk, BulkStats::getTotalOperations)); table.addCell(getOrNull(commonStats, CommonStats::getBulk, BulkStats::getTotalTime)); table.addCell(getOrNull(commonStats, CommonStats::getBulk, BulkStats::getTotalSizeInBytes)); diff --git a/server/src/test/java/org/elasticsearch/index/bulk/stats/BulkStatsTests.java b/server/src/test/java/org/elasticsearch/index/bulk/stats/BulkStatsTests.java index 74101a464e8cc..d5c70f7908fc2 100644 --- a/server/src/test/java/org/elasticsearch/index/bulk/stats/BulkStatsTests.java +++ b/server/src/test/java/org/elasticsearch/index/bulk/stats/BulkStatsTests.java @@ -19,24 +19,34 @@ package org.elasticsearch.index.bulk.stats; -import org.elasticsearch.common.io.stream.BytesStreamOutput; -import org.elasticsearch.common.io.stream.StreamInput; -import org.elasticsearch.test.ESTestCase; - -import java.io.IOException; - -public class BulkStatsTests extends ESTestCase { - - public void testSerialize() throws IOException { - BulkStats stats = new BulkStats(randomNonNegativeLong(), randomNonNegativeLong(), randomNonNegativeLong()); - BytesStreamOutput out = new BytesStreamOutput(); - stats.writeTo(out); - StreamInput input = out.bytes().streamInput(); - BulkStats read = new BulkStats(input); - assertEquals(-1, input.read()); - assertEquals(stats.getTotal(), read.getTotal()); - assertEquals(stats.getTotalTime(), read.getTotalTime()); - assertEquals(stats.getTotalSizeInBytes(), read.getTotalSizeInBytes()); +import org.elasticsearch.common.io.stream.Writeable; +import org.elasticsearch.test.AbstractWireSerializingTestCase; + +public class BulkStatsTests extends AbstractWireSerializingTestCase { + + @Override + protected Writeable.Reader instanceReader() { + return BulkStats::new; + } + + @Override + protected BulkStats createTestInstance() { + return new BulkStats(randomNonNegativeLong(), randomNonNegativeLong(), randomNonNegativeLong()); + } + + @Override + protected BulkStats mutateInstance(BulkStats instance) { + BulkStats mutateBulkStats = new BulkStats(randomNonNegativeLong(), randomNonNegativeLong(), randomNonNegativeLong()); + switch (between(0, 1)) { + case 0: + break; + case 1: + mutateBulkStats.add(instance); + break; + default: + throw new AssertionError("Illegal randomisation branch"); + } + return mutateBulkStats; } public void testAddTotals() { @@ -54,9 +64,10 @@ public void testAddTotals() { } private static void assertStats(BulkStats stats, long equalTo) { - assertEquals(equalTo, stats.getTotal()); + assertEquals(equalTo, stats.getTotalOperations()); assertEquals(equalTo, stats.getTotalTimeInMillis()); assertEquals(equalTo, stats.getTotalSizeInBytes()); } + } diff --git a/server/src/test/java/org/elasticsearch/indices/stats/IndexStatsIT.java b/server/src/test/java/org/elasticsearch/indices/stats/IndexStatsIT.java index ff44d35dc6d73..c8a5ebf3f2a31 100644 --- a/server/src/test/java/org/elasticsearch/indices/stats/IndexStatsIT.java +++ b/server/src/test/java/org/elasticsearch/indices/stats/IndexStatsIT.java @@ -1095,11 +1095,11 @@ public void testBulkStats() throws Exception { } IndicesStatsResponse stats = client().admin().indices().prepareStats(index).setBulk(true).get(); - assertThat(stats.getTotal().bulk.getTotal(), equalTo(4L)); + assertThat(stats.getTotal().bulk.getTotalOperations(), equalTo(4L)); assertThat(stats.getTotal().bulk.getTotalTimeInMillis(), greaterThan(0L)); assertThat(stats.getTotal().bulk.getTotalSizeInBytes(), greaterThan(0L)); - assertThat(stats.getPrimaries().bulk.getTotal(), equalTo(2L)); + assertThat(stats.getPrimaries().bulk.getTotalOperations(), equalTo(2L)); assertThat(stats.getPrimaries().bulk.getTotalTimeInMillis(), greaterThan(0L)); assertThat(stats.getPrimaries().bulk.getTotalSizeInBytes(), greaterThan(0L)); }