opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/registry/gossip/BucketStoreActor.java

   1 /*
   2  * Copyright (c) 2014, 2017 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.remote.rpc.registry.gossip;
   9
  10 import static com.google.common.base.Preconditions.checkState;
  11 import static com.google.common.base.Verify.verify;
  12 import static java.util.Objects.requireNonNull;
  13 import static org.opendaylight.controller.remote.rpc.registry.gossip.BucketStoreAccess.Singletons.GET_ALL_BUCKETS;
  14 import static org.opendaylight.controller.remote.rpc.registry.gossip.BucketStoreAccess.Singletons.GET_BUCKET_VERSIONS;
  15
  16 import akka.actor.ActorRef;
  17 import akka.actor.ActorRefProvider;
  18 import akka.actor.Address;
  19 import akka.actor.PoisonPill;
  20 import akka.actor.Terminated;
  21 import akka.cluster.ClusterActorRefProvider;
  22 import akka.persistence.DeleteSnapshotsFailure;
  23 import akka.persistence.DeleteSnapshotsSuccess;
  24 import akka.persistence.RecoveryCompleted;
  25 import akka.persistence.SaveSnapshotFailure;
  26 import akka.persistence.SaveSnapshotSuccess;
  27 import akka.persistence.SnapshotOffer;
  28 import akka.persistence.SnapshotSelectionCriteria;
  29 import com.google.common.annotations.VisibleForTesting;
  30 import com.google.common.collect.HashMultimap;
  31 import com.google.common.collect.ImmutableMap;
  32 import com.google.common.collect.SetMultimap;
  33 import java.util.Collection;
  34 import java.util.HashMap;
  35 import java.util.Map;
  36 import java.util.Map.Entry;
  37 import java.util.Optional;
  38 import java.util.function.Consumer;
  39 import org.opendaylight.controller.cluster.common.actor.AbstractUntypedPersistentActorWithMetering;
  40 import org.opendaylight.controller.remote.rpc.RemoteOpsProviderConfig;
  41
  42 /**
  43  * A store that syncs its data across nodes in the cluster.
  44  * It maintains a {@link org.opendaylight.controller.remote.rpc.registry.gossip.Bucket} per node. Buckets are versioned.
  45  * A node can write ONLY to its bucket. This way, write conflicts are avoided.
  46  *
  47  * <p>
  48  * Buckets are sync'ed across nodes using Gossip protocol (http://en.wikipedia.org/wiki/Gossip_protocol).
  49  * This store uses a {@link org.opendaylight.controller.remote.rpc.registry.gossip.Gossiper}.
  50  */
  51 public abstract class BucketStoreActor<T extends BucketData<T>> extends
  52         AbstractUntypedPersistentActorWithMetering {
  53     // Internal marker interface for messages which are just bridges to execute a method
  54     @FunctionalInterface
  55     private interface ExecuteInActor extends Consumer<BucketStoreActor<?>> {
  56
  57     }
  58
  59     /**
  60      * Buckets owned by other known nodes in the cluster.
  61      */
  62     private final Map<Address, Bucket<T>> remoteBuckets = new HashMap<>();
  63
  64     /**
  65      * Bucket version for every known node in the cluster including this node.
  66      */
  67     private final Map<Address, Long> versions = new HashMap<>();
  68
  69     /**
  70      * {@link ActorRef}s being watched for liveness due to being referenced in bucket data. Each actor is monitored
  71      * once, possibly being tied to multiple addresses (and by extension, buckets).
  72      */
  73     private final SetMultimap<ActorRef, Address> watchedActors = HashMultimap.create(1, 1);
  74
  75     private final RemoteOpsProviderConfig config;
  76     private final String persistenceId;
  77
  78     /**
  79      * Cluster address for this node.
  80      */
  81     private Address selfAddress;
  82
  83     /**
  84      * Bucket owned by the node. Initialized during recovery (due to incarnation number).
  85      */
  86     private LocalBucket<T> localBucket;
  87     private T initialData;
  88     private Integer incarnation;
  89     private boolean persisting;
  90
  91     protected BucketStoreActor(final RemoteOpsProviderConfig config, final String persistenceId, final T initialData) {
  92         this.config = requireNonNull(config);
  93         this.initialData = requireNonNull(initialData);
  94         this.persistenceId = requireNonNull(persistenceId);
  95     }
  96
  97     static ExecuteInActor getBucketsByMembersMessage(final Collection<Address> members) {
  98         return actor -> actor.getBucketsByMembers(members);
  99     }
 100
 101     static ExecuteInActor removeBucketMessage(final Address addr) {
 102         return actor -> actor.removeBucket(addr);
 103     }
 104
 105     static ExecuteInActor updateRemoteBucketsMessage(final Map<Address, Bucket<?>> buckets) {
 106         return actor -> actor.updateRemoteBuckets(buckets);
 107     }
 108
 109     static ExecuteInActor getLocalDataMessage() {
 110         return actor -> actor.getSender().tell(actor.getLocalData(), actor.getSelf());
 111     }
 112
 113     static ExecuteInActor getRemoteBucketsMessage() {
 114         return actor -> actor.getSender().tell(ImmutableMap.copyOf(actor.getRemoteBuckets()), actor.getSelf());
 115     }
 116
 117     public final T getLocalData() {
 118         return getLocalBucket().getData();
 119     }
 120
 121     public final Map<Address, Bucket<T>> getRemoteBuckets() {
 122         return remoteBuckets;
 123     }
 124
 125     public final Map<Address, Long> getVersions() {
 126         return versions;
 127     }
 128
 129     @Override
 130     public final String persistenceId() {
 131         return persistenceId;
 132     }
 133
 134     @Override
 135     public void preStart() {
 136         ActorRefProvider provider = getContext().provider();
 137         selfAddress = provider.getDefaultAddress();
 138
 139         if (provider instanceof ClusterActorRefProvider) {
 140             getContext().actorOf(Gossiper.props(config).withMailbox(config.getMailBoxName()), "gossiper");
 141         }
 142     }
 143
 144     @Override
 145     protected void handleCommand(final Object message) throws Exception {
 146         if (GET_ALL_BUCKETS == message) {
 147             // GetAllBuckets is used only in testing
 148             getSender().tell(getAllBuckets(), self());
 149             return;
 150         }
 151
 152         if (persisting) {
 153             handleSnapshotMessage(message);
 154             return;
 155         }
 156
 157         if (message instanceof ExecuteInActor execute) {
 158             execute.accept(this);
 159         } else if (GET_BUCKET_VERSIONS == message) {
 160             // FIXME: do we need to send ourselves?
 161             getSender().tell(ImmutableMap.copyOf(versions), getSelf());
 162         } else if (message instanceof Terminated terminated) {
 163             actorTerminated(terminated);
 164         } else if (message instanceof DeleteSnapshotsSuccess deleteSuccess) {
 165             LOG.debug("{}: got command: {}", persistenceId(), deleteSuccess);
 166         } else if (message instanceof DeleteSnapshotsFailure deleteFailure) {
 167             LOG.warn("{}: failed to delete prior snapshots", persistenceId(), deleteFailure.cause());
 168         } else {
 169             LOG.debug("Unhandled message [{}]", message);
 170             unhandled(message);
 171         }
 172     }
 173
 174     private void handleSnapshotMessage(final Object message) {
 175         if (message instanceof SaveSnapshotFailure saveFailure) {
 176             LOG.error("{}: failed to persist state", persistenceId(), saveFailure.cause());
 177             persisting = false;
 178             self().tell(PoisonPill.getInstance(), ActorRef.noSender());
 179         } else if (message instanceof SaveSnapshotSuccess saveSuccess) {
 180             LOG.debug("{}: got command: {}", persistenceId(), saveSuccess);
 181             deleteSnapshots(new SnapshotSelectionCriteria(scala.Long.MaxValue(), saveSuccess.metadata().timestamp() - 1,
 182                 0L, 0L));
 183             persisting = false;
 184             unstash();
 185         } else {
 186             LOG.debug("{}: stashing command {}", persistenceId(), message);
 187             stash();
 188         }
 189     }
 190
 191     @Override
 192     protected final void handleRecover(final Object message) {
 193         if (message instanceof RecoveryCompleted) {
 194             if (incarnation != null) {
 195                 incarnation = incarnation + 1;
 196             } else {
 197                 incarnation = 0;
 198             }
 199
 200             this.localBucket = new LocalBucket<>(incarnation, initialData);
 201             initialData = null;
 202             LOG.debug("{}: persisting new incarnation {}", persistenceId(), incarnation);
 203             persisting = true;
 204             saveSnapshot(incarnation);
 205         } else if (message instanceof SnapshotOffer snapshotOffer) {
 206             incarnation = (Integer) snapshotOffer.snapshot();
 207             LOG.debug("{}: recovered incarnation {}", persistenceId(), incarnation);
 208         } else {
 209             LOG.warn("{}: ignoring recovery message {}", persistenceId(), message);
 210         }
 211     }
 212
 213     protected final RemoteOpsProviderConfig getConfig() {
 214         return config;
 215     }
 216
 217     protected final void updateLocalBucket(final T data) {
 218         final LocalBucket<T> local = getLocalBucket();
 219         final boolean bumpIncarnation = local.setData(data);
 220         versions.put(selfAddress, local.getVersion());
 221
 222         if (bumpIncarnation) {
 223             LOG.debug("Version wrapped. incrementing incarnation");
 224
 225             verify(incarnation < Integer.MAX_VALUE, "Ran out of incarnations, cannot continue");
 226             incarnation = incarnation + 1;
 227
 228             persisting = true;
 229             saveSnapshot(incarnation);
 230         }
 231     }
 232
 233     /**
 234      * Callback to subclasses invoked when a bucket is removed.
 235      *
 236      * @param address Remote address
 237      * @param bucket Bucket removed
 238      */
 239     protected abstract void onBucketRemoved(Address address, Bucket<T> bucket);
 240
 241     /**
 242      * Callback to subclasses invoked when the set of remote buckets is updated.
 243      *
 244      * @param newBuckets Map of address to new bucket. Never null, but can be empty.
 245      */
 246     protected abstract void onBucketsUpdated(Map<Address, Bucket<T>> newBuckets);
 247
 248     /**
 249      * Helper to collect all known buckets.
 250      *
 251      * @return self owned + remote buckets
 252      */
 253     private Map<Address, Bucket<T>> getAllBuckets() {
 254         Map<Address, Bucket<T>> all = new HashMap<>(remoteBuckets.size() + 1);
 255
 256         //first add the local bucket
 257         all.put(selfAddress, getLocalBucket().snapshot());
 258
 259         //then get all remote buckets
 260         all.putAll(remoteBuckets);
 261
 262         return all;
 263     }
 264
 265     /**
 266      * Helper to collect buckets for requested members.
 267      *
 268      * @param members requested members
 269      */
 270     private void getBucketsByMembers(final Collection<Address> members) {
 271         Map<Address, Bucket<T>> buckets = new HashMap<>();
 272
 273         //first add the local bucket if asked
 274         if (members.contains(selfAddress)) {
 275             buckets.put(selfAddress, getLocalBucket().snapshot());
 276         }
 277
 278         //then get buckets for requested remote nodes
 279         for (Address address : members) {
 280             if (remoteBuckets.containsKey(address)) {
 281                 buckets.put(address, remoteBuckets.get(address));
 282             }
 283         }
 284
 285         getSender().tell(buckets, getSelf());
 286     }
 287
 288     private void removeBucket(final Address addr) {
 289         final Bucket<T> bucket = remoteBuckets.remove(addr);
 290         if (bucket != null) {
 291             bucket.getWatchActor().ifPresent(ref -> removeWatch(addr, ref));
 292             onBucketRemoved(addr, bucket);
 293         }
 294         versions.remove(addr);
 295     }
 296
 297     /**
 298      * Update local copy of remote buckets where local copy's version is older.
 299      *
 300      * @param receivedBuckets buckets sent by remote
 301      *                        {@link org.opendaylight.controller.remote.rpc.registry.gossip.Gossiper}
 302      */
 303     @VisibleForTesting
 304     void updateRemoteBuckets(final Map<Address, Bucket<?>> receivedBuckets) {
 305         LOG.debug("{}: receiveUpdateRemoteBuckets: {}", selfAddress, receivedBuckets);
 306         if (receivedBuckets == null || receivedBuckets.isEmpty()) {
 307             //nothing to do
 308             return;
 309         }
 310
 311         final Map<Address, Bucket<T>> newBuckets = new HashMap<>(receivedBuckets.size());
 312         for (Entry<Address, Bucket<?>> entry : receivedBuckets.entrySet()) {
 313             final Address addr = entry.getKey();
 314
 315             if (selfAddress.equals(addr)) {
 316                 // Remote cannot update our bucket
 317                 continue;
 318             }
 319
 320             @SuppressWarnings("unchecked")
 321             final Bucket<T> receivedBucket = (Bucket<T>) entry.getValue();
 322             if (receivedBucket == null) {
 323                 LOG.debug("Ignoring null bucket from {}", addr);
 324                 continue;
 325             }
 326
 327             // update only if remote version is newer
 328             final long remoteVersion = receivedBucket.getVersion();
 329             final Long localVersion = versions.get(addr);
 330             if (localVersion != null && remoteVersion <= localVersion.longValue()) {
 331                 LOG.debug("Ignoring down-versioned bucket from {} ({} local {} remote)", addr, localVersion,
 332                     remoteVersion);
 333                 continue;
 334             }
 335             newBuckets.put(addr, receivedBucket);
 336             versions.put(addr, remoteVersion);
 337             final Bucket<T> prevBucket = remoteBuckets.put(addr, receivedBucket);
 338
 339             // Deal with DeathWatch subscriptions
 340             final Optional<ActorRef> prevRef = prevBucket != null ? prevBucket.getWatchActor() : Optional.empty();
 341             final Optional<ActorRef> curRef = receivedBucket.getWatchActor();
 342             if (!curRef.equals(prevRef)) {
 343                 prevRef.ifPresent(ref -> removeWatch(addr, ref));
 344                 curRef.ifPresent(ref -> addWatch(addr, ref));
 345             }
 346
 347             LOG.debug("Updating bucket from {} to version {}", entry.getKey(), remoteVersion);
 348         }
 349
 350         LOG.debug("State after update - Local Bucket [{}], Remote Buckets [{}]", localBucket, remoteBuckets);
 351
 352         onBucketsUpdated(newBuckets);
 353     }
 354
 355     private void addWatch(final Address addr, final ActorRef ref) {
 356         if (!watchedActors.containsKey(ref)) {
 357             getContext().watch(ref);
 358             LOG.debug("Watching {}", ref);
 359         }
 360         watchedActors.put(ref, addr);
 361     }
 362
 363     private void removeWatch(final Address addr, final ActorRef ref) {
 364         watchedActors.remove(ref, addr);
 365         if (!watchedActors.containsKey(ref)) {
 366             getContext().unwatch(ref);
 367             LOG.debug("No longer watching {}", ref);
 368         }
 369     }
 370
 371     private void actorTerminated(final Terminated message) {
 372         LOG.info("Actor termination {} received", message);
 373
 374         for (Address addr : watchedActors.removeAll(message.getActor())) {
 375             versions.remove(addr);
 376             final Bucket<T> bucket = remoteBuckets.remove(addr);
 377             if (bucket != null) {
 378                 LOG.debug("Source actor dead, removing bucket {} from {}", bucket, addr);
 379                 onBucketRemoved(addr, bucket);
 380             }
 381         }
 382     }
 383
 384     @VisibleForTesting
 385     protected boolean isPersisting() {
 386         return persisting;
 387     }
 388
 389     private LocalBucket<T> getLocalBucket() {
 390         checkState(localBucket != null, "Attempted to access local bucket before recovery completed");
 391         return localBucket;
 392     }
 393 }