Skip to content

Commit

Permalink
[7.x][Transform] provide exponential_avg* stats for batch transforms (#…
Browse files Browse the repository at this point in the history 
…52041) (#52323)

provide exponential_avg* stats for batch transforms, avoids confusion
why those values are all 0 otherwise
  • Loading branch information
Hendrik Muhs authored Feb 14, 2020
1 parent a669882 commit efd7542
Show file tree
Hide file tree
Showing 9 changed files with 222 additions and 120 deletions.
120 changes: 87 additions & 33 deletions ...rc/main/java/org/elasticsearch/xpack/core/transform/transforms/TransformIndexerStats.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@

import org.elasticsearch.Version;
import org.elasticsearch.common.ParseField;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.xcontent.ConstructingObjectParser;
Expand Down Expand Up @@ -36,23 +37,34 @@ public class TransformIndexerStats extends IndexerJobStats {
public static ParseField SEARCH_TOTAL = new ParseField("search_total");
public static ParseField SEARCH_FAILURES = new ParseField("search_failures");
public static ParseField INDEX_FAILURES = new ParseField("index_failures");
public static ParseField EXPONENTIAL_AVG_CHECKPOINT_DURATION_MS =
new ParseField("exponential_avg_checkpoint_duration_ms");
public static ParseField EXPONENTIAL_AVG_DOCUMENTS_INDEXED =
new ParseField("exponential_avg_documents_indexed");
public static ParseField EXPONENTIAL_AVG_DOCUMENTS_PROCESSED =
new ParseField("exponential_avg_documents_processed");
public static ParseField EXPONENTIAL_AVG_CHECKPOINT_DURATION_MS = new ParseField("exponential_avg_checkpoint_duration_ms");
public static ParseField EXPONENTIAL_AVG_DOCUMENTS_INDEXED = new ParseField("exponential_avg_documents_indexed");
public static ParseField EXPONENTIAL_AVG_DOCUMENTS_PROCESSED = new ParseField("exponential_avg_documents_processed");

// This changes how much "weight" past calculations have.
// The shorter the window, the less "smoothing" will occur.
private static final int EXP_AVG_WINDOW = 10;
private static final double ALPHA = 2.0/(EXP_AVG_WINDOW + 1);
private static final double ALPHA = 2.0 / (EXP_AVG_WINDOW + 1);

private static final ConstructingObjectParser<TransformIndexerStats, Void> LENIENT_PARSER = new ConstructingObjectParser<>(
NAME, true,
args -> new TransformIndexerStats(
(long) args[0], (long) args[1], (long) args[2], (long) args[3], (long) args[4], (long) args[5], (long) args[6],
(long) args[7], (long) args[8], (long) args[9], (Double) args[10], (Double) args[11], (Double) args[12]));
NAME,
true,
args -> new TransformIndexerStats(
(long) args[0],
(long) args[1],
(long) args[2],
(long) args[3],
(long) args[4],
(long) args[5],
(long) args[6],
(long) args[7],
(long) args[8],
(long) args[9],
(Double) args[10],
(Double) args[11],
(Double) args[12]
)
);

static {
LENIENT_PARSER.declareLong(constructorArg(), NUM_PAGES);
Expand All @@ -73,37 +85,62 @@ public class TransformIndexerStats extends IndexerJobStats {
private double expAvgCheckpointDurationMs;
private double expAvgDocumentsIndexed;
private double expAvgDocumentsProcessed;

/**
* Create with all stats set to zero
*/
public TransformIndexerStats() {
super();
}

public TransformIndexerStats(long numPages, long numInputDocuments, long numOutputDocuments,
long numInvocations, long indexTime, long searchTime, long indexTotal, long searchTotal,
long indexFailures, long searchFailures, Double expAvgCheckpointDurationMs,
Double expAvgDocumentsIndexed, Double expAvgDocumentsProcessed ) {
super(numPages, numInputDocuments, numOutputDocuments, numInvocations, indexTime, searchTime, indexTotal, searchTotal,
indexFailures, searchFailures);
public TransformIndexerStats(
long numPages,
long numInputDocuments,
long numOutputDocuments,
long numInvocations,
long indexTime,
long searchTime,
long indexTotal,
long searchTotal,
long indexFailures,
long searchFailures,
Double expAvgCheckpointDurationMs,
Double expAvgDocumentsIndexed,
Double expAvgDocumentsProcessed
) {
super(
numPages,
numInputDocuments,
numOutputDocuments,
numInvocations,
indexTime,
searchTime,
indexTotal,
searchTotal,
indexFailures,
searchFailures
);
this.expAvgCheckpointDurationMs = expAvgCheckpointDurationMs == null ? 0.0 : expAvgCheckpointDurationMs;
this.expAvgDocumentsIndexed = expAvgDocumentsIndexed == null ? 0.0 : expAvgDocumentsIndexed;
this.expAvgDocumentsProcessed = expAvgDocumentsProcessed == null ? 0.0 : expAvgDocumentsProcessed;
}

public TransformIndexerStats(long numPages, long numInputDocuments, long numOutputDocuments,
long numInvocations, long indexTime, long searchTime, long indexTotal, long searchTotal,
long indexFailures, long searchFailures) {
this(numPages, numInputDocuments, numOutputDocuments, numInvocations, indexTime, searchTime, indexTotal, searchTotal,
indexFailures, searchFailures, 0.0, 0.0, 0.0);
}

public TransformIndexerStats(TransformIndexerStats other) {
this(other.numPages, other.numInputDocuments, other.numOuputDocuments, other.numInvocations,
other.indexTime, other.searchTime, other.indexTotal, other.searchTotal, other.indexFailures, other.searchFailures);
this.expAvgCheckpointDurationMs = other.expAvgCheckpointDurationMs;
this.expAvgDocumentsIndexed = other.expAvgDocumentsIndexed;
this.expAvgDocumentsProcessed = other.expAvgDocumentsProcessed;
this(
other.numPages,
other.numInputDocuments,
other.numOuputDocuments,
other.numInvocations,
other.indexTime,
other.searchTime,
other.indexTotal,
other.searchTotal,
other.indexFailures,
other.searchFailures,
other.expAvgCheckpointDurationMs,
other.expAvgDocumentsIndexed,
other.expAvgDocumentsProcessed
);
}

public TransformIndexerStats(StreamInput in) throws IOException {
Expand Down Expand Up @@ -180,7 +217,7 @@ public void incrementCheckpointExponentialAverages(long checkpointDurationMs, lo
}

private double calculateExpAvg(double previousExpValue, double alpha, long observedValue) {
return alpha * observedValue + (1-alpha) * previousExpValue;
return alpha * observedValue + (1 - alpha) * previousExpValue;
}

@Override
Expand Down Expand Up @@ -212,9 +249,26 @@ public boolean equals(Object other) {

@Override
public int hashCode() {
return Objects.hash(numPages, numInputDocuments, numOuputDocuments, numInvocations,
indexTime, searchTime, indexFailures, searchFailures, indexTotal, searchTotal,
expAvgCheckpointDurationMs, expAvgDocumentsIndexed, expAvgDocumentsProcessed);
return Objects.hash(
numPages,
numInputDocuments,
numOuputDocuments,
numInvocations,
indexTime,
searchTime,
indexFailures,
searchFailures,
indexTotal,
searchTotal,
expAvgCheckpointDurationMs,
expAvgDocumentsIndexed,
expAvgDocumentsProcessed
);
}

@Override
public String toString() {
return Strings.toString(this);
}

public static TransformIndexerStats fromXContent(XContentParser parser) {
Expand Down
16 changes: 12 additions & 4 deletions ...st/java/org/elasticsearch/xpack/core/transform/transforms/TransformIndexerStatsTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -31,13 +31,21 @@ protected TransformIndexerStats doParseInstance(XContentParser parser) {
}

public static TransformIndexerStats randomStats() {
return new TransformIndexerStats(randomLongBetween(10L, 10000L),
randomLongBetween(0L, 10000L), randomLongBetween(0L, 10000L), randomLongBetween(0L, 10000L), randomLongBetween(0L, 10000L),
randomLongBetween(0L, 10000L), randomLongBetween(0L, 10000L), randomLongBetween(0L, 10000L), randomLongBetween(0L, 10000L),
return new TransformIndexerStats(
randomLongBetween(10L, 10000L),
randomLongBetween(0L, 10000L),
randomLongBetween(0L, 10000L),
randomLongBetween(0L, 10000L),
randomLongBetween(0L, 10000L),
randomLongBetween(0L, 10000L),
randomLongBetween(0L, 10000L),
randomLongBetween(0L, 10000L),
randomLongBetween(0L, 10000L),
randomLongBetween(0L, 10000L),
randomBoolean() ? randomDouble() : null,
randomBoolean() ? randomDouble() : null,
randomBoolean() ? randomDouble() : null);
randomBoolean() ? randomDouble() : null
);
}

public void testExpAvgIncrement() {
Expand Down
2 changes: 1 addition & 1 deletion .../src/test/java/org/elasticsearch/xpack/core/transform/transforms/TransformStatsTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,7 @@ public void testBwcWith73() throws IOException {
STARTED,
randomBoolean() ? null : randomAlphaOfLength(100),
randomBoolean() ? null : NodeAttributeTests.randomNodeAttributes(),
new TransformIndexerStats(1, 2, 3, 4, 5, 6, 7, 8, 9, 10),
new TransformIndexerStats(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 0.0, 0.0, 0.0),
new TransformCheckpointingInfo(
new TransformCheckpointStats(0, null, null, 10, 100),
new TransformCheckpointStats(0, null, null, 100, 1000),
Expand Down
1 change: 0 additions & 1 deletion x-pack/plugin/src/test/resources/rest-api-spec/test/transform/transforms_stats.yml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -247,7 +247,6 @@ teardown:
- gte: { transforms.0.stats.search_time_in_ms: 0 }
- lte: { transforms.0.stats.search_total: 1 }
- match: { transforms.0.stats.search_failures: 0 }
- match: { transforms.0.stats.exponential_avg_checkpoint_duration_ms: 0.0 }
- match: { transforms.0.stats.exponential_avg_documents_indexed: 0.0 }
- match: { transforms.0.stats.exponential_avg_documents_processed: 0.0 }

Expand Down
Loading

0 comments on commit efd7542

Please sign in to comment.