netconf/sal-netconf-connector/src/main/java/org/opendaylight/netconf/sal/connect/netconf/sal/AbstractNetconfDataTreeService.java

   1 /*
   2  * Copyright (c) 2020 PANTHEON.tech, s.r.o. 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 package org.opendaylight.netconf.sal.connect.netconf.sal;
   9
  10 import static com.google.common.base.Preconditions.checkArgument;
  11 import static java.util.Objects.requireNonNull;
  12
  13 import com.google.common.util.concurrent.FutureCallback;
  14 import com.google.common.util.concurrent.Futures;
  15 import com.google.common.util.concurrent.ListenableFuture;
  16 import com.google.common.util.concurrent.MoreExecutors;
  17 import com.google.common.util.concurrent.SettableFuture;
  18 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
  19 import java.util.Collection;
  20 import java.util.List;
  21 import java.util.Optional;
  22 import java.util.StringJoiner;
  23 import org.eclipse.jdt.annotation.NonNull;
  24 import org.eclipse.jdt.annotation.Nullable;
  25 import org.opendaylight.mdsal.common.api.CommitInfo;
  26 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  27 import org.opendaylight.mdsal.common.api.TransactionCommitFailedException;
  28 import org.opendaylight.mdsal.dom.api.DOMRpcResult;
  29 import org.opendaylight.mdsal.dom.api.DOMRpcService;
  30 import org.opendaylight.netconf.api.DocumentedException;
  31 import org.opendaylight.netconf.api.ModifyAction;
  32 import org.opendaylight.netconf.api.NetconfDocumentedException;
  33 import org.opendaylight.netconf.dom.api.NetconfDataTreeService;
  34 import org.opendaylight.netconf.sal.connect.netconf.listener.NetconfSessionPreferences;
  35 import org.opendaylight.netconf.sal.connect.netconf.util.NetconfBaseOps;
  36 import org.opendaylight.netconf.sal.connect.netconf.util.NetconfRpcFutureCallback;
  37 import org.opendaylight.netconf.sal.connect.util.RemoteDeviceId;
  38 import org.opendaylight.yangtools.rfc8528.data.api.MountPointContext;
  39 import org.opendaylight.yangtools.yang.common.RpcError;
  40 import org.opendaylight.yangtools.yang.common.RpcResult;
  41 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
  42 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  43 import org.opendaylight.yangtools.yang.data.api.schema.DataContainerChild;
  44 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  45 import org.slf4j.Logger;
  46 import org.slf4j.LoggerFactory;
  47
  48 public abstract class AbstractNetconfDataTreeService implements NetconfDataTreeService {
  49     private static final class Candidate extends AbstractNetconfDataTreeService {
  50         Candidate(final RemoteDeviceId id, final NetconfBaseOps netconfOps, final boolean rollbackSupport) {
  51             super(id, netconfOps, rollbackSupport);
  52         }
  53
  54         /**
  55          * This has to be non blocking since it is called from a callback on commit and it is netty threadpool that is
  56          * really sensitive to blocking calls.
  57          */
  58         @Override
  59         public void discardChanges() {
  60             netconfOps.discardChanges(new NetconfRpcFutureCallback("Discarding candidate", id));
  61         }
  62
  63         @Override
  64         ListenableFuture<? extends DOMRpcResult> lockSingle() {
  65             return netconfOps.lockCandidate(new NetconfRpcFutureCallback("Lock candidate", id) {
  66                 @Override
  67                 public void onFailure(final Throwable throwable) {
  68                     super.onFailure(throwable);
  69                     discardChanges();
  70                 }
  71             });
  72         }
  73
  74         @Override
  75         void unlockImpl() {
  76             netconfOps.unlockCandidate(new NetconfRpcFutureCallback("Unlock candidate", id));
  77         }
  78
  79         @Override
  80         ListenableFuture<? extends DOMRpcResult> editConfig(final DataContainerChild<?, ?> editStructure,
  81                 final ModifyAction defaultOperation) {
  82             final NetconfRpcFutureCallback callback = new NetconfRpcFutureCallback("Edit candidate", id);
  83             return defaultOperation == null ? netconfOps.editConfigCandidate(callback, editStructure, rollbackSupport)
  84                 : netconfOps.editConfigCandidate(callback, editStructure, defaultOperation, rollbackSupport);
  85         }
  86
  87         @Override
  88         ListenableFuture<RpcResult<Void>> commitImpl(final List<ListenableFuture<? extends DOMRpcResult>> results) {
  89             results.add(netconfOps.commit(new NetconfRpcFutureCallback("Commit", id)));
  90             final ListenableFuture<RpcResult<Void>> result = resultsToStatus(id, results);
  91             Futures.addCallback(result, new FutureCallback<>() {
  92                 @Override
  93                 public void onSuccess(final RpcResult<Void> result) {
  94                     // do nothing, as callback is only used to catch failures
  95                 }
  96
  97                 @Override
  98                 public void onFailure(final Throwable throwable) {
  99                     discardChanges();
 100                 }
 101             }, MoreExecutors.directExecutor());
 102             return result;
 103         }
 104     }
 105
 106     private static final class Running extends AbstractNetconfDataTreeService {
 107         Running(final RemoteDeviceId id, final NetconfBaseOps netconfOps, final boolean rollbackSupport) {
 108             super(id, netconfOps, rollbackSupport);
 109         }
 110
 111         @Override
 112         public void discardChanges() {
 113             // Changes cannot be discarded from running
 114         }
 115
 116         @Override
 117         ListenableFuture<? extends DOMRpcResult> lockSingle() {
 118             return netconfOps.lockRunning(new NetconfRpcFutureCallback("Lock running", id));
 119         }
 120
 121         @Override
 122         void unlockImpl() {
 123             netconfOps.unlockRunning(new NetconfRpcFutureCallback("Unlock running", id));
 124         }
 125
 126         @Override
 127         ListenableFuture<? extends DOMRpcResult> editConfig(final DataContainerChild<?, ?> editStructure,
 128                 final ModifyAction defaultOperation) {
 129             final NetconfRpcFutureCallback callback = new NetconfRpcFutureCallback("Edit running", id);
 130             return defaultOperation == null ? netconfOps.editConfigRunning(callback, editStructure, rollbackSupport)
 131                 : netconfOps.editConfigRunning(callback, editStructure, defaultOperation, rollbackSupport);
 132         }
 133
 134         @Override
 135         ListenableFuture<RpcResult<Void>> commitImpl(final List<ListenableFuture<? extends DOMRpcResult>> results) {
 136             unlock();
 137             return resultsToStatus(id, results);
 138         }
 139     }
 140
 141     private static final class CandidateWithRunning extends AbstractNetconfDataTreeService {
 142         private final Candidate candidate;
 143         private final Running running;
 144
 145         CandidateWithRunning(final RemoteDeviceId id, final NetconfBaseOps netconfOps,
 146                 final boolean rollbackSupport) {
 147             super(id, netconfOps, rollbackSupport);
 148             candidate = new Candidate(id, netconfOps, rollbackSupport);
 149             running = new Running(id, netconfOps, rollbackSupport);
 150         }
 151
 152         @Override
 153         public void discardChanges() {
 154             candidate.discardChanges();
 155         }
 156
 157         @Override
 158         ListenableFuture<? extends DOMRpcResult> lockSingle() {
 159             throw new UnsupportedOperationException();
 160         }
 161
 162         @Override
 163         List<ListenableFuture<? extends DOMRpcResult>> lockImpl() {
 164             return List.of(candidate.lockSingle(), running.lockSingle());
 165         }
 166
 167         @Override
 168         void unlockImpl() {
 169             running.unlock();
 170             candidate.unlock();
 171         }
 172
 173         @Override
 174         ListenableFuture<? extends DOMRpcResult> editConfig(final DataContainerChild<?, ?> editStructure,
 175                 final ModifyAction defaultOperation) {
 176             return candidate.editConfig(editStructure, defaultOperation);
 177         }
 178
 179         @Override
 180         ListenableFuture<RpcResult<Void>> commitImpl(final List<ListenableFuture<? extends DOMRpcResult>> results) {
 181             return candidate.commitImpl(results);
 182         }
 183     }
 184
 185     private static final Logger LOG = LoggerFactory.getLogger(AbstractNetconfDataTreeService.class);
 186
 187     final @NonNull RemoteDeviceId id;
 188     final NetconfBaseOps netconfOps;
 189     final boolean rollbackSupport;
 190
 191     // FIXME: what do we do with locks acquired before this got flipped?
 192     private volatile boolean isLockAllowed = true;
 193
 194     AbstractNetconfDataTreeService(final RemoteDeviceId id, final NetconfBaseOps netconfOps,
 195             final boolean rollbackSupport) {
 196         this.id = requireNonNull(id);
 197         this.netconfOps = requireNonNull(netconfOps);
 198         this.rollbackSupport = rollbackSupport;
 199     }
 200
 201     public static @NonNull AbstractNetconfDataTreeService of(final RemoteDeviceId id,
 202             final MountPointContext mountContext, final DOMRpcService rpc,
 203             final NetconfSessionPreferences netconfSessionPreferences) {
 204         final NetconfBaseOps netconfOps = new NetconfBaseOps(rpc, mountContext);
 205         final boolean rollbackSupport = netconfSessionPreferences.isRollbackSupported();
 206
 207         // Examine preferences and decide which implementation to use
 208         if (netconfSessionPreferences.isCandidateSupported()) {
 209             return netconfSessionPreferences.isRunningWritable()
 210                 ? new CandidateWithRunning(id, netconfOps, rollbackSupport)
 211                     : new Candidate(id, netconfOps, rollbackSupport);
 212         } else if (netconfSessionPreferences.isRunningWritable()) {
 213             return new Running(id, netconfOps, rollbackSupport);
 214         } else {
 215             throw new IllegalArgumentException("Device " + id.getName() + " has advertised neither :writable-running "
 216                 + "nor :candidate capability. Failed to establish session, as at least one of these must be "
 217                 + "advertised.");
 218         }
 219     }
 220
 221     @Override
 222     public synchronized List<ListenableFuture<? extends DOMRpcResult>> lock() {
 223         if (isLockAllowed) {
 224             return lockImpl();
 225         }
 226         LOG.trace("Lock is not allowed: {}", id);
 227         return List.of();
 228     }
 229
 230     List<ListenableFuture<? extends DOMRpcResult>> lockImpl() {
 231         return List.of(lockSingle());
 232     }
 233
 234     abstract ListenableFuture<? extends DOMRpcResult> lockSingle();
 235
 236     @Override
 237     // FIXME: this should be asynchronous as well
 238     public synchronized void unlock() {
 239         // FIXME: deal with lock with lifecycle?
 240         if (isLockAllowed) {
 241             unlockImpl();
 242         } else {
 243             LOG.trace("Unlock is not allowed: {}", id);
 244         }
 245     }
 246
 247     abstract void unlockImpl();
 248
 249     @Override
 250     public ListenableFuture<Optional<NormalizedNode<?, ?>>> get(final YangInstanceIdentifier path) {
 251         return netconfOps.getData(new NetconfRpcFutureCallback("Data read", id), Optional.ofNullable(path));
 252     }
 253
 254     @Override
 255     public ListenableFuture<Optional<NormalizedNode<?, ?>>> get(final YangInstanceIdentifier path,
 256             final List<YangInstanceIdentifier> fields) {
 257         return netconfOps.getData(new NetconfRpcFutureCallback("Data read", id), Optional.ofNullable(path), fields);
 258     }
 259
 260     @Override
 261     public ListenableFuture<Optional<NormalizedNode<?, ?>>> getConfig(final YangInstanceIdentifier path) {
 262         return netconfOps.getConfigRunningData(new NetconfRpcFutureCallback("Data read", id),
 263             Optional.ofNullable(path));
 264     }
 265
 266     @Override
 267     public ListenableFuture<Optional<NormalizedNode<?, ?>>> getConfig(final YangInstanceIdentifier path,
 268             final List<YangInstanceIdentifier> fields) {
 269         return netconfOps.getConfigRunningData(new NetconfRpcFutureCallback("Data read", id),
 270             Optional.ofNullable(path), fields);
 271     }
 272
 273     @Override
 274     public synchronized ListenableFuture<? extends DOMRpcResult> merge(final LogicalDatastoreType store,
 275             final YangInstanceIdentifier path, final NormalizedNode<?, ?> data,
 276             final Optional<ModifyAction> defaultOperation) {
 277         checkEditable(store);
 278         return editConfig(
 279             netconfOps.createEditConfigStrcture(Optional.ofNullable(data), Optional.of(ModifyAction.MERGE), path),
 280             defaultOperation.orElse(null));
 281     }
 282
 283     @Override
 284     public synchronized ListenableFuture<? extends DOMRpcResult> replace(final LogicalDatastoreType store,
 285             final YangInstanceIdentifier path, final NormalizedNode<?, ?> data,
 286             final Optional<ModifyAction> defaultOperation) {
 287         checkEditable(store);
 288         return editConfig(
 289             netconfOps.createEditConfigStrcture(Optional.ofNullable(data), Optional.of(ModifyAction.REPLACE), path),
 290             defaultOperation.orElse(null));
 291     }
 292
 293     @Override
 294     public synchronized ListenableFuture<? extends DOMRpcResult> create(final LogicalDatastoreType store,
 295             final YangInstanceIdentifier path, final NormalizedNode<?, ?> data,
 296             final Optional<ModifyAction> defaultOperation) {
 297         checkEditable(store);
 298         return editConfig(
 299             netconfOps.createEditConfigStrcture(Optional.ofNullable(data), Optional.of(ModifyAction.CREATE), path),
 300             defaultOperation.orElse(null));
 301     }
 302
 303     @Override
 304     public synchronized ListenableFuture<? extends DOMRpcResult> delete(final LogicalDatastoreType store,
 305             final YangInstanceIdentifier path) {
 306         return editConfig(netconfOps.createEditConfigStrcture(Optional.empty(), Optional.of(ModifyAction.DELETE), path),
 307             null);
 308     }
 309
 310     @Override
 311     public synchronized ListenableFuture<? extends DOMRpcResult> remove(final LogicalDatastoreType store,
 312             final YangInstanceIdentifier path) {
 313         return editConfig(netconfOps.createEditConfigStrcture(Optional.empty(), Optional.of(ModifyAction.REMOVE), path),
 314             null);
 315     }
 316
 317     @Override
 318     public synchronized ListenableFuture<? extends CommitInfo> commit(
 319             final List<ListenableFuture<? extends DOMRpcResult>> resultsFutures) {
 320         final SettableFuture<CommitInfo> resultFuture = SettableFuture.create();
 321         Futures.addCallback(commitImpl(resultsFutures), new FutureCallback<>() {
 322             @Override
 323             public void onSuccess(final RpcResult<Void> result) {
 324                 if (!result.isSuccessful()) {
 325                     final Collection<RpcError> errors = result.getErrors();
 326                     resultFuture.setException(new TransactionCommitFailedException(
 327                             String.format("Commit of transaction %s failed", this),
 328                             errors.toArray(new RpcError[errors.size()])));
 329                     return;
 330                 }
 331                 unlock();
 332                 resultFuture.set(CommitInfo.empty());
 333             }
 334
 335             @Override
 336             public void onFailure(final Throwable failure) {
 337                 unlock();
 338                 resultFuture.setException(new TransactionCommitFailedException(
 339                         String.format("Commit of transaction %s failed", this), failure));
 340             }
 341         }, MoreExecutors.directExecutor());
 342         return resultFuture;
 343     }
 344
 345     abstract ListenableFuture<RpcResult<Void>> commitImpl(List<ListenableFuture<? extends DOMRpcResult>> results);
 346
 347     @Override
 348     public final Object getDeviceId() {
 349         return id;
 350     }
 351
 352     final void setLockAllowed(final boolean isLockAllowedOrig) {
 353         this.isLockAllowed = isLockAllowedOrig;
 354     }
 355
 356     abstract ListenableFuture<? extends DOMRpcResult> editConfig(DataContainerChild<?, ?> editStructure,
 357         @Nullable ModifyAction defaultOperation);
 358
 359     private static void checkEditable(final LogicalDatastoreType store) {
 360         checkArgument(store == LogicalDatastoreType.CONFIGURATION, "Can only edit configuration data, not %s", store);
 361     }
 362
 363     @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD",
 364         justification = "https://github.com/spotbugs/spotbugs/issues/811")
 365     private static ListenableFuture<RpcResult<Void>> resultsToStatus(
 366             final RemoteDeviceId id, final List<ListenableFuture<? extends DOMRpcResult>> resultsFutures) {
 367         final SettableFuture<RpcResult<Void>> transformed = SettableFuture.create();
 368
 369         Futures.addCallback(Futures.allAsList(resultsFutures), new FutureCallback<>() {
 370             @Override
 371             public void onSuccess(final List<DOMRpcResult> domRpcResults) {
 372                 if (!transformed.isDone()) {
 373                     extractResult(domRpcResults, transformed, id);
 374                 }
 375             }
 376
 377             @Override
 378             public void onFailure(final Throwable throwable) {
 379                 final NetconfDocumentedException exception =
 380                         new NetconfDocumentedException(
 381                                 id + ":RPC during tx returned an exception" + throwable.getMessage(),
 382                                 new Exception(throwable),
 383                                 DocumentedException.ErrorType.APPLICATION,
 384                                 DocumentedException.ErrorTag.OPERATION_FAILED,
 385                                 DocumentedException.ErrorSeverity.ERROR);
 386                 transformed.setException(exception);
 387             }
 388         }, MoreExecutors.directExecutor());
 389
 390         return transformed;
 391     }
 392
 393     @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD",
 394             justification = "https://github.com/spotbugs/spotbugs/issues/811")
 395     private static void extractResult(final List<DOMRpcResult> domRpcResults,
 396                                       final SettableFuture<RpcResult<Void>> transformed,
 397                                       final RemoteDeviceId id) {
 398         DocumentedException.ErrorType errType = DocumentedException.ErrorType.APPLICATION;
 399         DocumentedException.ErrorSeverity errSeverity = DocumentedException.ErrorSeverity.ERROR;
 400         StringJoiner msgBuilder = new StringJoiner(" ");
 401         boolean errorsEncouneterd = false;
 402         String errorTag = "operation-failed";
 403
 404         for (final DOMRpcResult domRpcResult : domRpcResults) {
 405             if (!domRpcResult.getErrors().isEmpty()) {
 406                 errorsEncouneterd = true;
 407                 final RpcError error = domRpcResult.getErrors().iterator().next();
 408                 final RpcError.ErrorType errorType = error.getErrorType();
 409                 switch (errorType) {
 410                     case RPC:
 411                         errType = DocumentedException.ErrorType.RPC;
 412                         break;
 413                     case PROTOCOL:
 414                         errType = DocumentedException.ErrorType.PROTOCOL;
 415                         break;
 416                     case TRANSPORT:
 417                         errType = DocumentedException.ErrorType.TRANSPORT;
 418                         break;
 419                     case APPLICATION:
 420                     default:
 421                         errType = DocumentedException.ErrorType.APPLICATION;
 422                         break;
 423                 }
 424                 final RpcError.ErrorSeverity severity = error.getSeverity();
 425                 switch (severity) {
 426                     case WARNING:
 427                         errSeverity = DocumentedException.ErrorSeverity.WARNING;
 428                         break;
 429                     case ERROR:
 430                     default:
 431                         errSeverity = DocumentedException.ErrorSeverity.ERROR;
 432                         break;
 433                 }
 434                 msgBuilder.add(error.getMessage());
 435                 msgBuilder.add(error.getInfo());
 436                 errorTag = error.getTag();
 437             }
 438         }
 439         if (errorsEncouneterd) {
 440             final NetconfDocumentedException exception = new NetconfDocumentedException(id
 441                     + ":RPC during tx failed. " + msgBuilder.toString(),
 442                     errType,
 443                     DocumentedException.ErrorTag.from(errorTag),
 444                     errSeverity);
 445             transformed.setException(exception);
 446             return;
 447         }
 448         transformed.set(RpcResultBuilder.<Void>success().build());
 449     }
 450 }