diff options
Diffstat (limited to 'clustercontroller-core/src/main/java/com/yahoo')
24 files changed, 231 insertions, 193 deletions
diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterInfo.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterInfo.java index c25dd5a5965..2cfaf64fe83 100644 --- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterInfo.java +++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterInfo.java @@ -4,6 +4,7 @@ package com.yahoo.vespa.clustercontroller.core; import com.yahoo.vdslib.distribution.ConfiguredNode; import com.yahoo.vdslib.distribution.Distribution; import com.yahoo.vdslib.state.Node; +import com.yahoo.vespa.clustercontroller.core.listeners.NodeListener; import java.util.Collection; import java.util.Collections; @@ -40,11 +41,11 @@ public class ClusterInfo { /** Returns information about the given node id, or null if this node does not exist */ public NodeInfo getNodeInfo(Node node) { return allNodeInfo.get(node); } - Collection<DistributorNodeInfo> getDistributorNodeInfo() { return Collections.unmodifiableCollection(distributorNodeInfo.values()); } + Collection<DistributorNodeInfo> getDistributorNodeInfos() { return Collections.unmodifiableCollection(distributorNodeInfo.values()); } - Collection<StorageNodeInfo> getStorageNodeInfo() { return Collections.unmodifiableCollection(storageNodeInfo.values()); } + Collection<StorageNodeInfo> getStorageNodeInfos() { return Collections.unmodifiableCollection(storageNodeInfo.values()); } - Collection<NodeInfo> getAllNodeInfo() { return Collections.unmodifiableCollection(allNodeInfo.values()); } + Collection<NodeInfo> getAllNodeInfos() { return Collections.unmodifiableCollection(allNodeInfo.values()); } /** Returns the configured nodes of this as a read-only map indexed on node index (distribution key) */ Map<Integer, ConfiguredNode> getConfiguredNodes() { return Collections.unmodifiableMap(nodes); } @@ -52,15 +53,23 @@ public class ClusterInfo { boolean hasConfiguredNode(int index) { return nodes.containsKey(index); } /** Sets the nodes which belongs to this cluster */ - void setNodes(Collection<ConfiguredNode> newNodes, ContentCluster owner, Distribution distribution) { + void setNodes(Collection<ConfiguredNode> newNodes, ContentCluster owner, + Distribution distribution, NodeListener nodeListener) { // Remove info for removed nodes Set<ConfiguredNode> newNodesSet = new HashSet<>(newNodes); for (ConfiguredNode existingNode : this.nodes.values()) { if ( ! newNodesSet.contains(existingNode)) { - Node existingStorageNode = storageNodeInfo.remove(existingNode.index()).getNode(); - Node existingDistributorNode = distributorNodeInfo.remove(existingNode.index()).getNode(); - allNodeInfo.remove(existingDistributorNode); - allNodeInfo.remove(existingStorageNode); + { + Node existingStorageNode = storageNodeInfo.remove(existingNode.index()).getNode(); + allNodeInfo.remove(existingStorageNode); + nodeListener.handleRemovedNode(existingStorageNode); + } + + { + Node existingDistributorNode = distributorNodeInfo.remove(existingNode.index()).getNode(); + allNodeInfo.remove(existingDistributorNode); + nodeListener.handleRemovedNode(existingDistributorNode); + } } } diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStateGenerator.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStateGenerator.java index f984c3cb3a2..75c6dbe6cec 100644 --- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStateGenerator.java +++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStateGenerator.java @@ -129,7 +129,7 @@ public class ClusterStateGenerator { final ClusterState workingState = ClusterState.emptyState(); final Map<Node, NodeStateReason> nodeStateReasons = new HashMap<>(); - for (final NodeInfo nodeInfo : cluster.getNodeInfo()) { + for (final NodeInfo nodeInfo : cluster.getNodeInfos()) { final NodeState nodeState = computeEffectiveNodeState(nodeInfo, params, nodeStateReasons); workingState.setNodeState(nodeInfo.getNode(), nodeState); } diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ContentCluster.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ContentCluster.java index fabc4999fe5..f2a4a9736c3 100644 --- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ContentCluster.java +++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ContentCluster.java @@ -10,6 +10,7 @@ import com.yahoo.vdslib.state.Node; import com.yahoo.vdslib.state.NodeState; import com.yahoo.vdslib.state.NodeType; import com.yahoo.vdslib.state.State; +import com.yahoo.vespa.clustercontroller.core.listeners.NodeListener; import com.yahoo.vespa.clustercontroller.core.status.statuspage.HtmlTable; import com.yahoo.vespa.clustercontroller.core.status.statuspage.VdsClusterHtmlRenderer; import com.yahoo.vespa.clustercontroller.utils.staterestapi.requests.SetUnitStateRequest; @@ -42,7 +43,7 @@ public class ContentCluster { if (configuredNodes == null) throw new IllegalArgumentException("Nodes must be set"); this.clusterName = clusterName; this.distribution = distribution; - setNodes(configuredNodes); + setNodes(configuredNodes, new NodeListener() {}); } // TODO move out, this doesn't belong in a domain model class @@ -104,14 +105,14 @@ public class ContentCluster { public void setDistribution(Distribution distribution) { this.distribution = distribution; - for (NodeInfo info : clusterInfo.getAllNodeInfo()) { + for (NodeInfo info : clusterInfo.getAllNodeInfos()) { info.setGroup(distribution); } } /** Sets the configured nodes of this cluster */ - public final void setNodes(Collection<ConfiguredNode> configuredNodes) { - clusterInfo.setNodes(configuredNodes, this, distribution); + public final void setNodes(Collection<ConfiguredNode> configuredNodes, NodeListener nodeListener) { + clusterInfo.setNodes(configuredNodes, this, distribution, nodeListener); } public void setStartTimestamp(Node n, long startTimestamp) { @@ -128,7 +129,7 @@ public class ContentCluster { } public void clearStates() { - for (NodeInfo info : clusterInfo.getAllNodeInfo()) { + for (NodeInfo info : clusterInfo.getAllNodeInfos()) { info.setReportedState(null, 0); } } @@ -145,8 +146,8 @@ public class ContentCluster { return clusterInfo.getConfiguredNodes(); } - public Collection<NodeInfo> getNodeInfo() { - return Collections.unmodifiableCollection(clusterInfo.getAllNodeInfo()); + public Collection<NodeInfo> getNodeInfos() { + return Collections.unmodifiableCollection(clusterInfo.getAllNodeInfos()); } public ClusterInfo clusterInfo() { return clusterInfo; } @@ -158,7 +159,7 @@ public class ContentCluster { public String toString() { StringBuilder sb = new StringBuilder(); sb.append("ContentCluster(").append(clusterName).append(") {"); - for (NodeInfo node : clusterInfo.getAllNodeInfo()) { + for (NodeInfo node : clusterInfo.getAllNodeInfos()) { sb.append("\n ").append(node); } sb.append("\n}"); @@ -197,14 +198,14 @@ public class ContentCluster { switch (state) { case MAINTENANCE: // Orchestrator's ALLOWED_TO_BE_DOWN case DOWN: // Orchestrator's PERMANENTLY_DOWN - return clusterInfo.getStorageNodeInfo().stream() - .filter(storageNodeInfo -> { + return clusterInfo.getStorageNodeInfos().stream() + .filter(storageNodeInfo -> { NodeState userWantedState = storageNodeInfo.getUserWantedState(); return userWantedState.getState() == state && Objects.equals(userWantedState.getDescription(), ORCHESTRATOR_RESERVED_DESCRIPTION); }) - .map(NodeInfo::getNodeIndex) - .collect(Collectors.toList()); + .map(NodeInfo::getNodeIndex) + .collect(Collectors.toList()); default: // Note: There is no trace left if the Orchestrator set the state to UP, so that's handled // like any other state: diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/FleetController.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/FleetController.java index 3137dfff606..7f385c6077c 100644 --- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/FleetController.java +++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/FleetController.java @@ -12,8 +12,8 @@ import com.yahoo.vdslib.state.State; import com.yahoo.vespa.clustercontroller.core.database.DatabaseHandler; import com.yahoo.vespa.clustercontroller.core.database.ZooKeeperDatabaseFactory; import com.yahoo.vespa.clustercontroller.core.hostinfo.HostInfo; -import com.yahoo.vespa.clustercontroller.core.listeners.NodeAddedOrRemovedListener; -import com.yahoo.vespa.clustercontroller.core.listeners.NodeStateOrHostInfoChangeHandler; +import com.yahoo.vespa.clustercontroller.core.listeners.SlobrokListener; +import com.yahoo.vespa.clustercontroller.core.listeners.NodeListener; import com.yahoo.vespa.clustercontroller.core.listeners.SystemStateListener; import com.yahoo.vespa.clustercontroller.core.rpc.RPCCommunicator; import com.yahoo.vespa.clustercontroller.core.rpc.RpcServer; @@ -47,7 +47,7 @@ import java.util.logging.Logger; import java.util.stream.Collectors; import java.util.stream.Stream; -public class FleetController implements NodeStateOrHostInfoChangeHandler, NodeAddedOrRemovedListener, SystemStateListener, +public class FleetController implements NodeListener, SlobrokListener, SystemStateListener, Runnable, RemoteClusterControllerTaskScheduler { private static final Logger logger = Logger.getLogger(FleetController.class.getName()); @@ -332,6 +332,13 @@ public class FleetController implements NodeStateOrHostInfoChangeHandler, NodeAd } @Override + public void handleRemovedNode(Node node) { + verifyInControllerThread(); + // Prune orphaned wanted states + wantedStateChanged = true; + } + + @Override public void handleUpdatedHostInfo(NodeInfo nodeInfo, HostInfo newHostInfo) { verifyInControllerThread(); triggerBundleRecomputationIfResourceExhaustionStateChanged(nodeInfo, newHostInfo); @@ -380,7 +387,7 @@ public class FleetController implements NodeStateOrHostInfoChangeHandler, NodeAd ClusterState baselineState = stateBundle.getBaselineClusterState(); newStates.add(stateBundle); metricUpdater.updateClusterStateMetrics(cluster, baselineState, - ResourceUsageStats.calculateFrom(cluster.getNodeInfo(), options.clusterFeedBlockLimit, stateBundle.getFeedBlock())); + ResourceUsageStats.calculateFrom(cluster.getNodeInfos(), options.clusterFeedBlockLimit, stateBundle.getFeedBlock())); lastMetricUpdateCycleCount = cycleCount; systemStateBroadcaster.handleNewClusterStates(stateBundle); // Iff master, always store new version in ZooKeeper _before_ publishing to any @@ -399,7 +406,7 @@ public class FleetController implements NodeStateOrHostInfoChangeHandler, NodeAd ClusterStateBundle stateBundle = stateVersionTracker.getVersionedClusterStateBundle(); ClusterState baselineState = stateBundle.getBaselineClusterState(); metricUpdater.updateClusterStateMetrics(cluster, baselineState, - ResourceUsageStats.calculateFrom(cluster.getNodeInfo(), options.clusterFeedBlockLimit, stateBundle.getFeedBlock())); + ResourceUsageStats.calculateFrom(cluster.getNodeInfos(), options.clusterFeedBlockLimit, stateBundle.getFeedBlock())); lastMetricUpdateCycleCount = cycleCount; return true; } else { @@ -511,7 +518,7 @@ public class FleetController implements NodeStateOrHostInfoChangeHandler, NodeAd eventLog.setMaxSize(options.eventLogMaxSize, options.eventNodeLogMaxSize); cluster.setPollingFrequency(options.statePollingFrequency); cluster.setDistribution(options.storageDistribution); - cluster.setNodes(options.nodes); + cluster.setNodes(options.nodes, databaseContext.getNodeStateUpdateListener()); database.setZooKeeperAddress(options.zooKeeperServerAddress, databaseContext); database.setZooKeeperSessionTimeout(options.zooKeeperSessionTimeout, databaseContext); stateGatherer.setMaxSlobrokDisconnectGracePeriod(options.maxSlobrokDisconnectGracePeriod); @@ -790,8 +797,8 @@ public class FleetController implements NodeStateOrHostInfoChangeHandler, NodeAd @Override public boolean inMasterMoratorium() { return inMasterMoratorium; } }; - context.nodeStateOrHostInfoChangeHandler = this; - context.nodeAddedOrRemovedListener = this; + context.nodeListener = this; + context.slobrokListener = this; return context; } @@ -806,10 +813,10 @@ public class FleetController implements NodeStateOrHostInfoChangeHandler, NodeAd if (bundle == null) { return List.of(); } - return cluster.getNodeInfo().stream(). - filter(n -> effectiveActivatedStateVersion(n, bundle) < version). - map(NodeInfo::getNode). - collect(Collectors.toList()); + return cluster.getNodeInfos().stream(). + filter(n -> effectiveActivatedStateVersion(n, bundle) < version). + map(NodeInfo::getNode). + collect(Collectors.toList()); } private static <E> String stringifyListWithLimits(List<E> list, int limit) { @@ -939,7 +946,7 @@ public class FleetController implements NodeStateOrHostInfoChangeHandler, NodeAd .stateDeriver(createBucketSpaceStateDeriver()) .deferredActivation(options.enableTwoPhaseClusterStateActivation) .feedBlock(createResourceExhaustionCalculator() - .inferContentClusterFeedBlockOrNull(cluster.getNodeInfo())) + .inferContentClusterFeedBlockOrNull(cluster.getNodeInfos())) .deriveAndBuild(); stateVersionTracker.updateLatestCandidateStateBundle(candidateBundle); invokeCandidateStateListeners(candidateBundle); @@ -1095,7 +1102,7 @@ public class FleetController implements NodeStateOrHostInfoChangeHandler, NodeAd didWork = true; } if (wantedStateChanged) { - database.saveWantedStates(databaseContext); + didWork |= database.saveWantedStates(databaseContext); wantedStateChanged = false; } } else { @@ -1150,9 +1157,9 @@ public class FleetController implements NodeStateOrHostInfoChangeHandler, NodeAd @Override public FleetController getFleetController() { return FleetController.this; } @Override - public NodeAddedOrRemovedListener getNodeAddedOrRemovedListener() { return FleetController.this; } + public SlobrokListener getNodeAddedOrRemovedListener() { return FleetController.this; } @Override - public NodeStateOrHostInfoChangeHandler getNodeStateUpdateListener() { return FleetController.this; } + public NodeListener getNodeStateUpdateListener() { return FleetController.this; } }; public void waitForCompleteCycle(long timeoutMS) { @@ -1183,7 +1190,7 @@ public class FleetController implements NodeStateOrHostInfoChangeHandler, NodeAd synchronized (monitor) { while (true) { int ackedNodes = 0; - for (NodeInfo node : cluster.getNodeInfo()) { + for (NodeInfo node : cluster.getNodeInfos()) { if (node.getClusterStateVersionBundleAcknowledged() >= version) { ++ackedNodes; } @@ -1206,7 +1213,7 @@ public class FleetController implements NodeStateOrHostInfoChangeHandler, NodeAd synchronized (monitor) { while (true) { int distCount = 0, storCount = 0; - for (NodeInfo info : cluster.getNodeInfo()) { + for (NodeInfo info : cluster.getNodeInfos()) { if (!info.isRpcAddressOutdated()) { if (info.isDistributor()) ++distCount; else ++storCount; diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/NodeLookup.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/NodeLookup.java index cdf8b24e72d..882ae8894fa 100644 --- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/NodeLookup.java +++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/NodeLookup.java @@ -1,7 +1,7 @@ // Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. package com.yahoo.vespa.clustercontroller.core; -import com.yahoo.vespa.clustercontroller.core.listeners.NodeAddedOrRemovedListener; +import com.yahoo.vespa.clustercontroller.core.listeners.SlobrokListener; /** * Interface for a node lookup service, such as slobrok, config, or tier controller. @@ -10,7 +10,7 @@ public interface NodeLookup { void shutdown(); - boolean updateCluster(ContentCluster cluster, NodeAddedOrRemovedListener listener); + boolean updateCluster(ContentCluster cluster, SlobrokListener listener); /** * Returns whether the lookup instance has been able to bootstrap itself with information about nodes. diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/NodeStateChangeChecker.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/NodeStateChangeChecker.java index eb0368749f0..3ed03e94fda 100644 --- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/NodeStateChangeChecker.java +++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/NodeStateChangeChecker.java @@ -359,7 +359,7 @@ public class NodeStateChangeChecker { // This method verifies both storage nodes and distributors are up (or retired). // The complicated part is making a summary error message. - for (NodeInfo storageNodeInfo : clusterInfo.getStorageNodeInfo()) { + for (NodeInfo storageNodeInfo : clusterInfo.getStorageNodeInfos()) { State wantedState = storageNodeInfo.getUserWantedState().getState(); if (wantedState != State.UP && wantedState != State.RETIRED) { return Result.createDisallowed("Another storage node wants state " + @@ -373,7 +373,7 @@ public class NodeStateChangeChecker { } } - for (NodeInfo distributorNodeInfo : clusterInfo.getDistributorNodeInfo()) { + for (NodeInfo distributorNodeInfo : clusterInfo.getDistributorNodeInfos()) { State wantedState = distributorNodeInfo.getUserWantedState().getState(); if (wantedState != State.UP && wantedState != State.RETIRED) { return Result.createDisallowed("Another distributor wants state " + wantedState.toString().toUpperCase() + @@ -418,10 +418,10 @@ public class NodeStateChangeChecker { * @param clusterStateVersion the cluster state we expect distributors to have */ private Result checkDistributors(Node node, int clusterStateVersion) { - if (clusterInfo.getDistributorNodeInfo().isEmpty()) { + if (clusterInfo.getDistributorNodeInfos().isEmpty()) { return Result.createDisallowed("Not aware of any distributors, probably not safe to upgrade?"); } - for (DistributorNodeInfo distributorNodeInfo : clusterInfo.getDistributorNodeInfo()) { + for (DistributorNodeInfo distributorNodeInfo : clusterInfo.getDistributorNodeInfos()) { Integer distributorClusterStateVersion = distributorNodeInfo.getHostInfo().getClusterStateVersionOrNull(); if (distributorClusterStateVersion == null) { return Result.createDisallowed("Distributor node " + distributorNodeInfo.getNodeIndex() diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/NodeStateGatherer.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/NodeStateGatherer.java index e3fe371c05e..69e97de84f9 100644 --- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/NodeStateGatherer.java +++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/NodeStateGatherer.java @@ -7,7 +7,7 @@ import java.util.logging.Level; import com.yahoo.vdslib.state.NodeState; import com.yahoo.vdslib.state.State; import com.yahoo.vespa.clustercontroller.core.hostinfo.HostInfo; -import com.yahoo.vespa.clustercontroller.core.listeners.NodeStateOrHostInfoChangeHandler; +import com.yahoo.vespa.clustercontroller.core.listeners.NodeListener; import java.util.LinkedList; import java.util.List; @@ -54,10 +54,10 @@ public class NodeStateGatherer { * Sends state requests to nodes that does not have one pending and is due * for another attempt. */ - public boolean sendMessages(ContentCluster cluster, Communicator communicator, NodeStateOrHostInfoChangeHandler listener) { + public boolean sendMessages(ContentCluster cluster, Communicator communicator, NodeListener listener) { boolean sentAnyMessages = false; long currentTime = timer.getCurrentTimeInMillis(); - for (NodeInfo info : cluster.getNodeInfo()) { + for (NodeInfo info : cluster.getNodeInfos()) { Long requestTime = info.getLatestNodeStateRequestTime(); if (requestTime != null && (currentTime - requestTime < nodeStateRequestTimeoutMS)) continue; // pending request @@ -93,7 +93,7 @@ public class NodeStateGatherer { } /** Reads replies to get node state requests and create events. */ - public boolean processResponses(NodeStateOrHostInfoChangeHandler listener) { + public boolean processResponses(NodeListener listener) { boolean processedAnyResponses = false; long currentTime = timer.getCurrentTimeInMillis(); synchronized(monitor) { diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/RemoteClusterControllerTask.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/RemoteClusterControllerTask.java index 949ad6f56a2..3c2143818e3 100644 --- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/RemoteClusterControllerTask.java +++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/RemoteClusterControllerTask.java @@ -2,8 +2,8 @@ package com.yahoo.vespa.clustercontroller.core; import com.yahoo.vdslib.state.ClusterState; -import com.yahoo.vespa.clustercontroller.core.listeners.NodeAddedOrRemovedListener; -import com.yahoo.vespa.clustercontroller.core.listeners.NodeStateOrHostInfoChangeHandler; +import com.yahoo.vespa.clustercontroller.core.listeners.SlobrokListener; +import com.yahoo.vespa.clustercontroller.core.listeners.NodeListener; import java.time.Instant; import java.util.Optional; @@ -15,8 +15,8 @@ public abstract class RemoteClusterControllerTask { public ClusterState currentConsolidatedState; public ClusterStateBundle publishedClusterStateBundle; public MasterInterface masterInfo; - public NodeStateOrHostInfoChangeHandler nodeStateOrHostInfoChangeHandler; - public NodeAddedOrRemovedListener nodeAddedOrRemovedListener; + public NodeListener nodeListener; + public SlobrokListener slobrokListener; } private final Object monitor = new Object(); diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/StateChangeHandler.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/StateChangeHandler.java index 46fafddfade..4c832592422 100644 --- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/StateChangeHandler.java +++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/StateChangeHandler.java @@ -8,7 +8,7 @@ import com.yahoo.vdslib.state.NodeState; import com.yahoo.vdslib.state.NodeType; import com.yahoo.vdslib.state.State; import com.yahoo.vespa.clustercontroller.core.database.DatabaseHandler; -import com.yahoo.vespa.clustercontroller.core.listeners.NodeStateOrHostInfoChangeHandler; +import com.yahoo.vespa.clustercontroller.core.listeners.NodeListener; import java.util.Map; import java.util.Set; @@ -113,7 +113,7 @@ public class StateChangeHandler { public void handleNewReportedNodeState(final ClusterState currentClusterState, final NodeInfo node, final NodeState reportedState, - final NodeStateOrHostInfoChangeHandler nodeListener) + final NodeListener nodeListener) { final NodeState currentState = currentClusterState.getNodeState(node.getNode()); final Level level = (currentState.equals(reportedState) && node.getVersion() == 0) ? Level.FINEST : Level.FINE; @@ -164,7 +164,7 @@ public class StateChangeHandler { public void handleMissingNode(final ClusterState currentClusterState, final NodeInfo node, - final NodeStateOrHostInfoChangeHandler nodeListener) { + final NodeListener nodeListener) { final long timeNow = timer.getCurrentTimeInMillis(); if (node.getLatestNodeStateRequestTime() != null) { @@ -241,12 +241,12 @@ public class StateChangeHandler { // `--> this will require adding more event edges and premature crash handling to it. Which is fine. public boolean watchTimers(final ContentCluster cluster, final ClusterState currentClusterState, - final NodeStateOrHostInfoChangeHandler nodeListener) + final NodeListener nodeListener) { boolean triggeredAnyTimers = false; final long currentTime = timer.getCurrentTimeInMillis(); - for(NodeInfo node : cluster.getNodeInfo()) { + for(NodeInfo node : cluster.getNodeInfos()) { triggeredAnyTimers |= handleTimeDependentOpsForNode(currentClusterState, nodeListener, currentTime, node); } @@ -257,7 +257,7 @@ public class StateChangeHandler { } private boolean handleTimeDependentOpsForNode(final ClusterState currentClusterState, - final NodeStateOrHostInfoChangeHandler nodeListener, + final NodeListener nodeListener, final long currentTime, final NodeInfo node) { @@ -334,7 +334,7 @@ public class StateChangeHandler { } private boolean reportDownIfOutdatedSlobrokNode(ClusterState currentClusterState, - NodeStateOrHostInfoChangeHandler nodeListener, + NodeListener nodeListener, long currentTime, NodeInfo node, NodeState lastReportedState) @@ -379,7 +379,7 @@ public class StateChangeHandler { private void updateNodeInfoFromReportedState(final NodeInfo node, final NodeState currentState, final NodeState reportedState, - final NodeStateOrHostInfoChangeHandler nodeListener) { + final NodeListener nodeListener) { final long timeNow = timer.getCurrentTimeInMillis(); log.log(Level.FINE, () -> String.format("Finding new cluster state entry for %s switching state %s", node, currentState.getTextualDifference(reportedState))); @@ -400,7 +400,7 @@ public class StateChangeHandler { private void markNodeUnstableIfDownEdgeDuringInit(final NodeInfo node, final NodeState currentState, final NodeState reportedState, - final NodeStateOrHostInfoChangeHandler nodeListener, + final NodeListener nodeListener, final long timeNow) { if (currentState.getState().equals(State.INITIALIZING) && reportedState.getState().oneOf("ds") @@ -419,7 +419,7 @@ public class StateChangeHandler { private boolean handleImplicitCrashEdgeFromReverseInitProgress(final NodeInfo node, final NodeState currentState, final NodeState reportedState, - final NodeStateOrHostInfoChangeHandler nodeListener, + final NodeListener nodeListener, final long timeNow) { if (currentState.getState().equals(State.INITIALIZING) && (reportedState.getState().equals(State.INITIALIZING) && reportedState.getInitProgress() < currentState.getInitProgress())) @@ -438,7 +438,7 @@ public class StateChangeHandler { } private boolean handleReportedNodeCrashEdge(NodeInfo node, NodeState currentState, - NodeState reportedState, NodeStateOrHostInfoChangeHandler nodeListener, + NodeState reportedState, NodeListener nodeListener, long timeNow) { if (nodeUpToDownEdge(node, currentState, reportedState)) { node.setTransitionTime(timeNow); @@ -467,7 +467,7 @@ public class StateChangeHandler { && (node.getWantedState().getState().equals(State.RETIRED) || !reportedState.getState().equals(State.INITIALIZING)); } - private boolean handlePrematureCrash(NodeInfo node, NodeStateOrHostInfoChangeHandler changeListener) { + private boolean handlePrematureCrash(NodeInfo node, NodeListener changeListener) { node.setPrematureCrashCount(node.getPrematureCrashCount() + 1); if (disableUnstableNodes && node.getPrematureCrashCount() > maxPrematureCrashes) { NodeState wantedState = new NodeState(node.getNode().getType(), State.DOWN) diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/SystemStateBroadcaster.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/SystemStateBroadcaster.java index d061f7edbea..2359e4d8389 100644 --- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/SystemStateBroadcaster.java +++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/SystemStateBroadcaster.java @@ -188,18 +188,18 @@ public class SystemStateBroadcaster { } private List<NodeInfo> resolveStateVersionSendSet(DatabaseHandler.DatabaseContext dbContext) { - return dbContext.getCluster().getNodeInfo().stream() - .filter(this::nodeNeedsClusterStateBundle) - .filter(node -> !newestStateBundleAlreadySentToNode(node)) - .collect(Collectors.toList()); + return dbContext.getCluster().getNodeInfos().stream() + .filter(this::nodeNeedsClusterStateBundle) + .filter(node -> !newestStateBundleAlreadySentToNode(node)) + .collect(Collectors.toList()); } // Precondition: no nodes in the cluster need to receive the current cluster state version bundle private List<NodeInfo> resolveStateActivationSendSet(DatabaseHandler.DatabaseContext dbContext) { - return dbContext.getCluster().getNodeInfo().stream() - .filter(this::nodeNeedsClusterStateActivation) - .filter(node -> !newestStateActivationAlreadySentToNode(node)) - .collect(Collectors.toList()); + return dbContext.getCluster().getNodeInfos().stream() + .filter(this::nodeNeedsClusterStateActivation) + .filter(node -> !newestStateActivationAlreadySentToNode(node)) + .collect(Collectors.toList()); } private boolean newestStateBundleAlreadySentToNode(NodeInfo node) { @@ -222,9 +222,9 @@ public class SystemStateBroadcaster { return; // Nothing to do for the current state } final int currentStateVersion = clusterStateBundle.getVersion(); - boolean anyDistributorsNeedStateBundle = dbContext.getCluster().getNodeInfo().stream() - .filter(NodeInfo::isDistributor) - .anyMatch(this::nodeNeedsClusterStateBundle); + boolean anyDistributorsNeedStateBundle = dbContext.getCluster().getNodeInfos().stream() + .filter(NodeInfo::isDistributor) + .anyMatch(this::nodeNeedsClusterStateBundle); if (!anyDistributorsNeedStateBundle && (currentStateVersion > lastStateVersionBundleAcked)) { markCurrentClusterStateBundleAsReceivedByAllDistributors(); @@ -243,9 +243,9 @@ public class SystemStateBroadcaster { return; } - boolean anyDistributorsNeedActivation = dbContext.getCluster().getNodeInfo().stream() - .filter(NodeInfo::isDistributor) - .anyMatch(this::nodeNeedsClusterStateActivation); + boolean anyDistributorsNeedActivation = dbContext.getCluster().getNodeInfos().stream() + .filter(NodeInfo::isDistributor) + .anyMatch(this::nodeNeedsClusterStateActivation); if (!anyDistributorsNeedActivation && (currentStateVersion > lastClusterStateVersionConverged)) { markCurrentClusterStateAsConverged(database, dbContext, fleetController); @@ -352,7 +352,7 @@ public class SystemStateBroadcaster { private static ClusterState buildModifiedClusterState(ClusterState sourceState, DatabaseHandler.DatabaseContext dbContext) { ClusterState newState = sourceState.clone(); - for (NodeInfo n : dbContext.getCluster().getNodeInfo()) { + for (NodeInfo n : dbContext.getCluster().getNodeInfos()) { NodeState ns = newState.getNodeState(n.getNode()); if (!n.isDistributor() && ns.getStartTimestamp() == 0) { ns.setStartTimestamp(n.getStartTimestamp()); diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/database/DatabaseFactory.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/database/DatabaseFactory.java index 45a416fade4..31f6bbfe932 100644 --- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/database/DatabaseFactory.java +++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/database/DatabaseFactory.java @@ -1,8 +1,6 @@ // Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. package com.yahoo.vespa.clustercontroller.core.database; -import com.yahoo.vespa.clustercontroller.core.ContentCluster; - /** * Database factory to enable test mocking of DB features. In practice, this * will always be {@link ZooKeeperDatabase} due to rather heavy ZK feature @@ -11,14 +9,10 @@ import com.yahoo.vespa.clustercontroller.core.ContentCluster; public interface DatabaseFactory { class Params { - ContentCluster cluster; - int nodeIndex; String dbAddress; int dbSessionTimeout; Database.DatabaseListener listener; - Params cluster(ContentCluster c) { this.cluster = c; return this; } - Params nodeIndex(int i) { this.nodeIndex = i; return this; } Params databaseAddress(String address) { this.dbAddress = address; return this; } Params databaseSessionTimeout(int timeout) { this.dbSessionTimeout = timeout; return this; } Params databaseListener(Database.DatabaseListener listener) { this.listener = listener; return this; } diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/database/DatabaseHandler.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/database/DatabaseHandler.java index 0c6c773a9bc..01b8ed48c80 100644 --- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/database/DatabaseHandler.java +++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/database/DatabaseHandler.java @@ -10,8 +10,8 @@ import com.yahoo.vespa.clustercontroller.core.FleetControllerContext; import com.yahoo.vespa.clustercontroller.core.FleetController; import com.yahoo.vespa.clustercontroller.core.NodeInfo; import com.yahoo.vespa.clustercontroller.core.Timer; -import com.yahoo.vespa.clustercontroller.core.listeners.NodeAddedOrRemovedListener; -import com.yahoo.vespa.clustercontroller.core.listeners.NodeStateOrHostInfoChangeHandler; +import com.yahoo.vespa.clustercontroller.core.listeners.SlobrokListener; +import com.yahoo.vespa.clustercontroller.core.listeners.NodeListener; import org.apache.zookeeper.KeeperException; import java.io.PrintWriter; @@ -32,8 +32,8 @@ public class DatabaseHandler { public interface DatabaseContext { ContentCluster getCluster(); FleetController getFleetController(); - NodeAddedOrRemovedListener getNodeAddedOrRemovedListener(); - NodeStateOrHostInfoChangeHandler getNodeStateUpdateListener(); + SlobrokListener getNodeAddedOrRemovedListener(); + NodeListener getNodeStateUpdateListener(); } private static class Data { @@ -191,9 +191,8 @@ public class DatabaseHandler { // being called, but after receiving a database loss event. clearSessionMetaData(false); fleetControllerContext.log(logger, Level.INFO, "Setting up new ZooKeeper session at " + zooKeeperAddress); - DatabaseFactory.Params params = new DatabaseFactory.Params() - .cluster(cluster) - .nodeIndex(fleetControllerContext.id().index()) + DatabaseFactory.Params params = new DatabaseFactory + .Params() .databaseAddress(zooKeeperAddress) .databaseSessionTimeout(zooKeeperSessionTimeout) .databaseListener(dbListener); @@ -426,10 +425,10 @@ public class DatabaseHandler { } } - public void saveWantedStates(DatabaseContext databaseContext) { + public boolean saveWantedStates(DatabaseContext databaseContext) { fleetControllerContext.log(logger, Level.FINE, () -> "Checking whether wanted states have changed compared to zookeeper version."); Map<Node, NodeState> wantedStates = new TreeMap<>(); - for (NodeInfo info : databaseContext.getCluster().getNodeInfo()) { + for (NodeInfo info : databaseContext.getCluster().getNodeInfos()) { if (!info.getUserWantedState().equals(new NodeState(info.getNode().getType(), State.UP))) { wantedStates.put(info.getNode(), info.getUserWantedState()); } @@ -444,6 +443,9 @@ public class DatabaseHandler { fleetControllerContext.log(logger, Level.FINE, () -> "Scheduling new wanted states to be stored into zookeeper."); pendingStore.wantedStates = wantedStates; doNextZooKeeperTask(databaseContext); + return true; + } else { + return false; } } @@ -466,7 +468,11 @@ public class DatabaseHandler { boolean altered = false; for (Node node : wantedStates.keySet()) { NodeInfo nodeInfo = databaseContext.getCluster().getNodeInfo(node); - if (nodeInfo == null) continue; // ignore wanted state of nodes which doesn't exist + if (nodeInfo == null) { + databaseContext.getNodeStateUpdateListener().handleRemovedNode(node); + altered = true; + continue; + } NodeState wantedState = wantedStates.get(node); if ( ! nodeInfo.getUserWantedState().equals(wantedState)) { nodeInfo.setWantedState(wantedState); @@ -477,7 +483,7 @@ public class DatabaseHandler { } // Remove wanted state from any node having a wanted state set that is no longer valid - for (NodeInfo info : databaseContext.getCluster().getNodeInfo()) { + for (NodeInfo info : databaseContext.getCluster().getNodeInfos()) { NodeState wantedState = wantedStates.get(info.getNode()); if (wantedState == null && !info.getUserWantedState().equals(new NodeState(info.getNode().getType(), State.UP))) { info.setWantedState(null); diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/database/MasterDataGatherer.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/database/MasterDataGatherer.java index 907f2e0c5e9..0c32d8ef6c2 100644 --- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/database/MasterDataGatherer.java +++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/database/MasterDataGatherer.java @@ -24,7 +24,7 @@ public class MasterDataGatherer { return Integer.parseInt(nodeName.substring(lastSlash + 1)); } - private final String zooKeeperRoot; // The root path in zookeeper, typically /vespa/fleetcontroller/<clustername>/ + private final ZooKeeperPaths paths; private Map<Integer, Integer> masterData = new TreeMap<>(); // The master state last reported to the fleetcontroller private final Map<Integer, Integer> nextMasterData = new TreeMap<>(); // Temporary master state while gathering new info from zookeeper private final AsyncCallback.ChildrenCallback childListener = new DirCallback(); // Dir change listener @@ -46,7 +46,7 @@ public class MasterDataGatherer { switch (watchedEvent.getType()) { case NodeChildrenChanged: // Fleetcontrollers have either connected or disconnected to ZooKeeper log.log(Level.INFO, "Fleetcontroller " + nodeIndex + ": A change occurred in the list of registered fleetcontrollers. Requesting new information"); - session.getChildren(zooKeeperRoot + "indexes", this, childListener, null); + session.getChildren(paths.indexesRoot(), this, childListener, null); break; case NodeDataChanged: // A fleetcontroller has changed what node it is voting for log.log(Level.INFO, "Fleetcontroller " + nodeIndex + ": Altered data in node " + watchedEvent.getPath() + ". Requesting new vote"); @@ -54,7 +54,7 @@ public class MasterDataGatherer { synchronized (nextMasterData) { nextMasterData.put(index, null); } - session.getData(zooKeeperRoot + "indexes/" + index, this, nodeListener, null); + session.getData(paths.indexOf(index), this, nodeListener, null); break; case NodeCreated: // How can this happen? Can one leave watches on non-existing nodes? log.log(Level.WARNING, "Fleetcontroller " + nodeIndex + ": Got unexpected ZooKeeper event NodeCreated"); @@ -85,8 +85,8 @@ public class MasterDataGatherer { int index = Integer.parseInt(node); nextMasterData.put(index, null); log.log(Level.FINE, () -> "Fleetcontroller " + nodeIndex + ": Attempting to fetch data in node '" - + zooKeeperRoot + index + "' to see vote"); - session.getData(zooKeeperRoot + "indexes/" + index, changeWatcher, nodeListener, null); + + paths.indexOf(index) + "' to see vote"); + session.getData(paths.indexOf(index), changeWatcher, nodeListener, null); // Invocation of cycleCompleted() for fully accumulated election state will happen // as soon as all getData calls have been processed. } @@ -146,8 +146,8 @@ public class MasterDataGatherer { } /** Constructor setting up the various needed members, and initializing the first data fetch to start things up */ - public MasterDataGatherer(ZooKeeper session, String zooKeeperRoot, Database.DatabaseListener listener, int nodeIndex) { - this.zooKeeperRoot = zooKeeperRoot; + public MasterDataGatherer(ZooKeeper session, ZooKeeperPaths paths, Database.DatabaseListener listener, int nodeIndex) { + this.paths = paths; this.session = session; this.listener = listener; this.nodeIndex = nodeIndex; @@ -161,7 +161,7 @@ public class MasterDataGatherer { synchronized (nextMasterData) { masterData = new TreeMap<>(); nextMasterData.clear(); - session.getChildren(zooKeeperRoot + "indexes", changeWatcher, childListener, null); + session.getChildren(paths.indexesRoot(), changeWatcher, childListener, null); } } diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/database/ZooKeeperDatabase.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/database/ZooKeeperDatabase.java index 72c81489351..ea745a56066 100644 --- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/database/ZooKeeperDatabase.java +++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/database/ZooKeeperDatabase.java @@ -6,7 +6,6 @@ import com.yahoo.vdslib.state.NodeState; import com.yahoo.vdslib.state.State; import com.yahoo.vespa.clustercontroller.core.AnnotatedClusterState; import com.yahoo.vespa.clustercontroller.core.ClusterStateBundle; -import com.yahoo.vespa.clustercontroller.core.ContentCluster; import com.yahoo.vespa.clustercontroller.core.FleetControllerContext; import com.yahoo.vespa.clustercontroller.core.rpc.EnvelopedClusterStateBundleCodec; import com.yahoo.vespa.clustercontroller.core.rpc.SlimeClusterStateBundleCodec; @@ -38,13 +37,12 @@ public class ZooKeeperDatabase extends Database { private static final Charset utf8 = StandardCharsets.UTF_8; private static final List<ACL> acl = ZooDefs.Ids.OPEN_ACL_UNSAFE; - private final String zooKeeperRoot; + private final ZooKeeperPaths paths; private final Database.DatabaseListener listener; private final ZooKeeperWatcher watcher = new ZooKeeperWatcher(); private final ZooKeeper session; private boolean sessionOpen = true; private final FleetControllerContext context; - private final int nodeIndex; private final MasterDataGatherer masterDataGatherer; // Expected ZK znode versions. Note: these are _not_ -1 as that would match anything. // We expect the caller to invoke the load methods prior to calling any store methods. @@ -102,31 +100,30 @@ public class ZooKeeperDatabase extends Database { } } - public ZooKeeperDatabase(FleetControllerContext context, ContentCluster cluster, int nodeIndex, String address, int timeout, DatabaseListener zksl) throws IOException, KeeperException, InterruptedException { + public ZooKeeperDatabase(FleetControllerContext context, String address, int timeout, DatabaseListener zksl) throws IOException, KeeperException, InterruptedException { this.context = context; - this.nodeIndex = nodeIndex; - zooKeeperRoot = "/vespa/fleetcontroller/" + cluster.getName() + "/"; + this.paths = new ZooKeeperPaths(context.id()); session = new ZooKeeper(address, timeout, watcher, new ZkClientConfigBuilder().toConfig()); boolean completedOk = false; try{ this.listener = zksl; setupRoot(); context.log(log, Level.FINEST, "Asking for initial data on master election"); - masterDataGatherer = new MasterDataGatherer(session, zooKeeperRoot, listener, nodeIndex); + masterDataGatherer = new MasterDataGatherer(session, paths, listener, context.id().index()); completedOk = true; } finally { if (!completedOk) session.close(); } } - private void createNode(String prefix, String nodename, byte[] value) throws KeeperException, InterruptedException { + private void createNode(String path, byte[] value) throws KeeperException, InterruptedException { try{ - if (session.exists(prefix + nodename, false) != null) { - context.log(log, Level.FINE, () -> "Zookeeper node '" + prefix + nodename + "' already exists. Not creating it"); + if (session.exists(path, false) != null) { + context.log(log, Level.FINE, () -> "Zookeeper node '" + path + "' already exists. Not creating it"); return; } - session.create(prefix + nodename, value, acl, CreateMode.PERSISTENT); - context.log(log, Level.FINE, () -> "Created zookeeper node '" + prefix + nodename + "'"); + session.create(path, value, acl, CreateMode.PERSISTENT); + context.log(log, Level.FINE, () -> "Created zookeeper node '" + path + "'"); } catch (KeeperException.NodeExistsException e) { context.log(log, Level.FINE, "Node to create existed, but this is normal as other nodes " + "may create them at the same time."); @@ -134,21 +131,21 @@ public class ZooKeeperDatabase extends Database { } private void setupRoot() throws KeeperException, InterruptedException { - String[] pathElements = zooKeeperRoot.substring(1).split("/"); + String[] pathElements = paths.root().substring(1).split("/"); String path = ""; for (String elem : pathElements) { path += "/" + elem; - createNode("", path, new byte[0]); + createNode(path, new byte[0]); } - createNode(zooKeeperRoot, "indexes", new byte[0]); - createNode(zooKeeperRoot, "wantedstates", new byte[0]); - createNode(zooKeeperRoot, "starttimestamps", new byte[0]); - createNode(zooKeeperRoot, "latestversion", Integer.valueOf(0).toString().getBytes(utf8)); - createNode(zooKeeperRoot, "published_state_bundle", new byte[0]); // TODO dedupe string constants - byte[] val = String.valueOf(nodeIndex).getBytes(utf8); - deleteNodeIfExists(getMyIndexPath()); + createNode(paths.indexesRoot(), new byte[0]); + createNode(paths.wantedStates(), new byte[0]); + createNode(paths.startTimestamps(), new byte[0]); + createNode(paths.latestVersion(), Integer.valueOf(0).toString().getBytes(utf8)); + createNode(paths.publishedStateBundle(), new byte[0]); + byte[] val = String.valueOf(context.id().index()).getBytes(utf8); + deleteNodeIfExists(paths.indexOfMe()); context.log(log, Level.INFO, "Creating ephemeral master vote node with vote to self."); - session.create(getMyIndexPath(), val, acl, CreateMode.EPHEMERAL); + session.create(paths.indexOfMe(), val, acl, CreateMode.EPHEMERAL); } private void deleteNodeIfExists(String path) throws KeeperException, InterruptedException { @@ -158,10 +155,6 @@ public class ZooKeeperDatabase extends Database { } } - private String getMyIndexPath() { - return zooKeeperRoot + "indexes/" + nodeIndex; - } - /** * If this is called, we assume we're in shutdown situation, or we are doing it because we need a new session. * Thus we only need to free up resources, no need to notify anyone. @@ -192,8 +185,8 @@ public class ZooKeeperDatabase extends Database { public boolean storeMasterVote(int wantedMasterIndex) { byte[] val = String.valueOf(wantedMasterIndex).getBytes(utf8); try{ - session.setData(getMyIndexPath(), val, -1); - context.log(log, Level.INFO, "Stored new vote in ephemeral node. " + nodeIndex + " -> " + wantedMasterIndex); + session.setData(paths.indexOfMe(), val, -1); + context.log(log, Level.INFO, "Stored new vote in ephemeral node. " + context.id().index() + " -> " + wantedMasterIndex); return true; } catch (InterruptedException e) { throw new RuntimeException(e); @@ -206,7 +199,7 @@ public class ZooKeeperDatabase extends Database { byte[] data = Integer.toString(version).getBytes(utf8); try{ context.log(log, Level.INFO, "Storing new cluster state version in ZooKeeper: " + version); - var stat = session.setData(zooKeeperRoot + "latestversion", data, lastKnownStateVersionZNodeVersion); + var stat = session.setData(paths.latestVersion(), data, lastKnownStateVersionZNodeVersion); lastKnownStateVersionZNodeVersion = stat.getVersion(); return true; } catch (InterruptedException e) { @@ -222,17 +215,17 @@ public class ZooKeeperDatabase extends Database { public Integer retrieveLatestSystemStateVersion() { Stat stat = new Stat(); - context.log(log, Level.FINE, "Fetching latest cluster state at '%slatestversion'", zooKeeperRoot); + context.log(log, Level.FINE, "Fetching latest cluster state at '%s'", paths.latestVersion()); final byte[] data; try { - data = session.getData(zooKeeperRoot + "latestversion", false, stat); + data = session.getData(paths.latestVersion(), false, stat); } catch (KeeperException.NoNodeException e) { // Initial condition: No latest version has ever been written (or ZK state completely wiped!) lastKnownStateVersionZNodeVersion = 0; maybeLogExceptionWarning(e, "No latest system state found"); return null; } catch (InterruptedException | KeeperException e) { - throw new RuntimeException("Failed to get " + zooKeeperRoot + "latestversion", e); + throw new RuntimeException("Failed to get " + paths.latestVersion(), e); } lastKnownStateVersionZNodeVersion = stat.getVersion(); @@ -257,8 +250,8 @@ public class ZooKeeperDatabase extends Database { } byte[] val = sb.toString().getBytes(utf8); try{ - context.log(log, Level.FINE, () -> "Storing wanted states at '" + zooKeeperRoot + "wantedstates'"); - session.setData(zooKeeperRoot + "wantedstates", val, -1); + context.log(log, Level.FINE, () -> "Storing wanted states at '" + paths.wantedStates() + "'"); + session.setData(paths.wantedStates(), val, -1); return true; } catch (InterruptedException e) { throw new RuntimeException(e); @@ -270,9 +263,9 @@ public class ZooKeeperDatabase extends Database { public Map<Node, NodeState> retrieveWantedStates() { try{ - context.log(log, Level.FINE, () -> "Fetching wanted states at '" + zooKeeperRoot + "wantedstates'"); + context.log(log, Level.FINE, () -> "Fetching wanted states at '" + paths.wantedStates() + "'"); Stat stat = new Stat(); - byte[] data = session.getData(zooKeeperRoot + "wantedstates", false, stat); + byte[] data = session.getData(paths.wantedStates(), false, stat); Map<Node, NodeState> wanted = new TreeMap<>(); if (data != null && data.length > 0) { StringTokenizer st = new StringTokenizer(new String(data, utf8), "\n", false); @@ -308,8 +301,8 @@ public class ZooKeeperDatabase extends Database { } byte val[] = sb.toString().getBytes(utf8); try{ - context.log(log, Level.FINE, () -> "Storing start timestamps at '" + zooKeeperRoot + "starttimestamps"); - session.setData(zooKeeperRoot + "starttimestamps", val, -1); + context.log(log, Level.FINE, () -> "Storing start timestamps at '" + paths.startTimestamps() + "'"); + session.setData(paths.startTimestamps(), val, -1); return true; } catch (InterruptedException e) { throw new RuntimeException(e); @@ -322,9 +315,9 @@ public class ZooKeeperDatabase extends Database { @Override public Map<Node, Long> retrieveStartTimestamps() { try{ - context.log(log, Level.FINE, () -> "Fetching start timestamps at '" + zooKeeperRoot + "starttimestamps'"); + context.log(log, Level.FINE, () -> "Fetching start timestamps at '" + paths.startTimestamps() + "'"); Stat stat = new Stat(); - byte[] data = session.getData(zooKeeperRoot + "starttimestamps", false, stat); + byte[] data = session.getData(paths.startTimestamps(), false, stat); Map<Node, Long> wanted = new TreeMap<Node, Long>(); if (data != null && data.length > 0) { StringTokenizer st = new StringTokenizer(new String(data, utf8), "\n", false); @@ -357,10 +350,9 @@ public class ZooKeeperDatabase extends Database { try{ context.log(log, Level.FINE, - () -> String.format("Storing published state bundle %s at " + - "'%spublished_state_bundle' with expected znode version %d", - stateBundle, zooKeeperRoot, lastKnownStateBundleZNodeVersion)); - var stat = session.setData(zooKeeperRoot + "published_state_bundle", encodedBundle, lastKnownStateBundleZNodeVersion); + () -> String.format("Storing published state bundle %s at '%s' with expected znode version %d", + stateBundle, paths.publishedStateBundle(), lastKnownStateBundleZNodeVersion)); + var stat = session.setData(paths.publishedStateBundle(), encodedBundle, lastKnownStateBundleZNodeVersion); lastKnownStateBundleZNodeVersion = stat.getVersion(); } catch (InterruptedException e) { throw new RuntimeException(e); @@ -378,7 +370,7 @@ public class ZooKeeperDatabase extends Database { public ClusterStateBundle retrieveLastPublishedStateBundle() { Stat stat = new Stat(); try { - byte[] data = session.getData(zooKeeperRoot + "published_state_bundle", false, stat); + byte[] data = session.getData(paths.publishedStateBundle(), false, stat); lastKnownStateBundleZNodeVersion = stat.getVersion(); if (data != null && data.length != 0) { EnvelopedClusterStateBundleCodec envelopedBundleCodec = new SlimeClusterStateBundleCodec(); diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/database/ZooKeeperDatabaseFactory.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/database/ZooKeeperDatabaseFactory.java index 71f39135609..3263c06a95c 100644 --- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/database/ZooKeeperDatabaseFactory.java +++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/database/ZooKeeperDatabaseFactory.java @@ -13,8 +13,7 @@ public class ZooKeeperDatabaseFactory implements DatabaseFactory { @Override public Database create(Params params) throws Exception { - return new ZooKeeperDatabase(context, params.cluster, params.nodeIndex, params.dbAddress, - params.dbSessionTimeout, params.listener); + return new ZooKeeperDatabase(context, params.dbAddress, params.dbSessionTimeout, params.listener); } } diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/database/ZooKeeperPaths.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/database/ZooKeeperPaths.java new file mode 100644 index 00000000000..06a9b240175 --- /dev/null +++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/database/ZooKeeperPaths.java @@ -0,0 +1,26 @@ +// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. +package com.yahoo.vespa.clustercontroller.core.database; + +import com.yahoo.vespa.clustercontroller.core.FleetControllerId; + +/** + * @author hakonhall + */ +public class ZooKeeperPaths { + private final String root; + private final int myIndex; + + public ZooKeeperPaths(FleetControllerId id) { + this.root = "/vespa/fleetcontroller/" + id.clusterName(); + this.myIndex = id.index(); + } + + public String root() { return root; } + public String indexesRoot() { return root + "/indexes"; } + public String indexOf(int index) { return indexesRoot() + "/" + index; } + public String indexOfMe() { return indexOf(myIndex); } + public String wantedStates() { return root + "/wantedstates"; } + public String publishedStateBundle() { return root + "/published_state_bundle"; } + public String latestVersion() { return root + "/latestversion"; } + public String startTimestamps() { return root + "/starttimestamps"; } +} diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/listeners/NodeListener.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/listeners/NodeListener.java new file mode 100644 index 00000000000..351b68b3b57 --- /dev/null +++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/listeners/NodeListener.java @@ -0,0 +1,25 @@ +// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. +package com.yahoo.vespa.clustercontroller.core.listeners; + +import com.yahoo.vdslib.state.Node; +import com.yahoo.vdslib.state.NodeState; +import com.yahoo.vespa.clustercontroller.core.NodeInfo; +import com.yahoo.vespa.clustercontroller.core.hostinfo.HostInfo; + +/** + * Implemented by classes wanting events when there are node changes. + */ +public interface NodeListener { + + default void handleNewNodeState(NodeInfo currentInfo, NodeState newState) {} + default void handleNewWantedNodeState(NodeInfo node, NodeState newState) {} + + /** Invoked after NodeInfo has been removed from the content cluster. */ + default void handleRemovedNode(Node node) {} + + /** + * For every getnodestate RPC call, handleUpdatedHostInfo() will be called with the host info JSON string. + */ + default void handleUpdatedHostInfo(NodeInfo node, HostInfo newHostInfo) {} + +} diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/listeners/NodeStateOrHostInfoChangeHandler.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/listeners/NodeStateOrHostInfoChangeHandler.java deleted file mode 100644 index b76b46b216b..00000000000 --- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/listeners/NodeStateOrHostInfoChangeHandler.java +++ /dev/null @@ -1,21 +0,0 @@ -// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. -package com.yahoo.vespa.clustercontroller.core.listeners; - -import com.yahoo.vdslib.state.NodeState; -import com.yahoo.vespa.clustercontroller.core.NodeInfo; -import com.yahoo.vespa.clustercontroller.core.hostinfo.HostInfo; - -/** - * Implemented by classes wanting events when node states changes. - */ -public interface NodeStateOrHostInfoChangeHandler { - - void handleNewNodeState(NodeInfo currentInfo, NodeState newState); - void handleNewWantedNodeState(NodeInfo node, NodeState newState); - - /** - * For every getnodestate RPC call, handleUpdatedHostInfo() will be called with the host info JSON string. - */ - void handleUpdatedHostInfo(NodeInfo node, HostInfo newHostInfo); - -} diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/listeners/NodeAddedOrRemovedListener.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/listeners/SlobrokListener.java index d811708c999..5a397cc4935 100644 --- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/listeners/NodeAddedOrRemovedListener.java +++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/listeners/SlobrokListener.java @@ -4,9 +4,9 @@ package com.yahoo.vespa.clustercontroller.core.listeners; import com.yahoo.vespa.clustercontroller.core.NodeInfo; /** - * Listeners for new nodes detected. + * Implemented by classes that wants to be notified of Slobrok events. */ -public interface NodeAddedOrRemovedListener { +public interface SlobrokListener { void handleNewNode(NodeInfo node); void handleMissingNode(NodeInfo node); void handleNewRpcAddress(NodeInfo node); diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/restapiv2/requests/SetNodeStateRequest.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/restapiv2/requests/SetNodeStateRequest.java index 431c207af5c..ddbda2bf776 100644 --- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/restapiv2/requests/SetNodeStateRequest.java +++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/restapiv2/requests/SetNodeStateRequest.java @@ -11,7 +11,7 @@ import com.yahoo.vespa.clustercontroller.core.ContentCluster; import com.yahoo.vespa.clustercontroller.core.NodeInfo; import com.yahoo.vespa.clustercontroller.core.NodeStateChangeChecker; import com.yahoo.vespa.clustercontroller.core.RemoteClusterControllerTask; -import com.yahoo.vespa.clustercontroller.core.listeners.NodeStateOrHostInfoChangeHandler; +import com.yahoo.vespa.clustercontroller.core.listeners.NodeListener; import com.yahoo.vespa.clustercontroller.core.restapiv2.Id; import com.yahoo.vespa.clustercontroller.core.restapiv2.MissingIdException; import com.yahoo.vespa.clustercontroller.core.restapiv2.Request; @@ -62,7 +62,7 @@ public class SetNodeStateRequest extends Request<SetResponse> { condition, newStates, id.getNode(), - context.nodeStateOrHostInfoChangeHandler, + context.nodeListener, context.currentConsolidatedState, context.masterInfo.inMasterMoratorium(), probe); @@ -112,7 +112,7 @@ public class SetNodeStateRequest extends Request<SetResponse> { SetUnitStateRequest.Condition condition, Map<String, UnitState> newStates, Node node, - NodeStateOrHostInfoChangeHandler stateListener, + NodeListener stateListener, ClusterState currentClusterState, boolean inMasterMoratorium, boolean probe) throws StateRestApiException { @@ -159,7 +159,7 @@ public class SetNodeStateRequest extends Request<SetResponse> { SetUnitStateRequest.Condition condition, NodeInfo nodeInfo, ContentCluster cluster, - NodeStateOrHostInfoChangeHandler stateListener, + NodeListener stateListener, boolean probe) { if (result.settingWantedStateIsAllowed()) { setNewWantedState(nodeInfo, newWantedState, stateListener, probe); @@ -186,7 +186,7 @@ public class SetNodeStateRequest extends Request<SetResponse> { private static void setDistributorWantedState(ContentCluster cluster, int index, NodeState newStorageWantedState, - NodeStateOrHostInfoChangeHandler stateListener, + NodeListener stateListener, boolean probe) { Node distributorNode = new Node(NodeType.DISTRIBUTOR, index); NodeInfo nodeInfo = cluster.getNodeInfo(distributorNode); @@ -224,7 +224,7 @@ public class SetNodeStateRequest extends Request<SetResponse> { private static void setNewWantedState(NodeInfo nodeInfo, NodeState newWantedState, - NodeStateOrHostInfoChangeHandler stateListener, + NodeListener stateListener, boolean probe) { if (probe) return; nodeInfo.setWantedState(newWantedState); diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/restapiv2/requests/SetNodeStatesForClusterRequest.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/restapiv2/requests/SetNodeStatesForClusterRequest.java index 55ac75957bc..1cc8f2860c6 100644 --- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/restapiv2/requests/SetNodeStatesForClusterRequest.java +++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/restapiv2/requests/SetNodeStatesForClusterRequest.java @@ -70,7 +70,7 @@ public class SetNodeStatesForClusterRequest extends Request<SetResponse> { condition, newStates, node, - context.nodeStateOrHostInfoChangeHandler, + context.nodeListener, context.currentConsolidatedState, context.masterInfo.inMasterMoratorium(), probe); diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/restapiv2/requests/WantedStateSetter.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/restapiv2/requests/WantedStateSetter.java index 8095d37d641..51b2f1cfe4f 100644 --- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/restapiv2/requests/WantedStateSetter.java +++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/restapiv2/requests/WantedStateSetter.java @@ -4,7 +4,7 @@ package com.yahoo.vespa.clustercontroller.core.restapiv2.requests; import com.yahoo.vdslib.state.ClusterState; import com.yahoo.vdslib.state.Node; import com.yahoo.vespa.clustercontroller.core.ContentCluster; -import com.yahoo.vespa.clustercontroller.core.listeners.NodeStateOrHostInfoChangeHandler; +import com.yahoo.vespa.clustercontroller.core.listeners.NodeListener; import com.yahoo.vespa.clustercontroller.utils.staterestapi.errors.StateRestApiException; import com.yahoo.vespa.clustercontroller.utils.staterestapi.requests.SetUnitStateRequest; import com.yahoo.vespa.clustercontroller.utils.staterestapi.response.SetResponse; @@ -21,7 +21,7 @@ public interface WantedStateSetter { SetUnitStateRequest.Condition condition, Map<String, UnitState> newStates, Node node, - NodeStateOrHostInfoChangeHandler stateListener, + NodeListener stateListener, ClusterState currentClusterState, boolean inMasterMoratorium, boolean probe) throws StateRestApiException; diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/rpc/RpcServer.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/rpc/RpcServer.java index 65c8a9df28e..6e416ce4906 100644 --- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/rpc/RpcServer.java +++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/rpc/RpcServer.java @@ -27,8 +27,8 @@ import com.yahoo.vespa.clustercontroller.core.ContentCluster; import com.yahoo.vespa.clustercontroller.core.MasterElectionHandler; import com.yahoo.vespa.clustercontroller.core.NodeInfo; import com.yahoo.vespa.clustercontroller.core.Timer; -import com.yahoo.vespa.clustercontroller.core.listeners.NodeAddedOrRemovedListener; -import com.yahoo.vespa.clustercontroller.core.listeners.NodeStateOrHostInfoChangeHandler; +import com.yahoo.vespa.clustercontroller.core.listeners.SlobrokListener; +import com.yahoo.vespa.clustercontroller.core.listeners.NodeListener; import java.io.PrintWriter; import java.io.StringWriter; @@ -186,8 +186,8 @@ public class RpcServer { } public boolean handleRpcRequests(ContentCluster cluster, ClusterState systemState, - NodeStateOrHostInfoChangeHandler changeListener, - NodeAddedOrRemovedListener addedListener) + NodeListener changeListener, + SlobrokListener addedListener) { boolean handledAnyRequests = false; if (!isConnected()) { @@ -234,7 +234,7 @@ public class RpcServer { log.log(Level.FINE, "Resolving RPC getNodeList request"); List<String> slobrok = new ArrayList<String>(); List<String> rpc = new ArrayList<String>(); - for(NodeInfo node : cluster.getNodeInfo()) { + for(NodeInfo node : cluster.getNodeInfos()) { String s1 = node.getSlobrokAddress(); String s2 = node.getRpcAddress(); assert(s1 != null); diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/rpc/SlobrokClient.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/rpc/SlobrokClient.java index 7487f9546b7..c88bf71af09 100644 --- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/rpc/SlobrokClient.java +++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/rpc/SlobrokClient.java @@ -13,7 +13,7 @@ import com.yahoo.vespa.clustercontroller.core.FleetControllerContext; import com.yahoo.vespa.clustercontroller.core.NodeInfo; import com.yahoo.vespa.clustercontroller.core.NodeLookup; import com.yahoo.vespa.clustercontroller.core.Timer; -import com.yahoo.vespa.clustercontroller.core.listeners.NodeAddedOrRemovedListener; +import com.yahoo.vespa.clustercontroller.core.listeners.SlobrokListener; import java.util.Iterator; import java.util.LinkedList; @@ -78,7 +78,7 @@ public class SlobrokClient implements NodeLookup { } @Override - public boolean updateCluster(ContentCluster cluster, NodeAddedOrRemovedListener listener) { + public boolean updateCluster(ContentCluster cluster, SlobrokListener listener) { if (mirror == null) return false; int mirrorVersion = mirror.updates(); if (freshMirror) { @@ -149,7 +149,7 @@ public class SlobrokClient implements NodeLookup { } } cluster.setSlobrokGenerationCount(mirrorVersion); - for (NodeInfo nodeInfo : cluster.getNodeInfo()) { + for (NodeInfo nodeInfo : cluster.getNodeInfos()) { if (slobrokNodes.containsKey(nodeInfo.getNode()) && nodeInfo.isRpcAddressOutdated()) { context.log(log, Level.WARNING, @@ -174,7 +174,7 @@ public class SlobrokClient implements NodeLookup { List<SlobrokData> alteredRpcAddress, List<NodeInfo> returningRpcAddressNodeInfos) { - Iterator<NodeInfo> oldIt = oldCluster.getNodeInfo().iterator(); + Iterator<NodeInfo> oldIt = oldCluster.getNodeInfos().iterator(); Iterator<SlobrokData> newIt = slobrokNodes.values().iterator(); NodeInfo oldNext = null; SlobrokData newNext = null; |