opendaylight/netconf/config-persister-impl/src/main/java/org/opendaylight/controller/netconf/persist/impl/ConfigPusher.java

   1 /*
   2  * Copyright (c) 2013 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.netconf.persist.impl;
  10
  11 import com.google.common.base.Function;
  12 import com.google.common.base.Preconditions;
  13 import com.google.common.base.Stopwatch;
  14 import com.google.common.collect.Collections2;
  15 import org.opendaylight.controller.config.api.ConflictingVersionException;
  16 import org.opendaylight.controller.config.persist.api.ConfigSnapshotHolder;
  17 import org.opendaylight.controller.netconf.api.NetconfDocumentedException;
  18 import org.opendaylight.controller.netconf.api.NetconfMessage;
  19 import org.opendaylight.controller.netconf.mapping.api.Capability;
  20 import org.opendaylight.controller.netconf.mapping.api.HandlingPriority;
  21 import org.opendaylight.controller.netconf.mapping.api.NetconfOperation;
  22 import org.opendaylight.controller.netconf.mapping.api.NetconfOperationChainedExecution;
  23 import org.opendaylight.controller.netconf.mapping.api.NetconfOperationService;
  24 import org.opendaylight.controller.netconf.mapping.api.NetconfOperationServiceFactory;
  25 import org.opendaylight.controller.netconf.util.NetconfUtil;
  26 import org.opendaylight.controller.netconf.util.xml.XmlElement;
  27 import org.opendaylight.controller.netconf.util.xml.XmlNetconfConstants;
  28 import org.opendaylight.controller.netconf.util.xml.XmlUtil;
  29 import org.slf4j.Logger;
  30 import org.slf4j.LoggerFactory;
  31 import org.w3c.dom.Document;
  32 import org.w3c.dom.Element;
  33 import org.xml.sax.SAXException;
  34
  35 import javax.annotation.concurrent.Immutable;
  36 import java.io.IOException;
  37 import java.io.InputStream;
  38 import java.util.Collection;
  39 import java.util.HashSet;
  40 import java.util.LinkedHashMap;
  41 import java.util.List;
  42 import java.util.Map.Entry;
  43 import java.util.Set;
  44 import java.util.TreeMap;
  45 import java.util.concurrent.TimeUnit;
  46
  47 @Immutable
  48 public class ConfigPusher {
  49     private static final Logger logger = LoggerFactory.getLogger(ConfigPusher.class);
  50
  51     private final long maxWaitForCapabilitiesMillis;
  52     private final long conflictingVersionTimeoutMillis;
  53     private final NetconfOperationServiceFactory configNetconfConnector;
  54
  55     public ConfigPusher(NetconfOperationServiceFactory configNetconfConnector, long maxWaitForCapabilitiesMillis,
  56                         long conflictingVersionTimeoutMillis) {
  57         this.configNetconfConnector = configNetconfConnector;
  58         this.maxWaitForCapabilitiesMillis = maxWaitForCapabilitiesMillis;
  59         this.conflictingVersionTimeoutMillis = conflictingVersionTimeoutMillis;
  60     }
  61
  62     public synchronized LinkedHashMap<ConfigSnapshotHolder, EditAndCommitResponse> pushConfigs(List<ConfigSnapshotHolder> configs) throws NetconfDocumentedException {
  63         logger.debug("Last config snapshots to be pushed to netconf: {}", configs);
  64         LinkedHashMap<ConfigSnapshotHolder, EditAndCommitResponse> result = new LinkedHashMap<>();
  65         // start pushing snapshots:
  66         for (ConfigSnapshotHolder configSnapshotHolder : configs) {
  67             EditAndCommitResponse editAndCommitResponseWithRetries = pushConfigWithConflictingVersionRetries(configSnapshotHolder);
  68             logger.debug("Config snapshot pushed successfully: {}, result: {}", configSnapshotHolder, result);
  69             result.put(configSnapshotHolder, editAndCommitResponseWithRetries);
  70         }
  71         logger.debug("All configuration snapshots have been pushed successfully.");
  72         return result;
  73     }
  74
  75     /**
  76      * First calls {@link #getOperationServiceWithRetries(java.util.Set, String)} in order to wait until
  77      * expected capabilities are present, then tries to push configuration. If {@link ConflictingVersionException}
  78      * is caught, whole process is retried - new service instance need to be obtained from the factory. Closes
  79      * {@link NetconfOperationService} after each use.
  80      */
  81     private synchronized EditAndCommitResponse pushConfigWithConflictingVersionRetries(ConfigSnapshotHolder configSnapshotHolder) throws NetconfDocumentedException {
  82         ConflictingVersionException lastException;
  83         Stopwatch stopwatch = new Stopwatch().start();
  84         do {
  85             try (NetconfOperationService operationService = getOperationServiceWithRetries(configSnapshotHolder.getCapabilities(), configSnapshotHolder.toString())) {
  86                 return pushConfig(configSnapshotHolder, operationService);
  87             } catch (ConflictingVersionException e) {
  88                 lastException = e;
  89                 logger.debug("Conflicting version detected, will retry after timeout");
  90                 sleep();
  91             }
  92         } while (stopwatch.elapsed(TimeUnit.MILLISECONDS) < conflictingVersionTimeoutMillis);
  93         throw new IllegalStateException("Max wait for conflicting version stabilization timeout after " + stopwatch.elapsed(TimeUnit.MILLISECONDS) + " ms",
  94                 lastException);
  95     }
  96
  97     private NetconfOperationService getOperationServiceWithRetries(Set<String> expectedCapabilities, String idForReporting) {
  98         Stopwatch stopwatch = new Stopwatch().start();
  99         NotEnoughCapabilitiesException lastException;
 100         do {
 101             try {
 102                 return getOperationService(expectedCapabilities, idForReporting);
 103             } catch (NotEnoughCapabilitiesException e) {
 104                 logger.debug("Not enough capabilities: " + e.toString());
 105                 lastException = e;
 106                 sleep();
 107             }
 108         } while (stopwatch.elapsed(TimeUnit.MILLISECONDS) < maxWaitForCapabilitiesMillis);
 109         throw new IllegalStateException("Max wait for capabilities reached." + lastException.getMessage(), lastException);
 110     }
 111
 112     private static class NotEnoughCapabilitiesException extends Exception {
 113         private NotEnoughCapabilitiesException(String message, Throwable cause) {
 114             super(message, cause);
 115         }
 116
 117         private NotEnoughCapabilitiesException(String message) {
 118             super(message);
 119         }
 120     }
 121
 122     /**
 123      * Get NetconfOperationService iif all required capabilities are present.
 124      *
 125      * @param expectedCapabilities that must be provided by configNetconfConnector
 126      * @param idForReporting
 127      * @return service if capabilities are present, otherwise absent value
 128      */
 129     private NetconfOperationService getOperationService(Set<String> expectedCapabilities, String idForReporting) throws NotEnoughCapabilitiesException {
 130         NetconfOperationService serviceCandidate;
 131         try {
 132             serviceCandidate = configNetconfConnector.createService(idForReporting);
 133         } catch(RuntimeException e) {
 134             throw new NotEnoughCapabilitiesException("Netconf service not stable for " + idForReporting, e);
 135         }
 136         Set<String> notFoundDiff = computeNotFoundCapabilities(expectedCapabilities, serviceCandidate);
 137         if (notFoundDiff.isEmpty()) {
 138             return serviceCandidate;
 139         } else {
 140             serviceCandidate.close();
 141             logger.trace("Netconf server did not provide required capabilities for {} " +
 142                     "Expected but not found: {}, all expected {}, current {}",
 143                     idForReporting, notFoundDiff, expectedCapabilities, serviceCandidate.getCapabilities()
 144             );
 145             throw new NotEnoughCapabilitiesException("Not enough capabilities for " + idForReporting + ". Expected but not found: " + notFoundDiff);
 146         }
 147     }
 148
 149     private static Set<String> computeNotFoundCapabilities(Set<String> expectedCapabilities, NetconfOperationService serviceCandidate) {
 150         Collection<String> actual = Collections2.transform(serviceCandidate.getCapabilities(), new Function<Capability, String>() {
 151             @Override
 152             public String apply(Capability input) {
 153                 return input.getCapabilityUri();
 154             }
 155         });
 156         Set<String> allNotFound = new HashSet<>(expectedCapabilities);
 157         allNotFound.removeAll(actual);
 158         return allNotFound;
 159     }
 160
 161
 162
 163     private void sleep() {
 164         try {
 165             Thread.sleep(100);
 166         } catch (InterruptedException e) {
 167             Thread.currentThread().interrupt();
 168             throw new IllegalStateException(e);
 169         }
 170     }
 171
 172     /**
 173      * Sends two RPCs to the netconf server: edit-config and commit.
 174      *
 175      * @param configSnapshotHolder
 176      * @throws ConflictingVersionException if commit fails on optimistic lock failure inside of config-manager
 177      * @throws java.lang.RuntimeException  if edit-config or commit fails otherwise
 178      */
 179     private synchronized EditAndCommitResponse pushConfig(ConfigSnapshotHolder configSnapshotHolder, NetconfOperationService operationService)
 180             throws ConflictingVersionException, NetconfDocumentedException {
 181
 182         Element xmlToBePersisted;
 183         try {
 184             xmlToBePersisted = XmlUtil.readXmlToElement(configSnapshotHolder.getConfigSnapshot());
 185         } catch (SAXException | IOException e) {
 186             throw new IllegalStateException("Cannot parse " + configSnapshotHolder);
 187         }
 188         logger.trace("Pushing last configuration to netconf: {}", configSnapshotHolder);
 189         Stopwatch stopwatch = new Stopwatch().start();
 190         NetconfMessage editConfigMessage = createEditConfigMessage(xmlToBePersisted);
 191
 192         Document editResponseMessage = sendRequestGetResponseCheckIsOK(editConfigMessage, operationService,
 193                 "edit-config", configSnapshotHolder.toString());
 194
 195         Document commitResponseMessage = sendRequestGetResponseCheckIsOK(getCommitMessage(), operationService,
 196                 "commit", configSnapshotHolder.toString());
 197
 198         if (logger.isTraceEnabled()) {
 199             StringBuilder response = new StringBuilder("editConfig response = {");
 200             response.append(XmlUtil.toString(editResponseMessage));
 201             response.append("}");
 202             response.append("commit response = {");
 203             response.append(XmlUtil.toString(commitResponseMessage));
 204             response.append("}");
 205             logger.trace("Last configuration loaded successfully");
 206             logger.trace("Detailed message {}", response);
 207             logger.trace("Total time spent {} ms", stopwatch.elapsed(TimeUnit.MILLISECONDS));
 208         }
 209         return new EditAndCommitResponse(editResponseMessage, commitResponseMessage);
 210     }
 211
 212     private NetconfOperation findOperation(NetconfMessage request, NetconfOperationService operationService) throws NetconfDocumentedException {
 213         TreeMap<HandlingPriority, NetconfOperation> allOperations = new TreeMap<>();
 214         Set<NetconfOperation> netconfOperations = operationService.getNetconfOperations();
 215         if (netconfOperations.isEmpty()) {
 216             throw new IllegalStateException("Possible code error: no config operations");
 217         }
 218         for (NetconfOperation netconfOperation : netconfOperations) {
 219             HandlingPriority handlingPriority = netconfOperation.canHandle(request.getDocument());
 220             allOperations.put(handlingPriority, netconfOperation);
 221         }
 222         Entry<HandlingPriority, NetconfOperation> highestEntry = allOperations.lastEntry();
 223         if (highestEntry.getKey().isCannotHandle()) {
 224             throw new IllegalStateException("Possible code error: operation with highest priority is CANNOT_HANDLE");
 225         }
 226         return highestEntry.getValue();
 227     }
 228
 229     private Document sendRequestGetResponseCheckIsOK(NetconfMessage request, NetconfOperationService operationService,
 230                                                      String operationNameForReporting, String configIdForReporting)
 231             throws ConflictingVersionException, NetconfDocumentedException {
 232
 233         NetconfOperation operation = findOperation(request, operationService);
 234         Document response;
 235         try {
 236             response = operation.handle(request.getDocument(), NetconfOperationChainedExecution.EXECUTION_TERMINATION_POINT);
 237         } catch (NetconfDocumentedException | RuntimeException e) {
 238             if (e instanceof NetconfDocumentedException && e.getCause() instanceof ConflictingVersionException) {
 239                 throw (ConflictingVersionException) e.getCause();
 240             }
 241             throw new IllegalStateException("Failed to send " + operationNameForReporting +
 242                     " for configuration " + configIdForReporting, e);
 243         }
 244         return NetconfUtil.checkIsMessageOk(response);
 245     }
 246
 247     // load editConfig.xml template, populate /rpc/edit-config/config with parameter
 248     private static NetconfMessage createEditConfigMessage(Element dataElement) throws NetconfDocumentedException {
 249         String editConfigResourcePath = "/netconfOp/editConfig.xml";
 250         try (InputStream stream = ConfigPersisterNotificationHandler.class.getResourceAsStream(editConfigResourcePath)) {
 251             Preconditions.checkNotNull(stream, "Unable to load resource " + editConfigResourcePath);
 252
 253             Document doc = XmlUtil.readXmlToDocument(stream);
 254
 255             XmlElement editConfigElement = XmlElement.fromDomDocument(doc).getOnlyChildElement();
 256             XmlElement configWrapper = editConfigElement.getOnlyChildElement(XmlNetconfConstants.CONFIG_KEY);
 257             editConfigElement.getDomElement().removeChild(configWrapper.getDomElement());
 258             for (XmlElement el : XmlElement.fromDomElement(dataElement).getChildElements()) {
 259                 boolean deep = true;
 260                 configWrapper.appendChild((Element) doc.importNode(el.getDomElement(), deep));
 261             }
 262             editConfigElement.appendChild(configWrapper.getDomElement());
 263             return new NetconfMessage(doc);
 264         } catch (IOException | SAXException e) {
 265             // error reading the xml file bundled into the jar
 266             throw new IllegalStateException("Error while opening local resource " + editConfigResourcePath, e);
 267         }
 268     }
 269
 270     private static NetconfMessage getCommitMessage() {
 271         String resource = "/netconfOp/commit.xml";
 272         try (InputStream stream = ConfigPusher.class.getResourceAsStream(resource)) {
 273             Preconditions.checkNotNull(stream, "Unable to load resource " + resource);
 274             return new NetconfMessage(XmlUtil.readXmlToDocument(stream));
 275         } catch (SAXException | IOException e) {
 276             // error reading the xml file bundled into the jar
 277             throw new IllegalStateException("Error while opening local resource " + resource, e);
 278         }
 279     }
 280
 281     static class EditAndCommitResponse {
 282         private final Document editResponse, commitResponse;
 283
 284         EditAndCommitResponse(Document editResponse, Document commitResponse) {
 285             this.editResponse = editResponse;
 286             this.commitResponse = commitResponse;
 287         }
 288
 289         public Document getEditResponse() {
 290             return editResponse;
 291         }
 292
 293         public Document getCommitResponse() {
 294             return commitResponse;
 295         }
 296
 297         @Override
 298         public String toString() {
 299             return "EditAndCommitResponse{" +
 300                     "editResponse=" + editResponse +
 301                     ", commitResponse=" + commitResponse +
 302                     '}';
 303         }
 304     }
 305 }