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