opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActorContextImpl.java

   1 /*
   2  * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.controller.cluster.raft;
  10
  11 import akka.actor.ActorContext;
  12 import akka.actor.ActorRef;
  13 import akka.actor.ActorSelection;
  14 import akka.actor.ActorSystem;
  15 import akka.actor.Props;
  16 import akka.cluster.Cluster;
  17 import com.google.common.annotations.VisibleForTesting;
  18 import com.google.common.base.Preconditions;
  19 import java.util.ArrayList;
  20 import java.util.Collection;
  21 import java.util.HashMap;
  22 import java.util.HashSet;
  23 import java.util.List;
  24 import java.util.Map;
  25 import java.util.Optional;
  26 import java.util.Set;
  27 import java.util.function.LongSupplier;
  28 import org.opendaylight.controller.cluster.DataPersistenceProvider;
  29 import org.opendaylight.controller.cluster.raft.behaviors.RaftActorBehavior;
  30 import org.opendaylight.controller.cluster.raft.persisted.ServerConfigurationPayload;
  31 import org.opendaylight.controller.cluster.raft.persisted.ServerInfo;
  32 import org.opendaylight.controller.cluster.raft.policy.RaftPolicy;
  33 import org.slf4j.Logger;
  34
  35 /**
  36  * Implementation of the RaftActorContext interface.
  37  *
  38  * @author Moiz Raja
  39  * @author Thomas Pantelis
  40  */
  41 public class RaftActorContextImpl implements RaftActorContext {
  42     private static final LongSupplier JVM_MEMORY_RETRIEVER = () -> Runtime.getRuntime().maxMemory();
  43
  44     private final ActorRef actor;
  45
  46     private final ActorContext context;
  47
  48     private final String id;
  49
  50     private final ElectionTerm termInformation;
  51
  52     private long commitIndex;
  53
  54     private long lastApplied;
  55
  56     private ReplicatedLog replicatedLog;
  57
  58     private final Map<String, PeerInfo> peerInfoMap = new HashMap<>();
  59
  60     private final Logger log;
  61
  62     private ConfigParams configParams;
  63
  64     private boolean dynamicServerConfiguration = false;
  65
  66     @VisibleForTesting
  67     private LongSupplier totalMemoryRetriever = JVM_MEMORY_RETRIEVER;
  68
  69     // Snapshot manager will need to be created on demand as it needs raft actor context which cannot
  70     // be passed to it in the constructor
  71     private SnapshotManager snapshotManager;
  72
  73     private final DataPersistenceProvider persistenceProvider;
  74
  75     private short payloadVersion;
  76
  77     private boolean votingMember = true;
  78
  79     private RaftActorBehavior currentBehavior;
  80
  81     private int numVotingPeers = -1;
  82
  83     private Optional<Cluster> cluster;
  84
  85     public RaftActorContextImpl(ActorRef actor, ActorContext context, String id,
  86             ElectionTerm termInformation, long commitIndex, long lastApplied, Map<String, String> peerAddresses,
  87             ConfigParams configParams, DataPersistenceProvider persistenceProvider, Logger logger) {
  88         this.actor = actor;
  89         this.context = context;
  90         this.id = id;
  91         this.termInformation = termInformation;
  92         this.commitIndex = commitIndex;
  93         this.lastApplied = lastApplied;
  94         this.configParams = configParams;
  95         this.persistenceProvider = persistenceProvider;
  96         this.log = logger;
  97
  98         for (Map.Entry<String, String> e: peerAddresses.entrySet()) {
  99             peerInfoMap.put(e.getKey(), new PeerInfo(e.getKey(), e.getValue(), VotingState.VOTING));
 100         }
 101     }
 102
 103     @VisibleForTesting
 104     public void setPayloadVersion(short payloadVersion) {
 105         this.payloadVersion = payloadVersion;
 106     }
 107
 108     @Override
 109     public short getPayloadVersion() {
 110         return payloadVersion;
 111     }
 112
 113     public void setConfigParams(ConfigParams configParams) {
 114         this.configParams = configParams;
 115     }
 116
 117     @Override
 118     public ActorRef actorOf(Props props) {
 119         return context.actorOf(props);
 120     }
 121
 122     @Override
 123     public ActorSelection actorSelection(String path) {
 124         return context.actorSelection(path);
 125     }
 126
 127     @Override
 128     public String getId() {
 129         return id;
 130     }
 131
 132     @Override
 133     public ActorRef getActor() {
 134         return actor;
 135     }
 136
 137     @Override
 138     @SuppressWarnings("checkstyle:IllegalCatch")
 139     public Optional<Cluster> getCluster() {
 140         if (cluster == null) {
 141             try {
 142                 cluster = Optional.of(Cluster.get(getActorSystem()));
 143             } catch (Exception e) {
 144                 // An exception means there's no cluster configured. This will only happen in unit tests.
 145                 log.debug("{}: Could not obtain Cluster: {}", getId(), e);
 146                 cluster = Optional.empty();
 147             }
 148         }
 149
 150         return cluster;
 151     }
 152
 153     @Override
 154     public ElectionTerm getTermInformation() {
 155         return termInformation;
 156     }
 157
 158     @Override
 159     public long getCommitIndex() {
 160         return commitIndex;
 161     }
 162
 163     @Override public void setCommitIndex(long commitIndex) {
 164         this.commitIndex = commitIndex;
 165     }
 166
 167     @Override
 168     public long getLastApplied() {
 169         return lastApplied;
 170     }
 171
 172     @Override
 173     public void setLastApplied(long lastApplied) {
 174         this.lastApplied = lastApplied;
 175     }
 176
 177     @Override
 178     public void setReplicatedLog(ReplicatedLog replicatedLog) {
 179         this.replicatedLog = replicatedLog;
 180     }
 181
 182     @Override
 183     public ReplicatedLog getReplicatedLog() {
 184         return replicatedLog;
 185     }
 186
 187     @Override public ActorSystem getActorSystem() {
 188         return context.system();
 189     }
 190
 191     @Override public Logger getLogger() {
 192         return this.log;
 193     }
 194
 195     @Override
 196     public Collection<String> getPeerIds() {
 197         return peerInfoMap.keySet();
 198     }
 199
 200     @Override
 201     public Collection<PeerInfo> getPeers() {
 202         return peerInfoMap.values();
 203     }
 204
 205     @Override
 206     public PeerInfo getPeerInfo(String peerId) {
 207         return peerInfoMap.get(peerId);
 208     }
 209
 210     @Override
 211     public String getPeerAddress(String peerId) {
 212         String peerAddress;
 213         PeerInfo peerInfo = peerInfoMap.get(peerId);
 214         if (peerInfo != null) {
 215             peerAddress = peerInfo.getAddress();
 216             if (peerAddress == null) {
 217                 peerAddress = configParams.getPeerAddressResolver().resolve(peerId);
 218                 peerInfo.setAddress(peerAddress);
 219             }
 220         } else {
 221             peerAddress = configParams.getPeerAddressResolver().resolve(peerId);
 222         }
 223
 224         return peerAddress;
 225     }
 226
 227     @Override
 228     public void updatePeerIds(ServerConfigurationPayload serverConfig) {
 229         votingMember = true;
 230         boolean foundSelf = false;
 231         Set<String> currentPeers = new HashSet<>(this.getPeerIds());
 232         for (ServerInfo server : serverConfig.getServerConfig()) {
 233             if (getId().equals(server.getId())) {
 234                 foundSelf = true;
 235                 if (!server.isVoting()) {
 236                     votingMember = false;
 237                 }
 238             } else {
 239                 VotingState votingState = server.isVoting() ? VotingState.VOTING : VotingState.NON_VOTING;
 240                 if (!currentPeers.contains(server.getId())) {
 241                     this.addToPeers(server.getId(), null, votingState);
 242                 } else {
 243                     this.getPeerInfo(server.getId()).setVotingState(votingState);
 244                     currentPeers.remove(server.getId());
 245                 }
 246             }
 247         }
 248
 249         for (String peerIdToRemove : currentPeers) {
 250             this.removePeer(peerIdToRemove);
 251         }
 252
 253         if (!foundSelf) {
 254             votingMember = false;
 255         }
 256
 257         log.debug("{}: Updated server config: isVoting: {}, peers: {}", id, votingMember, peerInfoMap.values());
 258
 259         setDynamicServerConfigurationInUse();
 260     }
 261
 262     @Override public ConfigParams getConfigParams() {
 263         return configParams;
 264     }
 265
 266     @Override
 267     public void addToPeers(String peerId, String address, VotingState votingState) {
 268         peerInfoMap.put(peerId, new PeerInfo(peerId, address, votingState));
 269         numVotingPeers = -1;
 270     }
 271
 272     @Override
 273     public void removePeer(String name) {
 274         if (getId().equals(name)) {
 275             votingMember = false;
 276         } else {
 277             peerInfoMap.remove(name);
 278             numVotingPeers = -1;
 279         }
 280     }
 281
 282     @Override public ActorSelection getPeerActorSelection(String peerId) {
 283         String peerAddress = getPeerAddress(peerId);
 284         if (peerAddress != null) {
 285             return actorSelection(peerAddress);
 286         }
 287         return null;
 288     }
 289
 290     @Override
 291     public void setPeerAddress(String peerId, String peerAddress) {
 292         PeerInfo peerInfo = peerInfoMap.get(peerId);
 293         if (peerInfo != null) {
 294             log.info("Peer address for peer {} set to {}", peerId, peerAddress);
 295             peerInfo.setAddress(peerAddress);
 296         }
 297     }
 298
 299     @Override
 300     public SnapshotManager getSnapshotManager() {
 301         if (snapshotManager == null) {
 302             snapshotManager = new SnapshotManager(this, log);
 303         }
 304         return snapshotManager;
 305     }
 306
 307     @Override
 308     public long getTotalMemory() {
 309         return totalMemoryRetriever.getAsLong();
 310     }
 311
 312     @Override
 313     public void setTotalMemoryRetriever(LongSupplier retriever) {
 314         totalMemoryRetriever = retriever == null ? JVM_MEMORY_RETRIEVER : retriever;
 315     }
 316
 317     @Override
 318     public boolean hasFollowers() {
 319         return !getPeerIds().isEmpty();
 320     }
 321
 322     @Override
 323     public DataPersistenceProvider getPersistenceProvider() {
 324         return persistenceProvider;
 325     }
 326
 327
 328     @Override
 329     public RaftPolicy getRaftPolicy() {
 330         return configParams.getRaftPolicy();
 331     }
 332
 333     @Override
 334     public boolean isDynamicServerConfigurationInUse() {
 335         return dynamicServerConfiguration;
 336     }
 337
 338     @Override
 339     public void setDynamicServerConfigurationInUse() {
 340         this.dynamicServerConfiguration = true;
 341     }
 342
 343     @Override
 344     public ServerConfigurationPayload getPeerServerInfo(boolean includeSelf) {
 345         if (!isDynamicServerConfigurationInUse()) {
 346             return null;
 347         }
 348         Collection<PeerInfo> peers = getPeers();
 349         List<ServerInfo> newConfig = new ArrayList<>(peers.size() + 1);
 350         for (PeerInfo peer: peers) {
 351             newConfig.add(new ServerInfo(peer.getId(), peer.isVoting()));
 352         }
 353
 354         if (includeSelf) {
 355             newConfig.add(new ServerInfo(getId(), votingMember));
 356         }
 357
 358         return new ServerConfigurationPayload(newConfig);
 359     }
 360
 361     @Override
 362     public boolean isVotingMember() {
 363         return votingMember;
 364     }
 365
 366     @Override
 367     public boolean anyVotingPeers() {
 368         if (numVotingPeers < 0) {
 369             numVotingPeers = 0;
 370             for (PeerInfo info: getPeers()) {
 371                 if (info.isVoting()) {
 372                     numVotingPeers++;
 373                 }
 374             }
 375         }
 376
 377         return numVotingPeers > 0;
 378     }
 379
 380     @Override
 381     public RaftActorBehavior getCurrentBehavior() {
 382         return currentBehavior;
 383     }
 384
 385     void setCurrentBehavior(final RaftActorBehavior behavior) {
 386         this.currentBehavior = Preconditions.checkNotNull(behavior);
 387     }
 388
 389     @SuppressWarnings("checkstyle:IllegalCatch")
 390     void close() {
 391         if (currentBehavior != null) {
 392             try {
 393                 currentBehavior.close();
 394             } catch (Exception e) {
 395                 log.debug("{}: Error closing behavior {}", getId(), currentBehavior.state(), e);
 396             }
 397         }
 398     }
 399 }