Add info logging and improve error reporting in clustering-it
[controller.git] / opendaylight / md-sal / samples / clustering-test-app / provider / src / main / java / org / opendaylight / controller / clustering / it / provider / impl / WriteTransactionsHandler.java
1 /*
2  * Copyright (c) 2017 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.clustering.it.provider.impl;
10
11 import com.google.common.base.Preconditions;
12 import com.google.common.util.concurrent.ListenableFuture;
13 import com.google.common.util.concurrent.SettableFuture;
14 import java.util.LinkedHashSet;
15 import java.util.Set;
16 import java.util.SplittableRandom;
17 import java.util.concurrent.ConcurrentHashMap;
18 import java.util.concurrent.ExecutionException;
19 import java.util.concurrent.TimeUnit;
20 import java.util.concurrent.TimeoutException;
21 import java.util.concurrent.atomic.AtomicLong;
22 import org.opendaylight.controller.md.sal.common.api.data.AsyncTransaction;
23 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
24 import org.opendaylight.controller.md.sal.common.api.data.OptimisticLockFailedException;
25 import org.opendaylight.controller.md.sal.common.api.data.TransactionChain;
26 import org.opendaylight.controller.md.sal.common.api.data.TransactionChainListener;
27 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
28 import org.opendaylight.controller.md.sal.dom.api.DOMDataBroker;
29 import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction;
30 import org.opendaylight.controller.md.sal.dom.api.DOMTransactionChain;
31 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.WriteTransactionsInput;
32 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.WriteTransactionsOutput;
33 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.WriteTransactionsOutputBuilder;
34 import org.opendaylight.yangtools.yang.common.RpcError;
35 import org.opendaylight.yangtools.yang.common.RpcResult;
36 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
37 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
38 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
39 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
40 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
41 import org.opendaylight.yangtools.yang.data.api.schema.MapNode;
42 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
43 import org.opendaylight.yangtools.yang.data.impl.schema.builder.api.CollectionNodeBuilder;
44 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder;
45 import org.slf4j.Logger;
46 import org.slf4j.LoggerFactory;
47
48 public abstract class WriteTransactionsHandler extends AbstractTransactionHandler {
49     private static final class Chained extends WriteTransactionsHandler implements TransactionChainListener {
50         private final SplittableRandom random = new SplittableRandom();
51         private final DOMTransactionChain transactionChain;
52
53         Chained(final DOMDataBroker dataBroker, final YangInstanceIdentifier idListItem,
54             final WriteTransactionsInput input) {
55             super(idListItem, input);
56             transactionChain = dataBroker.createTransactionChain(this);
57         }
58
59         @Override
60         DOMDataWriteTransaction createTransaction() {
61             return transactionChain.newWriteOnlyTransaction();
62         }
63
64         @Override
65         int nextInt(final int bound) {
66             return random.nextInt(bound);
67         }
68
69         @Override
70         public void onTransactionChainFailed(final TransactionChain<?, ?> chain,
71                 final AsyncTransaction<?, ?> transaction, final Throwable cause) {
72             // This is expected to happen frequently in isolation testing.
73             LOG.debug("Transaction chain failed.", cause);
74             // Do not return RPC here, rely on transaction failure to call runFailed.
75         }
76
77         @Override
78         public void onTransactionChainSuccessful(final TransactionChain<?, ?> chain) {
79             LOG.debug("Transaction chain closed successfully.");
80         }
81     }
82
83     private static final class Simple extends WriteTransactionsHandler {
84         private final LinkedHashSet<Integer> previousNumbers = new LinkedHashSet<>();
85         private final SplittableRandom random = new SplittableRandom();
86         private final DOMDataBroker dataBroker;
87
88         Simple(final DOMDataBroker dataBroker, final YangInstanceIdentifier idListItem,
89             final WriteTransactionsInput input) {
90             super(idListItem, input);
91             this.dataBroker = Preconditions.checkNotNull(dataBroker);
92         }
93
94         @Override
95         DOMDataWriteTransaction createTransaction() {
96             return dataBroker.newWriteOnlyTransaction();
97         }
98
99         @Override
100         int nextInt(final int bound) {
101             int nextInt;
102             do {
103                 nextInt = random.nextInt(bound);
104             } while (previousNumbers.contains(nextInt));
105
106             if (previousNumbers.size() > 100000) {
107                 previousNumbers.iterator().remove();
108             }
109             previousNumbers.add(nextInt);
110
111             return nextInt;
112         }
113     }
114
115     private static final Logger LOG = LoggerFactory.getLogger(WriteTransactionsHandler.class);
116
117     final SettableFuture<RpcResult<WriteTransactionsOutput>> completionFuture = SettableFuture.create();
118     private final Set<Integer> usedValues = ConcurrentHashMap.newKeySet();
119     private final YangInstanceIdentifier idListItem;
120
121     private final AtomicLong insertTx = new AtomicLong();
122     private final AtomicLong deleteTx = new AtomicLong();
123
124     WriteTransactionsHandler(final YangInstanceIdentifier idListItem, final WriteTransactionsInput input) {
125         super(input);
126         this.idListItem = Preconditions.checkNotNull(idListItem);
127     }
128
129     public static ListenableFuture<RpcResult<WriteTransactionsOutput>> start(final DOMDataBroker domDataBroker,
130             final WriteTransactionsInput input) {
131         LOG.info("Starting write transactions with input {}", input);
132
133         final String id = input.getId();
134         final MapEntryNode entry = ImmutableNodes.mapEntryBuilder(ID_INT, ID, id)
135                 .withChild(ImmutableNodes.mapNodeBuilder(ITEM).build())
136                 .build();
137         final YangInstanceIdentifier idListItem = ID_INT_YID.node(entry.getIdentifier());
138
139         final ContainerNode containerNode = ImmutableContainerNodeBuilder.create()
140                 .withNodeIdentifier(new NodeIdentifier(ID_INTS))
141                 .withChild(ImmutableNodes.mapNodeBuilder(ID_INT).build())
142                 .build();
143
144         DOMDataWriteTransaction tx = domDataBroker.newWriteOnlyTransaction();
145         // write only the top list
146         tx.merge(LogicalDatastoreType.CONFIGURATION, ID_INTS_YID, containerNode);
147         try {
148             tx.submit().checkedGet(INIT_TX_TIMEOUT_SECONDS, TimeUnit.SECONDS);
149         } catch (final OptimisticLockFailedException e) {
150             // when multiple write-transactions are executed concurrently we need to ignore this.
151             // If we get optimistic lock here it means id-ints already exists and we can continue.
152             LOG.debug("Got an optimistic lock when writing initial top level list element.", e);
153         } catch (final TransactionCommitFailedException | TimeoutException e) {
154             LOG.error("Error writing top-level path {}: {}", ID_INTS_YID, containerNode, e);
155             return RpcResultBuilder.<WriteTransactionsOutput>failed().withError(RpcError.ErrorType.APPLICATION,
156                 String.format("Could not start write transactions - error writing top-level path %s:  %s",
157                     ID_INTS_YID, containerNode), e).buildFuture();
158         }
159
160         tx = domDataBroker.newWriteOnlyTransaction();
161         tx.merge(LogicalDatastoreType.CONFIGURATION, idListItem, entry);
162
163         try {
164             tx.submit().get(INIT_TX_TIMEOUT_SECONDS, TimeUnit.SECONDS);
165         } catch (InterruptedException | ExecutionException | TimeoutException e) {
166             LOG.error("Error writing top-level path {}: {}", idListItem, entry, e);
167             return RpcResultBuilder.<WriteTransactionsOutput>failed().withError(RpcError.ErrorType.APPLICATION,
168                 String.format("Could not start write transactions - error writing list entry path %s: %s",
169                     idListItem, entry), e).buildFuture();
170         }
171
172         LOG.debug("Filling the item list with initial values.");
173
174         final CollectionNodeBuilder<MapEntryNode, MapNode> mapBuilder = ImmutableNodes.mapNodeBuilder(ITEM);
175
176         final YangInstanceIdentifier itemListId = idListItem.node(ITEM);
177         tx = domDataBroker.newWriteOnlyTransaction();
178         final MapNode itemListNode = mapBuilder.build();
179         tx.put(LogicalDatastoreType.CONFIGURATION, itemListId, itemListNode);
180
181         try {
182             tx.submit().get(INIT_TX_TIMEOUT_SECONDS, TimeUnit.SECONDS);
183         } catch (InterruptedException | ExecutionException | TimeoutException e) {
184             LOG.error("Error filling initial item list path {}: {}", itemListId, itemListNode, e);
185             return RpcResultBuilder.<WriteTransactionsOutput>failed().withError(RpcError.ErrorType.APPLICATION,
186                 String.format("Could not start write transactions - error filling initial item list path %s: %s",
187                     itemListId, itemListNode), e).buildFuture();
188         }
189
190         final WriteTransactionsHandler handler;
191         if (input.isChainedTransactions()) {
192             handler = new Chained(domDataBroker, idListItem, input);
193         } else {
194             handler = new Simple(domDataBroker, idListItem, input);
195         }
196
197         handler.doStart();
198
199         LOG.info("Write transactions successfully started");
200         return handler.completionFuture;
201     }
202
203     @Override
204     ListenableFuture<Void> execWrite(final long txId) {
205         final int i = nextInt(MAX_ITEM + 1);
206
207         final YangInstanceIdentifier entryId =
208                 idListItem.node(ITEM).node(new YangInstanceIdentifier.NodeIdentifierWithPredicates(ITEM, NUMBER, i));
209
210         final DOMDataWriteTransaction tx = createTransaction();
211
212         if (usedValues.contains(i)) {
213             LOG.debug("Deleting item: {}", i);
214             deleteTx.incrementAndGet();
215             tx.delete(LogicalDatastoreType.CONFIGURATION, entryId);
216             usedValues.remove(i);
217
218         } else {
219             LOG.debug("Inserting item: {}", i);
220             insertTx.incrementAndGet();
221             final MapEntryNode entry = ImmutableNodes.mapEntry(ITEM, NUMBER, i);
222             tx.put(LogicalDatastoreType.CONFIGURATION, entryId, entry);
223             usedValues.add(i);
224         }
225
226         return tx.submit();
227     }
228
229     @Override
230     void runFailed(final Throwable cause, final long txId) {
231         completionFuture.set(RpcResultBuilder.<WriteTransactionsOutput>failed()
232             .withError(RpcError.ErrorType.APPLICATION, "Commit failed for tx # " + txId, cause).build());
233     }
234
235     @Override
236     void runSuccessful(final long allTx) {
237         final WriteTransactionsOutput output = new WriteTransactionsOutputBuilder()
238                 .setAllTx(allTx)
239                 .setInsertTx(insertTx.get())
240                 .setDeleteTx(deleteTx.get())
241                 .build();
242
243         completionFuture.set(RpcResultBuilder.<WriteTransactionsOutput>success()
244                 .withResult(output).build());
245     }
246
247     @Override
248     void runTimedOut(final String cause) {
249         completionFuture.set(RpcResultBuilder.<WriteTransactionsOutput>failed()
250             .withError(RpcError.ErrorType.APPLICATION, cause).build());
251     }
252
253     abstract DOMDataWriteTransaction createTransaction();
254
255     abstract int nextInt(int bound);
256 }

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