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             LOG.debug("Sending PeerAddressResolved for peer {} with address {} to {}", peerId,
 123                     peerAddress, actor.path());
 124
 125             actor.tell(new PeerAddressResolved(peerId, peerAddress), sender);
 126         }
 127
 128         notifyOnShardInitializedCallbacks();
 129     }
 130
 131     void peerDown(MemberName memberName, String peerId, ActorRef sender) {
 132         if (actor != null) {
 133             actor.tell(new PeerDown(memberName, peerId), sender);
 134         }
 135     }
 136
 137     void peerUp(MemberName memberName, String peerId, ActorRef sender) {
 138         if (actor != null) {
 139             actor.tell(new PeerUp(memberName, peerId), sender);
 140         }
 141     }
 142
 143     boolean isShardReady() {
 144         return !RaftState.Candidate.name().equals(role) && !Strings.isNullOrEmpty(role);
 145     }
 146
 147     boolean isShardReadyWithLeaderId() {
 148         return leaderAvailable && isShardReady() && !RaftState.IsolatedLeader.name().equals(role)
 149                 && !RaftState.PreLeader.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.equals(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     String getRole() {
 213         return role;
 214     }
 215
 216     void setFollowerSyncStatus(boolean syncStatus) {
 217         this.followerSyncStatus = syncStatus;
 218     }
 219
 220     boolean isInSync() {
 221         if (RaftState.Follower.name().equals(this.role)) {
 222             return followerSyncStatus;
 223         } else if (RaftState.Leader.name().equals(this.role)) {
 224             return true;
 225         }
 226
 227         return false;
 228     }
 229
 230     boolean setLeaderId(String leaderId) {
 231         final boolean changed = !Objects.equals(this.leaderId, leaderId);
 232         this.leaderId = leaderId;
 233         if (leaderId != null) {
 234             this.leaderAvailable = true;
 235         }
 236         notifyOnShardInitializedCallbacks();
 237
 238         return changed;
 239     }
 240
 241     String getLeaderId() {
 242         return leaderId;
 243     }
 244
 245     void setLeaderAvailable(boolean leaderAvailable) {
 246         this.leaderAvailable = leaderAvailable;
 247
 248         if (leaderAvailable) {
 249             notifyOnShardInitializedCallbacks();
 250         }
 251     }
 252
 253     short getLeaderVersion() {
 254         return leaderVersion;
 255     }
 256
 257     void setLeaderVersion(short leaderVersion) {
 258         this.leaderVersion = leaderVersion;
 259     }
 260
 261     boolean isActiveMember() {
 262         return isActiveMember;
 263     }
 264
 265     void setActiveMember(boolean isActiveMember) {
 266         this.isActiveMember = isActiveMember;
 267     }
 268 }