You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "emitskevich-blp (via GitHub)" <gi...@apache.org> on 2024/05/08 02:26:12 UTC
[PR] Fix Rate window size [kafka]
emitskevich-blp opened a new pull request, #15889:
URL: https://github.com/apache/kafka/pull/15889
`Rate` metric reports value in the form of `sumOrCount`/`timePeriod`. It has a bug in `timePeriod` calculation, which leads to spikes in result values (see real example: `io-wait-ratio` metric has ~100% -> ~50% spikes):
<img width="1119" alt="Screenshot 2024-05-07 at 10 10 33 PM" src="https://github.com/apache/kafka/assets/135832807/24800c04-b754-4c60-891d-5174823890dd">
`timePeriod` consists of several time windows (see [SampledStat](https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/metrics/stats/SampledStat.java) for ref). Time window ends after a fixed time following the first event. New window starts by the next event after prev window end. There is always some gap between windows, while no event is reported.
Before reporting value, `Rate` purges (`sumOrCount := 0`) all windows, started before configured interval. However, it’s only reflected in numerator. Denominator is always the whole interval.
Consider the picture, corresponding to default config (`numWindows = 2`):
<img width="932" alt="Screenshot 2024-05-07 at 9 48 26 PM" src="https://github.com/apache/kafka/assets/135832807/b5d2bebb-b158-4489-a490-0910c5c460f4">
When report is requested (`measure()`) at the end of right window, the left window is purged for numerator, but not for denominator. It is the reason for spikes on the chart above: when we move to such zone, numerator is reduced by 2 times, while denominator remains the same.
This PR fixes such behavior. Now, if some window is purged for numerator, it reduces denominator as well.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "emitskevich-blp (via GitHub)" <gi...@apache.org>.
emitskevich-blp commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1610653813
##########
clients/src/test/java/org/apache/kafka/common/metrics/stats/SampledStatTest.java:
##########
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.metrics.stats;
+
+import org.apache.kafka.common.metrics.MetricConfig;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.common.utils.Time;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class SampledStatTest {
+
+ private SampledStat stat;
+ private Time time;
+
+ @BeforeEach
+ public void setup() {
+ stat = new SampleCount(0);
+ time = new MockTime();
+ }
+
+ @Test
+ @DisplayName("Sample should be purged if doesn't overlap the window")
+ public void testSampleIsPurgedIfDoesntOverlap() {
+ MetricConfig config = new MetricConfig().timeWindow(1, SECONDS).samples(2);
+
+ // Monitored window: 2s. Complete a sample and wait 2.5s after.
+ completeSample(config);
+ time.sleep(2500);
+
+ double numSamples = stat.measure(config, time.milliseconds());
+ assertEquals(0, numSamples);
+ }
+
+ @Test
+ @DisplayName("Sample should be kept if overlaps the window")
+ public void testSampleIsKeptIfOverlaps() {
+ MetricConfig config = new MetricConfig().timeWindow(1, SECONDS).samples(2);
+
+ // Monitored window: 2s. Complete a sample and wait 1.5s after.
+ completeSample(config);
+ time.sleep(1500);
+
+ double numSamples = stat.measure(config, time.milliseconds());
+ assertEquals(1, numSamples);
+ }
+
+ @Test
+ @DisplayName("Sample should be kept if overlaps the window and is n+1")
+ public void testSampleIsKeptIfOverlapsAndExtra() {
+ MetricConfig config = new MetricConfig().timeWindow(1, SECONDS).samples(2);
+
+ // Monitored window: 2s. Create 2 samples with gaps in between and
+ // take a measurement at 2.2s from the start.
+ completeSample(config);
+ time.sleep(100);
+ completeSample(config);
+ time.sleep(100);
+ stat.record(config, 1, time.milliseconds());
+
+ double numSamples = stat.measure(config, time.milliseconds());
+ assertEquals(3, numSamples);
+ }
+
+ // Creates a sample with events at the start and at the end. Positions clock at the end.
+ private void completeSample(MetricConfig config) {
+ stat.record(config, 1, time.milliseconds());
+ time.sleep(config.timeWindowMs() - 1);
+ stat.record(config, 1, time.milliseconds());
+ time.sleep(1);
+ }
+
+ // measure() of this impl returns the number of samples
+ static class SampleCount extends SampledStat {
+
+ SampleCount(double initialValue) {
Review Comment:
Applied
##########
clients/src/main/java/org/apache/kafka/common/metrics/stats/SampledStat.java:
##########
@@ -40,7 +40,7 @@ public abstract class SampledStat implements MeasurableStat {
public SampledStat(double initialValue) {
this.initialValue = initialValue;
- this.samples = new ArrayList<>(2);
+ this.samples = new ArrayList<>(3);
Review Comment:
Addressed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "emitskevich-blp (via GitHub)" <gi...@apache.org>.
emitskevich-blp commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1617770678
##########
clients/src/test/java/org/apache/kafka/common/metrics/stats/FrequenciesTest.java:
##########
@@ -89,8 +88,17 @@ public void testBooleanFrequencies() {
}
assertEquals(0.25, falseMetric.stat().measure(config, time.milliseconds()), DELTA);
assertEquals(0.75, trueMetric.stat().measure(config, time.milliseconds()), DELTA);
+ }
- // Record 2 more windows worth of values
+ @Test
+ public void testBooleanFrequenciesStrategy2() {
Review Comment:
Tried just to keep all test scenarios as is to limit the scope of PR. So didn't judge it deeply. Should we?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "emitskevich-blp (via GitHub)" <gi...@apache.org>.
emitskevich-blp commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1610223223
##########
clients/src/test/java/org/apache/kafka/common/metrics/stats/RateTest.java:
##########
@@ -64,4 +69,30 @@ public void testRateWithNoPriorAvailableSamples(int numSample, int sampleWindowS
double expectedRatePerSec = sampleValue / windowSize;
assertEquals(expectedRatePerSec, observedRate, EPS);
}
+
+ // Record an event every 100 ms on average, moving some 1 ms back or forth for fine-grained
+ // window control. The expected rate, hence, is 10-11 events/sec depending on the moment of
+ // measurement. Start assertions from the second window.
+ @Test
+ public void testRateIsConsistentAfterTheFirstWindow() {
+ MetricConfig config = new MetricConfig().timeWindow(1, SECONDS).samples(2);
+ List<Integer> steps = Arrays.asList(0, 99, 100, 100, 100, 100, 100, 100, 100, 100, 100);
+
+ // start the first window and record events at 0,99,199,...,999 ms
+ for (int stepMs : steps) {
+ time.sleep(stepMs);
+ rate.record(config, 1, time.milliseconds());
+ }
+
+ // making a gap of 100 ms between windows
+ time.sleep(101);
+
+ // start the second window and record events at 0,99,199,...,999 ms
+ for (int stepMs : steps) {
+ time.sleep(stepMs);
+ rate.record(config, 1, time.milliseconds());
+ double observedRate = rate.measure(config, time.milliseconds());
Review Comment:
I’m not feeling strongly, could you please advise?
My thoughts:
Now Rate.measure() provides repeatable results when called at/with the same timestamp. And it did so before. That non-repeatable state, previously discovered in current PR, was never committed to trunk. So it wouldn’t be clear for future readers, what’s the rationale behind this double-checking in the test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #15889:
URL: https://github.com/apache/kafka/pull/15889#issuecomment-2127744276
Most failed tests can be fixed by #16044
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "emitskevich-blp (via GitHub)" <gi...@apache.org>.
emitskevich-blp commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1593310007
##########
clients/src/main/java/org/apache/kafka/common/metrics/stats/SampledStat.java:
##########
@@ -106,44 +106,55 @@ public String toString() {
public abstract double combine(List<Sample> samples, MetricConfig config, long now);
- /* Timeout any windows that have expired in the absence of any events */
- protected void purgeObsoleteSamples(MetricConfig config, long now) {
+ /**
+ * Purges any windows that started before the configured period.
+ * Returns the end of the latest purged window.
+ */
+ protected long purgeObsoleteSamples(MetricConfig config, long now) {
long expireAge = config.samples() * config.timeWindowMs();
+ long purgedUpToMs = 0;
for (Sample sample : samples) {
- if (now - sample.lastWindowMs >= expireAge)
+ if (now - sample.startTimeMs >= expireAge) {
+ purgedUpToMs = Math.max(purgedUpToMs, sample.endTimeMs(config));
sample.reset(now);
+ }
}
+ return purgedUpToMs;
}
protected static class Sample {
public double initialValue;
public long eventCount;
- public long lastWindowMs;
+ public long startTimeMs;
Review Comment:
It always played a role of start time. Just renaming, no functional changes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1608743345
##########
clients/src/main/java/org/apache/kafka/common/metrics/stats/SampledStat.java:
##########
@@ -50,10 +50,11 @@ public void record(MetricConfig config, double value, long timeMs) {
sample = advance(config, timeMs);
update(sample, config, value, timeMs);
sample.eventCount += 1;
+ sample.lastEventMs = timeMs;
}
private Sample advance(MetricConfig config, long timeMs) {
- this.current = (this.current + 1) % config.samples();
+ this.current = (this.current + 1) % (config.samples() + 1);
Review Comment:
It would be useful to add a comment to explain why we keep an additional sample than configured.
##########
clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java:
##########
@@ -72,29 +67,12 @@ public long windowSize(MetricConfig config, long now) {
stat.purgeObsoleteSamples(config, now);
/*
- * Here we check the total amount of time elapsed since the oldest non-obsolete window.
- * This give the total windowSize of the batch which is the time used for Rate computation.
- * However, there is an issue if we do not have sufficient data for e.g. if only 1 second has elapsed in a 30 second
- * window, the measured rate will be very high.
- * Hence we assume that the elapsed time is always N-1 complete windows plus whatever fraction of the final window is complete.
- *
- * Note that we could simply count the amount of time elapsed in the current window and add n-1 windows to get the total time,
- * but this approach does not account for sleeps. SampledStat only creates samples whenever record is called,
- * if no record is called for a period of time that time is not accounted for in windowSize and produces incorrect results.
+ * Purging process above guarantees to keep all events starting from
+ * earliest(monitoredWindow start, oldestSample start). Use the largest as windowSize.
*/
- long totalElapsedTimeMs = now - stat.oldest(now).lastWindowMs;
- // Check how many full windows of data we have currently retained
- int numFullWindows = (int) (totalElapsedTimeMs / config.timeWindowMs());
- int minFullWindows = config.samples() - 1;
-
- // If the available windows are less than the minimum required, add the difference to the totalElapsedTime
- if (numFullWindows < minFullWindows)
- totalElapsedTimeMs += (minFullWindows - numFullWindows) * config.timeWindowMs();
-
- // If window size is being calculated at the exact beginning of the window with no prior samples, the window size
- // will result in a value of 0. Calculation of rate over a window is size 0 is undefined, hence, we assume the
- // minimum window size to be at least 1ms.
- return Math.max(totalElapsedTimeMs, 1);
+ long monitoredWindow = config.timeWindowMs() * config.samples();
Review Comment:
Hmm, the changes the existing logic a bit. The existing logic makes sure that we include at least config.samples() - 1 full windows. The last one could be partial.
##########
clients/src/test/java/org/apache/kafka/common/metrics/stats/RateTest.java:
##########
@@ -64,4 +69,30 @@ public void testRateWithNoPriorAvailableSamples(int numSample, int sampleWindowS
double expectedRatePerSec = sampleValue / windowSize;
assertEquals(expectedRatePerSec, observedRate, EPS);
}
+
+ // Record an event every 100 ms on average, moving some 1 ms back or forth for fine-grained
+ // window control. The expected rate, hence, is 10-11 events/sec depending on the moment of
+ // measurement. Start assertions from the second window.
+ @Test
+ public void testRateIsConsistentAfterTheFirstWindow() {
+ MetricConfig config = new MetricConfig().timeWindow(1, SECONDS).samples(2);
+ List<Integer> steps = Arrays.asList(0, 99, 100, 100, 100, 100, 100, 100, 100, 100, 100);
+
+ // start the first window and record events at 0,99,199,...,999 ms
+ for (int stepMs : steps) {
+ time.sleep(stepMs);
+ rate.record(config, 1, time.milliseconds());
+ }
+
+ // making a gap of 100 ms between windows
+ time.sleep(101);
+
+ // start the second window and record events at 0,99,199,...,999 ms
+ for (int stepMs : steps) {
+ time.sleep(stepMs);
+ rate.record(config, 1, time.milliseconds());
+ double observedRate = rate.measure(config, time.milliseconds());
Review Comment:
Yes, it's probably useful to assert that taking a second measurement with no time change leads to the same value. This is more for preventing future incorrect changes and it's also low overhead.
##########
clients/src/test/java/org/apache/kafka/common/metrics/stats/RateTest.java:
##########
@@ -64,4 +69,31 @@ public void testRateWithNoPriorAvailableSamples(int numSample, int sampleWindowS
double expectedRatePerSec = sampleValue / windowSize;
assertEquals(expectedRatePerSec, observedRate, EPS);
}
+
+ // Record an event every 100 ms on average, moving some 1 ms back or forth for fine-grained
+ // window control. The expected rate, hence, is 10-11 events/sec depending on the moment of
+ // measurement. Start assertions from the second window. This test is to address past issue,
+ // when measurements in the end of the sample led to value spikes.
Review Comment:
How about changing "This test is to address past issue, when measurements in the end of the sample led to value spikes." to sth like "This test covers the case where a sample window partially overlaps with the monitored window." ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao merged PR #15889:
URL: https://github.com/apache/kafka/pull/15889
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "emitskevich-blp (via GitHub)" <gi...@apache.org>.
emitskevich-blp commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1593310007
##########
clients/src/main/java/org/apache/kafka/common/metrics/stats/SampledStat.java:
##########
@@ -106,44 +106,55 @@ public String toString() {
public abstract double combine(List<Sample> samples, MetricConfig config, long now);
- /* Timeout any windows that have expired in the absence of any events */
- protected void purgeObsoleteSamples(MetricConfig config, long now) {
+ /**
+ * Purges any windows that started before the configured period.
+ * Returns the end of the latest purged window.
+ */
+ protected long purgeObsoleteSamples(MetricConfig config, long now) {
long expireAge = config.samples() * config.timeWindowMs();
+ long purgedUpToMs = 0;
for (Sample sample : samples) {
- if (now - sample.lastWindowMs >= expireAge)
+ if (now - sample.startTimeMs >= expireAge) {
+ purgedUpToMs = Math.max(purgedUpToMs, sample.endTimeMs(config));
sample.reset(now);
+ }
}
+ return purgedUpToMs;
}
protected static class Sample {
public double initialValue;
public long eventCount;
- public long lastWindowMs;
+ public long startTimeMs;
Review Comment:
It always played a role of start time. Just renaming, not a functional change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "emitskevich-blp (via GitHub)" <gi...@apache.org>.
emitskevich-blp commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1610656953
##########
clients/src/main/java/org/apache/kafka/common/metrics/stats/SampledStat.java:
##########
@@ -50,10 +50,13 @@ public void record(MetricConfig config, double value, long timeMs) {
sample = advance(config, timeMs);
update(sample, config, value, timeMs);
sample.eventCount += 1;
+ sample.lastEventMs = timeMs;
}
private Sample advance(MetricConfig config, long timeMs) {
- this.current = (this.current + 1) % config.samples();
+ // need to keep one extra sample (see purgeObsoleteSamples() logic)
Review Comment:
I slightly changed the suggestion, please check if it's readable. Because technically this extra space is not to remember the last recording time, it's to keep the entire extra sample with all collected data.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1611395173
##########
clients/src/main/java/org/apache/kafka/common/metrics/stats/SampledStat.java:
##########
@@ -50,10 +50,13 @@ public void record(MetricConfig config, double value, long timeMs) {
sample = advance(config, timeMs);
update(sample, config, value, timeMs);
sample.eventCount += 1;
+ sample.lastEventMs = timeMs;
}
private Sample advance(MetricConfig config, long timeMs) {
- this.current = (this.current + 1) % config.samples();
+ // need to keep one extra sample (see purgeObsoleteSamples() logic)
Review Comment:
your comment is great! thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "emitskevich-blp (via GitHub)" <gi...@apache.org>.
emitskevich-blp commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1617776104
##########
clients/src/test/java/org/apache/kafka/common/metrics/MetricsTest.java:
##########
@@ -475,28 +464,13 @@ public void testPercentiles() {
Metric p50 = this.metrics.metrics().get(metrics.metricName("test.p50", "grp1"));
Metric p75 = this.metrics.metrics().get(metrics.metricName("test.p75", "grp1"));
- // record two windows worth of sequential values
- for (int i = 0; i < buckets; i++)
+ // record 1-100 sequential values
+ for (int i = 0; i < buckets; i++) {
sensor.record(i);
-
- assertEquals(25, (Double) p25.metricValue(), 1.0);
- assertEquals(50, (Double) p50.metricValue(), 1.0);
- assertEquals(75, (Double) p75.metricValue(), 1.0);
-
- for (int i = 0; i < buckets; i++)
Review Comment:
Yes, it breaks the test. It relies on the knowledge of internal samples/windows implementation and effectively tests windows expiration (`SampledStat`, `SampledStatTest`) instead of testing `Percentiles`. Another option was to adapt the test for recent changes in windows expiration, but it seems better to avoid using such knowledge about internals at all. What do you think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "emitskevich-blp (via GitHub)" <gi...@apache.org>.
emitskevich-blp commented on PR #15889:
URL: https://github.com/apache/kafka/pull/15889#issuecomment-2138093800
> Are the failed tests related to this PR, especially the following one?
Not related, they are all flaky tests
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "emitskevich-blp (via GitHub)" <gi...@apache.org>.
emitskevich-blp commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1593310007
##########
clients/src/main/java/org/apache/kafka/common/metrics/stats/SampledStat.java:
##########
@@ -106,44 +106,55 @@ public String toString() {
public abstract double combine(List<Sample> samples, MetricConfig config, long now);
- /* Timeout any windows that have expired in the absence of any events */
- protected void purgeObsoleteSamples(MetricConfig config, long now) {
+ /**
+ * Purges any windows that started before the configured period.
+ * Returns the end of the latest purged window.
+ */
+ protected long purgeObsoleteSamples(MetricConfig config, long now) {
long expireAge = config.samples() * config.timeWindowMs();
+ long purgedUpToMs = 0;
for (Sample sample : samples) {
- if (now - sample.lastWindowMs >= expireAge)
+ if (now - sample.startTimeMs >= expireAge) {
+ purgedUpToMs = Math.max(purgedUpToMs, sample.endTimeMs(config));
sample.reset(now);
+ }
}
+ return purgedUpToMs;
}
protected static class Sample {
public double initialValue;
public long eventCount;
- public long lastWindowMs;
+ public long startTimeMs;
Review Comment:
It always played a role of start time. Not a functional change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "emitskevich-blp (via GitHub)" <gi...@apache.org>.
emitskevich-blp commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1593308659
##########
clients/src/test/java/org/apache/kafka/common/metrics/stats/RateTest.java:
##########
@@ -64,4 +67,31 @@ public void testRateWithNoPriorAvailableSamples(int numSample, int sampleWindowS
double expectedRatePerSec = sampleValue / windowSize;
assertEquals(expectedRatePerSec, observedRate, EPS);
}
+
+
+ // Record an event every 100 ms on average, moving some 1 ms back or forth for fine-grained
+ // window control. The expected rate, hence, is 10-11 events/sec depending on the moment of
+ // measurement. Start assertions from the second window.
+ @Test
+ public void testRateIsConsistentAfterTheFirstWindow() {
+ MetricConfig config = new MetricConfig().timeWindow(1, SECONDS).samples(2);
+ List<Integer> steps = Arrays.asList(0, 99, 100, 100, 100, 100, 100, 100, 100, 100, 100);
+
+ // start the first window and record events at 0,99,199,...,999 ms
+ for (int stepMs : steps) {
+ time.sleep(stepMs);
+ rate.record(config, 1, time.milliseconds());
+ }
+
+ // making a gap of 100 ms between windows
+ time.sleep(101);
+
+ // start the second window and record events at 0,99,199,...,999 ms
+ for (int stepMs : steps) {
+ time.sleep(stepMs);
+ rate.record(config, 1, time.milliseconds());
+ double observedRate = rate.measure(config, time.milliseconds());
+ assertTrue(observedRate >= 10 && observedRate <= 11);
Review Comment:
The last measurement here was affected by `Rate` bug and showed ~5 instead of ~10. Now it's consistent.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1602125187
##########
clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java:
##########
@@ -63,33 +63,38 @@ public void record(MetricConfig config, double value, long timeMs) {
@Override
public double measure(MetricConfig config, long now) {
+ long windowSizeMs = windowSize(config, now);
double value = stat.measure(config, now);
- return value / convert(windowSize(config, now), unit);
+ return value / convert(windowSizeMs, unit);
}
public long windowSize(MetricConfig config, long now) {
// purge old samples before we compute the window size
- stat.purgeObsoleteSamples(config, now);
+ long purgedUpToMs = stat.purgeObsoleteSamples(config, now);
/*
* Here we check the total amount of time elapsed since the oldest non-obsolete window.
* This give the total windowSize of the batch which is the time used for Rate computation.
- * However, there is an issue if we do not have sufficient data for e.g. if only 1 second has elapsed in a 30 second
+ * However, there is an issue if we do not have sufficient data for e.g. if only 1 second has elapsed in a 30-second
* window, the measured rate will be very high.
- * Hence we assume that the elapsed time is always N-1 complete windows plus whatever fraction of the final window is complete.
+ * Hence, we assume that the elapsed time is always N-1 complete windows plus whatever fraction of the final window is complete.
*
* Note that we could simply count the amount of time elapsed in the current window and add n-1 windows to get the total time,
* but this approach does not account for sleeps. SampledStat only creates samples whenever record is called,
* if no record is called for a period of time that time is not accounted for in windowSize and produces incorrect results.
*/
- long totalElapsedTimeMs = now - stat.oldest(now).lastWindowMs;
+ long totalElapsedTimeMs = now - stat.oldest(now).startTimeMs;
// Check how many full windows of data we have currently retained
int numFullWindows = (int) (totalElapsedTimeMs / config.timeWindowMs());
int minFullWindows = config.samples() - 1;
// If the available windows are less than the minimum required, add the difference to the totalElapsedTime
- if (numFullWindows < minFullWindows)
+ if (numFullWindows < minFullWindows) {
totalElapsedTimeMs += (minFullWindows - numFullWindows) * config.timeWindowMs();
+ }
+
+ // if some part of considered interval was just purged with its data, exclude it
+ totalElapsedTimeMs = Math.min(totalElapsedTimeMs, now - purgedUpToMs);
Review Comment:
This approach works for the newly added test case. However, I have two concerns.
1. It's not very repeatable. For example, if `measure()` is called a second time after a very short window, no samples will be purged and we won't be compensating for the purgedUpToMs. Then, we will fall back to use the larger window, which will result in a much lower measured value.
2. The logic is bit hard to follow now.
Here is an alternative. Note that in the new test case, the oldest window actually partially overlaps with the total sample window since the last recording falls in it. So, instead of purging it, we could keep the overlapping sample and include it in both the measure and windowSize calculation. This could make the sample window a bit larger than what's configured, but it probably provides a more accurate/consistent measurement. We need to extend `Sample` to remember the last recording time. This probably addresses both of the above concerns. What do you think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1605350575
##########
clients/src/main/java/org/apache/kafka/common/metrics/stats/SampledStat.java:
##########
@@ -110,40 +111,43 @@ public String toString() {
protected void purgeObsoleteSamples(MetricConfig config, long now) {
long expireAge = config.samples() * config.timeWindowMs();
for (Sample sample : samples) {
- if (now - sample.lastWindowMs >= expireAge)
+ if (now - sample.lastEventMs >= expireAge)
sample.reset(now);
}
}
protected static class Sample {
public double initialValue;
public long eventCount;
- public long lastWindowMs;
+ public long startTimeMs;
+ public long lastEventMs;
public double value;
public Sample(double initialValue, long now) {
this.initialValue = initialValue;
this.eventCount = 0;
- this.lastWindowMs = now;
+ this.startTimeMs = now;
+ this.lastEventMs = now;
this.value = initialValue;
}
public void reset(long now) {
this.eventCount = 0;
- this.lastWindowMs = now;
+ this.startTimeMs = now;
+ this.lastEventMs = now;
this.value = initialValue;
}
public boolean isComplete(long timeMs, MetricConfig config) {
- return timeMs - lastWindowMs >= config.timeWindowMs() || eventCount >= config.eventWindow();
+ return timeMs - startTimeMs >= config.timeWindowMs() || eventCount >= config.eventWindow();
}
@Override
public String toString() {
return "Sample(" +
"value=" + value +
", eventCount=" + eventCount +
- ", lastWindowMs=" + lastWindowMs +
+ ", startTimeMs=" + startTimeMs +
Review Comment:
Should we add lastEventMs?
##########
clients/src/main/java/org/apache/kafka/common/metrics/stats/SampledStat.java:
##########
@@ -110,40 +111,43 @@ public String toString() {
protected void purgeObsoleteSamples(MetricConfig config, long now) {
long expireAge = config.samples() * config.timeWindowMs();
for (Sample sample : samples) {
- if (now - sample.lastWindowMs >= expireAge)
+ if (now - sample.lastEventMs >= expireAge)
Review Comment:
Could we add a comment that we don't purge overlapping samples?
##########
clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java:
##########
@@ -82,7 +82,7 @@ public long windowSize(MetricConfig config, long now) {
* but this approach does not account for sleeps. SampledStat only creates samples whenever record is called,
* if no record is called for a period of time that time is not accounted for in windowSize and produces incorrect results.
*/
- long totalElapsedTimeMs = now - stat.oldest(now).lastWindowMs;
+ long totalElapsedTimeMs = now - stat.oldest(now).startTimeMs;
Review Comment:
Could we add some comments to note that the total elapse time could be larger than the window size since we keep the overlapping samples?
##########
clients/src/test/java/org/apache/kafka/common/metrics/stats/RateTest.java:
##########
@@ -64,4 +69,30 @@ public void testRateWithNoPriorAvailableSamples(int numSample, int sampleWindowS
double expectedRatePerSec = sampleValue / windowSize;
assertEquals(expectedRatePerSec, observedRate, EPS);
}
+
+ // Record an event every 100 ms on average, moving some 1 ms back or forth for fine-grained
+ // window control. The expected rate, hence, is 10-11 events/sec depending on the moment of
+ // measurement. Start assertions from the second window.
+ @Test
+ public void testRateIsConsistentAfterTheFirstWindow() {
+ MetricConfig config = new MetricConfig().timeWindow(1, SECONDS).samples(2);
+ List<Integer> steps = Arrays.asList(0, 99, 100, 100, 100, 100, 100, 100, 100, 100, 100);
+
+ // start the first window and record events at 0,99,199,...,999 ms
+ for (int stepMs : steps) {
+ time.sleep(stepMs);
+ rate.record(config, 1, time.milliseconds());
+ }
+
+ // making a gap of 100 ms between windows
+ time.sleep(101);
+
+ // start the second window and record events at 0,99,199,...,999 ms
+ for (int stepMs : steps) {
+ time.sleep(stepMs);
+ rate.record(config, 1, time.milliseconds());
+ double observedRate = rate.measure(config, time.milliseconds());
Review Comment:
Should we do a second measurement and check that the result doesn't change?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "emitskevich-blp (via GitHub)" <gi...@apache.org>.
emitskevich-blp commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1610219832
##########
clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java:
##########
@@ -82,7 +82,7 @@ public long windowSize(MetricConfig config, long now) {
* but this approach does not account for sleeps. SampledStat only creates samples whenever record is called,
* if no record is called for a period of time that time is not accounted for in windowSize and produces incorrect results.
*/
- long totalElapsedTimeMs = now - stat.oldest(now).lastWindowMs;
+ long totalElapsedTimeMs = now - stat.oldest(now).startTimeMs;
Review Comment:
Applied
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "emitskevich-blp (via GitHub)" <gi...@apache.org>.
emitskevich-blp commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1610394442
##########
clients/src/main/java/org/apache/kafka/common/metrics/stats/SampledStat.java:
##########
@@ -50,10 +50,11 @@ public void record(MetricConfig config, double value, long timeMs) {
sample = advance(config, timeMs);
update(sample, config, value, timeMs);
sample.eventCount += 1;
+ sample.lastEventMs = timeMs;
}
private Sample advance(MetricConfig config, long timeMs) {
- this.current = (this.current + 1) % config.samples();
+ this.current = (this.current + 1) % (config.samples() + 1);
Review Comment:
Yes, thank you, already added. Seems to be a line from intermediate state as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1611984550
##########
clients/src/main/java/org/apache/kafka/common/metrics/stats/SampledStat.java:
##########
@@ -106,44 +109,51 @@ public String toString() {
public abstract double combine(List<Sample> samples, MetricConfig config, long now);
- /* Timeout any windows that have expired in the absence of any events */
+ // purge any samples that lack observed events within the monitored window
protected void purgeObsoleteSamples(MetricConfig config, long now) {
long expireAge = config.samples() * config.timeWindowMs();
for (Sample sample : samples) {
- if (now - sample.lastWindowMs >= expireAge)
+ // samples overlapping the monitored window are kept,
+ // even if they started before it
+ if (now - sample.lastEventMs >= expireAge) {
sample.reset(now);
+ }
}
}
protected static class Sample {
public double initialValue;
public long eventCount;
- public long lastWindowMs;
+ public long startTimeMs;
+ public long lastEventMs;
public double value;
public Sample(double initialValue, long now) {
this.initialValue = initialValue;
this.eventCount = 0;
- this.lastWindowMs = now;
+ this.startTimeMs = now;
+ this.lastEventMs = now;
this.value = initialValue;
}
public void reset(long now) {
this.eventCount = 0;
- this.lastWindowMs = now;
+ this.startTimeMs = now;
+ this.lastEventMs = now;
this.value = initialValue;
}
public boolean isComplete(long timeMs, MetricConfig config) {
- return timeMs - lastWindowMs >= config.timeWindowMs() || eventCount >= config.eventWindow();
+ return timeMs - startTimeMs >= config.timeWindowMs() || eventCount >= config.eventWindow();
Review Comment:
Yes, when the oldest window overlaps with the monitored window, the old approach ignores it and the current approach keeps it. Either one could be considered correct.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "emitskevich-blp (via GitHub)" <gi...@apache.org>.
emitskevich-blp commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1610221659
##########
clients/src/main/java/org/apache/kafka/common/metrics/stats/SampledStat.java:
##########
@@ -110,40 +111,43 @@ public String toString() {
protected void purgeObsoleteSamples(MetricConfig config, long now) {
long expireAge = config.samples() * config.timeWindowMs();
for (Sample sample : samples) {
- if (now - sample.lastWindowMs >= expireAge)
+ if (now - sample.lastEventMs >= expireAge)
Review Comment:
Applied
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1617667802
##########
clients/src/test/java/org/apache/kafka/common/metrics/stats/FrequenciesTest.java:
##########
@@ -102,58 +110,69 @@ public void testBooleanFrequencies() {
}
@Test
- public void testUseWithMetrics() {
- MetricName name1 = name("1");
- MetricName name2 = name("2");
- MetricName name3 = name("3");
- MetricName name4 = name("4");
- Frequencies frequencies = new Frequencies(4, 1.0, 4.0,
- new Frequency(name1, 1.0),
- new Frequency(name2, 2.0),
- new Frequency(name3, 3.0),
- new Frequency(name4, 4.0));
+ public void testWithMetricsStrategy1() {
+ Frequencies frequencies = new Frequencies(4, 1.0, 4.0, freq("1", 1.0),
+ freq("2", 2.0), freq("3", 3.0), freq("4", 4.0));
Sensor sensor = metrics.sensor("test", config);
sensor.add(frequencies);
- Metric metric1 = this.metrics.metrics().get(name1);
- Metric metric2 = this.metrics.metrics().get(name2);
- Metric metric3 = this.metrics.metrics().get(name3);
- Metric metric4 = this.metrics.metrics().get(name4);
- // Record 2 windows worth of values
- for (int i = 0; i != 100; ++i) {
+ // Record 100 events uniformly between all buckets
+ for (int i = 0; i < 100; ++i) {
frequencies.record(config, i % 4 + 1, time.milliseconds());
}
- assertEquals(0.25, (Double) metric1.metricValue(), DELTA);
- assertEquals(0.25, (Double) metric2.metricValue(), DELTA);
- assertEquals(0.25, (Double) metric3.metricValue(), DELTA);
- assertEquals(0.25, (Double) metric4.metricValue(), DELTA);
+ assertEquals(0.25, metricValueFor("1"), DELTA);
+ assertEquals(0.25, metricValueFor("2"), DELTA);
+ assertEquals(0.25, metricValueFor("3"), DELTA);
+ assertEquals(0.25, metricValueFor("4"), DELTA);
+ }
+
+ @Test
+ public void testWithMetricsStrategy2() {
+ Frequencies frequencies = new Frequencies(4, 1.0, 4.0, freq("1", 1.0),
+ freq("2", 2.0), freq("3", 3.0), freq("4", 4.0));
+ Sensor sensor = metrics.sensor("test", config);
+ sensor.add(frequencies);
+
+ // Record 100 events half-half between 1st and 2nd buckets
+ for (int i = 0; i < 100; ++i) {
+ frequencies.record(config, i % 2 + 1, time.milliseconds());
+ }
+ assertEquals(0.50, metricValueFor("1"), DELTA);
+ assertEquals(0.50, metricValueFor("2"), DELTA);
+ assertEquals(0.00, metricValueFor("3"), DELTA);
+ assertEquals(0.00, metricValueFor("4"), DELTA);
+ }
- // Record 2 windows worth of values
- for (int i = 0; i != 100; ++i) {
+ @Test
+ public void testWithMetricsStrategy3() {
+ Frequencies frequencies = new Frequencies(4, 1.0, 4.0, freq("1", 1.0),
+ freq("2", 2.0), freq("3", 3.0), freq("4", 4.0));
+ Sensor sensor = metrics.sensor("test", config);
+ sensor.add(frequencies);
+
+ // Record 50 events half-half between 1st and 2nd buckets
+ for (int i = 0; i < 50; ++i) {
frequencies.record(config, i % 2 + 1, time.milliseconds());
}
- assertEquals(0.50, (Double) metric1.metricValue(), DELTA);
- assertEquals(0.50, (Double) metric2.metricValue(), DELTA);
- assertEquals(0.00, (Double) metric3.metricValue(), DELTA);
- assertEquals(0.00, (Double) metric4.metricValue(), DELTA);
-
- // Record 1 window worth of values to overlap with the last window
- // that is half 1.0 and half 2.0
- for (int i = 0; i != 50; ++i) {
+ // Record 50 events to 4th bucket
+ for (int i = 0; i < 50; ++i) {
frequencies.record(config, 4.0, time.milliseconds());
}
- assertEquals(0.25, (Double) metric1.metricValue(), DELTA);
- assertEquals(0.25, (Double) metric2.metricValue(), DELTA);
- assertEquals(0.00, (Double) metric3.metricValue(), DELTA);
- assertEquals(0.50, (Double) metric4.metricValue(), DELTA);
+ assertEquals(0.25, metricValueFor("1"), DELTA);
+ assertEquals(0.25, metricValueFor("2"), DELTA);
+ assertEquals(0.00, metricValueFor("3"), DELTA);
+ assertEquals(0.50, metricValueFor("4"), DELTA);
}
- protected MetricName name(String metricName) {
+ private MetricName name(String metricName) {
Review Comment:
name => toMetricName ?
##########
clients/src/test/java/org/apache/kafka/common/metrics/stats/FrequenciesTest.java:
##########
@@ -73,14 +72,14 @@ public void testMoreFrequencyParametersThanBuckets() {
}
@Test
- public void testBooleanFrequencies() {
+ public void testBooleanFrequenciesStrategy1() {
Review Comment:
Just curious. It seems that we just break the existing tests into smaller ones. Did we change any logic for the tests to pass?
##########
clients/src/test/java/org/apache/kafka/common/metrics/MetricsTest.java:
##########
@@ -475,28 +464,13 @@ public void testPercentiles() {
Metric p50 = this.metrics.metrics().get(metrics.metricName("test.p50", "grp1"));
Metric p75 = this.metrics.metrics().get(metrics.metricName("test.p75", "grp1"));
- // record two windows worth of sequential values
- for (int i = 0; i < buckets; i++)
+ // record 1-100 sequential values
+ for (int i = 0; i < buckets; i++) {
Review Comment:
In some other places in this class, we don't use {} for single line statements. Should we be consistent?
##########
clients/src/test/java/org/apache/kafka/common/metrics/MetricsTest.java:
##########
@@ -475,28 +464,13 @@ public void testPercentiles() {
Metric p50 = this.metrics.metrics().get(metrics.metricName("test.p50", "grp1"));
Metric p75 = this.metrics.metrics().get(metrics.metricName("test.p75", "grp1"));
- // record two windows worth of sequential values
- for (int i = 0; i < buckets; i++)
+ // record 1-100 sequential values
+ for (int i = 0; i < buckets; i++) {
sensor.record(i);
-
- assertEquals(25, (Double) p25.metricValue(), 1.0);
- assertEquals(50, (Double) p50.metricValue(), 1.0);
- assertEquals(75, (Double) p75.metricValue(), 1.0);
-
- for (int i = 0; i < buckets; i++)
Review Comment:
Why do we remove this code? Does it break the test?
##########
clients/src/test/java/org/apache/kafka/common/metrics/stats/FrequenciesTest.java:
##########
@@ -89,8 +88,17 @@ public void testBooleanFrequencies() {
}
assertEquals(0.25, falseMetric.stat().measure(config, time.milliseconds()), DELTA);
assertEquals(0.75, trueMetric.stat().measure(config, time.milliseconds()), DELTA);
+ }
- // Record 2 more windows worth of values
+ @Test
+ public void testBooleanFrequenciesStrategy2() {
Review Comment:
Does `testBooleanFrequenciesStrategy2` cover anything different from `testBooleanFrequenciesStrategy1`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "emitskevich-blp (via GitHub)" <gi...@apache.org>.
emitskevich-blp commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1617767952
##########
clients/src/test/java/org/apache/kafka/common/metrics/stats/FrequenciesTest.java:
##########
@@ -73,14 +72,14 @@ public void testMoreFrequencyParametersThanBuckets() {
}
@Test
- public void testBooleanFrequencies() {
+ public void testBooleanFrequenciesStrategy1() {
Review Comment:
There were 3 scenarios inside single test, which effectively should have been reset the state of metric in between. But reset was based on the knowledge of internal implementation of samples/windows, which is not a part of class API and which was slightly changed. Now it is separated to different test methods to make reset explicit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "emitskevich-blp (via GitHub)" <gi...@apache.org>.
emitskevich-blp commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1617784166
##########
clients/src/test/java/org/apache/kafka/common/metrics/stats/FrequenciesTest.java:
##########
@@ -102,58 +110,69 @@ public void testBooleanFrequencies() {
}
@Test
- public void testUseWithMetrics() {
- MetricName name1 = name("1");
- MetricName name2 = name("2");
- MetricName name3 = name("3");
- MetricName name4 = name("4");
- Frequencies frequencies = new Frequencies(4, 1.0, 4.0,
- new Frequency(name1, 1.0),
- new Frequency(name2, 2.0),
- new Frequency(name3, 3.0),
- new Frequency(name4, 4.0));
+ public void testWithMetricsStrategy1() {
+ Frequencies frequencies = new Frequencies(4, 1.0, 4.0, freq("1", 1.0),
+ freq("2", 2.0), freq("3", 3.0), freq("4", 4.0));
Sensor sensor = metrics.sensor("test", config);
sensor.add(frequencies);
- Metric metric1 = this.metrics.metrics().get(name1);
- Metric metric2 = this.metrics.metrics().get(name2);
- Metric metric3 = this.metrics.metrics().get(name3);
- Metric metric4 = this.metrics.metrics().get(name4);
- // Record 2 windows worth of values
- for (int i = 0; i != 100; ++i) {
+ // Record 100 events uniformly between all buckets
+ for (int i = 0; i < 100; ++i) {
frequencies.record(config, i % 4 + 1, time.milliseconds());
}
- assertEquals(0.25, (Double) metric1.metricValue(), DELTA);
- assertEquals(0.25, (Double) metric2.metricValue(), DELTA);
- assertEquals(0.25, (Double) metric3.metricValue(), DELTA);
- assertEquals(0.25, (Double) metric4.metricValue(), DELTA);
+ assertEquals(0.25, metricValueFor("1"), DELTA);
+ assertEquals(0.25, metricValueFor("2"), DELTA);
+ assertEquals(0.25, metricValueFor("3"), DELTA);
+ assertEquals(0.25, metricValueFor("4"), DELTA);
+ }
+
+ @Test
+ public void testWithMetricsStrategy2() {
+ Frequencies frequencies = new Frequencies(4, 1.0, 4.0, freq("1", 1.0),
+ freq("2", 2.0), freq("3", 3.0), freq("4", 4.0));
+ Sensor sensor = metrics.sensor("test", config);
+ sensor.add(frequencies);
+
+ // Record 100 events half-half between 1st and 2nd buckets
+ for (int i = 0; i < 100; ++i) {
+ frequencies.record(config, i % 2 + 1, time.milliseconds());
+ }
+ assertEquals(0.50, metricValueFor("1"), DELTA);
+ assertEquals(0.50, metricValueFor("2"), DELTA);
+ assertEquals(0.00, metricValueFor("3"), DELTA);
+ assertEquals(0.00, metricValueFor("4"), DELTA);
+ }
- // Record 2 windows worth of values
- for (int i = 0; i != 100; ++i) {
+ @Test
+ public void testWithMetricsStrategy3() {
+ Frequencies frequencies = new Frequencies(4, 1.0, 4.0, freq("1", 1.0),
+ freq("2", 2.0), freq("3", 3.0), freq("4", 4.0));
+ Sensor sensor = metrics.sensor("test", config);
+ sensor.add(frequencies);
+
+ // Record 50 events half-half between 1st and 2nd buckets
+ for (int i = 0; i < 50; ++i) {
frequencies.record(config, i % 2 + 1, time.milliseconds());
}
- assertEquals(0.50, (Double) metric1.metricValue(), DELTA);
- assertEquals(0.50, (Double) metric2.metricValue(), DELTA);
- assertEquals(0.00, (Double) metric3.metricValue(), DELTA);
- assertEquals(0.00, (Double) metric4.metricValue(), DELTA);
-
- // Record 1 window worth of values to overlap with the last window
- // that is half 1.0 and half 2.0
- for (int i = 0; i != 50; ++i) {
+ // Record 50 events to 4th bucket
+ for (int i = 0; i < 50; ++i) {
frequencies.record(config, 4.0, time.milliseconds());
}
- assertEquals(0.25, (Double) metric1.metricValue(), DELTA);
- assertEquals(0.25, (Double) metric2.metricValue(), DELTA);
- assertEquals(0.00, (Double) metric3.metricValue(), DELTA);
- assertEquals(0.50, (Double) metric4.metricValue(), DELTA);
+ assertEquals(0.25, metricValueFor("1"), DELTA);
+ assertEquals(0.25, metricValueFor("2"), DELTA);
+ assertEquals(0.00, metricValueFor("3"), DELTA);
+ assertEquals(0.50, metricValueFor("4"), DELTA);
}
- protected MetricName name(String metricName) {
+ private MetricName name(String metricName) {
Review Comment:
Tried to make now `metricName`, `freq` and `metricValue` consistent in naming, please check if it's ok
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "emitskevich-blp (via GitHub)" <gi...@apache.org>.
emitskevich-blp commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1602477638
##########
clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java:
##########
@@ -63,33 +63,38 @@ public void record(MetricConfig config, double value, long timeMs) {
@Override
public double measure(MetricConfig config, long now) {
+ long windowSizeMs = windowSize(config, now);
double value = stat.measure(config, now);
- return value / convert(windowSize(config, now), unit);
+ return value / convert(windowSizeMs, unit);
}
public long windowSize(MetricConfig config, long now) {
// purge old samples before we compute the window size
- stat.purgeObsoleteSamples(config, now);
+ long purgedUpToMs = stat.purgeObsoleteSamples(config, now);
/*
* Here we check the total amount of time elapsed since the oldest non-obsolete window.
* This give the total windowSize of the batch which is the time used for Rate computation.
- * However, there is an issue if we do not have sufficient data for e.g. if only 1 second has elapsed in a 30 second
+ * However, there is an issue if we do not have sufficient data for e.g. if only 1 second has elapsed in a 30-second
* window, the measured rate will be very high.
- * Hence we assume that the elapsed time is always N-1 complete windows plus whatever fraction of the final window is complete.
+ * Hence, we assume that the elapsed time is always N-1 complete windows plus whatever fraction of the final window is complete.
*
* Note that we could simply count the amount of time elapsed in the current window and add n-1 windows to get the total time,
* but this approach does not account for sleeps. SampledStat only creates samples whenever record is called,
* if no record is called for a period of time that time is not accounted for in windowSize and produces incorrect results.
*/
- long totalElapsedTimeMs = now - stat.oldest(now).lastWindowMs;
+ long totalElapsedTimeMs = now - stat.oldest(now).startTimeMs;
// Check how many full windows of data we have currently retained
int numFullWindows = (int) (totalElapsedTimeMs / config.timeWindowMs());
int minFullWindows = config.samples() - 1;
// If the available windows are less than the minimum required, add the difference to the totalElapsedTime
- if (numFullWindows < minFullWindows)
+ if (numFullWindows < minFullWindows) {
totalElapsedTimeMs += (minFullWindows - numFullWindows) * config.timeWindowMs();
+ }
+
+ // if some part of considered interval was just purged with its data, exclude it
+ totalElapsedTimeMs = Math.min(totalElapsedTimeMs, now - purgedUpToMs);
Review Comment:
@junrao I updated the PR according to your suggestion, works fine
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "emitskevich-blp (via GitHub)" <gi...@apache.org>.
emitskevich-blp commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1610220213
##########
clients/src/main/java/org/apache/kafka/common/metrics/stats/SampledStat.java:
##########
@@ -110,40 +111,43 @@ public String toString() {
protected void purgeObsoleteSamples(MetricConfig config, long now) {
long expireAge = config.samples() * config.timeWindowMs();
for (Sample sample : samples) {
- if (now - sample.lastWindowMs >= expireAge)
+ if (now - sample.lastEventMs >= expireAge)
sample.reset(now);
}
}
protected static class Sample {
public double initialValue;
public long eventCount;
- public long lastWindowMs;
+ public long startTimeMs;
+ public long lastEventMs;
public double value;
public Sample(double initialValue, long now) {
this.initialValue = initialValue;
this.eventCount = 0;
- this.lastWindowMs = now;
+ this.startTimeMs = now;
+ this.lastEventMs = now;
this.value = initialValue;
}
public void reset(long now) {
this.eventCount = 0;
- this.lastWindowMs = now;
+ this.startTimeMs = now;
+ this.lastEventMs = now;
this.value = initialValue;
}
public boolean isComplete(long timeMs, MetricConfig config) {
- return timeMs - lastWindowMs >= config.timeWindowMs() || eventCount >= config.eventWindow();
+ return timeMs - startTimeMs >= config.timeWindowMs() || eventCount >= config.eventWindow();
}
@Override
public String toString() {
return "Sample(" +
"value=" + value +
", eventCount=" + eventCount +
- ", lastWindowMs=" + lastWindowMs +
+ ", startTimeMs=" + startTimeMs +
Review Comment:
Applied
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "emitskevich-blp (via GitHub)" <gi...@apache.org>.
emitskevich-blp commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1610393654
##########
clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java:
##########
@@ -72,29 +67,12 @@ public long windowSize(MetricConfig config, long now) {
stat.purgeObsoleteSamples(config, now);
/*
- * Here we check the total amount of time elapsed since the oldest non-obsolete window.
- * This give the total windowSize of the batch which is the time used for Rate computation.
- * However, there is an issue if we do not have sufficient data for e.g. if only 1 second has elapsed in a 30 second
- * window, the measured rate will be very high.
- * Hence we assume that the elapsed time is always N-1 complete windows plus whatever fraction of the final window is complete.
- *
- * Note that we could simply count the amount of time elapsed in the current window and add n-1 windows to get the total time,
- * but this approach does not account for sleeps. SampledStat only creates samples whenever record is called,
- * if no record is called for a period of time that time is not accounted for in windowSize and produces incorrect results.
+ * Purging process above guarantees to keep all events starting from
+ * earliest(monitoredWindow start, oldestSample start). Use the largest as windowSize.
*/
- long totalElapsedTimeMs = now - stat.oldest(now).lastWindowMs;
- // Check how many full windows of data we have currently retained
- int numFullWindows = (int) (totalElapsedTimeMs / config.timeWindowMs());
- int minFullWindows = config.samples() - 1;
-
- // If the available windows are less than the minimum required, add the difference to the totalElapsedTime
- if (numFullWindows < minFullWindows)
- totalElapsedTimeMs += (minFullWindows - numFullWindows) * config.timeWindowMs();
-
- // If window size is being calculated at the exact beginning of the window with no prior samples, the window size
- // will result in a value of 0. Calculation of rate over a window is size 0 is undefined, hence, we assume the
- // minimum window size to be at least 1ms.
- return Math.max(totalElapsedTimeMs, 1);
+ long monitoredWindow = config.timeWindowMs() * config.samples();
Review Comment:
@junrao looks like it is outdated state... I made several commits today and this line is no more in HEAD. Is there any chance you commented on intermediate state? Sorry for a little mess around commits here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "emitskevich-blp (via GitHub)" <gi...@apache.org>.
emitskevich-blp commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1610406308
##########
clients/src/test/java/org/apache/kafka/common/metrics/stats/RateTest.java:
##########
@@ -64,4 +69,31 @@ public void testRateWithNoPriorAvailableSamples(int numSample, int sampleWindowS
double expectedRatePerSec = sampleValue / windowSize;
assertEquals(expectedRatePerSec, observedRate, EPS);
}
+
+ // Record an event every 100 ms on average, moving some 1 ms back or forth for fine-grained
+ // window control. The expected rate, hence, is 10-11 events/sec depending on the moment of
+ // measurement. Start assertions from the second window. This test is to address past issue,
+ // when measurements in the end of the sample led to value spikes.
Review Comment:
Applied
##########
clients/src/test/java/org/apache/kafka/common/metrics/stats/RateTest.java:
##########
@@ -64,4 +69,30 @@ public void testRateWithNoPriorAvailableSamples(int numSample, int sampleWindowS
double expectedRatePerSec = sampleValue / windowSize;
assertEquals(expectedRatePerSec, observedRate, EPS);
}
+
+ // Record an event every 100 ms on average, moving some 1 ms back or forth for fine-grained
+ // window control. The expected rate, hence, is 10-11 events/sec depending on the moment of
+ // measurement. Start assertions from the second window.
+ @Test
+ public void testRateIsConsistentAfterTheFirstWindow() {
+ MetricConfig config = new MetricConfig().timeWindow(1, SECONDS).samples(2);
+ List<Integer> steps = Arrays.asList(0, 99, 100, 100, 100, 100, 100, 100, 100, 100, 100);
+
+ // start the first window and record events at 0,99,199,...,999 ms
+ for (int stepMs : steps) {
+ time.sleep(stepMs);
+ rate.record(config, 1, time.milliseconds());
+ }
+
+ // making a gap of 100 ms between windows
+ time.sleep(101);
+
+ // start the second window and record events at 0,99,199,...,999 ms
+ for (int stepMs : steps) {
+ time.sleep(stepMs);
+ rate.record(config, 1, time.milliseconds());
+ double observedRate = rate.measure(config, time.milliseconds());
Review Comment:
Applied
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1610336329
##########
clients/src/main/java/org/apache/kafka/common/metrics/stats/SampledStat.java:
##########
@@ -40,7 +40,7 @@ public abstract class SampledStat implements MeasurableStat {
public SampledStat(double initialValue) {
this.initialValue = initialValue;
- this.samples = new ArrayList<>(2);
+ this.samples = new ArrayList<>(3);
Review Comment:
It would be nice to add comments for this magic number :)
##########
clients/src/main/java/org/apache/kafka/common/metrics/stats/SampledStat.java:
##########
@@ -106,44 +109,51 @@ public String toString() {
public abstract double combine(List<Sample> samples, MetricConfig config, long now);
- /* Timeout any windows that have expired in the absence of any events */
+ // purge any samples that lack observed events within the monitored window
protected void purgeObsoleteSamples(MetricConfig config, long now) {
long expireAge = config.samples() * config.timeWindowMs();
for (Sample sample : samples) {
- if (now - sample.lastWindowMs >= expireAge)
+ // samples overlapping the monitored window are kept,
+ // even if they started before it
+ if (now - sample.lastEventMs >= expireAge) {
sample.reset(now);
+ }
}
}
protected static class Sample {
public double initialValue;
public long eventCount;
- public long lastWindowMs;
+ public long startTimeMs;
+ public long lastEventMs;
public double value;
public Sample(double initialValue, long now) {
this.initialValue = initialValue;
this.eventCount = 0;
- this.lastWindowMs = now;
+ this.startTimeMs = now;
+ this.lastEventMs = now;
this.value = initialValue;
}
public void reset(long now) {
this.eventCount = 0;
- this.lastWindowMs = now;
+ this.startTimeMs = now;
+ this.lastEventMs = now;
this.value = initialValue;
}
public boolean isComplete(long timeMs, MetricConfig config) {
- return timeMs - lastWindowMs >= config.timeWindowMs() || eventCount >= config.eventWindow();
+ return timeMs - startTimeMs >= config.timeWindowMs() || eventCount >= config.eventWindow();
Review Comment:
This is unrelated to this PR, but it seems `eventCount` could be another issue if we set a non-maximum value. For example, all samples are within a single `timeWindowMs` due to a bunch of records. in this case, `totalElapsedTimeMs` will be equal to `(config.samples() - 1) * timeWindowMs`, and then we will get smaller measure due to the larger denominator
Maybe we can remove `eventWindow` as it is unused in production. This can be another PR if it is valid
##########
clients/src/main/java/org/apache/kafka/common/metrics/stats/SampledStat.java:
##########
@@ -50,10 +50,13 @@ public void record(MetricConfig config, double value, long timeMs) {
sample = advance(config, timeMs);
update(sample, config, value, timeMs);
sample.eventCount += 1;
+ sample.lastEventMs = timeMs;
}
private Sample advance(MetricConfig config, long timeMs) {
- this.current = (this.current + 1) % config.samples();
+ // need to keep one extra sample (see purgeObsoleteSamples() logic)
Review Comment:
How about saying "we have one extra sample to remember the last recording time in order to keep the overlapping sample (see purgeObsoleteSamples() logic)"
##########
clients/src/test/java/org/apache/kafka/common/metrics/stats/SampledStatTest.java:
##########
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.metrics.stats;
+
+import org.apache.kafka.common.metrics.MetricConfig;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.common.utils.Time;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class SampledStatTest {
+
+ private SampledStat stat;
+ private Time time;
+
+ @BeforeEach
+ public void setup() {
+ stat = new SampleCount(0);
+ time = new MockTime();
+ }
+
+ @Test
+ @DisplayName("Sample should be purged if doesn't overlap the window")
+ public void testSampleIsPurgedIfDoesntOverlap() {
+ MetricConfig config = new MetricConfig().timeWindow(1, SECONDS).samples(2);
+
+ // Monitored window: 2s. Complete a sample and wait 2.5s after.
+ completeSample(config);
+ time.sleep(2500);
+
+ double numSamples = stat.measure(config, time.milliseconds());
+ assertEquals(0, numSamples);
+ }
+
+ @Test
+ @DisplayName("Sample should be kept if overlaps the window")
+ public void testSampleIsKeptIfOverlaps() {
+ MetricConfig config = new MetricConfig().timeWindow(1, SECONDS).samples(2);
+
+ // Monitored window: 2s. Complete a sample and wait 1.5s after.
+ completeSample(config);
+ time.sleep(1500);
+
+ double numSamples = stat.measure(config, time.milliseconds());
+ assertEquals(1, numSamples);
+ }
+
+ @Test
+ @DisplayName("Sample should be kept if overlaps the window and is n+1")
+ public void testSampleIsKeptIfOverlapsAndExtra() {
+ MetricConfig config = new MetricConfig().timeWindow(1, SECONDS).samples(2);
+
+ // Monitored window: 2s. Create 2 samples with gaps in between and
+ // take a measurement at 2.2s from the start.
+ completeSample(config);
+ time.sleep(100);
+ completeSample(config);
+ time.sleep(100);
+ stat.record(config, 1, time.milliseconds());
+
+ double numSamples = stat.measure(config, time.milliseconds());
+ assertEquals(3, numSamples);
+ }
+
+ // Creates a sample with events at the start and at the end. Positions clock at the end.
+ private void completeSample(MetricConfig config) {
+ stat.record(config, 1, time.milliseconds());
+ time.sleep(config.timeWindowMs() - 1);
+ stat.record(config, 1, time.milliseconds());
+ time.sleep(1);
+ }
+
+ // measure() of this impl returns the number of samples
+ static class SampleCount extends SampledStat {
+
+ SampleCount(double initialValue) {
Review Comment:
We don't use `initialValue` in testing, so it can be simplified to
```java
SampleCount() {
super(0);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "emitskevich-blp (via GitHub)" <gi...@apache.org>.
emitskevich-blp commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1617783141
##########
clients/src/test/java/org/apache/kafka/common/metrics/MetricsTest.java:
##########
@@ -475,28 +464,13 @@ public void testPercentiles() {
Metric p50 = this.metrics.metrics().get(metrics.metricName("test.p50", "grp1"));
Metric p75 = this.metrics.metrics().get(metrics.metricName("test.p75", "grp1"));
- // record two windows worth of sequential values
- for (int i = 0; i < buckets; i++)
+ // record 1-100 sequential values
+ for (int i = 0; i < buckets; i++) {
Review Comment:
Applied
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1611391288
##########
clients/src/main/java/org/apache/kafka/common/metrics/stats/SampledStat.java:
##########
@@ -106,44 +109,51 @@ public String toString() {
public abstract double combine(List<Sample> samples, MetricConfig config, long now);
- /* Timeout any windows that have expired in the absence of any events */
+ // purge any samples that lack observed events within the monitored window
protected void purgeObsoleteSamples(MetricConfig config, long now) {
long expireAge = config.samples() * config.timeWindowMs();
for (Sample sample : samples) {
- if (now - sample.lastWindowMs >= expireAge)
+ // samples overlapping the monitored window are kept,
+ // even if they started before it
+ if (now - sample.lastEventMs >= expireAge) {
sample.reset(now);
+ }
}
}
protected static class Sample {
public double initialValue;
public long eventCount;
- public long lastWindowMs;
+ public long startTimeMs;
+ public long lastEventMs;
public double value;
public Sample(double initialValue, long now) {
this.initialValue = initialValue;
this.eventCount = 0;
- this.lastWindowMs = now;
+ this.startTimeMs = now;
+ this.lastEventMs = now;
this.value = initialValue;
}
public void reset(long now) {
this.eventCount = 0;
- this.lastWindowMs = now;
+ this.startTimeMs = now;
+ this.lastEventMs = now;
this.value = initialValue;
}
public boolean isComplete(long timeMs, MetricConfig config) {
- return timeMs - lastWindowMs >= config.timeWindowMs() || eventCount >= config.eventWindow();
+ return timeMs - startTimeMs >= config.timeWindowMs() || eventCount >= config.eventWindow();
Review Comment:
Please ignore my comments. see following test
```java
private double test(long eventWindow) {
Rate rate = new Rate();
Time timer = new MockTime();
final MetricConfig config = new MetricConfig().samples(2).timeWindow(1, SECONDS).eventWindow(eventWindow);
// generate 9 records
IntStream.range(0, 9).forEach(__ -> {
rate.record(config, 1, timer.milliseconds());
timer.sleep(1);
});
// almost 1 second
timer.sleep(990);
return rate.measure(config, timer.milliseconds());
}
```
**before**: 3 / second
**after**: 4.5 / second
I neglect that this approach has a extra sample, and so the measure value can be larger than before.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "emitskevich-blp (via GitHub)" <gi...@apache.org>.
emitskevich-blp commented on PR #15889:
URL: https://github.com/apache/kafka/pull/15889#issuecomment-2106339696
@chia7712 @junrao maybe you can take a look here? Not invasive change, test failures are not related to PR changes
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "emitskevich-blp (via GitHub)" <gi...@apache.org>.
emitskevich-blp commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1602153395
##########
clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java:
##########
@@ -63,33 +63,38 @@ public void record(MetricConfig config, double value, long timeMs) {
@Override
public double measure(MetricConfig config, long now) {
+ long windowSizeMs = windowSize(config, now);
double value = stat.measure(config, now);
- return value / convert(windowSize(config, now), unit);
+ return value / convert(windowSizeMs, unit);
}
public long windowSize(MetricConfig config, long now) {
// purge old samples before we compute the window size
- stat.purgeObsoleteSamples(config, now);
+ long purgedUpToMs = stat.purgeObsoleteSamples(config, now);
/*
* Here we check the total amount of time elapsed since the oldest non-obsolete window.
* This give the total windowSize of the batch which is the time used for Rate computation.
- * However, there is an issue if we do not have sufficient data for e.g. if only 1 second has elapsed in a 30 second
+ * However, there is an issue if we do not have sufficient data for e.g. if only 1 second has elapsed in a 30-second
* window, the measured rate will be very high.
- * Hence we assume that the elapsed time is always N-1 complete windows plus whatever fraction of the final window is complete.
+ * Hence, we assume that the elapsed time is always N-1 complete windows plus whatever fraction of the final window is complete.
*
* Note that we could simply count the amount of time elapsed in the current window and add n-1 windows to get the total time,
* but this approach does not account for sleeps. SampledStat only creates samples whenever record is called,
* if no record is called for a period of time that time is not accounted for in windowSize and produces incorrect results.
*/
- long totalElapsedTimeMs = now - stat.oldest(now).lastWindowMs;
+ long totalElapsedTimeMs = now - stat.oldest(now).startTimeMs;
// Check how many full windows of data we have currently retained
int numFullWindows = (int) (totalElapsedTimeMs / config.timeWindowMs());
int minFullWindows = config.samples() - 1;
// If the available windows are less than the minimum required, add the difference to the totalElapsedTime
- if (numFullWindows < minFullWindows)
+ if (numFullWindows < minFullWindows) {
totalElapsedTimeMs += (minFullWindows - numFullWindows) * config.timeWindowMs();
+ }
+
+ // if some part of considered interval was just purged with its data, exclude it
+ totalElapsedTimeMs = Math.min(totalElapsedTimeMs, now - purgedUpToMs);
Review Comment:
> if measure() is called a second time after a very short window, no samples will be purged and we won't be compensating for the purgedUpToMs. Then, we will fall back to use the larger window, which will result in a much lower measured value.
Thank you, I missed it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1610602287
##########
clients/src/main/java/org/apache/kafka/common/metrics/stats/SampledStat.java:
##########
@@ -106,44 +109,51 @@ public String toString() {
public abstract double combine(List<Sample> samples, MetricConfig config, long now);
- /* Timeout any windows that have expired in the absence of any events */
+ // purge any samples that lack observed events within the monitored window
protected void purgeObsoleteSamples(MetricConfig config, long now) {
long expireAge = config.samples() * config.timeWindowMs();
for (Sample sample : samples) {
- if (now - sample.lastWindowMs >= expireAge)
+ // samples overlapping the monitored window are kept,
+ // even if they started before it
+ if (now - sample.lastEventMs >= expireAge) {
sample.reset(now);
+ }
}
}
protected static class Sample {
public double initialValue;
public long eventCount;
- public long lastWindowMs;
+ public long startTimeMs;
+ public long lastEventMs;
public double value;
public Sample(double initialValue, long now) {
this.initialValue = initialValue;
this.eventCount = 0;
- this.lastWindowMs = now;
+ this.startTimeMs = now;
+ this.lastEventMs = now;
this.value = initialValue;
}
public void reset(long now) {
this.eventCount = 0;
- this.lastWindowMs = now;
+ this.startTimeMs = now;
+ this.lastEventMs = now;
this.value = initialValue;
}
public boolean isComplete(long timeMs, MetricConfig config) {
- return timeMs - lastWindowMs >= config.timeWindowMs() || eventCount >= config.eventWindow();
+ return timeMs - startTimeMs >= config.timeWindowMs() || eventCount >= config.eventWindow();
Review Comment:
> in this case, totalElapsedTimeMs will be equal to (config.samples() - 1) * timeWindowMs, and then we will get smaller measure due to the larger denominator
@chia7712: Could you explain your point a bit more? In the current design, we always ensure at least `(config.samples() - 1) * timeWindowMs` in `totalElapsedTimeMs`. This is to avoid the potential extreme large measured value when the accumulated sample window is small (e.g. when the metric is first recorded).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Fix rate metric spikes [kafka]
Posted by "emitskevich-blp (via GitHub)" <gi...@apache.org>.
emitskevich-blp commented on code in PR #15889:
URL: https://github.com/apache/kafka/pull/15889#discussion_r1610656953
##########
clients/src/main/java/org/apache/kafka/common/metrics/stats/SampledStat.java:
##########
@@ -50,10 +50,13 @@ public void record(MetricConfig config, double value, long timeMs) {
sample = advance(config, timeMs);
update(sample, config, value, timeMs);
sample.eventCount += 1;
+ sample.lastEventMs = timeMs;
}
private Sample advance(MetricConfig config, long timeMs) {
- this.current = (this.current + 1) % config.samples();
+ // need to keep one extra sample (see purgeObsoleteSamples() logic)
Review Comment:
Addressed, but I slightly changed the suggestion, please check if it's readable. Because technically this extra space is not to remember the last recording time, it's to keep the entire extra sample with all collected data.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org