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

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.