c003d901af800c6df9505e25a52b3501c7a0e4ee
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / Shard.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
9 package org.opendaylight.controller.cluster.datastore;
10
11 import akka.actor.ActorRef;
12 import akka.actor.ActorSelection;
13 import akka.actor.Cancellable;
14 import akka.actor.Props;
15 import akka.actor.Status;
16 import akka.actor.Status.Failure;
17 import akka.serialization.Serialization;
18 import com.google.common.annotations.VisibleForTesting;
19 import com.google.common.base.Optional;
20 import com.google.common.base.Preconditions;
21 import com.google.common.base.Ticker;
22 import com.google.common.base.Verify;
23 import com.google.common.collect.ImmutableList;
24 import com.google.common.collect.ImmutableMap;
25 import com.google.common.collect.Range;
26 import java.io.IOException;
27 import java.util.Arrays;
28 import java.util.Collection;
29 import java.util.Collections;
30 import java.util.Map;
31 import java.util.concurrent.TimeUnit;
32 import javax.annotation.Nonnull;
33 import javax.annotation.Nullable;
34 import org.opendaylight.controller.cluster.access.ABIVersion;
35 import org.opendaylight.controller.cluster.access.commands.ConnectClientRequest;
36 import org.opendaylight.controller.cluster.access.commands.ConnectClientSuccess;
37 import org.opendaylight.controller.cluster.access.commands.LocalHistoryRequest;
38 import org.opendaylight.controller.cluster.access.commands.NotLeaderException;
39 import org.opendaylight.controller.cluster.access.commands.OutOfSequenceEnvelopeException;
40 import org.opendaylight.controller.cluster.access.commands.TransactionRequest;
41 import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier;
42 import org.opendaylight.controller.cluster.access.concepts.FrontendIdentifier;
43 import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifier;
44 import org.opendaylight.controller.cluster.access.concepts.Request;
45 import org.opendaylight.controller.cluster.access.concepts.RequestEnvelope;
46 import org.opendaylight.controller.cluster.access.concepts.RequestException;
47 import org.opendaylight.controller.cluster.access.concepts.RequestSuccess;
48 import org.opendaylight.controller.cluster.access.concepts.RetiredGenerationException;
49 import org.opendaylight.controller.cluster.access.concepts.RuntimeRequestException;
50 import org.opendaylight.controller.cluster.access.concepts.SliceableMessage;
51 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
52 import org.opendaylight.controller.cluster.access.concepts.UnsupportedRequestException;
53 import org.opendaylight.controller.cluster.common.actor.CommonConfig;
54 import org.opendaylight.controller.cluster.common.actor.Dispatchers;
55 import org.opendaylight.controller.cluster.common.actor.Dispatchers.DispatcherType;
56 import org.opendaylight.controller.cluster.common.actor.MessageTracker;
57 import org.opendaylight.controller.cluster.common.actor.MessageTracker.Error;
58 import org.opendaylight.controller.cluster.common.actor.MeteringBehavior;
59 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
60 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
61 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardMBeanFactory;
62 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
63 import org.opendaylight.controller.cluster.datastore.messages.AbortTransaction;
64 import org.opendaylight.controller.cluster.datastore.messages.ActorInitialized;
65 import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
66 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction;
67 import org.opendaylight.controller.cluster.datastore.messages.CloseTransactionChain;
68 import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction;
69 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
70 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
71 import org.opendaylight.controller.cluster.datastore.messages.ForwardedReadyTransaction;
72 import org.opendaylight.controller.cluster.datastore.messages.GetShardDataTree;
73 import org.opendaylight.controller.cluster.datastore.messages.MakeLeaderLocal;
74 import org.opendaylight.controller.cluster.datastore.messages.OnDemandShardState;
75 import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved;
76 import org.opendaylight.controller.cluster.datastore.messages.PersistAbortTransactionPayload;
77 import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransaction;
78 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener;
79 import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeChangeListener;
80 import org.opendaylight.controller.cluster.datastore.messages.ShardLeaderStateChanged;
81 import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
82 import org.opendaylight.controller.cluster.datastore.persisted.AbortTransactionPayload;
83 import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot;
84 import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot.ShardSnapshot;
85 import org.opendaylight.controller.cluster.messaging.MessageSlicer;
86 import org.opendaylight.controller.cluster.messaging.SliceOptions;
87 import org.opendaylight.controller.cluster.notifications.LeaderStateChanged;
88 import org.opendaylight.controller.cluster.notifications.RegisterRoleChangeListener;
89 import org.opendaylight.controller.cluster.notifications.RoleChangeNotifier;
90 import org.opendaylight.controller.cluster.raft.LeadershipTransferFailedException;
91 import org.opendaylight.controller.cluster.raft.RaftActor;
92 import org.opendaylight.controller.cluster.raft.RaftActorRecoveryCohort;
93 import org.opendaylight.controller.cluster.raft.RaftActorSnapshotCohort;
94 import org.opendaylight.controller.cluster.raft.RaftState;
95 import org.opendaylight.controller.cluster.raft.base.messages.FollowerInitialSyncUpStatus;
96 import org.opendaylight.controller.cluster.raft.client.messages.OnDemandRaftState;
97 import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
98 import org.opendaylight.controller.cluster.raft.messages.RequestLeadership;
99 import org.opendaylight.controller.cluster.raft.messages.ServerRemoved;
100 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
101 import org.opendaylight.yangtools.concepts.Identifier;
102 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailedException;
103 import org.opendaylight.yangtools.yang.data.api.schema.tree.TipProducingDataTree;
104 import org.opendaylight.yangtools.yang.data.api.schema.tree.TreeType;
105 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
106 import org.opendaylight.yangtools.yang.model.api.SchemaContextProvider;
107 import scala.concurrent.duration.Duration;
108 import scala.concurrent.duration.FiniteDuration;
109
110 /**
111  * A Shard represents a portion of the logical data tree.
112  *
113  * <p>
114  * Our Shard uses InMemoryDataTree as it's internal representation and delegates all requests it
115  */
116 public class Shard extends RaftActor {
117
118     @VisibleForTesting
119     static final Object TX_COMMIT_TIMEOUT_CHECK_MESSAGE = new Object() {
120         @Override
121         public String toString() {
122             return "txCommitTimeoutCheck";
123         }
124     };
125
126     @VisibleForTesting
127     static final Object GET_SHARD_MBEAN_MESSAGE = new Object() {
128         @Override
129         public String toString() {
130             return "getShardMBeanMessage";
131         }
132     };
133
134     static final Object RESUME_NEXT_PENDING_TRANSACTION = new Object() {
135         @Override
136         public String toString() {
137             return "resumeNextPendingTransaction";
138         }
139     };
140
141     // FIXME: shard names should be encapsulated in their own class and this should be exposed as a constant.
142     public static final String DEFAULT_NAME = "default";
143
144     private static final Collection<ABIVersion> SUPPORTED_ABIVERSIONS;
145
146     static {
147         final ABIVersion[] values = ABIVersion.values();
148         final ABIVersion[] real = Arrays.copyOfRange(values, 1, values.length - 1);
149         SUPPORTED_ABIVERSIONS = ImmutableList.copyOf(real).reverse();
150     }
151
152     // FIXME: make this a dynamic property based on mailbox size and maximum number of clients
153     private static final int CLIENT_MAX_MESSAGES = 1000;
154
155     // The state of this Shard
156     private final ShardDataTree store;
157
158     /// The name of this shard
159     private final String name;
160
161     private final ShardStats shardMBean;
162
163     private DatastoreContext datastoreContext;
164
165     private final ShardCommitCoordinator commitCoordinator;
166
167     private long transactionCommitTimeout;
168
169     private Cancellable txCommitTimeoutCheckSchedule;
170
171     private final Optional<ActorRef> roleChangeNotifier;
172
173     private final MessageTracker appendEntriesReplyTracker;
174
175     private final ShardTransactionActorFactory transactionActorFactory;
176
177     private final ShardSnapshotCohort snapshotCohort;
178
179     private final DataTreeChangeListenerSupport treeChangeSupport = new DataTreeChangeListenerSupport(this);
180     private final DataChangeListenerSupport changeSupport = new DataChangeListenerSupport(this);
181
182
183     private ShardSnapshot restoreFromSnapshot;
184
185     private final ShardTransactionMessageRetrySupport messageRetrySupport;
186
187     private final FrontendMetadata frontendMetadata;
188     private Map<FrontendIdentifier, LeaderFrontendState> knownFrontends = ImmutableMap.of();
189     private boolean paused;
190
191     private final MessageSlicer responseMessageSlicer;
192     private final Dispatchers dispatchers;
193
194     protected Shard(final AbstractBuilder<?, ?> builder) {
195         super(builder.getId().toString(), builder.getPeerAddresses(),
196                 Optional.of(builder.getDatastoreContext().getShardRaftConfig()), DataStoreVersions.CURRENT_VERSION);
197
198         this.name = builder.getId().toString();
199         this.datastoreContext = builder.getDatastoreContext();
200         this.restoreFromSnapshot = builder.getRestoreFromSnapshot();
201         this.frontendMetadata = new FrontendMetadata(name);
202
203         setPersistence(datastoreContext.isPersistent());
204
205         LOG.info("Shard created : {}, persistent : {}", name, datastoreContext.isPersistent());
206
207         ShardDataTreeChangeListenerPublisherActorProxy treeChangeListenerPublisher =
208                 new ShardDataTreeChangeListenerPublisherActorProxy(getContext(), name + "-DTCL-publisher", name);
209         ShardDataChangeListenerPublisherActorProxy dataChangeListenerPublisher =
210                 new ShardDataChangeListenerPublisherActorProxy(getContext(), name + "-DCL-publisher", name);
211         if (builder.getDataTree() != null) {
212             store = new ShardDataTree(this, builder.getSchemaContext(), builder.getDataTree(),
213                     treeChangeListenerPublisher, dataChangeListenerPublisher, name, frontendMetadata);
214         } else {
215             store = new ShardDataTree(this, builder.getSchemaContext(), builder.getTreeType(),
216                     builder.getDatastoreContext().getStoreRoot(), treeChangeListenerPublisher,
217                     dataChangeListenerPublisher, name, frontendMetadata);
218         }
219
220         shardMBean = ShardMBeanFactory.getShardStatsMBean(name, datastoreContext.getDataStoreMXBeanType(), this);
221
222         if (isMetricsCaptureEnabled()) {
223             getContext().become(new MeteringBehavior(this));
224         }
225
226         commitCoordinator = new ShardCommitCoordinator(store, LOG, this.name);
227
228         setTransactionCommitTimeout();
229
230         // create a notifier actor for each cluster member
231         roleChangeNotifier = createRoleChangeNotifier(name);
232
233         appendEntriesReplyTracker = new MessageTracker(AppendEntriesReply.class,
234                 getRaftActorContext().getConfigParams().getIsolatedCheckIntervalInMillis());
235
236         dispatchers = new Dispatchers(context().system().dispatchers());
237         transactionActorFactory = new ShardTransactionActorFactory(store, datastoreContext,
238             dispatchers.getDispatcherPath(Dispatchers.DispatcherType.Transaction),
239                 self(), getContext(), shardMBean, builder.getId().getShardName());
240
241         snapshotCohort = ShardSnapshotCohort.create(getContext(), builder.getId().getMemberName(), store, LOG,
242             this.name);
243
244         messageRetrySupport = new ShardTransactionMessageRetrySupport(this);
245
246         responseMessageSlicer = MessageSlicer.builder().logContext(this.name)
247                 .messageSliceSize(datastoreContext.getMaximumMessageSliceSize())
248                 .fileBackedStreamFactory(getRaftActorContext().getFileBackedOutputStreamFactory())
249                 .expireStateAfterInactivity(2, TimeUnit.MINUTES).build();
250     }
251
252     private void setTransactionCommitTimeout() {
253         transactionCommitTimeout = TimeUnit.MILLISECONDS.convert(
254                 datastoreContext.getShardTransactionCommitTimeoutInSeconds(), TimeUnit.SECONDS) / 2;
255     }
256
257     private Optional<ActorRef> createRoleChangeNotifier(final String shardId) {
258         ActorRef shardRoleChangeNotifier = this.getContext().actorOf(
259             RoleChangeNotifier.getProps(shardId), shardId + "-notifier");
260         return Optional.of(shardRoleChangeNotifier);
261     }
262
263     @Override
264     public void postStop() {
265         LOG.info("Stopping Shard {}", persistenceId());
266
267         super.postStop();
268
269         messageRetrySupport.close();
270
271         if (txCommitTimeoutCheckSchedule != null) {
272             txCommitTimeoutCheckSchedule.cancel();
273         }
274
275         commitCoordinator.abortPendingTransactions("Transaction aborted due to shutdown.", this);
276
277         shardMBean.unregisterMBean();
278     }
279
280     @Override
281     protected void handleRecover(final Object message) {
282         LOG.debug("{}: onReceiveRecover: Received message {} from {}", persistenceId(), message.getClass(),
283             getSender());
284
285         super.handleRecover(message);
286         if (LOG.isTraceEnabled()) {
287             appendEntriesReplyTracker.begin();
288         }
289     }
290
291     @Override
292     protected void handleNonRaftCommand(final Object message) {
293         try (MessageTracker.Context context = appendEntriesReplyTracker.received(message)) {
294             final Optional<Error> maybeError = context.error();
295             if (maybeError.isPresent()) {
296                 LOG.trace("{} : AppendEntriesReply failed to arrive at the expected interval {}", persistenceId(),
297                     maybeError.get());
298             }
299
300             store.resetTransactionBatch();
301
302             if (message instanceof RequestEnvelope) {
303                 handleRequestEnvelope((RequestEnvelope)message);
304             } else if (message instanceof ConnectClientRequest) {
305                 handleConnectClient((ConnectClientRequest)message);
306             } else if (CreateTransaction.isSerializedType(message)) {
307                 handleCreateTransaction(message);
308             } else if (message instanceof BatchedModifications) {
309                 handleBatchedModifications((BatchedModifications)message);
310             } else if (message instanceof ForwardedReadyTransaction) {
311                 handleForwardedReadyTransaction((ForwardedReadyTransaction) message);
312             } else if (message instanceof ReadyLocalTransaction) {
313                 handleReadyLocalTransaction((ReadyLocalTransaction)message);
314             } else if (CanCommitTransaction.isSerializedType(message)) {
315                 handleCanCommitTransaction(CanCommitTransaction.fromSerializable(message));
316             } else if (CommitTransaction.isSerializedType(message)) {
317                 handleCommitTransaction(CommitTransaction.fromSerializable(message));
318             } else if (AbortTransaction.isSerializedType(message)) {
319                 handleAbortTransaction(AbortTransaction.fromSerializable(message));
320             } else if (CloseTransactionChain.isSerializedType(message)) {
321                 closeTransactionChain(CloseTransactionChain.fromSerializable(message));
322             } else if (message instanceof RegisterChangeListener) {
323                 changeSupport.onMessage((RegisterChangeListener) message, isLeader(), hasLeader());
324             } else if (message instanceof RegisterDataTreeChangeListener) {
325                 treeChangeSupport.onMessage((RegisterDataTreeChangeListener) message, isLeader(), hasLeader());
326             } else if (message instanceof UpdateSchemaContext) {
327                 updateSchemaContext((UpdateSchemaContext) message);
328             } else if (message instanceof PeerAddressResolved) {
329                 PeerAddressResolved resolved = (PeerAddressResolved) message;
330                 setPeerAddress(resolved.getPeerId(), resolved.getPeerAddress());
331             } else if (TX_COMMIT_TIMEOUT_CHECK_MESSAGE.equals(message)) {
332                 commitTimeoutCheck();
333             } else if (message instanceof DatastoreContext) {
334                 onDatastoreContext((DatastoreContext)message);
335             } else if (message instanceof RegisterRoleChangeListener) {
336                 roleChangeNotifier.get().forward(message, context());
337             } else if (message instanceof FollowerInitialSyncUpStatus) {
338                 shardMBean.setFollowerInitialSyncStatus(((FollowerInitialSyncUpStatus) message).isInitialSyncDone());
339                 context().parent().tell(message, self());
340             } else if (GET_SHARD_MBEAN_MESSAGE.equals(message)) {
341                 sender().tell(getShardMBean(), self());
342             } else if (message instanceof GetShardDataTree) {
343                 sender().tell(store.getDataTree(), self());
344             } else if (message instanceof ServerRemoved) {
345                 context().parent().forward(message, context());
346             } else if (ShardTransactionMessageRetrySupport.TIMER_MESSAGE_CLASS.isInstance(message)) {
347                 messageRetrySupport.onTimerMessage(message);
348             } else if (message instanceof DataTreeCohortActorRegistry.CohortRegistryCommand) {
349                 store.processCohortRegistryCommand(getSender(),
350                         (DataTreeCohortActorRegistry.CohortRegistryCommand) message);
351             } else if (message instanceof PersistAbortTransactionPayload) {
352                 final TransactionIdentifier txId = ((PersistAbortTransactionPayload) message).getTransactionId();
353                 persistPayload(txId, AbortTransactionPayload.create(txId), true);
354             } else if (message instanceof MakeLeaderLocal) {
355                 onMakeLeaderLocal();
356             } else if (RESUME_NEXT_PENDING_TRANSACTION.equals(message)) {
357                 store.resumeNextPendingTransaction();
358             } else if (!responseMessageSlicer.handleMessage(message)) {
359                 super.handleNonRaftCommand(message);
360             }
361         }
362     }
363
364     @SuppressWarnings("checkstyle:IllegalCatch")
365     private void handleRequestEnvelope(final RequestEnvelope envelope) {
366         final long now = ticker().read();
367         try {
368             final RequestSuccess<?, ?> success = handleRequest(envelope, now);
369             if (success != null) {
370                 final long executionTimeNanos = ticker().read() - now;
371                 if (success instanceof SliceableMessage) {
372                     dispatchers.getDispatcher(DispatcherType.Serialization).execute(() ->
373                         responseMessageSlicer.slice(SliceOptions.builder().identifier(success.getTarget())
374                             .message(envelope.newSuccessEnvelope(success, executionTimeNanos))
375                             .sendTo(envelope.getMessage().getReplyTo()).replyTo(self())
376                             .onFailureCallback(t -> {
377                                 LOG.warn("Error slicing response {}", success, t);
378                             }).build()));
379                 } else {
380                     envelope.sendSuccess(success, executionTimeNanos);
381                 }
382             }
383         } catch (RequestException e) {
384             LOG.debug("{}: request {} failed", persistenceId(), envelope, e);
385             envelope.sendFailure(e, ticker().read() - now);
386         } catch (Exception e) {
387             LOG.debug("{}: request {} caused failure", persistenceId(), envelope, e);
388             envelope.sendFailure(new RuntimeRequestException("Request failed to process", e),
389                 ticker().read() - now);
390         }
391     }
392
393     private void commitTimeoutCheck() {
394         store.checkForExpiredTransactions(transactionCommitTimeout, this::updateAccess);
395         commitCoordinator.checkForExpiredTransactions(transactionCommitTimeout, this);
396     }
397
398     private Optional<Long> updateAccess(final SimpleShardDataTreeCohort cohort) {
399         final FrontendIdentifier frontend = cohort.getIdentifier().getHistoryId().getClientId().getFrontendId();
400         final LeaderFrontendState state = knownFrontends.get(frontend);
401         if (state == null) {
402             // Not tell-based protocol, do nothing
403             return Optional.absent();
404         }
405
406         if (isIsolatedLeader()) {
407             // We are isolated and no new request can come through until we emerge from it. We are still updating
408             // liveness of frontend when we see it attempting to communicate. Use the last access timer.
409             return Optional.of(state.getLastSeenTicks());
410         }
411
412         // If this frontend has freshly connected, give it some time to catch up before killing its transactions.
413         return Optional.of(state.getLastConnectTicks());
414     }
415
416     private void onMakeLeaderLocal() {
417         LOG.debug("{}: onMakeLeaderLocal received", persistenceId());
418         if (isLeader()) {
419             getSender().tell(new Status.Success(null), getSelf());
420             return;
421         }
422
423         final ActorSelection leader = getLeader();
424
425         if (leader == null) {
426             // Leader is not present. The cluster is most likely trying to
427             // elect a leader and we should let that run its normal course
428
429             // TODO we can wait for the election to complete and retry the
430             // request. We can also let the caller retry by sending a flag
431             // in the response indicating the request is "reTryable".
432             getSender().tell(new Failure(
433                     new LeadershipTransferFailedException("We cannot initiate leadership transfer to local node. "
434                             + "Currently there is no leader for " + persistenceId())),
435                     getSelf());
436             return;
437         }
438
439         leader.tell(new RequestLeadership(getId(), getSender()), getSelf());
440     }
441
442     // Acquire our frontend tracking handle and verify generation matches
443     @Nullable
444     private LeaderFrontendState findFrontend(final ClientIdentifier clientId) throws RequestException {
445         final LeaderFrontendState existing = knownFrontends.get(clientId.getFrontendId());
446         if (existing != null) {
447             final int cmp = Long.compareUnsigned(existing.getIdentifier().getGeneration(), clientId.getGeneration());
448             if (cmp == 0) {
449                 existing.touch();
450                 return existing;
451             }
452             if (cmp > 0) {
453                 LOG.debug("{}: rejecting request from outdated client {}", persistenceId(), clientId);
454                 throw new RetiredGenerationException(existing.getIdentifier().getGeneration());
455             }
456
457             LOG.info("{}: retiring state {}, outdated by request from client {}", persistenceId(), existing, clientId);
458             existing.retire();
459             knownFrontends.remove(clientId.getFrontendId());
460         } else {
461             LOG.debug("{}: client {} is not yet known", persistenceId(), clientId);
462         }
463
464         return null;
465     }
466
467     private LeaderFrontendState getFrontend(final ClientIdentifier clientId) throws RequestException {
468         final LeaderFrontendState ret = findFrontend(clientId);
469         if (ret != null) {
470             return ret;
471         }
472
473         // TODO: a dedicated exception would be better, but this is technically true, too
474         throw new OutOfSequenceEnvelopeException(0);
475     }
476
477     private static @Nonnull ABIVersion selectVersion(final ConnectClientRequest message) {
478         final Range<ABIVersion> clientRange = Range.closed(message.getMinVersion(), message.getMaxVersion());
479         for (ABIVersion v : SUPPORTED_ABIVERSIONS) {
480             if (clientRange.contains(v)) {
481                 return v;
482             }
483         }
484
485         throw new IllegalArgumentException(String.format(
486             "No common version between backend versions %s and client versions %s", SUPPORTED_ABIVERSIONS,
487             clientRange));
488     }
489
490     @SuppressWarnings("checkstyle:IllegalCatch")
491     private void handleConnectClient(final ConnectClientRequest message) {
492         try {
493             final ClientIdentifier clientId = message.getTarget();
494             final LeaderFrontendState existing = findFrontend(clientId);
495             if (existing != null) {
496                 existing.touch();
497             }
498
499             if (!isLeader() || !isLeaderActive()) {
500                 LOG.info("{}: not currently leader, rejecting request {}. isLeader: {}, isLeaderActive: {},"
501                                 + "isLeadershipTransferInProgress: {}.",
502                         persistenceId(), message, isLeader(), isLeaderActive(), isLeadershipTransferInProgress());
503                 throw new NotLeaderException(getSelf());
504             }
505
506             final ABIVersion selectedVersion = selectVersion(message);
507             final LeaderFrontendState frontend;
508             if (existing == null) {
509                 frontend = new LeaderFrontendState(persistenceId(), clientId, store);
510                 knownFrontends.put(clientId.getFrontendId(), frontend);
511                 LOG.debug("{}: created state {} for client {}", persistenceId(), frontend, clientId);
512             } else {
513                 frontend = existing;
514             }
515
516             frontend.reconnect();
517             message.getReplyTo().tell(new ConnectClientSuccess(message.getTarget(), message.getSequence(), getSelf(),
518                 ImmutableList.of(), store.getDataTree(), CLIENT_MAX_MESSAGES).toVersion(selectedVersion),
519                 ActorRef.noSender());
520         } catch (RequestException | RuntimeException e) {
521             message.getReplyTo().tell(new Failure(e), ActorRef.noSender());
522         }
523     }
524
525     private @Nullable RequestSuccess<?, ?> handleRequest(final RequestEnvelope envelope, final long now)
526             throws RequestException {
527         // We are not the leader, hence we want to fail-fast.
528         if (!isLeader() || paused || !isLeaderActive()) {
529             LOG.debug("{}: not currently active leader, rejecting request {}. isLeader: {}, isLeaderActive: {},"
530                             + "isLeadershipTransferInProgress: {}, paused: {}",
531                     persistenceId(), envelope, isLeader(), isLeaderActive(), isLeadershipTransferInProgress(), paused);
532             throw new NotLeaderException(getSelf());
533         }
534
535         final Request<?, ?> request = envelope.getMessage();
536         if (request instanceof TransactionRequest) {
537             final TransactionRequest<?> txReq = (TransactionRequest<?>)request;
538             final ClientIdentifier clientId = txReq.getTarget().getHistoryId().getClientId();
539             return getFrontend(clientId).handleTransactionRequest(txReq, envelope, now);
540         } else if (request instanceof LocalHistoryRequest) {
541             final LocalHistoryRequest<?> lhReq = (LocalHistoryRequest<?>)request;
542             final ClientIdentifier clientId = lhReq.getTarget().getClientId();
543             return getFrontend(clientId).handleLocalHistoryRequest(lhReq, envelope, now);
544         } else {
545             LOG.warn("{}: rejecting unsupported request {}", persistenceId(), request);
546             throw new UnsupportedRequestException(request);
547         }
548     }
549
550     private boolean hasLeader() {
551         return getLeaderId() != null;
552     }
553
554     public int getPendingTxCommitQueueSize() {
555         return store.getQueueSize();
556     }
557
558     public int getCohortCacheSize() {
559         return commitCoordinator.getCohortCacheSize();
560     }
561
562     @Override
563     protected Optional<ActorRef> getRoleChangeNotifier() {
564         return roleChangeNotifier;
565     }
566
567     @Override
568     protected LeaderStateChanged newLeaderStateChanged(final String memberId, final String leaderId,
569             final short leaderPayloadVersion) {
570         return isLeader() ? new ShardLeaderStateChanged(memberId, leaderId, store.getDataTree(), leaderPayloadVersion)
571                 : new ShardLeaderStateChanged(memberId, leaderId, leaderPayloadVersion);
572     }
573
574     protected void onDatastoreContext(final DatastoreContext context) {
575         datastoreContext = context;
576
577         setTransactionCommitTimeout();
578
579         setPersistence(datastoreContext.isPersistent());
580
581         updateConfigParams(datastoreContext.getShardRaftConfig());
582     }
583
584     // applyState() will be invoked once consensus is reached on the payload
585     void persistPayload(final Identifier id, final Payload payload, final boolean batchHint) {
586         boolean canSkipPayload = !hasFollowers() && !persistence().isRecoveryApplicable();
587         if (canSkipPayload) {
588             applyState(self(), id, payload);
589         } else {
590             // We are faking the sender
591             persistData(self(), id, payload, batchHint);
592         }
593     }
594
595     private void handleCommitTransaction(final CommitTransaction commit) {
596         if (isLeader()) {
597             commitCoordinator.handleCommit(commit.getTransactionId(), getSender(), this);
598         } else {
599             ActorSelection leader = getLeader();
600             if (leader == null) {
601                 messageRetrySupport.addMessageToRetry(commit, getSender(),
602                         "Could not commit transaction " + commit.getTransactionId());
603             } else {
604                 LOG.debug("{}: Forwarding CommitTransaction to leader {}", persistenceId(), leader);
605                 leader.forward(commit, getContext());
606             }
607         }
608     }
609
610     private void handleCanCommitTransaction(final CanCommitTransaction canCommit) {
611         LOG.debug("{}: Can committing transaction {}", persistenceId(), canCommit.getTransactionId());
612
613         if (isLeader()) {
614             commitCoordinator.handleCanCommit(canCommit.getTransactionId(), getSender(), this);
615         } else {
616             ActorSelection leader = getLeader();
617             if (leader == null) {
618                 messageRetrySupport.addMessageToRetry(canCommit, getSender(),
619                         "Could not canCommit transaction " + canCommit.getTransactionId());
620             } else {
621                 LOG.debug("{}: Forwarding CanCommitTransaction to leader {}", persistenceId(), leader);
622                 leader.forward(canCommit, getContext());
623             }
624         }
625     }
626
627     @SuppressWarnings("checkstyle:IllegalCatch")
628     protected void handleBatchedModificationsLocal(final BatchedModifications batched, final ActorRef sender) {
629         try {
630             commitCoordinator.handleBatchedModifications(batched, sender, this);
631         } catch (Exception e) {
632             LOG.error("{}: Error handling BatchedModifications for Tx {}", persistenceId(),
633                     batched.getTransactionId(), e);
634             sender.tell(new Failure(e), getSelf());
635         }
636     }
637
638     private void handleBatchedModifications(final BatchedModifications batched) {
639         // This message is sent to prepare the modifications transaction directly on the Shard as an
640         // optimization to avoid the extra overhead of a separate ShardTransaction actor. On the last
641         // BatchedModifications message, the caller sets the ready flag in the message indicating
642         // modifications are complete. The reply contains the cohort actor path (this actor) for the caller
643         // to initiate the 3-phase commit. This also avoids the overhead of sending an additional
644         // ReadyTransaction message.
645
646         // If we're not the leader then forward to the leader. This is a safety measure - we shouldn't
647         // normally get here if we're not the leader as the front-end (TransactionProxy) should determine
648         // the primary/leader shard. However with timing and caching on the front-end, there's a small
649         // window where it could have a stale leader during leadership transitions.
650         //
651         boolean isLeaderActive = isLeaderActive();
652         if (isLeader() && isLeaderActive) {
653             handleBatchedModificationsLocal(batched, getSender());
654         } else {
655             ActorSelection leader = getLeader();
656             if (!isLeaderActive || leader == null) {
657                 messageRetrySupport.addMessageToRetry(batched, getSender(),
658                         "Could not commit transaction " + batched.getTransactionId());
659             } else {
660                 // If this is not the first batch and leadership changed in between batched messages,
661                 // we need to reconstruct previous BatchedModifications from the transaction
662                 // DataTreeModification, honoring the max batched modification count, and forward all the
663                 // previous BatchedModifications to the new leader.
664                 Collection<BatchedModifications> newModifications = commitCoordinator
665                         .createForwardedBatchedModifications(batched,
666                                 datastoreContext.getShardBatchedModificationCount());
667
668                 LOG.debug("{}: Forwarding {} BatchedModifications to leader {}", persistenceId(),
669                         newModifications.size(), leader);
670
671                 for (BatchedModifications bm : newModifications) {
672                     leader.forward(bm, getContext());
673                 }
674             }
675         }
676     }
677
678     private boolean failIfIsolatedLeader(final ActorRef sender) {
679         if (isIsolatedLeader()) {
680             sender.tell(new Failure(new NoShardLeaderException(String.format(
681                     "Shard %s was the leader but has lost contact with all of its followers. Either all"
682                     + " other follower nodes are down or this node is isolated by a network partition.",
683                     persistenceId()))), getSelf());
684             return true;
685         }
686
687         return false;
688     }
689
690     protected boolean isIsolatedLeader() {
691         return getRaftState() == RaftState.IsolatedLeader;
692     }
693
694     @SuppressWarnings("checkstyle:IllegalCatch")
695     private void handleReadyLocalTransaction(final ReadyLocalTransaction message) {
696         LOG.debug("{}: handleReadyLocalTransaction for {}", persistenceId(), message.getTransactionId());
697
698         boolean isLeaderActive = isLeaderActive();
699         if (isLeader() && isLeaderActive) {
700             try {
701                 commitCoordinator.handleReadyLocalTransaction(message, getSender(), this);
702             } catch (Exception e) {
703                 LOG.error("{}: Error handling ReadyLocalTransaction for Tx {}", persistenceId(),
704                         message.getTransactionId(), e);
705                 getSender().tell(new Failure(e), getSelf());
706             }
707         } else {
708             ActorSelection leader = getLeader();
709             if (!isLeaderActive || leader == null) {
710                 messageRetrySupport.addMessageToRetry(message, getSender(),
711                         "Could not commit transaction " + message.getTransactionId());
712             } else {
713                 LOG.debug("{}: Forwarding ReadyLocalTransaction to leader {}", persistenceId(), leader);
714                 message.setRemoteVersion(getCurrentBehavior().getLeaderPayloadVersion());
715                 leader.forward(message, getContext());
716             }
717         }
718     }
719
720     private void handleForwardedReadyTransaction(final ForwardedReadyTransaction forwardedReady) {
721         LOG.debug("{}: handleForwardedReadyTransaction for {}", persistenceId(), forwardedReady.getTransactionId());
722
723         boolean isLeaderActive = isLeaderActive();
724         if (isLeader() && isLeaderActive) {
725             commitCoordinator.handleForwardedReadyTransaction(forwardedReady, getSender(), this);
726         } else {
727             ActorSelection leader = getLeader();
728             if (!isLeaderActive || leader == null) {
729                 messageRetrySupport.addMessageToRetry(forwardedReady, getSender(),
730                         "Could not commit transaction " + forwardedReady.getTransactionId());
731             } else {
732                 LOG.debug("{}: Forwarding ForwardedReadyTransaction to leader {}", persistenceId(), leader);
733
734                 ReadyLocalTransaction readyLocal = new ReadyLocalTransaction(forwardedReady.getTransactionId(),
735                         forwardedReady.getTransaction().getSnapshot(), forwardedReady.isDoImmediateCommit());
736                 readyLocal.setRemoteVersion(getCurrentBehavior().getLeaderPayloadVersion());
737                 leader.forward(readyLocal, getContext());
738             }
739         }
740     }
741
742     private void handleAbortTransaction(final AbortTransaction abort) {
743         doAbortTransaction(abort.getTransactionId(), getSender());
744     }
745
746     void doAbortTransaction(final Identifier transactionID, final ActorRef sender) {
747         commitCoordinator.handleAbort(transactionID, sender, this);
748     }
749
750     private void handleCreateTransaction(final Object message) {
751         if (isLeader()) {
752             createTransaction(CreateTransaction.fromSerializable(message));
753         } else if (getLeader() != null) {
754             getLeader().forward(message, getContext());
755         } else {
756             getSender().tell(new Failure(new NoShardLeaderException(
757                     "Could not create a shard transaction", persistenceId())), getSelf());
758         }
759     }
760
761     private void closeTransactionChain(final CloseTransactionChain closeTransactionChain) {
762         final LocalHistoryIdentifier id = closeTransactionChain.getIdentifier();
763         store.closeTransactionChain(id, null);
764         store.purgeTransactionChain(id, null);
765     }
766
767     @SuppressWarnings("checkstyle:IllegalCatch")
768     private void createTransaction(final CreateTransaction createTransaction) {
769         try {
770             if (TransactionType.fromInt(createTransaction.getTransactionType()) != TransactionType.READ_ONLY
771                     && failIfIsolatedLeader(getSender())) {
772                 return;
773             }
774
775             ActorRef transactionActor = createTransaction(createTransaction.getTransactionType(),
776                 createTransaction.getTransactionId());
777
778             getSender().tell(new CreateTransactionReply(Serialization.serializedActorPath(transactionActor),
779                     createTransaction.getTransactionId(), createTransaction.getVersion()).toSerializable(), getSelf());
780         } catch (Exception e) {
781             getSender().tell(new Failure(e), getSelf());
782         }
783     }
784
785     private ActorRef createTransaction(final int transactionType, final TransactionIdentifier transactionId) {
786         LOG.debug("{}: Creating transaction : {} ", persistenceId(), transactionId);
787         return transactionActorFactory.newShardTransaction(TransactionType.fromInt(transactionType),
788             transactionId);
789     }
790
791     private void updateSchemaContext(final UpdateSchemaContext message) {
792         updateSchemaContext(message.getSchemaContext());
793     }
794
795     @VisibleForTesting
796     void updateSchemaContext(final SchemaContext schemaContext) {
797         store.updateSchemaContext(schemaContext);
798     }
799
800     private boolean isMetricsCaptureEnabled() {
801         CommonConfig config = new CommonConfig(getContext().system().settings().config());
802         return config.isMetricCaptureEnabled();
803     }
804
805     @Override
806     @VisibleForTesting
807     public RaftActorSnapshotCohort getRaftActorSnapshotCohort() {
808         return snapshotCohort;
809     }
810
811     @Override
812     @Nonnull
813     protected RaftActorRecoveryCohort getRaftActorRecoveryCohort() {
814         return new ShardRecoveryCoordinator(store,
815             restoreFromSnapshot != null ? restoreFromSnapshot.getSnapshot() : null, persistenceId(), LOG);
816     }
817
818     @Override
819     protected void onRecoveryComplete() {
820         restoreFromSnapshot = null;
821
822         //notify shard manager
823         getContext().parent().tell(new ActorInitialized(), getSelf());
824
825         // Being paranoid here - this method should only be called once but just in case...
826         if (txCommitTimeoutCheckSchedule == null) {
827             // Schedule a message to be periodically sent to check if the current in-progress
828             // transaction should be expired and aborted.
829             FiniteDuration period = Duration.create(transactionCommitTimeout / 3, TimeUnit.MILLISECONDS);
830             txCommitTimeoutCheckSchedule = getContext().system().scheduler().schedule(
831                     period, period, getSelf(),
832                     TX_COMMIT_TIMEOUT_CHECK_MESSAGE, getContext().dispatcher(), ActorRef.noSender());
833         }
834     }
835
836     @Override
837     protected void applyState(final ActorRef clientActor, final Identifier identifier, final Object data) {
838         if (data instanceof Payload) {
839             try {
840                 store.applyReplicatedPayload(identifier, (Payload)data);
841             } catch (DataValidationFailedException | IOException e) {
842                 LOG.error("{}: Error applying replica {}", persistenceId(), identifier, e);
843             }
844         } else {
845             LOG.error("{}: Unknown state for {} received {}", persistenceId(), identifier, data);
846         }
847     }
848
849     @Override
850     protected void onStateChanged() {
851         boolean isLeader = isLeader();
852         boolean hasLeader = hasLeader();
853         changeSupport.onLeadershipChange(isLeader, hasLeader);
854         treeChangeSupport.onLeadershipChange(isLeader, hasLeader);
855
856         // If this actor is no longer the leader close all the transaction chains
857         if (!isLeader) {
858             if (LOG.isDebugEnabled()) {
859                 LOG.debug(
860                     "{}: onStateChanged: Closing all transaction chains because shard {} is no longer the leader",
861                     persistenceId(), getId());
862             }
863
864             paused = false;
865             store.purgeLeaderState();
866         }
867
868         if (hasLeader && !isIsolatedLeader()) {
869             messageRetrySupport.retryMessages();
870         }
871     }
872
873     @Override
874     protected void onLeaderChanged(final String oldLeader, final String newLeader) {
875         shardMBean.incrementLeadershipChangeCount();
876         paused = false;
877
878         if (!isLeader()) {
879             if (!knownFrontends.isEmpty()) {
880                 LOG.debug("{}: removing frontend state for {}", persistenceId(), knownFrontends.keySet());
881                 knownFrontends = ImmutableMap.of();
882             }
883
884             if (!hasLeader()) {
885                 // No leader anywhere, nothing else to do
886                 return;
887             }
888
889             // Another leader was elected. If we were the previous leader and had pending transactions, convert
890             // them to transaction messages and send to the new leader.
891             ActorSelection leader = getLeader();
892             if (leader != null) {
893                 Collection<?> messagesToForward = convertPendingTransactionsToMessages();
894
895                 if (!messagesToForward.isEmpty()) {
896                     LOG.debug("{}: Forwarding {} pending transaction messages to leader {}", persistenceId(),
897                             messagesToForward.size(), leader);
898
899                     for (Object message : messagesToForward) {
900                         leader.tell(message, self());
901                     }
902                 }
903             } else {
904                 commitCoordinator.abortPendingTransactions("The transacton was aborted due to inflight leadership "
905                         + "change and the leader address isn't available.", this);
906             }
907         } else {
908             // We have become the leader, we need to reconstruct frontend state
909             knownFrontends = Verify.verifyNotNull(frontendMetadata.toLeaderState(this));
910             LOG.debug("{}: became leader with frontend state for {}", persistenceId(), knownFrontends.keySet());
911         }
912
913         if (!isIsolatedLeader()) {
914             messageRetrySupport.retryMessages();
915         }
916     }
917
918     /**
919      * Clears all pending transactions and converts them to messages to be forwarded to a new leader.
920      *
921      * @return the converted messages
922      */
923     public Collection<?> convertPendingTransactionsToMessages() {
924         return commitCoordinator.convertPendingTransactionsToMessages(
925                 datastoreContext.getShardBatchedModificationCount());
926     }
927
928     @Override
929     protected void pauseLeader(final Runnable operation) {
930         LOG.debug("{}: In pauseLeader, operation: {}", persistenceId(), operation);
931         paused = true;
932
933         // Tell-based protocol can replay transaction state, so it is safe to blow it up when we are paused.
934         knownFrontends.values().forEach(LeaderFrontendState::retire);
935         knownFrontends = ImmutableMap.of();
936
937         store.setRunOnPendingTransactionsComplete(operation);
938     }
939
940     @Override
941     protected void unpauseLeader() {
942         LOG.debug("{}: In unpauseLeader", persistenceId());
943         paused = false;
944
945         store.setRunOnPendingTransactionsComplete(null);
946
947         // Restore tell-based protocol state as if we were becoming the leader
948         knownFrontends = Verify.verifyNotNull(frontendMetadata.toLeaderState(this));
949     }
950
951     @Override
952     protected OnDemandRaftState.AbstractBuilder<?> newOnDemandRaftStateBuilder() {
953         return OnDemandShardState.newBuilder().treeChangeListenerActors(treeChangeSupport.getListenerActors())
954                 .dataChangeListenerActors(changeSupport.getListenerActors())
955                 .commitCohortActors(store.getCohortActors());
956     }
957
958     @Override
959     public String persistenceId() {
960         return this.name;
961     }
962
963     @VisibleForTesting
964     ShardCommitCoordinator getCommitCoordinator() {
965         return commitCoordinator;
966     }
967
968     public DatastoreContext getDatastoreContext() {
969         return datastoreContext;
970     }
971
972     @VisibleForTesting
973     public ShardDataTree getDataStore() {
974         return store;
975     }
976
977     @VisibleForTesting
978     ShardStats getShardMBean() {
979         return shardMBean;
980     }
981
982     public static Builder builder() {
983         return new Builder();
984     }
985
986     public abstract static class AbstractBuilder<T extends AbstractBuilder<T, S>, S extends Shard> {
987         private final Class<S> shardClass;
988         private ShardIdentifier id;
989         private Map<String, String> peerAddresses = Collections.emptyMap();
990         private DatastoreContext datastoreContext;
991         private SchemaContextProvider schemaContextProvider;
992         private DatastoreSnapshot.ShardSnapshot restoreFromSnapshot;
993         private TipProducingDataTree dataTree;
994         private volatile boolean sealed;
995
996         protected AbstractBuilder(final Class<S> shardClass) {
997             this.shardClass = shardClass;
998         }
999
1000         protected void checkSealed() {
1001             Preconditions.checkState(!sealed, "Builder isalready sealed - further modifications are not allowed");
1002         }
1003
1004         @SuppressWarnings("unchecked")
1005         private T self() {
1006             return (T) this;
1007         }
1008
1009         public T id(final ShardIdentifier newId) {
1010             checkSealed();
1011             this.id = newId;
1012             return self();
1013         }
1014
1015         public T peerAddresses(final Map<String, String> newPeerAddresses) {
1016             checkSealed();
1017             this.peerAddresses = newPeerAddresses;
1018             return self();
1019         }
1020
1021         public T datastoreContext(final DatastoreContext newDatastoreContext) {
1022             checkSealed();
1023             this.datastoreContext = newDatastoreContext;
1024             return self();
1025         }
1026
1027         public T schemaContextProvider(final SchemaContextProvider schemaContextProvider) {
1028             checkSealed();
1029             this.schemaContextProvider = Preconditions.checkNotNull(schemaContextProvider);
1030             return self();
1031         }
1032
1033         public T restoreFromSnapshot(final DatastoreSnapshot.ShardSnapshot newRestoreFromSnapshot) {
1034             checkSealed();
1035             this.restoreFromSnapshot = newRestoreFromSnapshot;
1036             return self();
1037         }
1038
1039         public T dataTree(final TipProducingDataTree newDataTree) {
1040             checkSealed();
1041             this.dataTree = newDataTree;
1042             return self();
1043         }
1044
1045         public ShardIdentifier getId() {
1046             return id;
1047         }
1048
1049         public Map<String, String> getPeerAddresses() {
1050             return peerAddresses;
1051         }
1052
1053         public DatastoreContext getDatastoreContext() {
1054             return datastoreContext;
1055         }
1056
1057         public SchemaContext getSchemaContext() {
1058             return Verify.verifyNotNull(schemaContextProvider.getSchemaContext());
1059         }
1060
1061         public DatastoreSnapshot.ShardSnapshot getRestoreFromSnapshot() {
1062             return restoreFromSnapshot;
1063         }
1064
1065         public TipProducingDataTree getDataTree() {
1066             return dataTree;
1067         }
1068
1069         public TreeType getTreeType() {
1070             switch (datastoreContext.getLogicalStoreType()) {
1071                 case CONFIGURATION:
1072                     return TreeType.CONFIGURATION;
1073                 case OPERATIONAL:
1074                     return TreeType.OPERATIONAL;
1075                 default:
1076                     throw new IllegalStateException("Unhandled logical store type "
1077                             + datastoreContext.getLogicalStoreType());
1078             }
1079         }
1080
1081         protected void verify() {
1082             Preconditions.checkNotNull(id, "id should not be null");
1083             Preconditions.checkNotNull(peerAddresses, "peerAddresses should not be null");
1084             Preconditions.checkNotNull(datastoreContext, "dataStoreContext should not be null");
1085             Preconditions.checkNotNull(schemaContextProvider, "schemaContextProvider should not be null");
1086         }
1087
1088         public Props props() {
1089             sealed = true;
1090             verify();
1091             return Props.create(shardClass, this);
1092         }
1093     }
1094
1095     public static class Builder extends AbstractBuilder<Builder, Shard> {
1096         private Builder() {
1097             super(Shard.class);
1098         }
1099     }
1100
1101     Ticker ticker() {
1102         return Ticker.systemTicker();
1103     }
1104
1105     void scheduleNextPendingTransaction() {
1106         self().tell(RESUME_NEXT_PENDING_TRANSACTION, ActorRef.noSender());
1107     }
1108 }

©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.