diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/DataTree.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/DataTree.java index 1f93f18c778..6d55fc29e36 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/DataTree.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/DataTree.java @@ -62,6 +62,7 @@ import org.apache.zookeeper.data.StatPersisted; import org.apache.zookeeper.server.watch.IWatchManager; import org.apache.zookeeper.server.watch.WatchManagerFactory; +import org.apache.zookeeper.server.watch.WatchRegistration; import org.apache.zookeeper.server.watch.WatcherMode; import org.apache.zookeeper.server.watch.WatcherOrBitSet; import org.apache.zookeeper.server.watch.WatchesPathReport; @@ -1429,6 +1430,36 @@ public synchronized WatchesPathReport getWatchesByPath() { return dataWatches.getWatchesByPath(); } + /** + * Returns bounded Data Watch registrations. + * + * @param path exact path filter, or null for every path + * @param sessionIds Session ID filter, or null for every session + * @param maxResults maximum registrations to return + * @return matching Data Watch registrations + */ + public List getDataWatchRegistrations( + String path, + Set sessionIds, + int maxResults) { + return dataWatches.getWatchRegistrations(path, sessionIds, maxResults); + } + + /** + * Returns bounded Children Watch registrations. + * + * @param path exact path filter, or null for every path + * @param sessionIds Session ID filter, or null for every session + * @param maxResults maximum registrations to return + * @return matching Children Watch registrations + */ + public List getChildWatchRegistrations( + String path, + Set sessionIds, + int maxResults) { + return childWatches.getWatchRegistrations(path, sessionIds, maxResults); + } + /** * Returns a watch summary. * diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/DumbWatcher.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/DumbWatcher.java index fc4648758cf..ea3c0a1495a 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/DumbWatcher.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/DumbWatcher.java @@ -73,7 +73,7 @@ public long getMostRecentZxid() { @Override - int getSessionTimeout() { + public int getSessionTimeout() { return 0; } diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerCnxn.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerCnxn.java index eb31b5e9254..728924fc2e2 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerCnxn.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerCnxn.java @@ -132,7 +132,12 @@ public ServerCnxn(final ZooKeeperServer zkServer) { */ private volatile boolean invalid = false; - abstract int getSessionTimeout(); + /** + * Returns the negotiated session timeout in milliseconds. + * + * @return session timeout in milliseconds + */ + public abstract int getSessionTimeout(); public void incrOutstandingAndCheckThrottle(RequestHeader h) { if (h.getXid() <= 0) { diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/admin/Commands.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/admin/Commands.java index af69008986c..6bb33e629f5 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/admin/Commands.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/admin/Commands.java @@ -29,11 +29,15 @@ import java.io.InputStream; import java.net.InetSocketAddress; import java.nio.charset.StandardCharsets; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.Date; import java.util.HashMap; import java.util.HashSet; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Properties; import java.util.Set; @@ -48,10 +52,12 @@ import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.Version; import org.apache.zookeeper.ZooDefs; +import org.apache.zookeeper.common.PathUtils; import org.apache.zookeeper.data.ACL; import org.apache.zookeeper.data.Id; import org.apache.zookeeper.server.DataNode; import org.apache.zookeeper.server.DataTree; +import org.apache.zookeeper.server.ServerCnxn; import org.apache.zookeeper.server.ServerCnxnFactory; import org.apache.zookeeper.server.ServerMetrics; import org.apache.zookeeper.server.ZooKeeperServer; @@ -72,6 +78,7 @@ import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier; import org.apache.zookeeper.server.util.RateLimiter; import org.apache.zookeeper.server.util.ZxidUtils; +import org.apache.zookeeper.server.watch.WatchRegistration; import org.eclipse.jetty.http.HttpStatus; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -304,6 +311,7 @@ public static Command getCommand(String cmdName) { registerCommand(new SystemPropertiesCommand()); registerCommand(new VotingViewCommand()); registerCommand(new WatchCommand()); + registerCommand(new WatchDetailsCommand()); registerCommand(new WatchesByPathCommand()); registerCommand(new WatchSummaryCommand()); registerCommand(new ZabStateCommand()); @@ -1138,6 +1146,309 @@ public CommandResponse runGet(ZooKeeperServer zkServer, Map kwar } + /** + * Detailed information about active watch registrations on this server. + */ + public static class WatchDetailsCommand extends GetCommand { + + private static final int DEFAULT_LIMIT = 100; + private static final int MIN_LIMIT = 1; + private static final int MAX_LIMIT = 1000; + private static final String DATA_WATCH_KIND = "data"; + private static final String CHILDREN_WATCH_KIND = "children"; + + public WatchDetailsCommand() { + super( + Arrays.asList("watch_details", "wchd"), + true); + } + + @Override + public CommandResponse runGet(ZooKeeperServer zkServer, Map kwargs) { + WatchDetailsQuery query; + try { + query = parseQuery(kwargs == null ? Collections.emptyMap() : kwargs); + } catch (IllegalArgumentException e) { + return new CommandResponse(getPrimaryName(), e.getMessage(), HttpServletResponse.SC_BAD_REQUEST); + } + + try { + return execute(zkServer, query); + } catch (RuntimeException e) { + LOG.warn("Failed to query watch details", e); + return new CommandResponse( + getPrimaryName(), + "Failed to query watch details", + HttpServletResponse.SC_INTERNAL_SERVER_ERROR); + } + } + + private CommandResponse execute(ZooKeeperServer zkServer, WatchDetailsQuery query) { + Map connections = snapshotConnections(zkServer); + Set candidateSessionIds = new HashSet<>(); + for (ConnectionSnapshot connection : connections.values()) { + if ((query.sessionId == null || query.sessionId == connection.sessionId) + && (query.clientIp == null || query.clientIp.equals(connection.clientIp))) { + candidateSessionIds.add(connection.sessionId); + } + } + + Map aggregatedWatchDetails = new LinkedHashMap<>(); + int maxResults = query.limit + 1; + DataTree dataTree = zkServer.getZKDatabase().getDataTree(); + List dataRegistrations; + try { + dataRegistrations = dataTree.getDataWatchRegistrations( + query.path, + candidateSessionIds, + maxResults); + } catch (UnsupportedOperationException e) { + return unsupportedWatchManagerResponse(); + } + mergeWatchDetails( + aggregatedWatchDetails, + dataRegistrations, + connections, + DATA_WATCH_KIND); + + List childRegistrations; + try { + childRegistrations = dataTree.getChildWatchRegistrations( + query.path, + candidateSessionIds, + maxResults); + } catch (UnsupportedOperationException e) { + return unsupportedWatchManagerResponse(); + } + mergeWatchDetails( + aggregatedWatchDetails, + childRegistrations, + connections, + CHILDREN_WATCH_KIND); + + List> watchDetails = new ArrayList<>(aggregatedWatchDetails.size()); + for (WatchDetail detail : aggregatedWatchDetails.values()) { + watchDetails.add(detail.toMap()); + } + + boolean truncated = watchDetails.size() > query.limit; + if (truncated) { + watchDetails = new ArrayList<>(watchDetails.subList(0, query.limit)); + } + CommandResponse response = initializeResponse(); + response.put("server_id", zkServer.getServerId()); + response.put("returned_count", watchDetails.size()); + response.put("truncated", truncated); + response.put("watches", watchDetails); + return response; + } + + private CommandResponse unsupportedWatchManagerResponse() { + return new CommandResponse( + getPrimaryName(), + "Watch details are not supported by the configured WatchManager", + HttpServletResponse.SC_NOT_IMPLEMENTED); + } + + private static WatchDetailsQuery parseQuery(Map kwargs) { + String path = kwargs.get("path"); + if (path != null) { + try { + PathUtils.validatePath(path); + } catch (IllegalArgumentException e) { + throw new IllegalArgumentException("Invalid path: " + path); + } + } + + String sessionId = kwargs.get("session_id"); + Long parsedSessionId = null; + if (sessionId != null) { + try { + if (sessionId.startsWith("0x") || sessionId.startsWith("0X")) { + parsedSessionId = Long.parseUnsignedLong(sessionId.substring(2), 16); + } else { + parsedSessionId = Long.parseUnsignedLong(sessionId); + } + } catch (NumberFormatException e) { + throw new IllegalArgumentException("Invalid session_id: " + sessionId); + } + } + + String clientIp = kwargs.get("client_ip"); + if (clientIp != null && clientIp.trim().isEmpty()) { + throw new IllegalArgumentException("client_ip must not be empty"); + } + clientIp = clientIp == null ? null : clientIp.trim(); + + String rawLimit = kwargs.get("limit"); + int limit = DEFAULT_LIMIT; + if (rawLimit != null) { + try { + limit = Integer.parseInt(rawLimit); + } catch (NumberFormatException e) { + throw new IllegalArgumentException("limit must be an integer"); + } + } + if (limit < MIN_LIMIT || limit > MAX_LIMIT) { + throw new IllegalArgumentException("limit must be between 1 and 1000"); + } + return new WatchDetailsQuery(path, parsedSessionId, clientIp, limit); + } + + private static Map snapshotConnections(ZooKeeperServer zkServer) { + Map snapshots = new HashMap<>(); + addConnectionSnapshots(snapshots, zkServer.getServerCnxnFactory()); + addConnectionSnapshots(snapshots, zkServer.getSecureServerCnxnFactory()); + return snapshots; + } + + private static void addConnectionSnapshots( + Map snapshots, + ServerCnxnFactory factory) { + if (factory == null) { + return; + } + Iterable connections = factory.getConnections(); + if (connections == null) { + return; + } + for (ServerCnxn connection : connections) { + if (connection == null || connection.isStale() || connection.getSessionId() == 0) { + continue; + } + InetSocketAddress remoteAddress = connection.getRemoteSocketAddress(); + if (remoteAddress == null || remoteAddress.getAddress() == null) { + continue; + } + String clientIp = remoteAddress.getAddress().getHostAddress(); + Date established = connection.getEstablished(); + if (clientIp == null || clientIp.isEmpty() || established == null) { + continue; + } + ConnectionSnapshot snapshot = new ConnectionSnapshot( + connection.getSessionId(), + clientIp, + remoteAddress.getPort(), + established.getTime(), + connection.getSessionTimeout(), + connection.isSecure()); + if (connection.isStale()) { + continue; + } + ConnectionSnapshot current = snapshots.get(snapshot.sessionId); + if (current == null || snapshot.establishedAt > current.establishedAt) { + snapshots.put(snapshot.sessionId, snapshot); + } + } + } + + private static void mergeWatchDetails( + Map watchDetails, + List registrations, + Map connections, + String watchKind) { + for (WatchRegistration registration : registrations) { + ConnectionSnapshot connection = connections.get(registration.getSessionId()); + if (connection == null) { + continue; + } + WatchDetail detail = watchDetails.get(registration); + if (detail == null) { + detail = new WatchDetail(registration, connection); + watchDetails.put(registration, detail); + } + detail.includeKind(watchKind); + } + } + + private static final class WatchDetail { + + private final WatchRegistration registration; + private final ConnectionSnapshot connection; + private boolean data; + private boolean children; + + private WatchDetail(WatchRegistration registration, ConnectionSnapshot connection) { + this.registration = registration; + this.connection = connection; + } + + private void includeKind(String watchKind) { + if (registration.getWatcherMode().isPersistent()) { + data = true; + children = true; + } else if (DATA_WATCH_KIND.equals(watchKind)) { + data = true; + } else if (CHILDREN_WATCH_KIND.equals(watchKind)) { + children = true; + } + } + + private Map toMap() { + List watchKinds = new ArrayList<>(2); + if (data) { + watchKinds.add(DATA_WATCH_KIND); + } + if (children) { + watchKinds.add(CHILDREN_WATCH_KIND); + } + + Map detail = new LinkedHashMap<>(); + detail.put("path", registration.getPath()); + detail.put("session_id", "0x" + Long.toHexString(connection.sessionId)); + detail.put("client_ip", connection.clientIp); + detail.put("client_port", connection.clientPort); + detail.put("watch_kind", watchKinds); + detail.put("watch_mode", registration.getWatcherMode().name().toLowerCase(Locale.ROOT)); + detail.put("connection_established_at", connection.establishedAt); + detail.put("session_timeout_ms", connection.sessionTimeout); + detail.put("secure", connection.secure); + return detail; + } + } + + private static final class WatchDetailsQuery { + + private final String path; + private final Long sessionId; + private final String clientIp; + private final int limit; + + private WatchDetailsQuery(String path, Long sessionId, String clientIp, int limit) { + this.path = path; + this.sessionId = sessionId; + this.clientIp = clientIp; + this.limit = limit; + } + } + + private static final class ConnectionSnapshot { + + private final long sessionId; + private final String clientIp; + private final int clientPort; + private final long establishedAt; + private final int sessionTimeout; + private final boolean secure; + + private ConnectionSnapshot( + long sessionId, + String clientIp, + int clientPort, + long establishedAt, + int sessionTimeout, + boolean secure) { + this.sessionId = sessionId; + this.clientIp = clientIp; + this.clientPort = clientPort; + this.establishedAt = establishedAt; + this.sessionTimeout = sessionTimeout; + this.secure = secure; + } + } + + } + /** * Watch information aggregated by path. Returned Map contains: * - "path_to_session_ids": Map<String, Set<Long>> path -> session IDs of sessions watching path diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/IWatchManager.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/IWatchManager.java index 3e27698555d..920c9f55d2c 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/IWatchManager.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/IWatchManager.java @@ -20,6 +20,7 @@ import java.io.PrintWriter; import java.util.List; +import java.util.Set; import javax.annotation.Nullable; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.Watcher.Event.EventType; @@ -171,6 +172,18 @@ default boolean removeWatcher(String path, Watcher watcher, WatcherMode watcherM */ WatchesPathReport getWatchesByPath(); + /** + * Returns bounded Watch registration details. + * + * @param path exact path filter, or null for every path + * @param sessionIds Session ID filter, or null for every session + * @param maxResults maximum registrations to return + * @return Watch registrations matching the filters + */ + default List getWatchRegistrations(String path, Set sessionIds, int maxResults) { + throw new UnsupportedOperationException("Watch registration details are not supported"); + } + /** * String representation of watches. Warning, may be large! * diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatchManager.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatchManager.java index 014c944fc84..c17ba2e07c5 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatchManager.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatchManager.java @@ -19,10 +19,12 @@ package org.apache.zookeeper.server.watch; import java.io.PrintWriter; +import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Map.Entry; @@ -326,6 +328,66 @@ Map> getWatch2Paths() { return watch2Paths; } + @Override + public synchronized List getWatchRegistrations( + String path, + Set sessionIds, + int maxResults) { + if (maxResults <= 0 || (sessionIds != null && sessionIds.isEmpty())) { + return Collections.emptyList(); + } + Set registrations = new LinkedHashSet<>(Math.min(maxResults, 1024)); + if (path != null) { + collectWatchRegistrations(path, watchTable.get(path), sessionIds, maxResults, registrations); + return new ArrayList<>(registrations); + } + for (Entry> entry : watchTable.entrySet()) { + if (collectWatchRegistrations( + entry.getKey(), + entry.getValue(), + sessionIds, + maxResults, + registrations)) { + break; + } + } + return new ArrayList<>(registrations); + } + + private boolean collectWatchRegistrations( + String path, + Set watchers, + Set sessionIds, + int maxResults, + Set registrations) { + if (watchers == null) { + return false; + } + for (Watcher watcher : watchers) { + if (!(watcher instanceof ServerCnxn) || isDeadWatcher(watcher)) { + continue; + } + long sessionId = ((ServerCnxn) watcher).getSessionId(); + if (sessionId == 0 || (sessionIds != null && !sessionIds.contains(sessionId))) { + continue; + } + Map paths = watch2Paths.get(watcher); + WatchStats stats = paths == null ? null : paths.get(path); + if (stats == null) { + continue; + } + for (WatcherMode watcherMode : WatcherMode.values()) { + if (stats.hasMode(watcherMode)) { + boolean added = registrations.add(new WatchRegistration(path, sessionId, watcherMode)); + if (added && registrations.size() >= maxResults) { + return true; + } + } + } + } + return false; + } + @Override public synchronized WatchesReport getWatches() { Map> id2paths = new HashMap<>(); diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatchManagerOptimized.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatchManagerOptimized.java index 84cf2c73d13..2c81cae05f9 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatchManagerOptimized.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatchManagerOptimized.java @@ -19,9 +19,12 @@ package org.apache.zookeeper.server.watch; import java.io.PrintWriter; +import java.util.ArrayList; import java.util.BitSet; +import java.util.Collections; import java.util.HashMap; import java.util.HashSet; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Map.Entry; @@ -309,6 +312,61 @@ public WatchesSummary getWatchesSummary() { return new WatchesSummary(watcherBitIdMap.size(), pathSize(), size()); } + @Override + public List getWatchRegistrations( + String path, + Set sessionIds, + int maxResults) { + if (maxResults <= 0 || (sessionIds != null && sessionIds.isEmpty())) { + return Collections.emptyList(); + } + Set registrations = new LinkedHashSet<>(Math.min(maxResults, 1024)); + if (path != null) { + collectWatchRegistrations(path, pathWatches.get(path), sessionIds, maxResults, registrations); + return new ArrayList<>(registrations); + } + for (Entry entry : pathWatches.entrySet()) { + if (collectWatchRegistrations( + entry.getKey(), + entry.getValue(), + sessionIds, + maxResults, + registrations)) { + break; + } + } + return new ArrayList<>(registrations); + } + + private boolean collectWatchRegistrations( + String path, + BitHashSet watchers, + Set sessionIds, + int maxResults, + Set registrations) { + if (watchers == null) { + return false; + } + synchronized (watchers) { + for (Integer watcherBit : watchers) { + Watcher watcher = watcherBitIdMap.get(watcherBit); + if (!(watcher instanceof ServerCnxn) || isDeadWatcher(watcher)) { + continue; + } + long sessionId = ((ServerCnxn) watcher).getSessionId(); + if (sessionId == 0 || (sessionIds != null && !sessionIds.contains(sessionId))) { + continue; + } + boolean added = registrations.add( + new WatchRegistration(path, sessionId, WatcherMode.STANDARD)); + if (added && registrations.size() >= maxResults) { + return true; + } + } + } + return false; + } + @Override public WatchesReport getWatches() { Map> id2paths = new HashMap<>(); diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatchRegistration.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatchRegistration.java new file mode 100644 index 00000000000..d76b514caee --- /dev/null +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatchRegistration.java @@ -0,0 +1,84 @@ +/* + * 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.zookeeper.server.watch; + +import java.util.Objects; + +/** + * A single Watch registration held by an {@link IWatchManager}. + */ +public final class WatchRegistration { + + private final String path; + private final long sessionId; + private final WatcherMode watcherMode; + + /** + * Creates a Watch registration. + * + * @param path watched znode path + * @param sessionId owning Session ID + * @param watcherMode Watch registration mode + */ + public WatchRegistration(String path, long sessionId, WatcherMode watcherMode) { + this.path = path; + this.sessionId = sessionId; + this.watcherMode = watcherMode; + } + + /** + * @return watched znode path + */ + public String getPath() { + return path; + } + + /** + * @return owning Session ID + */ + public long getSessionId() { + return sessionId; + } + + /** + * @return Watch registration mode + */ + public WatcherMode getWatcherMode() { + return watcherMode; + } + + @Override + public boolean equals(Object other) { + if (this == other) { + return true; + } + if (!(other instanceof WatchRegistration)) { + return false; + } + WatchRegistration that = (WatchRegistration) other; + return sessionId == that.sessionId + && Objects.equals(path, that.path) + && watcherMode == that.watcherMode; + } + + @Override + public int hashCode() { + return Objects.hash(path, sessionId, watcherMode); + } +} diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/DataTreeTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/DataTreeTest.java index 89a2f17b75f..ee7f62ba297 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/server/DataTreeTest.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/server/DataTreeTest.java @@ -25,6 +25,8 @@ import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.io.DataInputStream; @@ -33,7 +35,9 @@ import java.io.PrintWriter; import java.io.StringWriter; import java.lang.reflect.Field; +import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; @@ -50,6 +54,8 @@ import org.apache.zookeeper.common.PathTrie; import org.apache.zookeeper.data.Stat; import org.apache.zookeeper.metrics.MetricsUtils; +import org.apache.zookeeper.server.watch.WatchManager; +import org.apache.zookeeper.server.watch.WatchRegistration; import org.apache.zookeeper.txn.CreateTxn; import org.apache.zookeeper.txn.TxnHeader; import org.junit.jupiter.api.Test; @@ -129,6 +135,60 @@ public void testRootWatchTriggered() throws Exception { assertTrue(fire.isDone(), "Root node watch not triggered"); } + @Test + public void testGetDataAndChildWatchRegistrations() throws Exception { + String property = org.apache.zookeeper.server.watch.WatchManagerFactory.ZOOKEEPER_WATCH_MANAGER_NAME; + String previous = System.getProperty(property); + System.setProperty(property, WatchManager.class.getName()); + DataTree dataTree; + try { + dataTree = new DataTree(); + } finally { + if (previous == null) { + System.clearProperty(property); + } else { + System.setProperty(property, previous); + } + } + + try { + ServerCnxn connection = mock(ServerCnxn.class); + when(connection.getSessionId()).thenReturn(0x55L); + when(connection.isStale()).thenReturn(false); + + dataTree.statNode("/", connection); + dataTree.getChildren("/", null, connection); + dataTree.addWatch("/persistent", connection, ZooDefs.AddWatchModes.persistent); + dataTree.addWatch("/recursive", connection, ZooDefs.AddWatchModes.persistentRecursive); + + List dataRegistrations = dataTree.getDataWatchRegistrations(null, null, 10); + List childRegistrations = dataTree.getChildWatchRegistrations(null, null, 10); + + Set data = new java.util.HashSet<>(); + for (WatchRegistration registration : dataRegistrations) { + data.add(registration.getPath() + ":" + registration.getWatcherMode()); + } + Set children = new java.util.HashSet<>(); + for (WatchRegistration registration : childRegistrations) { + children.add(registration.getPath() + ":" + registration.getWatcherMode()); + } + + assertEquals( + new java.util.HashSet<>(java.util.Arrays.asList( + "/:STANDARD", + "/persistent:PERSISTENT", + "/recursive:PERSISTENT_RECURSIVE")), + data); + assertEquals( + new java.util.HashSet<>(java.util.Arrays.asList( + "/:STANDARD", + "/persistent:PERSISTENT")), + children); + } finally { + dataTree.shutdownWatcher(); + } + } + /** * For ZOOKEEPER-1046 test if cversion is getting incremented correctly. */ diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/MockServerCnxn.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/MockServerCnxn.java index af095921dad..4e7248d9f90 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/server/MockServerCnxn.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/server/MockServerCnxn.java @@ -39,7 +39,7 @@ public MockServerCnxn() { } @Override - int getSessionTimeout() { + public int getSessionTimeout() { return 0; } diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/admin/CommandsTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/admin/CommandsTest.java index ef80e7778e4..6fcf4ceb552 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/server/admin/CommandsTest.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/server/admin/CommandsTest.java @@ -28,23 +28,34 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; import java.io.ByteArrayInputStream; import java.io.IOException; import java.io.InputStream; +import java.net.InetSocketAddress; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; +import java.util.Date; import java.util.HashMap; +import java.util.List; import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; import javax.servlet.http.HttpServletResponse; +import org.apache.zookeeper.ZooDefs; import org.apache.zookeeper.metrics.MetricsUtils; +import org.apache.zookeeper.server.DataTree; +import org.apache.zookeeper.server.ServerCnxn; import org.apache.zookeeper.server.ServerCnxnFactory; import org.apache.zookeeper.server.ServerStats; import org.apache.zookeeper.server.ZKDatabase; import org.apache.zookeeper.server.ZooKeeperServer; import org.apache.zookeeper.server.quorum.BufferStats; +import org.apache.zookeeper.server.watch.WatchManager; +import org.apache.zookeeper.server.watch.WatchManagerFactory; import org.apache.zookeeper.test.ClientBase; import org.junit.jupiter.api.Test; @@ -314,6 +325,465 @@ public void testWatchSummary() throws IOException, InterruptedException { testCommand("watch_summary", new Field("num_connections", Integer.class), new Field("num_paths", Integer.class), new Field("num_total_watches", Integer.class)); } + @Test + public void testWatchDetailsRegistrationMetadata() { + Command command = Commands.getCommand("watch_details"); + + assertNotNull(command); + assertSame(command, Commands.getCommand("wchd")); + assertEquals("watch_details", command.getPrimaryName()); + assertTrue(Commands.getPrimaryNames().contains("watch_details")); + assertNull(command.getAuthRequest()); + } + + @Test + public void testWatchDetailsQueryValidationAndEmptyResponse() { + DataTree dataTree = mock(DataTree.class); + when(dataTree.getDataWatchRegistrations("/", Collections.emptySet(), 1001)) + .thenReturn(Collections.emptyList()); + when(dataTree.getChildWatchRegistrations("/", Collections.emptySet(), 1001)) + .thenReturn(Collections.emptyList()); + when(dataTree.getDataWatchRegistrations("/", Collections.emptySet(), 2)) + .thenReturn(Collections.emptyList()); + when(dataTree.getChildWatchRegistrations("/", Collections.emptySet(), 2)) + .thenReturn(Collections.emptyList()); + ZooKeeperServer zkServer = mockWatchDetailsServer(dataTree, null, null); + Commands.WatchDetailsCommand command = new Commands.WatchDetailsCommand(); + + assertWatchDetailsBadRequest(command, zkServer, "path", "relative", "Invalid path: relative"); + assertWatchDetailsBadRequest(command, zkServer, "session_id", "not-a-session", "Invalid session_id: not-a-session"); + assertWatchDetailsBadRequest(command, zkServer, "client_ip", " ", "client_ip must not be empty"); + assertWatchDetailsBadRequest(command, zkServer, "limit", "abc", "limit must be an integer"); + assertWatchDetailsBadRequest(command, zkServer, "limit", "0", "limit must be between 1 and 1000"); + assertWatchDetailsBadRequest(command, zkServer, "limit", "1001", "limit must be between 1 and 1000"); + + Map kwargs = new HashMap<>(); + kwargs.put("path", "/"); + kwargs.put("session_id", "0xffffffffffffffff"); + kwargs.put("client_ip", "127.0.0.1"); + kwargs.put("limit", "1000"); + CommandResponse response = command.runGet(zkServer, kwargs); + + assertEquals(HttpServletResponse.SC_OK, response.getStatusCode()); + assertEquals(7L, response.toMap().get("server_id")); + assertEquals(0, response.toMap().get("returned_count")); + assertEquals(false, response.toMap().get("truncated")); + assertEquals(new ArrayList<>(), response.toMap().get("watches")); + + kwargs.put("limit", "1"); + assertEquals(HttpServletResponse.SC_OK, command.runGet(zkServer, kwargs).getStatusCode()); + } + + @Test + public void testWatchDetailsDoesNotRequireAuthorization() { + ZooKeeperServer zkServer = serverFactory.getZooKeeperServer(); + CommandResponse response = Commands.runGetCommand( + "watch_details", + zkServer, + new HashMap<>(), + null, + null); + + assertEquals(HttpServletResponse.SC_OK, response.getStatusCode()); + assertNull(response.getError()); + } + + @Test + public void testWatchDetailsIncludesKindsModesAndConnectionFields() throws Exception { + DataTree dataTree = newWatchDetailsDataTree(); + try { + ServerCnxn plain = mockWatchConnection( + 0x11L, + "10.10.1.11", + 5111, + 1000L, + 20000, + false, + new AtomicBoolean(false)); + ServerCnxn secure = mockWatchConnection( + -1L, + "10.10.1.25", + 51324, + 1784112305123L, + 30000, + true, + new AtomicBoolean(false)); + + dataTree.statNode("/", plain); + dataTree.getChildren("/", null, secure); + dataTree.addWatch("/persistent", secure, ZooDefs.AddWatchModes.persistent); + dataTree.addWatch("/recursive", secure, ZooDefs.AddWatchModes.persistentRecursive); + + ZooKeeperServer zkServer = mockWatchDetailsServer( + dataTree, + Collections.singletonList(plain), + Collections.singletonList(secure)); + Map kwargs = new HashMap<>(); + kwargs.put("limit", "1000"); + CommandResponse response = new Commands.WatchDetailsCommand().runGet(zkServer, kwargs); + List> watches = watchDetails(response); + + assertEquals(4, watches.size()); + Map plainWatch = findWatch(watches, "/", "0x11", "standard"); + assertEquals("0x11", plainWatch.get("session_id")); + assertEquals("10.10.1.11", plainWatch.get("client_ip")); + assertEquals(5111, plainWatch.get("client_port")); + assertEquals(Arrays.asList("data"), plainWatch.get("watch_kind")); + assertEquals("standard", plainWatch.get("watch_mode")); + assertEquals(1000L, plainWatch.get("connection_established_at")); + assertEquals(20000, plainWatch.get("session_timeout_ms")); + assertEquals(false, plainWatch.get("secure")); + + Map childWatch = findWatch( + watches, + "/", + "0xffffffffffffffff", + "standard"); + assertEquals(Arrays.asList("children"), childWatch.get("watch_kind")); + + Map recursiveWatch = findWatch( + watches, + "/recursive", + "0xffffffffffffffff", + "persistent_recursive"); + assertEquals("0xffffffffffffffff", recursiveWatch.get("session_id")); + assertEquals(Arrays.asList("data", "children"), recursiveWatch.get("watch_kind")); + assertEquals("persistent_recursive", recursiveWatch.get("watch_mode")); + assertEquals(true, recursiveWatch.get("secure")); + + kwargs.put("path", "/persistent"); + kwargs.put("session_id", "0xffffffffffffffff"); + kwargs.put("client_ip", "10.10.1.25"); + watches = watchDetails(new Commands.WatchDetailsCommand().runGet(zkServer, kwargs)); + assertEquals(1, watches.size()); + Map persistentWatch = findWatch( + watches, + "/persistent", + "0xffffffffffffffff", + "persistent"); + assertEquals(Arrays.asList("data", "children"), persistentWatch.get("watch_kind")); + + kwargs.remove("path"); + kwargs.put("session_id", "18446744073709551615"); + assertEquals(3, watchDetails(new Commands.WatchDetailsCommand().runGet(zkServer, kwargs)).size()); + } finally { + dataTree.shutdownWatcher(); + } + } + + @Test + public void testWatchDetailsAggregatesStandardDataAndChildKinds() throws Exception { + DataTree dataTree = newWatchDetailsDataTree(); + try { + ServerCnxn connection = mockWatchConnection( + 0x12L, + "10.10.1.12", + 5112, + 1200L, + 20000, + false, + new AtomicBoolean(false)); + dataTree.statNode("/", connection); + dataTree.getChildren("/", null, connection); + + ZooKeeperServer zkServer = mockWatchDetailsServer( + dataTree, + Collections.singletonList(connection), + null); + Map kwargs = new HashMap<>(); + kwargs.put("path", "/"); + kwargs.put("limit", "1"); + + CommandResponse response = new Commands.WatchDetailsCommand().runGet(zkServer, kwargs); + List> watches = watchDetails(response); + + assertEquals(1, response.toMap().get("returned_count")); + assertEquals(false, response.toMap().get("truncated")); + assertEquals(1, watches.size()); + Map watch = findWatch(watches, "/", "0x12", "standard"); + assertEquals(Arrays.asList("data", "children"), watch.get("watch_kind")); + } finally { + dataTree.shutdownWatcher(); + } + } + + @Test + public void testWatchDetailsUsesNewestConnectionAndFiltersInactiveWatches() throws Exception { + DataTree dataTree = newWatchDetailsDataTree(); + try { + ServerCnxn oldConnection = mockWatchConnection( + 0x22L, + "10.10.2.1", + 5201, + 1000L, + 20000, + false, + new AtomicBoolean(false)); + ServerCnxn newConnection = mockWatchConnection( + 0x22L, + "10.10.2.2", + 5202, + 2000L, + 30000, + false, + new AtomicBoolean(false)); + AtomicBoolean staleState = new AtomicBoolean(false); + ServerCnxn staleConnection = mockWatchConnection( + 0x33L, + "10.10.3.3", + 5303, + 3000L, + 30000, + false, + staleState); + ServerCnxn disconnected = mockWatchConnection( + 0x44L, + "10.10.4.4", + 5404, + 4000L, + 30000, + false, + new AtomicBoolean(false)); + + dataTree.statNode("/", oldConnection); + dataTree.addWatch("/stale", staleConnection, ZooDefs.AddWatchModes.persistentRecursive); + dataTree.addWatch("/disconnected", disconnected, ZooDefs.AddWatchModes.persistentRecursive); + staleState.set(true); + + ZooKeeperServer zkServer = mockWatchDetailsServer( + dataTree, + Arrays.asList(oldConnection, newConnection, staleConnection), + null); + Map kwargs = new HashMap<>(); + kwargs.put("session_id", "34"); + kwargs.put("client_ip", "10.10.2.2"); + List> watches = watchDetails( + new Commands.WatchDetailsCommand().runGet(zkServer, kwargs)); + + assertEquals(1, watches.size()); + Map watch = watches.get(0); + assertEquals("10.10.2.2", watch.get("client_ip")); + assertEquals(5202, watch.get("client_port")); + assertEquals(2000L, watch.get("connection_established_at")); + assertEquals(30000, watch.get("session_timeout_ms")); + + watches = watchDetails(new Commands.WatchDetailsCommand().runGet(zkServer, new HashMap<>())); + assertEquals(1, watches.size()); + assertEquals("/", watches.get(0).get("path")); + } finally { + dataTree.shutdownWatcher(); + } + } + + @Test + public void testWatchDetailsLimitAndTruncation() { + DataTree dataTree = newWatchDetailsDataTree(); + try { + ServerCnxn connection = mockWatchConnection( + 0x55L, + "10.10.5.5", + 5505, + 5000L, + 30000, + false, + new AtomicBoolean(false)); + for (int i = 0; i < 101; i++) { + dataTree.addWatch("/watch-" + i, connection, ZooDefs.AddWatchModes.persistentRecursive); + } + ZooKeeperServer zkServer = mockWatchDetailsServer( + dataTree, + Collections.singletonList(connection), + null); + + CommandResponse response = new Commands.WatchDetailsCommand().runGet(zkServer, new HashMap<>()); + assertEquals(100, response.toMap().get("returned_count")); + assertEquals(true, response.toMap().get("truncated")); + assertEquals(100, watchDetails(response).size()); + + Map kwargs = new HashMap<>(); + kwargs.put("limit", "1000"); + response = new Commands.WatchDetailsCommand().runGet(zkServer, kwargs); + assertEquals(101, response.toMap().get("returned_count")); + assertEquals(false, response.toMap().get("truncated")); + assertEquals(101, watchDetails(response).size()); + } finally { + dataTree.shutdownWatcher(); + } + } + + @Test + public void testWatchDetailsReturnsNotImplementedForUnsupportedManager() { + DataTree dataTree = mock(DataTree.class); + when(dataTree.getDataWatchRegistrations("/", Collections.singleton(0x66L), 101)) + .thenThrow(new UnsupportedOperationException("Watch registration details are not supported")); + ServerCnxn connection = mockWatchConnection( + 0x66L, + "10.10.6.6", + 5606, + 6000L, + 30000, + false, + new AtomicBoolean(false)); + ZooKeeperServer zkServer = mockWatchDetailsServer( + dataTree, + Collections.singletonList(connection), + null); + Map kwargs = new HashMap<>(); + kwargs.put("path", "/"); + + CommandResponse response = new Commands.WatchDetailsCommand().runGet(zkServer, kwargs); + + assertEquals(HttpServletResponse.SC_NOT_IMPLEMENTED, response.getStatusCode()); + assertEquals("Watch details are not supported by the configured WatchManager", response.getError()); + } + + @Test + public void testWatchDetailsReturnsNotImplementedWithoutActiveConnections() { + DataTree dataTree = mock(DataTree.class); + when(dataTree.getDataWatchRegistrations(null, Collections.emptySet(), 101)) + .thenThrow(new UnsupportedOperationException("Watch registration details are not supported")); + ZooKeeperServer zkServer = mockWatchDetailsServer(dataTree, null, null); + + CommandResponse response = new Commands.WatchDetailsCommand().runGet( + zkServer, + Collections.emptyMap()); + + assertEquals(HttpServletResponse.SC_NOT_IMPLEMENTED, response.getStatusCode()); + assertEquals("Watch details are not supported by the configured WatchManager", response.getError()); + } + + @Test + public void testWatchDetailsReturnsInternalServerErrorForUnexpectedFailure() { + DataTree dataTree = mock(DataTree.class); + when(dataTree.getDataWatchRegistrations("/", Collections.singleton(0x77L), 101)) + .thenThrow(new IllegalStateException("unexpected")); + ServerCnxn connection = mockWatchConnection( + 0x77L, + "10.10.7.7", + 5707, + 7000L, + 30000, + false, + new AtomicBoolean(false)); + ZooKeeperServer zkServer = mockWatchDetailsServer( + dataTree, + Collections.singletonList(connection), + null); + Map kwargs = new HashMap<>(); + kwargs.put("path", "/"); + + CommandResponse response = new Commands.WatchDetailsCommand().runGet(zkServer, kwargs); + + assertEquals(HttpServletResponse.SC_INTERNAL_SERVER_ERROR, response.getStatusCode()); + assertEquals("Failed to query watch details", response.getError()); + } + + @Test + public void testWatchDetailsDoesNotTreatConnectionFailureAsUnsupportedManager() { + ServerCnxnFactory factory = mock(ServerCnxnFactory.class); + when(factory.getConnections()).thenThrow(new UnsupportedOperationException("unexpected")); + ZooKeeperServer zkServer = mock(ZooKeeperServer.class); + when(zkServer.getServerCnxnFactory()).thenReturn(factory); + + CommandResponse response = new Commands.WatchDetailsCommand().runGet( + zkServer, + Collections.emptyMap()); + + assertEquals(HttpServletResponse.SC_INTERNAL_SERVER_ERROR, response.getStatusCode()); + assertEquals("Failed to query watch details", response.getError()); + } + + private void assertWatchDetailsBadRequest( + Commands.WatchDetailsCommand command, + ZooKeeperServer zkServer, + String key, + String value, + String expectedError) { + Map kwargs = new HashMap<>(); + kwargs.put(key, value); + + CommandResponse response = command.runGet(zkServer, kwargs); + + assertEquals(HttpServletResponse.SC_BAD_REQUEST, response.getStatusCode()); + assertEquals(expectedError, response.getError()); + } + + private static DataTree newWatchDetailsDataTree() { + String property = WatchManagerFactory.ZOOKEEPER_WATCH_MANAGER_NAME; + String previous = System.getProperty(property); + System.setProperty(property, WatchManager.class.getName()); + try { + return new DataTree(); + } finally { + if (previous == null) { + System.clearProperty(property); + } else { + System.setProperty(property, previous); + } + } + } + + private static ServerCnxn mockWatchConnection( + long sessionId, + String clientIp, + int clientPort, + long establishedAt, + int sessionTimeout, + boolean secure, + AtomicBoolean stale) { + ServerCnxn connection = mock(ServerCnxn.class); + when(connection.getSessionId()).thenReturn(sessionId); + when(connection.getRemoteSocketAddress()).thenReturn(new InetSocketAddress(clientIp, clientPort)); + when(connection.getEstablished()).thenReturn(new Date(establishedAt)); + when(connection.getSessionTimeout()).thenReturn(sessionTimeout); + when(connection.isSecure()).thenReturn(secure); + when(connection.isStale()).thenAnswer(invocation -> stale.get()); + return connection; + } + + private static ZooKeeperServer mockWatchDetailsServer( + DataTree dataTree, + Iterable plainConnections, + Iterable secureConnections) { + ZooKeeperServer zkServer = mock(ZooKeeperServer.class); + ZKDatabase zkDatabase = mock(ZKDatabase.class); + when(zkServer.getServerId()).thenReturn(7L); + when(zkServer.getZKDatabase()).thenReturn(zkDatabase); + when(zkDatabase.getDataTree()).thenReturn(dataTree); + if (plainConnections != null) { + ServerCnxnFactory factory = mock(ServerCnxnFactory.class); + when(factory.getConnections()).thenReturn(plainConnections); + when(zkServer.getServerCnxnFactory()).thenReturn(factory); + } + if (secureConnections != null) { + ServerCnxnFactory factory = mock(ServerCnxnFactory.class); + when(factory.getConnections()).thenReturn(secureConnections); + when(zkServer.getSecureServerCnxnFactory()).thenReturn(factory); + } + return zkServer; + } + + @SuppressWarnings("unchecked") + private static List> watchDetails(CommandResponse response) { + return (List>) response.toMap().get("watches"); + } + + private static Map findWatch( + List> watches, + String path, + String sessionId, + String watchMode) { + for (Map watch : watches) { + if (path.equals(watch.get("path")) + && sessionId.equals(watch.get("session_id")) + && watchMode.equals(watch.get("watch_mode"))) { + return watch; + } + } + throw new AssertionError( + "Watch not found for path " + path + ", session " + sessionId + ", and mode " + watchMode); + } + @Test public void testVotingViewCommand() throws IOException, InterruptedException { testCommand("voting_view", diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/watch/WatchManagerTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/watch/WatchManagerTest.java index 51bbb94d88a..94307ee7d74 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/server/watch/WatchManagerTest.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/server/watch/WatchManagerTest.java @@ -21,6 +21,8 @@ import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; import java.io.IOException; import java.util.ArrayList; import java.util.HashSet; @@ -29,6 +31,7 @@ import java.util.Random; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Stream; import org.apache.zookeeper.Watcher; @@ -82,6 +85,134 @@ public DumbWatcher createOrGetWatcher(int watcherId) { return watchers.get(watcherId); } + @Test + public void testGetWatchRegistrationsExpandsModesAndFilters() { + WatchManager manager = new WatchManager(); + ServerCnxn active = mock(ServerCnxn.class); + when(active.getSessionId()).thenReturn(0x11L); + when(active.isStale()).thenReturn(false); + + ServerCnxn otherSession = mock(ServerCnxn.class); + when(otherSession.getSessionId()).thenReturn(0x22L); + when(otherSession.isStale()).thenReturn(false); + + AtomicBoolean stale = new AtomicBoolean(false); + ServerCnxn staleConnection = mock(ServerCnxn.class); + when(staleConnection.getSessionId()).thenReturn(0x11L); + when(staleConnection.isStale()).thenAnswer(invocation -> stale.get()); + + manager.addWatch("/node1", active, WatcherMode.STANDARD); + manager.addWatch("/node1", active, WatcherMode.PERSISTENT); + manager.addWatch("/node1", active, WatcherMode.PERSISTENT_RECURSIVE); + manager.addWatch("/node2", active, WatcherMode.STANDARD); + manager.addWatch("/node1", otherSession, WatcherMode.STANDARD); + manager.addWatch("/node1", staleConnection, WatcherMode.STANDARD); + stale.set(true); + + List registrations = manager.getWatchRegistrations( + "/node1", + java.util.Collections.singleton(0x11L), + 10); + + assertEquals(3, registrations.size()); + Set modes = new HashSet<>(); + for (WatchRegistration registration : registrations) { + assertEquals("/node1", registration.getPath()); + assertEquals(0x11L, registration.getSessionId()); + modes.add(registration.getWatcherMode()); + } + assertEquals( + new HashSet<>(java.util.Arrays.asList( + WatcherMode.STANDARD, + WatcherMode.PERSISTENT, + WatcherMode.PERSISTENT_RECURSIVE)), + modes); + } + + @Test + public void testGetWatchRegistrationsStopsAtMaxResults() { + WatchManager manager = new WatchManager(); + ServerCnxn connection = mock(ServerCnxn.class); + when(connection.getSessionId()).thenReturn(0x33L); + when(connection.isStale()).thenReturn(false); + manager.addWatch("/node1", connection); + manager.addWatch("/node2", connection); + + List registrations = manager.getWatchRegistrations(null, null, 1); + + assertEquals(1, registrations.size()); + } + + @Test + public void testGetWatchRegistrationsDeduplicatesSessionRegistrations() { + WatchManager manager = new WatchManager(); + ServerCnxn firstConnection = mock(ServerCnxn.class); + when(firstConnection.getSessionId()).thenReturn(0x34L); + when(firstConnection.isStale()).thenReturn(false); + ServerCnxn secondConnection = mock(ServerCnxn.class); + when(secondConnection.getSessionId()).thenReturn(0x34L); + when(secondConnection.isStale()).thenReturn(false); + manager.addWatch("/node1", firstConnection); + manager.addWatch("/node1", secondConnection); + + List registrations = manager.getWatchRegistrations(null, null, 10); + + assertEquals(1, registrations.size()); + } + + @Test + public void testOptimizedGetWatchRegistrationsFiltersAndStopsEarly() { + WatchManagerOptimized manager = new WatchManagerOptimized(); + try { + ServerCnxn active = mock(ServerCnxn.class); + when(active.getSessionId()).thenReturn(0x44L); + when(active.isStale()).thenReturn(false); + + AtomicBoolean stale = new AtomicBoolean(false); + ServerCnxn staleConnection = mock(ServerCnxn.class); + when(staleConnection.getSessionId()).thenReturn(0x44L); + when(staleConnection.isStale()).thenAnswer(invocation -> stale.get()); + + manager.addWatch("/node1", active); + manager.addWatch("/node2", active); + manager.addWatch("/node1", staleConnection); + stale.set(true); + + List registrations = manager.getWatchRegistrations( + "/node1", + java.util.Collections.singleton(0x44L), + 1); + + assertEquals(1, registrations.size()); + assertEquals("/node1", registrations.get(0).getPath()); + assertEquals(0x44L, registrations.get(0).getSessionId()); + assertEquals(WatcherMode.STANDARD, registrations.get(0).getWatcherMode()); + } finally { + manager.shutdown(); + } + } + + @Test + public void testOptimizedGetWatchRegistrationsDeduplicatesSessionRegistrations() { + WatchManagerOptimized manager = new WatchManagerOptimized(); + try { + ServerCnxn firstConnection = mock(ServerCnxn.class); + when(firstConnection.getSessionId()).thenReturn(0x45L); + when(firstConnection.isStale()).thenReturn(false); + ServerCnxn secondConnection = mock(ServerCnxn.class); + when(secondConnection.getSessionId()).thenReturn(0x45L); + when(secondConnection.isStale()).thenReturn(false); + manager.addWatch("/node1", firstConnection); + manager.addWatch("/node1", secondConnection); + + List registrations = manager.getWatchRegistrations(null, null, 10); + + assertEquals(1, registrations.size()); + } finally { + manager.shutdown(); + } + } + public class AddWatcherWorker extends Thread { private final IWatchManager manager; diff --git a/zookeeper-website/app/pages/_docs/docs/_mdx/admin-ops/administrators-guide/commands.mdx b/zookeeper-website/app/pages/_docs/docs/_mdx/admin-ops/administrators-guide/commands.mdx index 001875aa26d..9edc67ee897 100644 --- a/zookeeper-website/app/pages/_docs/docs/_mdx/admin-ops/administrators-guide/commands.mdx +++ b/zookeeper-website/app/pages/_docs/docs/_mdx/admin-ops/administrators-guide/commands.mdx @@ -503,6 +503,23 @@ Available commands include: Summarized watch information. Returns "num_total_watches", "num_paths", and "num_connections". +- _watch_details/wchd_ : + Returns Watch registration details for the current ZooKeeper server and + associates each result with its active client connection. + Optional query parameters are "path" (exact znode path), "session_id" + (decimal or hexadecimal), "client_ip" (exact address), and "limit" + (1 through 1000, default 100). Multiple filters are combined with AND. + Returns "server_id", "returned_count", "truncated", and "watches". + Each Watch entry contains "path", "session_id", "client_ip", "client_port", + "watch_kind", "watch_mode", "connection_established_at", + "session_timeout_ms", and "secure". + "watch_kind" is an array ordered as "data", then "children". Entries with + the same path, session, and mode are merged, so Standard Data and Children + Watches on the same znode are returned as `["data", "children"]`. Persistent + and Persistent Recursive Watches also cover both kinds. + Results are local to the queried server, are not ordered, and may reflect + concurrent Watch or connection changes during the query. + - _zabstate_ : The current phase of Zab protocol that peer is running and whether it is a voting member.