opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/shardmanager/ShardInformation.java

   1 /*
   2  * Copyright (c) 2016 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 package org.opendaylight.controller.cluster.datastore.shardmanager;
   9
  10 import akka.actor.ActorRef;
  11 import akka.actor.Props;
  12 import akka.serialization.Serialization;
  13 import com.google.common.base.Preconditions;
  14 import com.google.common.base.Strings;
  15 import java.util.HashSet;
  16 import java.util.Iterator;
  17 import java.util.Map;
  18 import java.util.Objects;
  19 import java.util.Optional;
  20 import java.util.Set;
  21 import javax.annotation.Nullable;
  22 import org.opendaylight.controller.cluster.access.concepts.MemberName;
  23 import org.opendaylight.controller.cluster.datastore.DatastoreContext;
  24 import org.opendaylight.controller.cluster.datastore.Shard;
  25 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
  26 import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved;
  27 import org.opendaylight.controller.cluster.datastore.messages.PeerDown;
  28 import org.opendaylight.controller.cluster.datastore.messages.PeerUp;
  29 import org.opendaylight.controller.cluster.datastore.shardmanager.ShardManager.OnShardInitialized;
  30 import org.opendaylight.controller.cluster.datastore.shardmanager.ShardManager.OnShardReady;
  31 import org.opendaylight.controller.cluster.raft.RaftState;
  32 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
  33 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  34 import org.slf4j.Logger;
  35 import org.slf4j.LoggerFactory;
  36
  37 final class ShardInformation {
  38     private static final Logger LOG = LoggerFactory.getLogger(ShardInformation.class);
  39
  40     private final Set<OnShardInitialized> onShardInitializedSet = new HashSet<>();
  41     private final Map<String, String> initialPeerAddresses;
  42     private final ShardPeerAddressResolver addressResolver;
  43     private final ShardIdentifier shardId;
  44     private final String shardName;
  45     private ActorRef actor;
  46     private Optional<DataTree> localShardDataTree;
  47     private boolean leaderAvailable = false;
  48
  49     // flag that determines if the actor is ready for business
  50     private boolean actorInitialized = false;
  51
  52     private boolean followerSyncStatus = false;
  53
  54     private String role ;
  55     private String leaderId;
  56     private short leaderVersion;
  57
  58     private DatastoreContext datastoreContext;
  59     private Shard.AbstractBuilder<?, ?> builder;
  60     private boolean isActiveMember = true;
  61
  62     ShardInformation(String shardName, ShardIdentifier shardId,
  63             Map<String, String> initialPeerAddresses, DatastoreContext datastoreContext,
  64             Shard.AbstractBuilder<?, ?> builder, ShardPeerAddressResolver addressResolver) {
  65         this.shardName = shardName;
  66         this.shardId = shardId;
  67         this.initialPeerAddresses = initialPeerAddresses;
  68         this.datastoreContext = datastoreContext;
  69         this.builder = builder;
  70         this.addressResolver = addressResolver;
  71     }
  72
  73     Props newProps(SchemaContext schemaContext) {
  74         Preconditions.checkNotNull(builder);
  75         Props props = builder.id(shardId).peerAddresses(initialPeerAddresses).datastoreContext(datastoreContext).
  76                 schemaContext(schemaContext).props();
  77         builder = null;
  78         return props;
  79     }
  80
  81     String getShardName() {
  82         return shardName;
  83     }
  84
  85     @Nullable
  86     ActorRef getActor(){
  87         return actor;
  88     }
  89
  90     void setActor(ActorRef actor) {
  91         this.actor = actor;
  92     }
  93
  94     ShardIdentifier getShardId() {
  95         return shardId;
  96     }
  97
  98     void setLocalDataTree(Optional<DataTree> localShardDataTree) {
  99         this.localShardDataTree = localShardDataTree;
 100     }
 101
 102     Optional<DataTree> getLocalShardDataTree() {
 103         return localShardDataTree;
 104     }
 105
 106     DatastoreContext getDatastoreContext() {
 107         return datastoreContext;
 108     }
 109
 110     void setDatastoreContext(DatastoreContext datastoreContext, ActorRef sender) {
 111         this.datastoreContext = datastoreContext;
 112         if (actor != null) {
 113             LOG.debug("Sending new DatastoreContext to {}", shardId);
 114             actor.tell(this.datastoreContext, sender);
 115         }
 116     }
 117
 118     void updatePeerAddress(String peerId, String peerAddress, ActorRef sender){
 119         LOG.info("updatePeerAddress for peer {} with address {}", peerId, peerAddress);
 120
 121         if(actor != null) {
 122             if(LOG.isDebugEnabled()) {
 123                 LOG.debug("Sending PeerAddressResolved for peer {} with address {} to {}",
 124                         peerId, peerAddress, actor.path());
 125             }
 126
 127             actor.tell(new PeerAddressResolved(peerId, peerAddress), sender);
 128         }
 129
 130         notifyOnShardInitializedCallbacks();
 131     }
 132
 133     void peerDown(MemberName memberName, String peerId, ActorRef sender) {
 134         if(actor != null) {
 135             actor.tell(new PeerDown(memberName, peerId), sender);
 136         }
 137     }
 138
 139     void peerUp(MemberName memberName, String peerId, ActorRef sender) {
 140         if(actor != null) {
 141             actor.tell(new PeerUp(memberName, peerId), sender);
 142         }
 143     }
 144
 145     boolean isShardReady() {
 146         return !RaftState.Candidate.name().equals(role) && !Strings.isNullOrEmpty(role);
 147     }
 148
 149     boolean isShardReadyWithLeaderId() {
 150         return leaderAvailable && isShardReady() && !RaftState.IsolatedLeader.name().equals(role) &&
 151                 (isLeader() || addressResolver.resolve(leaderId) != null);
 152     }
 153
 154     boolean isShardInitialized() {
 155         return getActor() != null && actorInitialized;
 156     }
 157
 158     boolean isLeader() {
 159         return Objects.equals(leaderId, shardId.toString());
 160     }
 161
 162     String getSerializedLeaderActor() {
 163         if(isLeader()) {
 164             return Serialization.serializedActorPath(getActor());
 165         } else {
 166             return addressResolver.resolve(leaderId);
 167         }
 168     }
 169
 170     void setActorInitialized() {
 171         LOG.debug("Shard {} is initialized", shardId);
 172
 173         this.actorInitialized = true;
 174
 175         notifyOnShardInitializedCallbacks();
 176     }
 177
 178     private void notifyOnShardInitializedCallbacks() {
 179         if(onShardInitializedSet.isEmpty()) {
 180             return;
 181         }
 182
 183         boolean ready = isShardReadyWithLeaderId();
 184
 185         LOG.debug("Shard {} is {} - notifying {} OnShardInitialized callbacks", shardId,
 186             ready ? "ready" : "initialized", onShardInitializedSet.size());
 187
 188         Iterator<OnShardInitialized> iter = onShardInitializedSet.iterator();
 189         while(iter.hasNext()) {
 190             OnShardInitialized onShardInitialized = iter.next();
 191             if (!(onShardInitialized instanceof OnShardReady) || ready) {
 192                 iter.remove();
 193                 onShardInitialized.getTimeoutSchedule().cancel();
 194                 onShardInitialized.getReplyRunnable().run();
 195             }
 196         }
 197     }
 198
 199     void addOnShardInitialized(OnShardInitialized onShardInitialized) {
 200         onShardInitializedSet.add(onShardInitialized);
 201     }
 202
 203     void removeOnShardInitialized(OnShardInitialized onShardInitialized) {
 204         onShardInitializedSet.remove(onShardInitialized);
 205     }
 206
 207     void setRole(String newRole) {
 208         this.role = newRole;
 209
 210         notifyOnShardInitializedCallbacks();
 211     }
 212
 213     void setFollowerSyncStatus(boolean syncStatus){
 214         this.followerSyncStatus = syncStatus;
 215     }
 216
 217     boolean isInSync(){
 218         if(RaftState.Follower.name().equals(this.role)){
 219             return followerSyncStatus;
 220         } else if(RaftState.Leader.name().equals(this.role)){
 221             return true;
 222         }
 223
 224         return false;
 225     }
 226
 227     boolean setLeaderId(String leaderId) {
 228         boolean changed = !Objects.equals(this.leaderId, leaderId);
 229         this.leaderId = leaderId;
 230         if(leaderId != null) {
 231             this.leaderAvailable = true;
 232         }
 233         notifyOnShardInitializedCallbacks();
 234
 235         return changed;
 236     }
 237
 238     String getLeaderId() {
 239         return leaderId;
 240     }
 241
 242     void setLeaderAvailable(boolean leaderAvailable) {
 243         this.leaderAvailable = leaderAvailable;
 244
 245         if(leaderAvailable) {
 246             notifyOnShardInitializedCallbacks();
 247         }
 248     }
 249
 250     short getLeaderVersion() {
 251         return leaderVersion;
 252     }
 253
 254     void setLeaderVersion(short leaderVersion) {
 255         this.leaderVersion = leaderVersion;
 256     }
 257
 258     boolean isActiveMember() {
 259         return isActiveMember;
 260     }
 261
 262     void setActiveMember(boolean isActiveMember) {
 263         this.isActiveMember = isActiveMember;
 264     }
 265 }