Skip to content

ZOOKEEPER-4216: Fix a race condition in WatcherCleanerTest.testDeadWatcherMetrics #1989

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
PapaCharlie wants to merge 3 commits into from
Closed

ZOOKEEPER-4216: Fix a race condition in WatcherCleanerTest.testDeadWatcherMetrics #1989

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c1d57e0
Fix a race condition in WatcherCleanerTest.testDeadWatcherMetrics
PapaCharlie Mar 8, 2023
1052ee9
Leverage an existing method and refactor the rest of the code to match
PapaCharlie Mar 8, 2023
c0eb6fc
Update log message to be less alarming when polling for metric value
PapaCharlie May 24, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 43 additions & 3 deletions zookeeper-server/src/test/java/org/apache/zookeeper/ZKTestCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,12 @@
import static org.junit.jupiter.api.Assertions.fail;
import java.io.File;
import java.time.Instant;
import org.apache.zookeeper.metrics.MetricsUtils;
import org.apache.zookeeper.util.ServiceUtils;
import org.hamcrest.CustomMatcher;
import org.hamcrest.Description;
import org.hamcrest.Matcher;
import org.hamcrest.StringDescription;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
Expand All @@ -40,6 +45,8 @@ public class ZKTestCase {

protected static final File testBaseDir = new File(System.getProperty("build.test.dir", "build"));
private static final Logger LOG = LoggerFactory.getLogger(ZKTestCase.class);
public static final int DEFAULT_METRIC_TIMEOUT = 30;
public static final int METRIC_POLL_INTERVAL_MS = 100;

static {
// Disable System.exit in tests.
Expand Down Expand Up @@ -103,15 +110,48 @@ public interface WaitForCondition {
* @param timeout timeout in seconds
* @throws InterruptedException
*/
public void waitFor(String msg, WaitForCondition condition, int timeout) throws InterruptedException {
public static void waitFor(String msg, WaitForCondition condition, int timeout) throws InterruptedException {
final Instant deadline = Instant.now().plusSeconds(timeout);
while (Instant.now().isBefore(deadline)) {
if (condition.evaluate()) {
return;
}
Thread.sleep(100);
Thread.sleep(METRIC_POLL_INTERVAL_MS);
}
fail(msg);
}

}
public static <T> void waitForMetric(String metricKey, Matcher<T> matcher) throws InterruptedException {
waitForMetric(metricKey, matcher, DEFAULT_METRIC_TIMEOUT);
}

public static <T> void waitForMetric(String metricKey, Matcher<T> matcher, int timeoutInSeconds) throws InterruptedException {
String errorMessage = String.format("metric \"%s\" failed to match after %d seconds",
metricKey, timeoutInSeconds);
waitFor(errorMessage, () -> {
@SuppressWarnings("unchecked")
T actual = (T) MetricsUtils.currentServerMetrics().get(metricKey);
if (!matcher.matches(actual)) {
Description description = new StringDescription();
matcher.describeMismatch(actual, description);
LOG.info("metric {} does not yet match expected value: {}. Checking again in {}ms",
metricKey, description, METRIC_POLL_INTERVAL_MS);
return false;
}
return true;
}, timeoutInSeconds);
}

/**
* Functionally identical to {@link org.hamcrest.Matchers#closeTo} except that it accepts all numerical types
* instead of failing if the value is not a {@link Double}.
*/
public static Matcher<Number> closeTo(double operand, double error) {
return new CustomMatcher<Number>(String.format("A number within %s of %s", error, operand)) {
@Override
public boolean matches(Object actual) {
return Math.abs(operand - ((Number) actual).doubleValue()) <= error;
}
};
}
}
16 changes: 5 additions & 11 deletions zookeeper-server/src/test/java/org/apache/zookeeper/server/RequestThrottlerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,8 @@
package org.apache.zookeeper.server;

import static org.apache.zookeeper.test.ClientBase.CONNECTION_TIMEOUT;
import static org.hamcrest.Matchers.greaterThanOrEqualTo;
import static org.hamcrest.Matchers.is;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.io.File;
Expand Down Expand Up @@ -75,10 +77,6 @@ public class RequestThrottlerTest extends ZKTestCase {
ZooKeeper zk = null;
int connectionLossCount = 0;

private long getCounterMetric(String name) {
return (long) MetricsUtils.currentServerMetrics().get(name);
}

@BeforeEach
public void setup() throws Exception {
// start a server and create a client
Expand Down Expand Up @@ -222,11 +220,8 @@ public void testRequestThrottler() throws Exception {
submitted.await(5, TimeUnit.SECONDS);

// but only two requests can get into the pipeline because of the throttler
WaitForCondition requestQueued = () -> getCounterMetric("prep_processor_request_queued") == 2;
waitFor("request not queued", requestQueued, 5);

WaitForCondition throttleWait = () -> getCounterMetric("request_throttle_wait_count") >= 1;
waitFor("no throttle wait", throttleWait, 5);
waitForMetric("prep_processor_request_queued", is(2L));
waitForMetric("request_throttle_wait_count", greaterThanOrEqualTo(1L));

// let the requests go through the pipeline and the throttler will be waken up to allow more requests
// to enter the pipeline
Expand Down Expand Up @@ -387,8 +382,7 @@ public void testGlobalOutstandingRequestThrottlingWithRequestThrottlerDisabled()
// be GLOBAL_OUTSTANDING_LIMIT + 2.
//
// But due to leak of consistent view of number of outstanding requests, the number could be larger.
WaitForCondition requestQueued = () -> getCounterMetric("prep_processor_request_queued") >= Integer.parseInt(GLOBAL_OUTSTANDING_LIMIT) + 2;
waitFor("no enough requests queued", requestQueued, 5);
waitForMetric("prep_processor_request_queued", greaterThanOrEqualTo(Long.parseLong(GLOBAL_OUTSTANDING_LIMIT) + 2));

resumeProcess.countDown();
} catch (Exception e) {
Expand Down
3 changes: 1 addition & 2 deletions ...rver/src/test/java/org/apache/zookeeper/server/persistence/FileTxnSnapLogMetricsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,8 +79,7 @@ public void testFileTxnSnapLogMetrics() throws Exception {
}

// It is possible that above writes will trigger more than one snapshot due to randomization.
WaitForCondition newSnapshot = () -> (long) MetricsUtils.currentServerMetrics().get("cnt_snapshottime") >= 2L;
waitFor("no snapshot in 10s", newSnapshot, 10);
waitForMetric("cnt_snapshottime", greaterThanOrEqualTo(2L), 10);

// Pauses snapshot and logs more txns.
cnxnFactory.getZooKeeperServer().getTxnLogFactory().snapLog.close();
Expand Down
15 changes: 0 additions & 15 deletions zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/LearnerMetricsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,10 +26,8 @@
import org.apache.zookeeper.PortAssignment;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.metrics.MetricsUtils;
import org.apache.zookeeper.server.ServerMetrics;
import org.apache.zookeeper.test.ClientBase;
import org.hamcrest.Matcher;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
Expand All @@ -38,7 +36,6 @@

public class LearnerMetricsTest extends QuorumPeerTestBase {

private static final int TIMEOUT_SECONDS = 30;
private static final int SERVER_COUNT = 4; // 1 observer, 3 participants
private final QuorumPeerTestBase.MainThread[] mt = new QuorumPeerTestBase.MainThread[SERVER_COUNT];
private ZooKeeper zk_client;
Expand Down Expand Up @@ -113,18 +110,6 @@ public void testLearnerMetricsTest(boolean asyncSending) throws Exception {
waitForMetric("min_commit_propagation_latency", greaterThanOrEqualTo(0L));
}

private void waitForMetric(final String metricKey, final Matcher<Long> matcher) throws InterruptedException {
final String errorMessage = String.format("unable to match on metric: %s", metricKey);
waitFor(errorMessage, () -> {
long actual = (long) MetricsUtils.currentServerMetrics().get(metricKey);
if (!matcher.matches(actual)) {
LOG.info("match failed on {}, actual value: {}", metricKey, actual);
return false;
}
return true;
}, TIMEOUT_SECONDS);
}

@AfterEach
public void tearDown() throws Exception {
zk_client.close();
Expand Down
32 changes: 16 additions & 16 deletions zookeeper-server/src/test/java/org/apache/zookeeper/server/watch/WatcherCleanerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,8 @@

package org.apache.zookeeper.server.watch;

import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.number.OrderingComparison.greaterThan;
import static org.hamcrest.Matchers.greaterThan;
import static org.hamcrest.Matchers.is;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
Expand Down Expand Up @@ -140,7 +140,7 @@ public void testMaxInProcessingDeadWatchers() {
}

@Test
public void testDeadWatcherMetrics() {
public void testDeadWatcherMetrics() throws InterruptedException {
ServerMetrics.getMetrics().resetAll();
MyDeadWatcherListener listener = new MyDeadWatcherListener();
WatcherCleaner cleaner = new WatcherCleaner(listener, 1, 1, 1, 1);
Expand All @@ -156,19 +156,19 @@ public void testDeadWatcherMetrics() {
assertTrue(listener.wait(5000));

Map<String, Object> values = MetricsUtils.currentServerMetrics();
Comment thread
PapaCharlie marked this conversation as resolved.
assertThat("Adding dead watcher should be stalled twice", (Long) values.get("add_dead_watcher_stall_time"), greaterThan(0L));
assertEquals(3L, values.get("dead_watchers_queued"), "Total dead watchers added to the queue should be 3");
assertEquals(3L, values.get("dead_watchers_cleared"), "Total dead watchers cleared should be 3");

assertEquals(3L, values.get("cnt_dead_watchers_cleaner_latency"));

//Each latency should be a little over 20 ms, allow 20 ms deviation
assertEquals(20D, (Double) values.get("avg_dead_watchers_cleaner_latency"), 20);
assertEquals(20D, ((Long) values.get("min_dead_watchers_cleaner_latency")).doubleValue(), 20);
assertEquals(20D, ((Long) values.get("max_dead_watchers_cleaner_latency")).doubleValue(), 20);
assertEquals(20D, ((Long) values.get("p50_dead_watchers_cleaner_latency")).doubleValue(), 20);
assertEquals(20D, ((Long) values.get("p95_dead_watchers_cleaner_latency")).doubleValue(), 20);
assertEquals(20D, ((Long) values.get("p99_dead_watchers_cleaner_latency")).doubleValue(), 20);
// Adding dead watcher should be stalled twice
waitForMetric("add_dead_watcher_stall_time", greaterThan(0L));
waitForMetric("dead_watchers_queued", is(3L));
waitForMetric("dead_watchers_cleared", is(3L));
waitForMetric("cnt_dead_watchers_cleaner_latency", is(3L));

//Each latency should be a little over 20 ms, allow 5 ms deviation

@kezhuw kezhuw Mar 23, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Seems that we rollback ZOOKEEPER-4200(#1592) after waitForMetric. I am not sure whether this is a regression. But I think ZOOKEEPER-4200 could also caused by concurrency in avg assertion and not atomic DEAD_WATCHERS_CLEANER_LATENCY.add(latency). I am neutral to this change. But max is more likely to fail in loaded environment. Maybe we can treat max a little special ? What do you think ? @ztzg @eolivelli

zookeeper/zookeeper-server/src/main/java/org/apache/zookeeper/server/metric/AvgMinMaxCounter.java

Lines 44 to 49 in e50a0bb

public void addDataPoint(long value) {
total.addAndGet(value);
count.incrementAndGet();
setMin(value);
setMax(value);
}

zookeeper/zookeeper-server/src/main/java/org/apache/zookeeper/server/metric/AvgMinMaxCounter.java

Lines 65 to 76 in e50a0bb

public double getAvg() {
// There is possible race-condition but we don't need the stats to be
// extremely accurate.
long currentCount = count.get();
long currentTotal = total.get();
if (currentCount > 0) {
double avgLatency = currentTotal / (double) currentCount;
BigDecimal bg = new BigDecimal(avgLatency);
return bg.setScale(4, RoundingMode.HALF_UP).doubleValue();
}
return 0;
}

waitForMetric("avg_dead_watchers_cleaner_latency", closeTo(20, 5));
waitForMetric("min_dead_watchers_cleaner_latency", closeTo(20, 5));
waitForMetric("max_dead_watchers_cleaner_latency", closeTo(20, 5));
waitForMetric("p50_dead_watchers_cleaner_latency", closeTo(20, 5));
waitForMetric("p95_dead_watchers_cleaner_latency", closeTo(20, 5));
waitForMetric("p99_dead_watchers_cleaner_latency", closeTo(20, 5));
}

}