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

   1 /*
   2  * Copyright (c) 2014 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 akka.actor.ActorRef;
  11 import akka.actor.ActorRefProvider;
  12 import akka.actor.ActorSelection;
  13 import akka.actor.Address;
  14 import akka.actor.Cancellable;
  15 import akka.cluster.Cluster;
  16 import akka.cluster.ClusterActorRefProvider;
  17 import akka.cluster.ClusterEvent;
  18 import akka.cluster.Member;
  19 import akka.dispatch.Mapper;
  20 import akka.event.Logging;
  21 import akka.event.LoggingAdapter;
  22 import akka.pattern.Patterns;
  23 import org.opendaylight.controller.cluster.common.actor.AbstractUntypedActorWithMetering;
  24 import org.opendaylight.controller.remote.rpc.RemoteRpcProviderConfig;
  25 import scala.concurrent.Future;
  26 import scala.concurrent.duration.FiniteDuration;
  27
  28 import java.util.ArrayList;
  29 import java.util.HashSet;
  30 import java.util.List;
  31 import java.util.Map;
  32 import java.util.Set;
  33 import java.util.concurrent.ThreadLocalRandom;
  34 import java.util.concurrent.TimeUnit;
  35
  36 import static org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketVersions;
  37 import static org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketVersionsReply;
  38 import static org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketsByMembers;
  39 import static org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketsByMembersReply;
  40 import static org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.UpdateRemoteBuckets;
  41 import static org.opendaylight.controller.remote.rpc.registry.gossip.Messages.GossiperMessages.GossipEnvelope;
  42 import static org.opendaylight.controller.remote.rpc.registry.gossip.Messages.GossiperMessages.GossipStatus;
  43 import static org.opendaylight.controller.remote.rpc.registry.gossip.Messages.GossiperMessages.GossipTick;
  44
  45 /**
  46  * Gossiper that syncs bucket store across nodes in the cluster.
  47  * <p/>
  48  * It keeps a local scheduler that periodically sends Gossip ticks to
  49  * itself to send bucket store's bucket versions to a randomly selected remote
  50  * gossiper.
  51  * <p/>
  52  * When bucket versions are received from a remote gossiper, it is compared
  53  * with bucket store's bucket versions. Which ever buckets are newer
  54  * locally, are sent to remote gossiper. If any bucket is older in bucket store,
  55  * a gossip status is sent to remote gossiper so that it can send the newer buckets.
  56  * <p/>
  57  * When a bucket is received from a remote gossiper, its sent to the bucket store
  58  * for update.
  59  *
  60  */
  61
  62 public class Gossiper extends AbstractUntypedActorWithMetering {
  63
  64     final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
  65
  66     private Cluster cluster;
  67
  68     /**
  69      * ActorSystem's address for the current cluster node.
  70      */
  71     private Address selfAddress;
  72
  73     /**
  74      * All known cluster members
  75      */
  76     private List<Address> clusterMembers = new ArrayList<>();
  77
  78     private Cancellable gossipTask;
  79
  80     private Boolean autoStartGossipTicks = true;
  81
  82     private RemoteRpcProviderConfig config;
  83
  84     public Gossiper(){
  85         config = new RemoteRpcProviderConfig(getContext().system().settings().config());
  86     }
  87
  88     /**
  89      * Helpful for testing
  90      * @param autoStartGossipTicks used for turning off gossip ticks during testing.
  91      *                             Gossip tick can be manually sent.
  92      */
  93     public Gossiper(Boolean autoStartGossipTicks){
  94         this.autoStartGossipTicks = autoStartGossipTicks;
  95     }
  96
  97     @Override
  98     public void preStart(){
  99         ActorRefProvider provider = getContext().provider();
 100         selfAddress = provider.getDefaultAddress();
 101
 102         if ( provider instanceof ClusterActorRefProvider ) {
 103             cluster = Cluster.get(getContext().system());
 104             cluster.subscribe(getSelf(),
 105                     ClusterEvent.initialStateAsEvents(),
 106                     ClusterEvent.MemberEvent.class,
 107                     ClusterEvent.UnreachableMember.class);
 108         }
 109
 110         if (autoStartGossipTicks) {
 111             gossipTask = getContext().system().scheduler().schedule(
 112                     new FiniteDuration(1, TimeUnit.SECONDS),        //initial delay
 113                     config.getGossipTickInterval(),                 //interval
 114                     getSelf(),                                       //target
 115                     new Messages.GossiperMessages.GossipTick(),      //message
 116                     getContext().dispatcher(),                       //execution context
 117                     getSelf()                                        //sender
 118             );
 119         }
 120     }
 121
 122     @Override
 123     public void postStop(){
 124         if (cluster != null)
 125             cluster.unsubscribe(getSelf());
 126         if (gossipTask != null)
 127             gossipTask.cancel();
 128     }
 129
 130     @Override
 131     protected void handleReceive(Object message) throws Exception {
 132         //Usually sent by self via gossip task defined above. But its not enforced.
 133         //These ticks can be sent by another actor as well which is esp. useful while testing
 134         if (message instanceof GossipTick)
 135             receiveGossipTick();
 136
 137             //Message from remote gossiper with its bucket versions
 138         else if (message instanceof GossipStatus)
 139             receiveGossipStatus((GossipStatus) message);
 140
 141             //Message from remote gossiper with buckets. This is usually in response to GossipStatus message
 142             //The contained buckets are newer as determined by the remote gossiper by comparing the GossipStatus
 143             //message with its local versions
 144         else if (message instanceof GossipEnvelope)
 145             receiveGossip((GossipEnvelope) message);
 146
 147         else if (message instanceof ClusterEvent.MemberUp) {
 148             receiveMemberUp(((ClusterEvent.MemberUp) message).member());
 149
 150         } else if (message instanceof ClusterEvent.MemberRemoved) {
 151             receiveMemberRemoveOrUnreachable(((ClusterEvent.MemberRemoved) message).member());
 152
 153         } else if ( message instanceof ClusterEvent.UnreachableMember){
 154             receiveMemberRemoveOrUnreachable(((ClusterEvent.UnreachableMember) message).member());
 155
 156         } else
 157             unhandled(message);
 158     }
 159
 160     /**
 161      * Remove member from local copy of member list. If member down is self, then stop the actor
 162      *
 163      * @param member who went down
 164      */
 165     void receiveMemberRemoveOrUnreachable(Member member) {
 166         //if its self, then stop itself
 167         if (selfAddress.equals(member.address())){
 168             getContext().stop(getSelf());
 169             return;
 170         }
 171
 172         clusterMembers.remove(member.address());
 173         log.debug("Removed member [{}], Active member list [{}]", member.address(), clusterMembers);
 174     }
 175
 176     /**
 177      * Add member to the local copy of member list if it doesnt already
 178      * @param member
 179      */
 180     void receiveMemberUp(Member member) {
 181
 182         if (selfAddress.equals(member.address()))
 183             return; //ignore up notification for self
 184
 185         if (!clusterMembers.contains(member.address()))
 186             clusterMembers.add(member.address());
 187
 188         log.debug("Added member [{}], Active member list [{}]", member.address(), clusterMembers);
 189     }
 190
 191     /**
 192      * Sends Gossip status to other members in the cluster. <br/>
 193      * 1. If there are no member, ignore the tick. </br>
 194      * 2. If there's only 1 member, send gossip status (bucket versions) to it. <br/>
 195      * 3. If there are more than one member, randomly pick one and send gossip status (bucket versions) to it.
 196      */
 197     void receiveGossipTick(){
 198         if (clusterMembers.size() == 0) return; //no members to send gossip status to
 199
 200         Address remoteMemberToGossipTo;
 201
 202         if (clusterMembers.size() == 1)
 203             remoteMemberToGossipTo = clusterMembers.get(0);
 204         else {
 205             Integer randomIndex = ThreadLocalRandom.current().nextInt(0, clusterMembers.size());
 206             remoteMemberToGossipTo = clusterMembers.get(randomIndex);
 207         }
 208
 209         log.debug("Gossiping to [{}]", remoteMemberToGossipTo);
 210         getLocalStatusAndSendTo(remoteMemberToGossipTo);
 211     }
 212
 213     /**
 214      * Process gossip status received from a remote gossiper. Remote versions are compared with
 215      * the local copy. <p>
 216      *
 217      * For each bucket
 218      * <ul>
 219      *  <li>If local copy is newer, the newer buckets are sent in GossipEnvelope to remote</li>
 220      *  <li>If local is older, GossipStatus is sent to remote so that it can reply with GossipEnvelope</li>
 221      *  <li>If both are same, noop</li>
 222      * </ul>
 223      *
 224      * @param status bucket versions from a remote member
 225      */
 226     void receiveGossipStatus(GossipStatus status){
 227         //Don't accept messages from non-members
 228         if (!clusterMembers.contains(status.from()))
 229             return;
 230
 231         final ActorRef sender = getSender();
 232         Future<Object> futureReply =
 233                 Patterns.ask(getContext().parent(), new GetBucketVersions(), config.getAskDuration());
 234
 235         futureReply.map(getMapperToProcessRemoteStatus(sender, status), getContext().dispatcher());
 236
 237     }
 238
 239     /**
 240      * Sends the received buckets in the envelope to the parent Bucket store.
 241      *
 242      * @param envelope contains buckets from a remote gossiper
 243      */
 244     void receiveGossip(GossipEnvelope envelope){
 245         //TODO: Add more validations
 246         if (!selfAddress.equals(envelope.to())) {
 247             log.debug("Ignoring message intended for someone else. From [{}] to [{}]", envelope.from(), envelope.to());
 248             return;
 249         }
 250
 251         updateRemoteBuckets(envelope.getBuckets());
 252
 253     }
 254
 255     /**
 256      * Helper to send received buckets to bucket store
 257      *
 258      * @param buckets
 259      */
 260     void updateRemoteBuckets(Map<Address, Bucket> buckets) {
 261
 262         UpdateRemoteBuckets updateRemoteBuckets = new UpdateRemoteBuckets(buckets);
 263         getContext().parent().tell(updateRemoteBuckets, getSelf());
 264     }
 265
 266     /**
 267      * Gets the buckets from bucket store for the given node addresses and sends them to remote gossiper
 268      *
 269      * @param remote     remote node to send Buckets to
 270      * @param addresses  node addresses whose buckets needs to be sent
 271      */
 272     void sendGossipTo(final ActorRef remote, final Set<Address> addresses){
 273
 274         Future<Object> futureReply =
 275                 Patterns.ask(getContext().parent(), new GetBucketsByMembers(addresses), config.getAskDuration());
 276         futureReply.map(getMapperToSendGossip(remote), getContext().dispatcher());
 277     }
 278
 279     /**
 280      * Gets bucket versions from bucket store and sends to the supplied address
 281      *
 282      * @param remoteActorSystemAddress remote gossiper to send to
 283      */
 284     void getLocalStatusAndSendTo(Address remoteActorSystemAddress){
 285
 286         //Get local status from bucket store and send to remote
 287         Future<Object> futureReply =
 288                 Patterns.ask(getContext().parent(), new GetBucketVersions(), config.getAskDuration());
 289
 290         //Find gossiper on remote system
 291         ActorSelection remoteRef = getContext().system().actorSelection(
 292                 remoteActorSystemAddress.toString() + getSelf().path().toStringWithoutAddress());
 293
 294         log.debug("Sending bucket versions to [{}]", remoteRef);
 295
 296         futureReply.map(getMapperToSendLocalStatus(remoteRef), getContext().dispatcher());
 297
 298     }
 299
 300     /**
 301      * Helper to send bucket versions received from local store
 302      * @param remote        remote gossiper to send versions to
 303      * @param localVersions bucket versions received from local store
 304      */
 305     void sendGossipStatusTo(ActorRef remote, Map<Address, Long> localVersions){
 306
 307         GossipStatus status = new GossipStatus(selfAddress, localVersions);
 308         remote.tell(status, getSelf());
 309     }
 310
 311     void sendGossipStatusTo(ActorSelection remote, Map<Address, Long> localVersions){
 312
 313         GossipStatus status = new GossipStatus(selfAddress, localVersions);
 314         remote.tell(status, getSelf());
 315     }
 316
 317     ///
 318     /// Private factories to create mappers
 319     ///
 320
 321     private Mapper<Object, Void> getMapperToSendLocalStatus(final ActorSelection remote){
 322
 323         return new Mapper<Object, Void>() {
 324             @Override
 325             public Void apply(Object replyMessage) {
 326                 if (replyMessage instanceof GetBucketVersionsReply) {
 327                     GetBucketVersionsReply reply = (GetBucketVersionsReply) replyMessage;
 328                     Map<Address, Long> localVersions = reply.getVersions();
 329
 330                     sendGossipStatusTo(remote, localVersions);
 331
 332                 }
 333                 return null;
 334             }
 335         };
 336     }
 337
 338     /**
 339      * Process bucket versions received from
 340      * {@link org.opendaylight.controller.remote.rpc.registry.gossip.BucketStore}.
 341      * Then this method compares remote bucket versions with local bucket versions.
 342      * <ul>
 343      *     <li>The buckets that are newer locally, send
 344      *     {@link org.opendaylight.controller.remote.rpc.registry.gossip.Messages.GossiperMessages.GossipEnvelope}
 345      *     to remote
 346      *     <li>The buckets that are older locally, send
 347      *     {@link org.opendaylight.controller.remote.rpc.registry.gossip.Messages.GossiperMessages.GossipStatus}
 348      *     to remote so that remote sends GossipEnvelop.
 349      * </ul>
 350      *
 351      * @param sender the remote member
 352      * @param status bucket versions from a remote member
 353      * @return a {@link akka.dispatch.Mapper} that gets evaluated in future
 354      *
 355      */
 356     private Mapper<Object, Void> getMapperToProcessRemoteStatus(final ActorRef sender, final GossipStatus status){
 357
 358         final Map<Address, Long> remoteVersions = status.getVersions();
 359
 360         return new Mapper<Object, Void>() {
 361             @Override
 362             public Void apply(Object replyMessage) {
 363                 if (replyMessage instanceof GetBucketVersionsReply) {
 364                     GetBucketVersionsReply reply = (GetBucketVersionsReply) replyMessage;
 365                     Map<Address, Long> localVersions = reply.getVersions();
 366
 367                     //diff between remote list and local
 368                     Set<Address> localIsOlder = new HashSet<>();
 369                     localIsOlder.addAll(remoteVersions.keySet());
 370                     localIsOlder.removeAll(localVersions.keySet());
 371
 372                     //diff between local list and remote
 373                     Set<Address> localIsNewer = new HashSet<>();
 374                     localIsNewer.addAll(localVersions.keySet());
 375                     localIsNewer.removeAll(remoteVersions.keySet());
 376
 377
 378                     for (Address address : remoteVersions.keySet()){
 379
 380                         if (localVersions.get(address) == null || remoteVersions.get(address) == null)
 381                             continue; //this condition is taken care of by above diffs
 382                         if (localVersions.get(address) <  remoteVersions.get(address))
 383                             localIsOlder.add(address);
 384                         else if (localVersions.get(address) > remoteVersions.get(address))
 385                             localIsNewer.add(address);
 386                     }
 387
 388                     if (!localIsOlder.isEmpty())
 389                         sendGossipStatusTo(sender, localVersions );
 390
 391                     if (!localIsNewer.isEmpty())
 392                         sendGossipTo(sender, localIsNewer);//send newer buckets to remote
 393
 394                 }
 395                 return null;
 396             }
 397         };
 398     }
 399
 400     /**
 401      * Processes the message from {@link org.opendaylight.controller.remote.rpc.registry.gossip.BucketStore}
 402      * that contains {@link org.opendaylight.controller.remote.rpc.registry.gossip.Bucket}.
 403      * These buckets are sent to a remote member encapsulated in
 404      * {@link org.opendaylight.controller.remote.rpc.registry.gossip.Messages.GossiperMessages.GossipEnvelope}
 405      *
 406      * @param sender the remote member that sent
 407      *               {@link org.opendaylight.controller.remote.rpc.registry.gossip.Messages.GossiperMessages.GossipStatus}
 408      *               in reply to which bucket is being sent back
 409      * @return a {@link akka.dispatch.Mapper} that gets evaluated in future
 410      *
 411      */
 412     private Mapper<Object, Void> getMapperToSendGossip(final ActorRef sender) {
 413
 414         return new Mapper<Object, Void>() {
 415             @Override
 416             public Void apply(Object msg) {
 417                 if (msg instanceof GetBucketsByMembersReply) {
 418                     Map<Address, Bucket> buckets = ((GetBucketsByMembersReply) msg).getBuckets();
 419                     log.debug("Buckets to send from {}: {}", selfAddress, buckets);
 420                     GossipEnvelope envelope = new GossipEnvelope(selfAddress, sender.path().address(), buckets);
 421                     sender.tell(envelope, getSelf());
 422                 }
 423                 return null;
 424             }
 425         };
 426     }
 427
 428     ///
 429     ///Getter Setters
 430     ///
 431     List<Address> getClusterMembers() {
 432         return clusterMembers;
 433     }
 434
 435     void setClusterMembers(List<Address> clusterMembers) {
 436         this.clusterMembers = clusterMembers;
 437     }
 438
 439     Address getSelfAddress() {
 440         return selfAddress;
 441     }
 442
 443     void setSelfAddress(Address selfAddress) {
 444         this.selfAddress = selfAddress;
 445     }
 446 }