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