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