netconf/sal-netconf-connector/src/main/java/org/opendaylight/netconf/sal/connect/netconf/sal/tx/WriteCandidateTx.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.netconf.sal.connect.netconf.sal.tx;
  10
  11 import com.google.common.base.Function;
  12 import com.google.common.base.Optional;
  13 import com.google.common.base.Preconditions;
  14 import com.google.common.util.concurrent.CheckedFuture;
  15 import com.google.common.util.concurrent.FutureCallback;
  16 import com.google.common.util.concurrent.Futures;
  17 import com.google.common.util.concurrent.ListenableFuture;
  18 import org.opendaylight.controller.md.sal.common.api.TransactionStatus;
  19 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
  20 import org.opendaylight.controller.md.sal.dom.api.DOMRpcResult;
  21 import org.opendaylight.netconf.api.NetconfDocumentedException;
  22 import org.opendaylight.netconf.sal.connect.netconf.util.NetconfBaseOps;
  23 import org.opendaylight.netconf.sal.connect.netconf.util.NetconfRpcFutureCallback;
  24 import org.opendaylight.netconf.sal.connect.util.RemoteDeviceId;
  25 import org.opendaylight.yangtools.yang.common.RpcError;
  26 import org.opendaylight.yangtools.yang.common.RpcResult;
  27 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
  28 import org.opendaylight.yangtools.yang.data.api.ModifyAction;
  29 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  30 import org.opendaylight.yangtools.yang.data.api.schema.DataContainerChild;
  31 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  32 import org.slf4j.Logger;
  33 import org.slf4j.LoggerFactory;
  34
  35 /**
  36  * Tx implementation for netconf devices that support only candidate datastore and no writable running
  37  * The sequence goes as:
  38  * <ol>
  39  *   <li>Lock candidate datastore on tx construction
  40  *     <ul>
  41  *       <li>Lock has to succeed, if it does not, an attempt to discard changes is made</li>
  42  *       <li>Discard changes has to succeed</li>
  43  *       <li>If discard is successful, lock is reattempted</li>
  44  *       <li>Second lock attempt has to succeed</li>
  45  *     </ul>
  46  *   </li>
  47  *   <li>Edit-config in candidate N times
  48  *     <ul>
  49  *       <li>If any issue occurs during edit, datastore is discarded using discard-changes rpc, unlocked and an exception is thrown async</li>
  50  *     </ul>
  51  *   </li>
  52  *   <li>Commit and Unlock candidate datastore async</li>
  53  * </ol>
  54  */
  55 public class WriteCandidateTx extends AbstractWriteTx {
  56
  57     private static final Logger LOG  = LoggerFactory.getLogger(WriteCandidateTx.class);
  58
  59     private static final Function<DOMRpcResult, RpcResult<TransactionStatus>> RPC_RESULT_TO_TX_STATUS = new Function<DOMRpcResult, RpcResult<TransactionStatus>>() {
  60         @Override
  61         public RpcResult<TransactionStatus> apply(final DOMRpcResult input) {
  62             if (isSuccess(input)) {
  63                 return RpcResultBuilder.success(TransactionStatus.COMMITED).build();
  64             } else {
  65                 final RpcResultBuilder<TransactionStatus> failed = RpcResultBuilder.failed();
  66                 for (final RpcError rpcError : input.getErrors()) {
  67                     failed.withError(rpcError.getErrorType(), rpcError.getTag(), rpcError.getMessage(),
  68                             rpcError.getApplicationTag(), rpcError.getInfo(), rpcError.getCause());
  69                 }
  70                 return failed.build();
  71             }
  72         }
  73     };
  74
  75     public WriteCandidateTx(final RemoteDeviceId id, final NetconfBaseOps rpc, final boolean rollbackSupport) {
  76         super(rpc, id, rollbackSupport);
  77     }
  78
  79     @Override
  80     protected synchronized void init() {
  81         LOG.trace("{}: Initializing {} transaction", id, getClass().getSimpleName());
  82         lock();
  83     }
  84
  85     private void lock() {
  86         final FutureCallback<DOMRpcResult> lockCandidateCallback = new FutureCallback<DOMRpcResult>() {
  87             @Override
  88             public void onSuccess(DOMRpcResult result) {
  89                 if (isSuccess(result)) {
  90                     if (LOG.isTraceEnabled()) {
  91                         LOG.trace("Lock candidate succesfull");
  92                     }
  93                 } else {
  94                     LOG.warn("{}: lock candidate invoked unsuccessfully: {}", id, result.getErrors());
  95                 }
  96             }
  97
  98             @Override
  99             public void onFailure(Throwable t) {
 100                 LOG.warn("Lock candidate operation failed. {}", t);
 101                 NetconfDocumentedException e = new NetconfDocumentedException(id + ": Lock candidate operation failed.", NetconfDocumentedException.ErrorType.application,
 102                         NetconfDocumentedException.ErrorTag.operation_failed, NetconfDocumentedException.ErrorSeverity.warning);
 103                 discardChanges();
 104                 throw new RuntimeException(e);
 105             }
 106         };
 107         netOps.lockCandidate(lockCandidateCallback);
 108     }
 109
 110     @Override
 111     protected void cleanup() {
 112         discardChanges();
 113         cleanupOnSuccess();
 114     }
 115
 116     @Override
 117     protected void handleEditException(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data, final NetconfDocumentedException e, final String editType) {
 118         LOG.warn("{}: Error {} data to (candidate){}, data: {}, canceling", id, editType, path, data, e);
 119         cancel();
 120         throw new RuntimeException(id + ": Error while " + editType + ": (candidate)" + path, e);
 121     }
 122
 123     @Override
 124     public synchronized CheckedFuture<Void, TransactionCommitFailedException> submit() {
 125         final ListenableFuture<Void> commitFutureAsVoid = Futures.transform(commit(), new Function<RpcResult<TransactionStatus>, Void>() {
 126             @Override
 127             public Void apply(final RpcResult<TransactionStatus> input) {
 128                 Preconditions.checkArgument(input.isSuccessful() && input.getErrors().isEmpty(), "Submit failed with errors: %s", input.getErrors());
 129                 return null;
 130             }
 131         });
 132
 133         return Futures.makeChecked(commitFutureAsVoid, new Function<Exception, TransactionCommitFailedException>() {
 134             @Override
 135             public TransactionCommitFailedException apply(final Exception input) {
 136                 return new TransactionCommitFailedException("Submit of transaction " + getIdentifier() + " failed", input);
 137             }
 138         });
 139     }
 140
 141     /**
 142      * This has to be non blocking since it is called from a callback on commit and its netty threadpool that is really sensitive to blocking calls
 143      */
 144     private void discardChanges() {
 145         netOps.discardChanges(new NetconfRpcFutureCallback("Discarding candidate", id));
 146     }
 147
 148     @Override
 149     public synchronized ListenableFuture<RpcResult<TransactionStatus>> performCommit() {
 150         final ListenableFuture<DOMRpcResult> rpcResult = netOps.commit(new NetconfRpcFutureCallback("Commit", id) {
 151             @Override
 152             public void onSuccess(final DOMRpcResult result) {
 153                 super.onSuccess(result);
 154                 LOG.debug("{}: Write successful, transaction: {}. Unlocking", id, getIdentifier());
 155                 cleanupOnSuccess();
 156             }
 157
 158             @Override
 159             protected void onUnsuccess(final DOMRpcResult result) {
 160                 LOG.error("{}: Write failed, transaction {}, discarding changes, unlocking: {}", id, getIdentifier(), result.getErrors());
 161                 cleanup();
 162             }
 163
 164             @Override
 165             public void onFailure(final Throwable t) {
 166                 LOG.error("{}: Write failed, transaction {}, discarding changes, unlocking", id, getIdentifier(), t);
 167                 cleanup();
 168             }
 169         });
 170
 171         return Futures.transform(rpcResult, RPC_RESULT_TO_TX_STATUS);
 172     }
 173
 174     protected void cleanupOnSuccess() {
 175         unlock();
 176     }
 177
 178     @Override
 179     protected void editConfig(final YangInstanceIdentifier path,
 180                               final Optional<NormalizedNode<?, ?>> data,
 181                               final DataContainerChild<?, ?> editStructure,
 182                               final Optional<ModifyAction> defaultOperation,
 183                               final String operation) {
 184         FutureCallback<DOMRpcResult> editConfigCallback = new FutureCallback<DOMRpcResult>() {
 185             @Override
 186             public void onSuccess(DOMRpcResult result) {
 187                 if (isSuccess(result)) {
 188                     if (LOG.isTraceEnabled()) {
 189                         LOG.trace("Edit candidate succesfull");
 190                     }
 191                 } else {
 192                     LOG.warn("{}: Edit candidate invoked unsuccessfully: {}", id, result.getErrors());
 193                 }
 194             }
 195
 196             @Override
 197             public void onFailure(Throwable t) {
 198                 LOG.warn("Edit candidate operation failed. {}", t);
 199                 NetconfDocumentedException e = new NetconfDocumentedException(id + ": Edit candidate operation failed.", NetconfDocumentedException.ErrorType.application,
 200                         NetconfDocumentedException.ErrorTag.operation_failed, NetconfDocumentedException.ErrorSeverity.warning);
 201                 handleEditException(path, data.orNull(), e, operation);
 202             }
 203         };
 204         if (defaultOperation.isPresent()) {
 205             netOps.editConfigCandidate(editConfigCallback, editStructure, defaultOperation.get(), rollbackSupport);
 206         } else {
 207             netOps.editConfigCandidate(editConfigCallback, editStructure, rollbackSupport);
 208         }
 209     }
 210
 211     /**
 212      * This has to be non blocking since it is called from a callback on commit and its netty threadpool that is really sensitive to blocking calls
 213      */
 214     private void unlock() {
 215         netOps.unlockCandidate(new NetconfRpcFutureCallback("Unlock candidate", id));
 216     }
 217
 218 }