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