Merge "Fixed bug in cross-broker RPC routing"
[controller.git] / 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.Optional;
12 import com.google.common.base.Preconditions;
13 import io.netty.channel.EventLoopGroup;
14 import org.opendaylight.controller.config.api.ConflictingVersionException;
15 import org.opendaylight.controller.config.persist.api.ConfigSnapshotHolder;
16 import org.opendaylight.controller.netconf.api.NetconfMessage;
17 import org.opendaylight.controller.netconf.client.NetconfClient;
18 import org.opendaylight.controller.netconf.client.NetconfClientDispatcher;
19 import org.opendaylight.controller.netconf.util.xml.XmlElement;
20 import org.opendaylight.controller.netconf.util.xml.XmlNetconfConstants;
21 import org.opendaylight.controller.netconf.util.xml.XmlUtil;
22 import org.slf4j.Logger;
23 import org.slf4j.LoggerFactory;
24 import org.w3c.dom.Document;
25 import org.w3c.dom.Element;
26 import org.xml.sax.SAXException;
27
28 import javax.annotation.concurrent.Immutable;
29 import java.io.IOException;
30 import java.io.InputStream;
31 import java.net.InetSocketAddress;
32 import java.util.Collections;
33 import java.util.HashSet;
34 import java.util.List;
35 import java.util.Set;
36
37 @Immutable
38 public class ConfigPusher {
39     private static final Logger logger = LoggerFactory.getLogger(ConfigPersisterNotificationHandler.class);
40     private static final int NETCONF_SEND_ATTEMPT_MS_DELAY = 1000;
41     private static final int NETCONF_SEND_ATTEMPTS = 20;
42
43     private final InetSocketAddress address;
44     private final EventLoopGroup nettyThreadgroup;
45
46
47     public static final long DEFAULT_TIMEOUT = 120000L;// 120 seconds until netconf must be stable
48     private final long timeout;
49
50     public ConfigPusher(InetSocketAddress address, EventLoopGroup nettyThreadgroup) {
51         this(address, DEFAULT_TIMEOUT, nettyThreadgroup);
52
53     }
54
55     public ConfigPusher(InetSocketAddress address, long timeout, EventLoopGroup nettyThreadgroup) {
56         this.address = address;
57         this.timeout = timeout;
58
59         this.nettyThreadgroup = nettyThreadgroup;
60     }
61
62     public synchronized NetconfClient init(List<ConfigSnapshotHolder> configs) throws InterruptedException {
63         logger.debug("Last config snapshots to be pushed to netconf: {}", configs);
64         return pushAllConfigs(configs);
65     }
66
67     private synchronized NetconfClient pushAllConfigs(List<ConfigSnapshotHolder> configs) throws InterruptedException {
68         NetconfClient netconfClient = makeNetconfConnection(Collections.<String>emptySet(), Optional.<NetconfClient>absent());
69         for (ConfigSnapshotHolder configSnapshotHolder: configs){
70             netconfClient = pushSnapshotWithRetries(configSnapshotHolder, Optional.of(netconfClient));
71         }
72         return netconfClient;
73     }
74
75     private synchronized NetconfClient pushSnapshotWithRetries(ConfigSnapshotHolder configSnapshotHolder,
76                                                                Optional<NetconfClient> oldClientForPossibleReuse)
77             throws InterruptedException {
78
79         ConflictingVersionException lastException = null;
80         int maxAttempts = 30;
81         for(int i = 0 ; i < maxAttempts; i++) {
82             NetconfClient netconfClient = makeNetconfConnection(configSnapshotHolder.getCapabilities(), oldClientForPossibleReuse);
83             final String configSnapshot = configSnapshotHolder.getConfigSnapshot();
84             logger.trace("Pushing following xml to netconf {}", configSnapshot);
85             try {
86                 pushLastConfig(configSnapshotHolder, netconfClient);
87                 return netconfClient;
88             } catch(ConflictingVersionException e) {
89                 Util.closeClientAndDispatcher(netconfClient);
90                 lastException = e;
91                 Thread.sleep(1000);
92             } catch (SAXException | IOException e) {
93                 throw new IllegalStateException("Unable to load last config", e);
94             }
95         }
96         throw new IllegalStateException("Failed to push configuration, maximum attempt count has been reached: "
97                 + maxAttempts, lastException);
98     }
99
100     /**
101      * @param expectedCaps capabilities that server hello must contain. Will retry until all are found or throws RuntimeException.
102      *                     If empty set is provided, will only make sure netconf client successfuly connected to the server.
103      * @param oldClientForPossibleReuse if present, try to get expected capabilities from it before closing it and retrying with
104      *                                  new client connection.
105      * @return NetconfClient that has all required capabilities from server.
106      */
107     private synchronized NetconfClient makeNetconfConnection(Set<String> expectedCaps,
108                                                              Optional<NetconfClient> oldClientForPossibleReuse)
109             throws InterruptedException {
110
111         if (oldClientForPossibleReuse.isPresent()) {
112             NetconfClient oldClient = oldClientForPossibleReuse.get();
113             if (Util.isSubset(oldClient, expectedCaps)) {
114                 return oldClient;
115             } else {
116                 Util.closeClientAndDispatcher(oldClient);
117             }
118         }
119
120         // TODO think about moving capability subset check to netconf client
121         // could be utilized by integration tests
122
123         long pollingStart = System.currentTimeMillis();
124         int delay = 5000;
125
126         int attempt = 0;
127
128         long deadline = pollingStart + timeout;
129
130         Set<String> latestCapabilities = new HashSet<>();
131         while (System.currentTimeMillis() < deadline) {
132             attempt++;
133             NetconfClientDispatcher netconfClientDispatcher = new NetconfClientDispatcher(nettyThreadgroup, nettyThreadgroup);
134             NetconfClient netconfClient;
135             try {
136                 netconfClient = new NetconfClient(this.toString(), address, delay, netconfClientDispatcher);
137             } catch (IllegalStateException e) {
138                 logger.debug("Netconf {} was not initialized or is not stable, attempt {}", address, attempt, e);
139                 netconfClientDispatcher.close();
140                 Thread.sleep(delay);
141                 continue;
142             }
143             latestCapabilities = netconfClient.getCapabilities();
144             if (Util.isSubset(netconfClient, expectedCaps)) {
145                 logger.debug("Hello from netconf stable with {} capabilities", latestCapabilities);
146                 logger.info("Session id received from netconf server: {}", netconfClient.getClientSession());
147                 return netconfClient;
148             }
149             logger.debug("Polling hello from netconf, attempt {}, capabilities {}", attempt, latestCapabilities);
150             Util.closeClientAndDispatcher(netconfClient);
151             Thread.sleep(delay);
152         }
153         Set<String> allNotFound = new HashSet<>(expectedCaps);
154         allNotFound.removeAll(latestCapabilities);
155         logger.error("Netconf server did not provide required capabilities. Expected but not found: {}, all expected {}, current {}",
156                 allNotFound, expectedCaps, latestCapabilities);
157         throw new RuntimeException("Netconf server did not provide required capabilities. Expected but not found:" + allNotFound);
158     }
159
160
161     private synchronized void pushLastConfig(ConfigSnapshotHolder configSnapshotHolder, NetconfClient netconfClient)
162             throws ConflictingVersionException, IOException, SAXException {
163
164         Element xmlToBePersisted = XmlUtil.readXmlToElement(configSnapshotHolder.getConfigSnapshot());
165         logger.info("Pushing last configuration to netconf: {}", configSnapshotHolder);
166         StringBuilder response = new StringBuilder("editConfig response = {");
167
168
169         NetconfMessage message = createEditConfigMessage(xmlToBePersisted, "/netconfOp/editConfig.xml");
170
171         // sending message to netconf
172         NetconfMessage responseMessage = getResponse(message, netconfClient);
173
174         XmlElement element = XmlElement.fromDomDocument(responseMessage.getDocument());
175         Preconditions.checkState(element.getName().equals(XmlNetconfConstants.RPC_REPLY_KEY));
176         element = element.getOnlyChildElement();
177
178         Util.checkIsOk(element, responseMessage);
179         response.append(XmlUtil.toString(responseMessage.getDocument()));
180         response.append("}");
181         responseMessage = getResponse(getNetconfMessageFromResource("/netconfOp/commit.xml"), netconfClient);
182
183         element = XmlElement.fromDomDocument(responseMessage.getDocument());
184         Preconditions.checkState(element.getName().equals(XmlNetconfConstants.RPC_REPLY_KEY));
185         element = element.getOnlyChildElement();
186
187         Util.checkIsOk(element, responseMessage);
188         response.append("commit response = {");
189         response.append(XmlUtil.toString(responseMessage.getDocument()));
190         response.append("}");
191         logger.info("Last configuration loaded successfully");
192         logger.trace("Detailed message {}", response);
193     }
194
195     private static NetconfMessage getResponse(NetconfMessage request, NetconfClient netconfClient) {
196         try {
197             return netconfClient.sendMessage(request, NETCONF_SEND_ATTEMPTS, NETCONF_SEND_ATTEMPT_MS_DELAY);
198         } catch(RuntimeException e) {
199             logger.error("Error while sending message {} to {}", request, netconfClient);
200             throw e;
201         }
202     }
203
204     private static NetconfMessage createEditConfigMessage(Element dataElement, String editConfigResourcename) {
205         try (InputStream stream = ConfigPersisterNotificationHandler.class.getResourceAsStream(editConfigResourcename)) {
206             Preconditions.checkNotNull(stream, "Unable to load resource " + editConfigResourcename);
207
208             Document doc = XmlUtil.readXmlToDocument(stream);
209
210             doc.getDocumentElement();
211             XmlElement editConfigElement = XmlElement.fromDomDocument(doc).getOnlyChildElement();
212             XmlElement configWrapper = editConfigElement.getOnlyChildElement(XmlNetconfConstants.CONFIG_KEY);
213             editConfigElement.getDomElement().removeChild(configWrapper.getDomElement());
214             for (XmlElement el : XmlElement.fromDomElement(dataElement).getChildElements()) {
215                 configWrapper.appendChild((Element) doc.importNode(el.getDomElement(), true));
216             }
217             editConfigElement.appendChild(configWrapper.getDomElement());
218             return new NetconfMessage(doc);
219         } catch (IOException | SAXException e) {
220             throw new RuntimeException("Unable to parse message from resources " + editConfigResourcename, e);
221         }
222     }
223
224     private static NetconfMessage getNetconfMessageFromResource(String resource) {
225         try (InputStream stream = ConfigPusher.class.getResourceAsStream(resource)) {
226             Preconditions.checkNotNull(stream, "Unable to load resource " + resource);
227             return new NetconfMessage(XmlUtil.readXmlToDocument(stream));
228         } catch (SAXException | IOException e) {
229             throw new RuntimeException("Unable to parse message from resources " + resource, e);
230         }
231     }
232 }