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.Props;
14 import akka.event.Logging;
15 import akka.event.LoggingAdapter;
16 import akka.japi.Creator;
17 import akka.persistence.RecoveryFailure;
18 import akka.serialization.Serialization;
20 import com.google.common.base.Optional;
21 import com.google.common.base.Preconditions;
22 import com.google.common.util.concurrent.FutureCallback;
23 import com.google.common.util.concurrent.Futures;
24 import com.google.common.util.concurrent.ListenableFuture;
25 import com.google.protobuf.ByteString;
26 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
27 import org.opendaylight.controller.cluster.datastore.identifiers.ShardTransactionIdentifier;
28 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardMBeanFactory;
29 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
30 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
31 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
32 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionChain;
33 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionChainReply;
34 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
35 import org.opendaylight.controller.cluster.datastore.messages.EnableNotification;
36 import org.opendaylight.controller.cluster.datastore.messages.ForwardedCommitTransaction;
37 import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved;
38 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener;
39 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
40 import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
41 import org.opendaylight.controller.cluster.datastore.modification.Modification;
42 import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
43 import org.opendaylight.controller.cluster.raft.ConfigParams;
44 import org.opendaylight.controller.cluster.raft.DefaultConfigParamsImpl;
45 import org.opendaylight.controller.cluster.raft.RaftActor;
46 import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
47 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
48 import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStore;
49 import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStoreFactory;
50 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
51 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
52 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
53 import org.opendaylight.yangtools.concepts.ListenerRegistration;
54 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
55 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
56 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
58 import scala.concurrent.duration.FiniteDuration;
60 import java.util.ArrayList;
61 import java.util.Date;
62 import java.util.HashMap;
63 import java.util.List;
65 import java.util.concurrent.ExecutionException;
66 import java.util.concurrent.TimeUnit;
69 * A Shard represents a portion of the logical data tree <br/>
71 * Our Shard uses InMemoryDataStore as it's internal representation and delegates all requests it
74 public class Shard extends RaftActor {
76 private static final ConfigParams configParams = new ShardConfigParams();
78 public static final String DEFAULT_NAME = "default";
80 // The state of this Shard
81 private final InMemoryDOMDataStore store;
83 private final Map<Object, DOMStoreThreePhaseCommitCohort>
84 modificationToCohort = new HashMap<>();
86 private final LoggingAdapter LOG =
87 Logging.getLogger(getContext().system(), this);
89 // By default persistent will be true and can be turned off using the system
90 // property shard.persistent
91 private final boolean persistent;
93 /// The name of this shard
94 private final ShardIdentifier name;
96 private final ShardStats shardMBean;
98 private final List<ActorSelection> dataChangeListeners = new ArrayList<>();
100 private final DatastoreContext datastoreContext;
103 private SchemaContext schemaContext;
105 private Shard(ShardIdentifier name, Map<ShardIdentifier, String> peerAddresses,
106 DatastoreContext datastoreContext) {
107 super(name.toString(), mapPeerAddresses(peerAddresses), Optional.of(configParams));
110 this.datastoreContext = datastoreContext;
112 String setting = System.getProperty("shard.persistent");
114 this.persistent = !"false".equals(setting);
116 LOG.info("Shard created : {} persistent : {}", name, persistent);
118 store = InMemoryDOMDataStoreFactory.create(name.toString(), null,
119 datastoreContext.getDataStoreProperties());
121 shardMBean = ShardMBeanFactory.getShardStatsMBean(name.toString());
126 private static Map<String, String> mapPeerAddresses(
127 Map<ShardIdentifier, String> peerAddresses) {
128 Map<String, String> map = new HashMap<>();
130 for (Map.Entry<ShardIdentifier, String> entry : peerAddresses
132 map.put(entry.getKey().toString(), entry.getValue());
138 public static Props props(final ShardIdentifier name,
139 final Map<ShardIdentifier, String> peerAddresses,
140 DatastoreContext datastoreContext) {
141 Preconditions.checkNotNull(name, "name should not be null");
142 Preconditions.checkNotNull(peerAddresses, "peerAddresses should not be null");
143 Preconditions.checkNotNull(datastoreContext, "shardContext should not be null");
145 return Props.create(new ShardCreator(name, peerAddresses, datastoreContext));
148 @Override public void onReceiveRecover(Object message) {
149 LOG.debug("onReceiveRecover: Received message {} from {}", message.getClass().toString(),
152 if (message instanceof RecoveryFailure){
153 LOG.error(((RecoveryFailure) message).cause(), "Recovery failed because of this cause");
155 super.onReceiveRecover(message);
159 @Override public void onReceiveCommand(Object message) {
160 LOG.debug("onReceiveCommand: Received message {} from {}", message.getClass().toString(),
163 if (message.getClass()
164 .equals(CreateTransactionChain.SERIALIZABLE_CLASS)) {
166 createTransactionChain();
167 } else if (getLeader() != null) {
168 getLeader().forward(message, getContext());
170 } else if (message instanceof RegisterChangeListener) {
171 registerChangeListener((RegisterChangeListener) message);
172 } else if (message instanceof UpdateSchemaContext) {
173 updateSchemaContext((UpdateSchemaContext) message);
174 } else if (message instanceof ForwardedCommitTransaction) {
175 handleForwardedCommit((ForwardedCommitTransaction) message);
176 } else if (message.getClass()
177 .equals(CreateTransaction.SERIALIZABLE_CLASS)) {
179 createTransaction(CreateTransaction.fromSerializable(message));
180 } else if (getLeader() != null) {
181 getLeader().forward(message, getContext());
183 } else if (message instanceof PeerAddressResolved) {
184 PeerAddressResolved resolved = (PeerAddressResolved) message;
185 setPeerAddress(resolved.getPeerId().toString(),
186 resolved.getPeerAddress());
188 super.onReceiveCommand(message);
192 private ActorRef createTypedTransactionActor(
193 CreateTransaction createTransaction,
194 ShardTransactionIdentifier transactionId) {
195 if (createTransaction.getTransactionType()
196 == TransactionProxy.TransactionType.READ_ONLY.ordinal()) {
198 shardMBean.incrementReadOnlyTransactionCount();
200 return getContext().actorOf(
201 ShardTransaction.props(store.newReadOnlyTransaction(), getSelf(),
202 schemaContext,datastoreContext, name.toString()), transactionId.toString());
204 } else if (createTransaction.getTransactionType()
205 == TransactionProxy.TransactionType.READ_WRITE.ordinal()) {
207 shardMBean.incrementReadWriteTransactionCount();
209 return getContext().actorOf(
210 ShardTransaction.props(store.newReadWriteTransaction(), getSelf(),
211 schemaContext, datastoreContext,name.toString()), transactionId.toString());
214 } else if (createTransaction.getTransactionType()
215 == TransactionProxy.TransactionType.WRITE_ONLY.ordinal()) {
217 shardMBean.incrementWriteOnlyTransactionCount();
219 return getContext().actorOf(
220 ShardTransaction.props(store.newWriteOnlyTransaction(), getSelf(),
221 schemaContext, datastoreContext, name.toString()), transactionId.toString());
223 throw new IllegalArgumentException(
224 "Shard="+name + ":CreateTransaction message has unidentified transaction type="
225 + createTransaction.getTransactionType());
229 private void createTransaction(CreateTransaction createTransaction) {
231 ShardTransactionIdentifier transactionId =
232 ShardTransactionIdentifier.builder()
233 .remoteTransactionId(createTransaction.getTransactionId())
235 LOG.debug("Creating transaction : {} ", transactionId);
236 ActorRef transactionActor =
237 createTypedTransactionActor(createTransaction, transactionId);
240 .tell(new CreateTransactionReply(
241 Serialization.serializedActorPath(transactionActor),
242 createTransaction.getTransactionId()).toSerializable(),
247 private void commit(final ActorRef sender, Object serialized) {
248 Modification modification = MutableCompositeModification
249 .fromSerializable(serialized, schemaContext);
250 DOMStoreThreePhaseCommitCohort cohort =
251 modificationToCohort.remove(serialized);
252 if (cohort == null) {
254 "Could not find cohort for modification : {}. Writing modification using a new transaction",
256 DOMStoreReadWriteTransaction transaction =
257 store.newReadWriteTransaction();
258 modification.apply(transaction);
259 DOMStoreThreePhaseCommitCohort commitCohort = transaction.ready();
260 ListenableFuture<Void> future =
261 commitCohort.preCommit();
264 future = commitCohort.commit();
266 } catch (InterruptedException | ExecutionException e) {
267 shardMBean.incrementFailedTransactionsCount();
268 LOG.error("Failed to commit", e);
271 //we want to just apply the recovery commit and return
272 shardMBean.incrementCommittedTransactionCount();
276 final ListenableFuture<Void> future = cohort.commit();
277 final ActorRef self = getSelf();
279 Futures.addCallback(future, new FutureCallback<Void>() {
281 public void onSuccess(Void v) {
282 sender.tell(new CommitTransactionReply().toSerializable(), self);
283 shardMBean.incrementCommittedTransactionCount();
284 shardMBean.setLastCommittedTransactionTime(new Date());
288 public void onFailure(Throwable t) {
289 LOG.error(t, "An exception happened during commit");
290 shardMBean.incrementFailedTransactionsCount();
291 sender.tell(new akka.actor.Status.Failure(t), self);
297 private void handleForwardedCommit(ForwardedCommitTransaction message) {
298 Object serializedModification =
299 message.getModification().toSerializable();
302 .put(serializedModification, message.getCohort());
305 this.persistData(getSender(), "identifier",
306 new CompositeModificationPayload(serializedModification));
308 this.commit(getSender(), serializedModification);
312 private void updateSchemaContext(UpdateSchemaContext message) {
313 this.schemaContext = message.getSchemaContext();
314 store.onGlobalContextUpdated(message.getSchemaContext());
317 private void registerChangeListener(
318 RegisterChangeListener registerChangeListener) {
320 LOG.debug("registerDataChangeListener for {}", registerChangeListener
324 ActorSelection dataChangeListenerPath = getContext()
325 .system().actorSelection(
326 registerChangeListener.getDataChangeListenerPath());
329 // Notify the listener if notifications should be enabled or not
330 // If this shard is the leader then it will enable notifications else
332 dataChangeListenerPath
333 .tell(new EnableNotification(isLeader()), getSelf());
335 // Now store a reference to the data change listener so it can be notified
336 // at a later point if notifications should be enabled or disabled
337 dataChangeListeners.add(dataChangeListenerPath);
339 AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>
340 listener = new DataChangeListenerProxy(schemaContext, dataChangeListenerPath);
342 ListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>>
343 registration = store.registerChangeListener(registerChangeListener.getPath(),
344 listener, registerChangeListener.getScope());
345 ActorRef listenerRegistration =
346 getContext().actorOf(
347 DataChangeListenerRegistration.props(registration));
350 "registerDataChangeListener sending reply, listenerRegistrationPath = {} "
351 , listenerRegistration.path().toString());
354 .tell(new RegisterChangeListenerReply(listenerRegistration.path()),
358 private void createTransactionChain() {
359 DOMStoreTransactionChain chain = store.createTransactionChain();
360 ActorRef transactionChain = getContext().actorOf(
361 ShardTransactionChain.props(chain, schemaContext, datastoreContext,name.toString() ));
362 getSender().tell(new CreateTransactionChainReply(transactionChain.path()).toSerializable(),
366 @Override protected void applyState(ActorRef clientActor, String identifier,
369 if (data instanceof CompositeModificationPayload) {
370 Object modification =
371 ((CompositeModificationPayload) data).getModification();
373 if (modification != null) {
374 commit(clientActor, modification);
377 "modification is null - this is very unexpected, clientActor = {}, identifier = {}",
378 identifier, clientActor.path().toString());
382 LOG.error("Unknown state received {}", data);
386 ReplicatedLogEntry lastLogEntry = getLastLogEntry();
388 if (lastLogEntry != null) {
389 shardMBean.setLastLogIndex(lastLogEntry.getIndex());
390 shardMBean.setLastLogTerm(lastLogEntry.getTerm());
393 shardMBean.setCommitIndex(getCommitIndex());
394 shardMBean.setLastApplied(getLastApplied());
398 @Override protected void createSnapshot() {
399 throw new UnsupportedOperationException("createSnapshot");
402 @Override protected void applySnapshot(ByteString snapshot) {
403 throw new UnsupportedOperationException("applySnapshot");
406 @Override protected void onStateChanged() {
407 for (ActorSelection dataChangeListener : dataChangeListeners) {
409 .tell(new EnableNotification(isLeader()), getSelf());
412 if (getLeaderId() != null) {
413 shardMBean.setLeader(getLeaderId());
416 shardMBean.setRaftState(getRaftState().name());
417 shardMBean.setCurrentTerm(getCurrentTerm());
420 @Override public String persistenceId() {
421 return this.name.toString();
425 private static class ShardConfigParams extends DefaultConfigParamsImpl {
426 public static final FiniteDuration HEART_BEAT_INTERVAL =
427 new FiniteDuration(500, TimeUnit.MILLISECONDS);
429 @Override public FiniteDuration getHeartBeatInterval() {
430 return HEART_BEAT_INTERVAL;
434 private static class ShardCreator implements Creator<Shard> {
436 private static final long serialVersionUID = 1L;
438 final ShardIdentifier name;
439 final Map<ShardIdentifier, String> peerAddresses;
440 final DatastoreContext datastoreContext;
442 ShardCreator(ShardIdentifier name, Map<ShardIdentifier, String> peerAddresses,
443 DatastoreContext datastoreContext) {
445 this.peerAddresses = peerAddresses;
446 this.datastoreContext = datastoreContext;
450 public Shard create() throws Exception {
451 return new Shard(name, peerAddresses, datastoreContext);