2 * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
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
9 package org.opendaylight.controller.cluster.datastore;
11 import akka.actor.ActorRef;
12 import akka.actor.ActorSelection;
13 import akka.actor.PoisonPill;
14 import akka.actor.Props;
15 import akka.event.Logging;
16 import akka.event.LoggingAdapter;
17 import akka.japi.Creator;
18 import akka.persistence.RecoveryFailure;
19 import akka.serialization.Serialization;
20 import com.google.common.annotations.VisibleForTesting;
21 import com.google.common.base.Optional;
22 import com.google.common.base.Preconditions;
23 import com.google.common.util.concurrent.CheckedFuture;
24 import com.google.common.util.concurrent.FutureCallback;
25 import com.google.common.util.concurrent.Futures;
26 import com.google.common.util.concurrent.ListenableFuture;
27 import com.google.protobuf.ByteString;
28 import com.google.protobuf.InvalidProtocolBufferException;
29 import org.opendaylight.controller.cluster.common.actor.CommonConfig;
30 import org.opendaylight.controller.cluster.common.actor.MeteringBehavior;
31 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
32 import org.opendaylight.controller.cluster.datastore.identifiers.ShardTransactionIdentifier;
33 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardMBeanFactory;
34 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
35 import org.opendaylight.controller.cluster.datastore.messages.CloseTransactionChain;
36 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
37 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
38 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionChainReply;
39 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
40 import org.opendaylight.controller.cluster.datastore.messages.EnableNotification;
41 import org.opendaylight.controller.cluster.datastore.messages.ForwardedCommitTransaction;
42 import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved;
43 import org.opendaylight.controller.cluster.datastore.messages.ReadData;
44 import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply;
45 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener;
46 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
47 import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
48 import org.opendaylight.controller.cluster.datastore.modification.Modification;
49 import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
50 import org.opendaylight.controller.cluster.datastore.node.NormalizedNodeToNodeCodec;
51 import org.opendaylight.controller.cluster.raft.ConfigParams;
52 import org.opendaylight.controller.cluster.raft.DefaultConfigParamsImpl;
53 import org.opendaylight.controller.cluster.raft.RaftActor;
54 import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
55 import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply;
56 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.CompositeModificationPayload;
57 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
58 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
59 import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStore;
60 import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStoreFactory;
61 import org.opendaylight.controller.protobuff.messages.common.NormalizedNodeMessages;
62 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
63 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
64 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
65 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionFactory;
66 import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
67 import org.opendaylight.yangtools.concepts.ListenerRegistration;
68 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
69 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
70 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
71 import scala.concurrent.duration.FiniteDuration;
73 import java.util.ArrayList;
74 import java.util.HashMap;
75 import java.util.List;
77 import java.util.concurrent.ExecutionException;
78 import java.util.concurrent.TimeUnit;
81 * A Shard represents a portion of the logical data tree <br/>
83 * Our Shard uses InMemoryDataStore as it's internal representation and delegates all requests it
86 public class Shard extends RaftActor {
88 private static final ConfigParams configParams = new ShardConfigParams();
90 public static final String DEFAULT_NAME = "default";
92 // The state of this Shard
93 private final InMemoryDOMDataStore store;
95 private final Map<Object, DOMStoreThreePhaseCommitCohort>
96 modificationToCohort = new HashMap<>();
98 private final LoggingAdapter LOG =
99 Logging.getLogger(getContext().system(), this);
101 // By default persistent will be true and can be turned off using the system
102 // property shard.persistent
103 private final boolean persistent;
105 /// The name of this shard
106 private final ShardIdentifier name;
108 private final ShardStats shardMBean;
110 private final List<ActorSelection> dataChangeListeners = new ArrayList<>();
112 private final DatastoreContext datastoreContext;
114 private SchemaContext schemaContext;
116 private ActorRef createSnapshotTransaction;
118 private final Map<String, DOMStoreTransactionChain> transactionChains = new HashMap<>();
120 private Shard(ShardIdentifier name, Map<ShardIdentifier, String> peerAddresses,
121 DatastoreContext datastoreContext, SchemaContext schemaContext) {
122 super(name.toString(), mapPeerAddresses(peerAddresses), Optional.of(configParams));
125 this.datastoreContext = datastoreContext;
126 this.schemaContext = schemaContext;
128 String setting = System.getProperty("shard.persistent");
130 this.persistent = !"false".equals(setting);
132 LOG.info("Shard created : {} persistent : {}", name, persistent);
134 store = InMemoryDOMDataStoreFactory.create(name.toString(), null,
135 datastoreContext.getDataStoreProperties());
137 if(schemaContext != null) {
138 store.onGlobalContextUpdated(schemaContext);
141 shardMBean = ShardMBeanFactory.getShardStatsMBean(name.toString(),
142 datastoreContext.getDataStoreMXBeanType());
143 shardMBean.setDataStoreExecutor(store.getDomStoreExecutor());
144 shardMBean.setNotificationManager(store.getDataChangeListenerNotificationManager());
146 if (isMetricsCaptureEnabled()) {
147 getContext().become(new MeteringBehavior(this));
151 private static Map<String, String> mapPeerAddresses(
152 Map<ShardIdentifier, String> peerAddresses) {
153 Map<String, String> map = new HashMap<>();
155 for (Map.Entry<ShardIdentifier, String> entry : peerAddresses
157 map.put(entry.getKey().toString(), entry.getValue());
163 public static Props props(final ShardIdentifier name,
164 final Map<ShardIdentifier, String> peerAddresses,
165 DatastoreContext datastoreContext, SchemaContext schemaContext) {
166 Preconditions.checkNotNull(name, "name should not be null");
167 Preconditions.checkNotNull(peerAddresses, "peerAddresses should not be null");
168 Preconditions.checkNotNull(datastoreContext, "dataStoreContext should not be null");
169 Preconditions.checkNotNull(schemaContext, "schemaContext should not be null");
171 return Props.create(new ShardCreator(name, peerAddresses, datastoreContext, schemaContext));
174 @Override public void onReceiveRecover(Object message) {
175 LOG.debug("onReceiveRecover: Received message {} from {}", message.getClass().toString(),
178 if (message instanceof RecoveryFailure){
179 LOG.error(((RecoveryFailure) message).cause(), "Recovery failed because of this cause");
181 super.onReceiveRecover(message);
185 @Override public void onReceiveCommand(Object message) {
186 LOG.debug("onReceiveCommand: Received message {} from {}", message.getClass().toString(),
189 if(message.getClass().equals(ReadDataReply.SERIALIZABLE_CLASS)) {
190 // This must be for install snapshot. Don't want to open this up and trigger
193 .tell(new CaptureSnapshotReply(ReadDataReply.getNormalizedNodeByteString(message)),
196 createSnapshotTransaction = null;
197 // Send a PoisonPill instead of sending close transaction because we do not really need
199 getSender().tell(PoisonPill.getInstance(), self());
201 } else if (message.getClass().equals(CloseTransactionChain.SERIALIZABLE_CLASS)){
202 closeTransactionChain(CloseTransactionChain.fromSerializable(message));
203 } else if (message instanceof RegisterChangeListener) {
204 registerChangeListener((RegisterChangeListener) message);
205 } else if (message instanceof UpdateSchemaContext) {
206 updateSchemaContext((UpdateSchemaContext) message);
207 } else if (message instanceof ForwardedCommitTransaction) {
208 handleForwardedCommit((ForwardedCommitTransaction) message);
209 } else if (message.getClass()
210 .equals(CreateTransaction.SERIALIZABLE_CLASS)) {
212 createTransaction(CreateTransaction.fromSerializable(message));
213 } else if (getLeader() != null) {
214 getLeader().forward(message, getContext());
216 getSender().tell(new akka.actor.Status.Failure(new IllegalStateException(
217 "Could not find leader so transaction cannot be created")), getSelf());
219 } else if (message instanceof PeerAddressResolved) {
220 PeerAddressResolved resolved = (PeerAddressResolved) message;
221 setPeerAddress(resolved.getPeerId().toString(),
222 resolved.getPeerAddress());
224 super.onReceiveCommand(message);
228 private void closeTransactionChain(CloseTransactionChain closeTransactionChain) {
229 DOMStoreTransactionChain chain =
230 transactionChains.remove(closeTransactionChain.getTransactionChainId());
237 private ActorRef createTypedTransactionActor(
239 ShardTransactionIdentifier transactionId,
240 String transactionChainId ) {
242 DOMStoreTransactionFactory factory = store;
244 if(!transactionChainId.isEmpty()) {
245 factory = transactionChains.get(transactionChainId);
247 DOMStoreTransactionChain transactionChain = store.createTransactionChain();
248 transactionChains.put(transactionChainId, transactionChain);
249 factory = transactionChain;
253 if(this.schemaContext == null){
254 throw new NullPointerException("schemaContext should not be null");
258 == TransactionProxy.TransactionType.READ_ONLY.ordinal()) {
260 shardMBean.incrementReadOnlyTransactionCount();
262 return getContext().actorOf(
263 ShardTransaction.props(factory.newReadOnlyTransaction(), getSelf(),
264 schemaContext,datastoreContext, shardMBean), transactionId.toString());
266 } else if (transactionType
267 == TransactionProxy.TransactionType.READ_WRITE.ordinal()) {
269 shardMBean.incrementReadWriteTransactionCount();
271 return getContext().actorOf(
272 ShardTransaction.props(factory.newReadWriteTransaction(), getSelf(),
273 schemaContext, datastoreContext, shardMBean), transactionId.toString());
276 } else if (transactionType
277 == TransactionProxy.TransactionType.WRITE_ONLY.ordinal()) {
279 shardMBean.incrementWriteOnlyTransactionCount();
281 return getContext().actorOf(
282 ShardTransaction.props(factory.newWriteOnlyTransaction(), getSelf(),
283 schemaContext, datastoreContext, shardMBean), transactionId.toString());
285 throw new IllegalArgumentException(
286 "Shard="+name + ":CreateTransaction message has unidentified transaction type="
291 private void createTransaction(CreateTransaction createTransaction) {
292 createTransaction(createTransaction.getTransactionType(),
293 createTransaction.getTransactionId(), createTransaction.getTransactionChainId());
296 private ActorRef createTransaction(int transactionType, String remoteTransactionId, String transactionChainId) {
298 ShardTransactionIdentifier transactionId =
299 ShardTransactionIdentifier.builder()
300 .remoteTransactionId(remoteTransactionId)
302 LOG.debug("Creating transaction : {} ", transactionId);
303 ActorRef transactionActor =
304 createTypedTransactionActor(transactionType, transactionId, transactionChainId);
307 .tell(new CreateTransactionReply(
308 Serialization.serializedActorPath(transactionActor),
309 remoteTransactionId).toSerializable(),
312 return transactionActor;
315 private void syncCommitTransaction(DOMStoreWriteTransaction transaction)
316 throws ExecutionException, InterruptedException {
317 DOMStoreThreePhaseCommitCohort commitCohort = transaction.ready();
318 commitCohort.preCommit().get();
319 commitCohort.commit().get();
323 private void commit(final ActorRef sender, Object serialized) {
324 Modification modification = MutableCompositeModification
325 .fromSerializable(serialized, schemaContext);
326 DOMStoreThreePhaseCommitCohort cohort =
327 modificationToCohort.remove(serialized);
328 if (cohort == null) {
330 "Could not find cohort for modification : {}. Writing modification using a new transaction",
332 DOMStoreWriteTransaction transaction =
333 store.newWriteOnlyTransaction();
335 LOG.debug("Created new transaction {}", transaction.getIdentifier().toString());
337 modification.apply(transaction);
339 syncCommitTransaction(transaction);
340 } catch (InterruptedException | ExecutionException e) {
341 shardMBean.incrementFailedTransactionsCount();
342 LOG.error("Failed to commit", e);
345 //we want to just apply the recovery commit and return
346 shardMBean.incrementCommittedTransactionCount();
352 LOG.error("Commit failed. Sender cannot be null");
356 final ListenableFuture<Void> future = cohort.commit();
357 final ActorRef self = getSelf();
359 Futures.addCallback(future, new FutureCallback<Void>() {
361 public void onSuccess(Void v) {
362 sender.tell(new CommitTransactionReply().toSerializable(), self);
363 shardMBean.incrementCommittedTransactionCount();
364 shardMBean.setLastCommittedTransactionTime(System.currentTimeMillis());
368 public void onFailure(Throwable t) {
369 LOG.error(t, "An exception happened during commit");
370 shardMBean.incrementFailedTransactionsCount();
371 sender.tell(new akka.actor.Status.Failure(t), self);
377 private void handleForwardedCommit(ForwardedCommitTransaction message) {
378 Object serializedModification =
379 message.getModification().toSerializable();
382 .put(serializedModification, message.getCohort());
385 this.persistData(getSender(), "identifier",
386 new CompositeModificationPayload(serializedModification));
388 this.commit(getSender(), serializedModification);
392 private void updateSchemaContext(UpdateSchemaContext message) {
393 this.schemaContext = message.getSchemaContext();
394 updateSchemaContext(message.getSchemaContext());
395 store.onGlobalContextUpdated(message.getSchemaContext());
398 @VisibleForTesting void updateSchemaContext(SchemaContext schemaContext) {
399 store.onGlobalContextUpdated(schemaContext);
402 private void registerChangeListener(
403 RegisterChangeListener registerChangeListener) {
405 LOG.debug("registerDataChangeListener for {}", registerChangeListener
409 ActorSelection dataChangeListenerPath = getContext()
410 .system().actorSelection(
411 registerChangeListener.getDataChangeListenerPath());
414 // Notify the listener if notifications should be enabled or not
415 // If this shard is the leader then it will enable notifications else
417 dataChangeListenerPath
418 .tell(new EnableNotification(isLeader()), getSelf());
420 // Now store a reference to the data change listener so it can be notified
421 // at a later point if notifications should be enabled or disabled
422 dataChangeListeners.add(dataChangeListenerPath);
424 AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>
425 listener = new DataChangeListenerProxy(schemaContext, dataChangeListenerPath);
427 ListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>>
428 registration = store.registerChangeListener(registerChangeListener.getPath(),
429 listener, registerChangeListener.getScope());
430 ActorRef listenerRegistration =
431 getContext().actorOf(
432 DataChangeListenerRegistration.props(registration));
435 "registerDataChangeListener sending reply, listenerRegistrationPath = {} "
436 , listenerRegistration.path().toString());
439 .tell(new RegisterChangeListenerReply(listenerRegistration.path()),
443 private void createTransactionChain() {
444 DOMStoreTransactionChain chain = store.createTransactionChain();
445 ActorRef transactionChain = getContext().actorOf(
446 ShardTransactionChain.props(chain, schemaContext, datastoreContext, shardMBean));
447 getSender().tell(new CreateTransactionChainReply(transactionChain.path()).toSerializable(),
451 private boolean isMetricsCaptureEnabled(){
452 CommonConfig config = new CommonConfig(getContext().system().settings().config());
453 return config.isMetricCaptureEnabled();
456 @Override protected void applyState(ActorRef clientActor, String identifier,
459 if (data instanceof CompositeModificationPayload) {
460 Object modification =
461 ((CompositeModificationPayload) data).getModification();
463 if (modification != null) {
464 commit(clientActor, modification);
467 "modification is null - this is very unexpected, clientActor = {}, identifier = {}",
468 identifier, clientActor.path().toString());
472 LOG.error("Unknown state received {} Class loader = {} CompositeNodeMod.ClassLoader = {}", data, data.getClass().getClassLoader(), CompositeModificationPayload.class.getClassLoader());
476 ReplicatedLogEntry lastLogEntry = getLastLogEntry();
478 if (lastLogEntry != null) {
479 shardMBean.setLastLogIndex(lastLogEntry.getIndex());
480 shardMBean.setLastLogTerm(lastLogEntry.getTerm());
483 shardMBean.setCommitIndex(getCommitIndex());
484 shardMBean.setLastApplied(getLastApplied());
488 @Override protected void createSnapshot() {
489 if (createSnapshotTransaction == null) {
491 // Create a transaction. We are really going to treat the transaction as a worker
492 // so that this actor does not get block building the snapshot
493 createSnapshotTransaction = createTransaction(
494 TransactionProxy.TransactionType.READ_ONLY.ordinal(),
495 "createSnapshot", "");
497 createSnapshotTransaction.tell(
498 new ReadData(YangInstanceIdentifier.builder().build()).toSerializable(), self());
503 @VisibleForTesting @Override protected void applySnapshot(ByteString snapshot) {
504 // Since this will be done only on Recovery or when this actor is a Follower
505 // we can safely commit everything in here. We not need to worry about event notifications
506 // as they would have already been disabled on the follower
508 LOG.info("Applying snapshot");
510 DOMStoreWriteTransaction transaction = store.newWriteOnlyTransaction();
511 NormalizedNodeMessages.Node serializedNode = NormalizedNodeMessages.Node.parseFrom(snapshot);
512 NormalizedNode<?, ?> node = new NormalizedNodeToNodeCodec(schemaContext)
513 .decode(YangInstanceIdentifier.builder().build(), serializedNode);
515 // delete everything first
516 transaction.delete(YangInstanceIdentifier.builder().build());
518 // Add everything from the remote node back
519 transaction.write(YangInstanceIdentifier.builder().build(), node);
520 syncCommitTransaction(transaction);
521 } catch (InvalidProtocolBufferException | InterruptedException | ExecutionException e) {
522 LOG.error(e, "An exception occurred when applying snapshot");
524 LOG.info("Done applying snapshot");
528 @Override protected void onStateChanged() {
529 for (ActorSelection dataChangeListener : dataChangeListeners) {
531 .tell(new EnableNotification(isLeader()), getSelf());
535 shardMBean.setRaftState(getRaftState().name());
536 shardMBean.setCurrentTerm(getCurrentTerm());
538 // If this actor is no longer the leader close all the transaction chains
540 for(Map.Entry<String, DOMStoreTransactionChain> entry : transactionChains.entrySet()){
541 LOG.debug("onStateChanged: Closing transaction chain {} because shard {} is no longer the leader", entry.getKey(), getId());
542 entry.getValue().close();
545 transactionChains.clear();
549 @Override protected void onLeaderChanged(String oldLeader, String newLeader) {
550 if((oldLeader == null && newLeader == null) || (newLeader != null && newLeader.equals(oldLeader)) ){
553 LOG.info("Current state = {}, Leader = {}", getRaftState().name(), newLeader);
554 shardMBean.setLeader(newLeader);
557 @Override public String persistenceId() {
558 return this.name.toString();
562 private static class ShardConfigParams extends DefaultConfigParamsImpl {
563 public static final FiniteDuration HEART_BEAT_INTERVAL =
564 new FiniteDuration(500, TimeUnit.MILLISECONDS);
566 @Override public FiniteDuration getHeartBeatInterval() {
567 return HEART_BEAT_INTERVAL;
571 private static class ShardCreator implements Creator<Shard> {
573 private static final long serialVersionUID = 1L;
575 final ShardIdentifier name;
576 final Map<ShardIdentifier, String> peerAddresses;
577 final DatastoreContext datastoreContext;
578 final SchemaContext schemaContext;
580 ShardCreator(ShardIdentifier name, Map<ShardIdentifier, String> peerAddresses,
581 DatastoreContext datastoreContext, SchemaContext schemaContext) {
583 this.peerAddresses = peerAddresses;
584 this.datastoreContext = datastoreContext;
585 this.schemaContext = schemaContext;
589 public Shard create() throws Exception {
590 return new Shard(name, peerAddresses, datastoreContext, schemaContext);
594 @VisibleForTesting NormalizedNode readStore() throws ExecutionException, InterruptedException {
595 DOMStoreReadTransaction transaction = store.newReadOnlyTransaction();
597 CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> future =
598 transaction.read(YangInstanceIdentifier.builder().build());
600 NormalizedNode<?, ?> node = future.get().get();
607 @VisibleForTesting void writeToStore(YangInstanceIdentifier id, NormalizedNode node)
608 throws ExecutionException, InterruptedException {
609 DOMStoreWriteTransaction transaction = store.newWriteOnlyTransaction();
611 transaction.write(id, node);
613 syncCommitTransaction(transaction);