Skip to content

Commit e3fc87a

Browse files
jasontedorjasontedor
committed
Simplify stats
1 parent 42c309d commit e3fc87a

File tree

2 files changed

+41
-47
lines changed

2 files changed

+41
-47
lines changed

‎x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/CcrStatsAction.java

Lines changed: 4 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -15,9 +15,6 @@
1515
import org.elasticsearch.common.io.stream.StreamInput;
1616
import org.elasticsearch.common.io.stream.StreamOutput;
1717
import org.elasticsearch.common.io.stream.Writeable;
18-
import org.elasticsearch.common.unit.ByteSizeUnit;
19-
import org.elasticsearch.common.unit.ByteSizeValue;
20-
import org.elasticsearch.common.unit.TimeValue;
2118
import org.elasticsearch.common.xcontent.ToXContentObject;
2219
import org.elasticsearch.common.xcontent.XContentBuilder;
2320
import org.elasticsearch.index.shard.ShardId;
@@ -28,7 +25,6 @@
2825
import java.util.List;
2926
import java.util.Map;
3027
import java.util.TreeMap;
31-
import java.util.concurrent.TimeUnit;
3228

3329
public class CcrStatsAction extends Action<CcrStatsAction.TasksResponse> {
3430

@@ -63,6 +59,7 @@ public TasksResponse() {
6359

6460
@Override
6561
public XContentBuilder toXContent(final XContentBuilder builder, final Params params) throws IOException {
62+
// sort by index name, then shard ID
6663
final Map<String, Map<Integer, TaskResponse>> taskResponsesByIndex = new TreeMap<>();
6764
for (final TaskResponse taskResponse : taskResponses) {
6865
taskResponsesByIndex.computeIfAbsent(
@@ -72,45 +69,13 @@ public XContentBuilder toXContent(final XContentBuilder builder, final Params pa
7269
builder.startObject();
7370
{
7471
for (final Map.Entry<String, Map<Integer, TaskResponse>> index : taskResponsesByIndex.entrySet()) {
75-
builder.startObject(index.getKey());
72+
builder.startArray(index.getKey());
7673
{
7774
for (final Map.Entry<Integer, TaskResponse> shard : index.getValue().entrySet()) {
78-
builder.startObject(Integer.toString(shard.getKey()));
79-
{
80-
final ShardFollowNodeTask.Status status = shard.getValue().status();
81-
builder.field("leader_global_checkpoint", status.leaderGlobalCheckpoint());
82-
builder.field("leader_max_seq_no", status.leaderMaxSeqNo());
83-
builder.field("follower_global_checkpoint", status.followerGlobalCheckpoint());
84-
builder.field("follower_max_seq_no", status.followerMaxSeqNo());
85-
builder.field("last_requested_seq_no", status.lastRequestedSeqNo());
86-
builder.field("number_of_concurrent_reads", status.numberOfConcurrentReads());
87-
builder.field("number_of_concurrent_writes", status.numberOfConcurrentWrites());
88-
builder.field("number_of_queued_writes", status.numberOfQueuedWrites());
89-
builder.humanReadableField(
90-
"total_fetch_time_millis",
91-
"total_fetch_time",
92-
new TimeValue(status.totalFetchTimeNanos(), TimeUnit.NANOSECONDS));
93-
builder.field("number_of_successful_fetches", status.numberOfSuccessfulFetches());
94-
builder.field("number_of_fetches", status.numberOfSuccessfulFetches() + status.numberOfFailedFetches());
95-
builder.field("operations_received", status.operationsReceived());
96-
builder.humanReadableField(
97-
"total_transferred_bytes",
98-
"total_transferred",
99-
new ByteSizeValue(status.totalTransferredBytes(), ByteSizeUnit.BYTES));
100-
builder.humanReadableField(
101-
"total_index_time_millis",
102-
"total_index_time",
103-
new TimeValue(status.totalIndexTimeNanos(), TimeUnit.NANOSECONDS));
104-
builder.field("number_of_successful_bulk_operations", status.numberOfSuccessfulBulkOperations());
105-
builder.field(
106-
"number_of_bulk_operations",
107-
status.numberOfSuccessfulBulkOperations() + status.numberOfFailedBulkOperations());
108-
builder.field("number_of_operations_indexed", status.numberOfOperationsIndexed());
109-
}
110-
builder.endObject();
75+
shard.getValue().status().toXContent(builder, params);
11176
}
11277
}
113-
builder.endObject();
78+
builder.endArray();
11479
}
11580
}
11681
builder.endObject();

‎x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowNodeTask.java

Lines changed: 37 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,8 @@
1616
import org.elasticsearch.common.io.stream.StreamOutput;
1717
import org.elasticsearch.common.logging.Loggers;
1818
import org.elasticsearch.common.transport.NetworkExceptionHelper;
19+
import org.elasticsearch.common.unit.ByteSizeUnit;
20+
import org.elasticsearch.common.unit.ByteSizeValue;
1921
import org.elasticsearch.common.unit.TimeValue;
2022
import org.elasticsearch.common.xcontent.ConstructingObjectParser;
2123
import org.elasticsearch.common.xcontent.XContentBuilder;
@@ -36,6 +38,7 @@
3638
import java.util.Objects;
3739
import java.util.PriorityQueue;
3840
import java.util.Queue;
41+
import java.util.concurrent.TimeUnit;
3942
import java.util.concurrent.atomic.AtomicInteger;
4043
import java.util.function.BiConsumer;
4144
import java.util.function.Consumer;
@@ -380,6 +383,7 @@ public ShardId getFollowShardId() {
380383
@Override
381384
public synchronized Status getStatus() {
382385
return new Status(
386+
getFollowShardId().getId(),
383387
leaderGlobalCheckpoint,
384388
leaderMaxSeqNo,
385389
followerGlobalCheckpoint,
@@ -404,6 +408,7 @@ public static class Status implements Task.Status {
404408

405409
public static final String NAME = "shard-follow-node-task-status";
406410

411+
static final ParseField SHARD_ID = new ParseField("shard_id");
407412
static final ParseField LEADER_GLOBAL_CHECKPOINT_FIELD = new ParseField("leader_global_checkpoint");
408413
static final ParseField LEADER_MAX_SEQ_NO_FIELD = new ParseField("leader_max_seq_no");
409414
static final ParseField FOLLOWER_GLOBAL_CHECKPOINT_FIELD = new ParseField("follower_global_checkpoint");
@@ -425,15 +430,15 @@ public static class Status implements Task.Status {
425430

426431
static final ConstructingObjectParser<Status, Void> PARSER = new ConstructingObjectParser<>(NAME,
427432
args -> new Status(
428-
(long) args[0],
433+
(int) args[0],
429434
(long) args[1],
430435
(long) args[2],
431436
(long) args[3],
432437
(long) args[4],
433-
(int) args[5],
438+
(long) args[5],
434439
(int) args[6],
435440
(int) args[7],
436-
(long) args[8],
441+
(int) args[8],
437442
(long) args[9],
438443
(long) args[10],
439444
(long) args[11],
@@ -442,9 +447,11 @@ public static class Status implements Task.Status {
442447
(long) args[14],
443448
(long) args[15],
444449
(long) args[16],
445-
(long) args[17]));
450+
(long) args[17],
451+
(long) args[18]));
446452

447453
static {
454+
PARSER.declareInt(ConstructingObjectParser.constructorArg(), SHARD_ID);
448455
PARSER.declareLong(ConstructingObjectParser.constructorArg(), LEADER_GLOBAL_CHECKPOINT_FIELD);
449456
PARSER.declareLong(ConstructingObjectParser.constructorArg(), LEADER_MAX_SEQ_NO_FIELD);
450457
PARSER.declareLong(ConstructingObjectParser.constructorArg(), FOLLOWER_GLOBAL_CHECKPOINT_FIELD);
@@ -465,6 +472,12 @@ public static class Status implements Task.Status {
465472
PARSER.declareLong(ConstructingObjectParser.constructorArg(), NUMBER_OF_OPERATIONS_INDEXED_FIELD);
466473
}
467474

475+
private final int shardId;
476+
477+
public int getShardId() {
478+
return shardId;
479+
}
480+
468481
private final long leaderGlobalCheckpoint;
469482

470483
public long leaderGlobalCheckpoint() {
@@ -574,6 +587,7 @@ public long numberOfOperationsIndexed() {
574587
}
575588

576589
Status(
590+
final int shardId,
577591
final long leaderGlobalCheckpoint,
578592
final long leaderMaxSeqNo,
579593
final long followerGlobalCheckpoint,
@@ -592,6 +606,7 @@ public long numberOfOperationsIndexed() {
592606
final long numberOfSuccessfulBulkOperations,
593607
final long numberOfFailedBulkOperations,
594608
final long numberOfOperationsIndexed) {
609+
this.shardId = shardId;
595610
this.leaderGlobalCheckpoint = leaderGlobalCheckpoint;
596611
this.leaderMaxSeqNo = leaderMaxSeqNo;
597612
this.followerGlobalCheckpoint = followerGlobalCheckpoint;
@@ -613,6 +628,7 @@ public long numberOfOperationsIndexed() {
613628
}
614629

615630
public Status(final StreamInput in) throws IOException {
631+
this.shardId = in.readVInt();
616632
this.leaderGlobalCheckpoint = in.readZLong();
617633
this.leaderMaxSeqNo = in.readZLong();
618634
this.followerGlobalCheckpoint = in.readZLong();
@@ -640,6 +656,7 @@ public String getWriteableName() {
640656

641657
@Override
642658
public void writeTo(final StreamOutput out) throws IOException {
659+
out.writeVInt(shardId);
643660
out.writeZLong(leaderGlobalCheckpoint);
644661
out.writeZLong(leaderMaxSeqNo);
645662
out.writeZLong(followerGlobalCheckpoint);
@@ -664,6 +681,7 @@ public void writeTo(final StreamOutput out) throws IOException {
664681
public XContentBuilder toXContent(final XContentBuilder builder, final Params params) throws IOException {
665682
builder.startObject();
666683
{
684+
builder.field(SHARD_ID.getPreferredName(), shardId);
667685
builder.field(LEADER_GLOBAL_CHECKPOINT_FIELD.getPreferredName(), leaderGlobalCheckpoint);
668686
builder.field(LEADER_MAX_SEQ_NO_FIELD.getPreferredName(), leaderMaxSeqNo);
669687
builder.field(FOLLOWER_GLOBAL_CHECKPOINT_FIELD.getPreferredName(), followerGlobalCheckpoint);
@@ -673,12 +691,21 @@ public XContentBuilder toXContent(final XContentBuilder builder, final Params pa
673691
builder.field(NUMBER_OF_CONCURRENT_WRITES_FIELD.getPreferredName(), numberOfConcurrentWrites);
674692
builder.field(NUMBER_OF_QUEUED_WRITES_FIELD.getPreferredName(), numberOfQueuedWrites);
675693
builder.field(INDEX_METADATA_VERSION_FIELD.getPreferredName(), indexMetadataVersion);
676-
builder.field(TOTAL_FETCH_TIME_NANOS_FIELD.getPreferredName(), totalFetchTimeNanos);
694+
builder.humanReadableField(
695+
TOTAL_FETCH_TIME_NANOS_FIELD.getPreferredName(),
696+
"total_fetch_time",
697+
new TimeValue(totalFetchTimeNanos, TimeUnit.NANOSECONDS));
677698
builder.field(NUMBER_OF_SUCCESSFUL_FETCHES_FIELD.getPreferredName(), numberOfSuccessfulFetches);
678699
builder.field(NUMBER_OF_FAILED_FETCHES_FIELD.getPreferredName(), numberOfFailedFetches);
679700
builder.field(OPERATIONS_RECEIVED_FIELD.getPreferredName(), operationsReceived);
680-
builder.field(TOTAL_TRANSFERRED_BYTES.getPreferredName(), totalTransferredBytes);
681-
builder.field(TOTAL_INDEX_TIME_NANOS_FIELD.getPreferredName(), totalIndexTimeNanos);
701+
builder.humanReadableField(
702+
TOTAL_TRANSFERRED_BYTES.getPreferredName(),
703+
"total_transferred",
704+
new ByteSizeValue(totalTransferredBytes, ByteSizeUnit.BYTES));
705+
builder.humanReadableField(
706+
TOTAL_INDEX_TIME_NANOS_FIELD.getPreferredName(),
707+
"total_index_time",
708+
new TimeValue(totalIndexTimeNanos, TimeUnit.NANOSECONDS));
682709
builder.field(NUMBER_OF_SUCCESSFUL_BULK_OPERATIONS_FIELD.getPreferredName(), numberOfSuccessfulBulkOperations);
683710
builder.field(NUMBER_OF_FAILED_BULK_OPERATIONS_FIELD.getPreferredName(), numberOfFailedBulkOperations);
684711
builder.field(NUMBER_OF_OPERATIONS_INDEXED_FIELD.getPreferredName(), numberOfOperationsIndexed);
@@ -696,7 +723,8 @@ public boolean equals(final Object o) {
696723
if (this == o) return true;
697724
if (o == null || getClass() != o.getClass()) return false;
698725
final Status that = (Status) o;
699-
return leaderGlobalCheckpoint == that.leaderGlobalCheckpoint &&
726+
return shardId == that.shardId &&
727+
leaderGlobalCheckpoint == that.leaderGlobalCheckpoint &&
700728
leaderMaxSeqNo == that.leaderMaxSeqNo &&
701729
followerGlobalCheckpoint == that.followerGlobalCheckpoint &&
702730
followerMaxSeqNo == that.followerMaxSeqNo &&
@@ -717,6 +745,7 @@ public boolean equals(final Object o) {
717745
@Override
718746
public int hashCode() {
719747
return Objects.hash(
748+
shardId,
720749
leaderGlobalCheckpoint,
721750
leaderMaxSeqNo,
722751
followerGlobalCheckpoint,

0 commit comments

Comments
 (0)