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 com.google.common.annotations.VisibleForTesting;
  17 import com.google.common.base.Supplier;
  18 import java.util.ArrayList;
  19 import java.util.Collection;
  20 import java.util.HashMap;
  21 import java.util.HashSet;
  22 import java.util.List;
  23 import java.util.Map;
  24 import java.util.Set;
  25 import org.opendaylight.controller.cluster.DataPersistenceProvider;
  26 import org.opendaylight.controller.cluster.raft.ServerConfigurationPayload.ServerInfo;
  27 import org.opendaylight.controller.cluster.raft.policy.RaftPolicy;
  28 import org.slf4j.Logger;
  29
  30 public class RaftActorContextImpl implements RaftActorContext {
  31
  32     private final ActorRef actor;
  33
  34     private final ActorContext context;
  35
  36     private final String id;
  37
  38     private final ElectionTerm termInformation;
  39
  40     private long commitIndex;
  41
  42     private long lastApplied;
  43
  44     private ReplicatedLog replicatedLog;
  45
  46     private final Map<String, PeerInfo> peerInfoMap = new HashMap<>();
  47
  48     private final Logger LOG;
  49
  50     private ConfigParams configParams;
  51
  52     private boolean dynamicServerConfiguration = false;
  53
  54     @VisibleForTesting
  55     private Supplier<Long> totalMemoryRetriever;
  56
  57     // Snapshot manager will need to be created on demand as it needs raft actor context which cannot
  58     // be passed to it in the constructor
  59     private SnapshotManager snapshotManager;
  60
  61     private final DataPersistenceProvider persistenceProvider;
  62
  63     private short payloadVersion;
  64
  65     public RaftActorContextImpl(ActorRef actor, ActorContext context, String id,
  66             ElectionTerm termInformation, long commitIndex, long lastApplied, Map<String, String> peerAddresses,
  67             ConfigParams configParams, DataPersistenceProvider persistenceProvider, Logger logger) {
  68         this.actor = actor;
  69         this.context = context;
  70         this.id = id;
  71         this.termInformation = termInformation;
  72         this.commitIndex = commitIndex;
  73         this.lastApplied = lastApplied;
  74         this.configParams = configParams;
  75         this.persistenceProvider = persistenceProvider;
  76         this.LOG = logger;
  77
  78         for(Map.Entry<String, String> e: peerAddresses.entrySet()) {
  79             peerInfoMap.put(e.getKey(), new PeerInfo(e.getKey(), e.getValue(), VotingState.VOTING));
  80         }
  81     }
  82
  83     public void setPayloadVersion(short payloadVersion) {
  84         this.payloadVersion = payloadVersion;
  85     }
  86
  87     @Override
  88     public short getPayloadVersion() {
  89         return payloadVersion;
  90     }
  91
  92     public void setConfigParams(ConfigParams configParams) {
  93         this.configParams = configParams;
  94     }
  95
  96     @Override
  97     public ActorRef actorOf(Props props){
  98         return context.actorOf(props);
  99     }
 100
 101     @Override
 102     public ActorSelection actorSelection(String path){
 103         return context.actorSelection(path);
 104     }
 105
 106     @Override
 107     public String getId() {
 108         return id;
 109     }
 110
 111     @Override
 112     public ActorRef getActor() {
 113         return actor;
 114     }
 115
 116     @Override
 117     public ElectionTerm getTermInformation() {
 118         return termInformation;
 119     }
 120
 121     @Override
 122     public long getCommitIndex() {
 123         return commitIndex;
 124     }
 125
 126     @Override public void setCommitIndex(long commitIndex) {
 127         this.commitIndex = commitIndex;
 128     }
 129
 130     @Override
 131     public long getLastApplied() {
 132         return lastApplied;
 133     }
 134
 135     @Override
 136     public void setLastApplied(long lastApplied) {
 137         this.lastApplied = lastApplied;
 138     }
 139
 140     @Override
 141     public void setReplicatedLog(ReplicatedLog replicatedLog) {
 142         this.replicatedLog = replicatedLog;
 143     }
 144
 145     @Override
 146     public ReplicatedLog getReplicatedLog() {
 147         return replicatedLog;
 148     }
 149
 150     @Override public ActorSystem getActorSystem() {
 151         return context.system();
 152     }
 153
 154     @Override public Logger getLogger() {
 155         return this.LOG;
 156     }
 157
 158     @Override
 159     public Collection<String> getPeerIds() {
 160         return peerInfoMap.keySet();
 161     }
 162
 163     @Override
 164     public Collection<PeerInfo> getPeers() {
 165         return peerInfoMap.values();
 166     }
 167
 168     @Override
 169     public PeerInfo getPeerInfo(String peerId) {
 170         return peerInfoMap.get(peerId);
 171     }
 172
 173     @Override
 174     public String getPeerAddress(String peerId) {
 175         String peerAddress = null;
 176         PeerInfo peerInfo = peerInfoMap.get(peerId);
 177         if(peerInfo != null) {
 178             peerAddress = peerInfo.getAddress();
 179             if(peerAddress == null) {
 180                 peerAddress = configParams.getPeerAddressResolver().resolve(peerId);
 181                 peerInfo.setAddress(peerAddress);
 182             }
 183         } else {
 184             peerAddress = configParams.getPeerAddressResolver().resolve(peerId);
 185         }
 186
 187         return peerAddress;
 188     }
 189
 190     @Override
 191     public void updatePeerIds(ServerConfigurationPayload serverConfig){
 192
 193         Set<String> currentPeers = new HashSet<>(this.getPeerIds());
 194         for(ServerInfo server: serverConfig.getServerConfig()) {
 195             if(!getId().equals(server.getId())) {
 196                 VotingState votingState = server.isVoting() ? VotingState.VOTING: VotingState.NON_VOTING;
 197                 if(!currentPeers.contains(server.getId())) {
 198                     this.addToPeers(server.getId(), null, votingState);
 199                 } else {
 200                     this.getPeerInfo(server.getId()).setVotingState(votingState);
 201                     currentPeers.remove(server.getId());
 202                 }
 203             }
 204         }
 205
 206         for(String peerIdToRemove: currentPeers) {
 207             this.removePeer(peerIdToRemove);
 208         }
 209         setDynamicServerConfigurationInUse();
 210     }
 211
 212     @Override public ConfigParams getConfigParams() {
 213         return configParams;
 214     }
 215
 216     @Override
 217     public void addToPeers(String id, String address, VotingState votingState) {
 218         peerInfoMap.put(id, new PeerInfo(id, address, votingState));
 219     }
 220
 221     @Override public void removePeer(String name) {
 222         peerInfoMap.remove(name);
 223     }
 224
 225     @Override public ActorSelection getPeerActorSelection(String peerId) {
 226         String peerAddress = getPeerAddress(peerId);
 227         if(peerAddress != null){
 228             return actorSelection(peerAddress);
 229         }
 230         return null;
 231     }
 232
 233     @Override
 234     public void setPeerAddress(String peerId, String peerAddress) {
 235         PeerInfo peerInfo = peerInfoMap.get(peerId);
 236         if(peerInfo != null) {
 237             LOG.info("Peer address for peer {} set to {}", peerId, peerAddress);
 238             peerInfo.setAddress(peerAddress);
 239         }
 240     }
 241
 242     @Override
 243     public SnapshotManager getSnapshotManager() {
 244         if(snapshotManager == null){
 245             snapshotManager = new SnapshotManager(this, LOG);
 246         }
 247         return snapshotManager;
 248     }
 249
 250     @Override
 251     public long getTotalMemory() {
 252         return totalMemoryRetriever != null ? totalMemoryRetriever.get() : Runtime.getRuntime().totalMemory();
 253     }
 254
 255     @Override
 256     public void setTotalMemoryRetriever(Supplier<Long> retriever) {
 257         totalMemoryRetriever = retriever;
 258     }
 259
 260     @Override
 261     public boolean hasFollowers() {
 262         return getPeerIds().size() > 0;
 263     }
 264
 265     @Override
 266     public DataPersistenceProvider getPersistenceProvider() {
 267         return persistenceProvider;
 268     }
 269
 270
 271     @Override
 272     public RaftPolicy getRaftPolicy() {
 273         return configParams.getRaftPolicy();
 274     }
 275
 276     @Override
 277     public boolean isDynamicServerConfigurationInUse() {
 278         return dynamicServerConfiguration;
 279     }
 280
 281     @Override
 282     public void setDynamicServerConfigurationInUse() {
 283         this.dynamicServerConfiguration = true;
 284     }
 285
 286     @Override
 287     public ServerConfigurationPayload getPeerServerInfo() {
 288         if (!isDynamicServerConfigurationInUse()) {
 289             return null;
 290         }
 291         Collection<PeerInfo> peers = getPeers();
 292         List<ServerInfo> newConfig = new ArrayList<>(peers.size() + 1);
 293         for(PeerInfo peer: peers) {
 294             newConfig.add(new ServerInfo(peer.getId(), peer.isVoting()));
 295         }
 296
 297         newConfig.add(new ServerInfo(getId(), true));
 298         return (new ServerConfigurationPayload(newConfig));
 299     }
 300 }