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