Line data Source code
1 : /*
2 : * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
3 : * Copyright Contributors to the OpenSDN project
4 : */
5 :
6 : #include <stdint.h>
7 : #include <string.h>
8 : #include <pugixml/pugixml.hpp>
9 : #include <xml/xml_base.h>
10 : #include <xml/xml_pugi.h>
11 :
12 : #include <base/util.h>
13 : #include <base/logging.h>
14 : #include <xmpp/xmpp_channel.h>
15 : #include <ifmap/ifmap_agent_parser.h>
16 :
17 : #include <cmn/agent_cmn.h>
18 : #include <cmn/agent_stats.h>
19 :
20 : #include <controller/controller_init.h>
21 : #include <controller/controller_ifmap.h>
22 : #include <controller/controller_peer.h>
23 : #include <controller/controller_types.h>
24 :
25 :
26 : uint64_t AgentIfMapXmppChannel::seq_number_;
27 :
28 2 : AgentIfMapXmppChannel::AgentIfMapXmppChannel(Agent *agent, XmppChannel *channel,
29 2 : uint8_t cnt) : channel_(channel),
30 2 : xs_idx_(cnt),
31 2 : agent_(agent) {
32 2 : channel_->RegisterReceive(xmps::CONFIG,
33 : boost::bind(&AgentIfMapXmppChannel::ReceiveInternal,
34 : this, _1));
35 2 : config_cleanup_timer_.reset(new ConfigCleanupTimer(agent));
36 2 : end_of_config_timer_.reset(new EndOfConfigTimer(agent, this));
37 2 : }
38 :
39 4 : AgentIfMapXmppChannel::~AgentIfMapXmppChannel() {
40 2 : channel_->UnRegisterWriteReady(xmps::CONFIG);
41 2 : channel_->UnRegisterReceive(xmps::CONFIG);
42 2 : config_cleanup_timer_.reset();
43 2 : end_of_config_timer_.reset();
44 4 : }
45 :
46 38 : uint64_t AgentIfMapXmppChannel::NewSeqNumber() {
47 :
48 38 : seq_number_++;
49 :
50 38 : if (seq_number_ == 0) {
51 0 : return ++seq_number_;
52 : }
53 :
54 38 : CONTROLLER_TRACE(IFMapSeqTrace, GetSeqNumber(), seq_number_, "New Config Seq Num");
55 38 : return seq_number_;
56 : }
57 :
58 57 : ConfigCleanupTimer *AgentIfMapXmppChannel::config_cleanup_timer() {
59 57 : return config_cleanup_timer_.get();
60 : }
61 :
62 57 : EndOfConfigTimer *AgentIfMapXmppChannel::end_of_config_timer() {
63 57 : return end_of_config_timer_.get();
64 : }
65 :
66 0 : bool AgentIfMapXmppChannel::SendUpdate(const std::string &msg) {
67 0 : if (!channel_) return false;
68 0 : return channel_->Send((const uint8_t *)msg.data(), msg.size(), xmps::CONFIG,
69 0 : boost::bind(&AgentIfMapXmppChannel::WriteReadyCb, this, _1));
70 : }
71 :
72 0 : void AgentIfMapXmppChannel::ReceiveUpdate(const XmppStanza::XmppMessage *msg) {
73 0 : if (msg && msg->type == XmppStanza::IQ_STANZA) {
74 0 : std::unique_ptr<XmlBase> impl(XmppXmlImplFactory::Instance()->GetXmlImpl());
75 0 : XmlPugi *pugi = reinterpret_cast<XmlPugi *>(impl.get());
76 0 : XmlPugi *msg_pugi = reinterpret_cast<XmlPugi *>(msg->dom.get());
77 0 : pugi->LoadXmlDoc(msg_pugi->doc());
78 : boost::shared_ptr<ControllerXmppData> data(new ControllerXmppData(xmps::CONFIG,
79 : xmps::UNKNOWN,
80 0 : xs_idx_,
81 0 : std::move(impl),
82 0 : true));
83 0 : agent_->controller()->Enqueue(data);
84 0 : end_of_config_timer()->last_config_receive_time_ = UTCTimestampUsec();
85 0 : }
86 0 : }
87 :
88 0 : void AgentIfMapXmppChannel::ReceiveConfigMessage(std::unique_ptr<XmlBase> impl) {
89 :
90 0 : if (GetXmppServerIdx() != agent_->ifmap_active_xmpp_server_index()) {
91 0 : LOG(WARN, "IFMap config on non primary channel");
92 0 : return;
93 : }
94 :
95 0 : XmlPugi *pugi = reinterpret_cast<XmlPugi *>(impl.get());
96 0 : pugi::xml_node node = pugi->FindNode("config");
97 0 : IFMapAgentParser *parser = agent_->ifmap_parser();
98 0 : assert(parser);
99 0 : parser->ConfigParse(node, seq_number_);
100 : }
101 :
102 0 : void AgentIfMapXmppChannel::ReceiveInternal(const XmppStanza::XmppMessage *msg) {
103 0 : if (agent_->stats())
104 0 : agent_->stats()->incr_xmpp_config_in_msgs(GetXmppServerIdx());
105 0 : ReceiveUpdate(msg);
106 0 : }
107 :
108 0 : std::string AgentIfMapXmppChannel::ToString() const {
109 0 : return channel_->ToString();
110 : }
111 :
112 0 : void AgentIfMapXmppChannel::WriteReadyCb(const boost::system::error_code &ec) {
113 0 : }
114 :
115 0 : void AgentIfMapXmppChannel::StartEndOfConfigTimer() {
116 : //First start for end of config identification.
117 0 : end_of_config_timer()->Start(agent_->controller_xmpp_channel(xs_idx_));
118 0 : }
119 :
120 0 : void AgentIfMapXmppChannel::StopEndOfConfigTimer() {
121 : //First start for end of config identification.
122 0 : end_of_config_timer()->Cancel();
123 0 : }
124 :
125 0 : void AgentIfMapXmppChannel::StartConfigCleanupTimer() {
126 0 : config_cleanup_timer()->Start(agent_->controller_xmpp_channel(xs_idx_));
127 0 : }
128 :
129 0 : void AgentIfMapXmppChannel::StopConfigCleanupTimer() {
130 0 : config_cleanup_timer()->Cancel();
131 0 : }
132 :
133 19 : void AgentIfMapXmppChannel::EnqueueEndOfConfig() {
134 19 : EndOfConfigDataPtr data(new EndOfConfigData(this));
135 : VNController::ControllerWorkQueueDataType base_data =
136 19 : boost::static_pointer_cast<ControllerWorkQueueData>(data);
137 19 : agent_->controller()->Enqueue(base_data);
138 19 : }
139 :
140 0 : void AgentIfMapXmppChannel::ProcessEndOfConfig() {
141 0 : config_cleanup_timer()->Start(agent_->controller_xmpp_channel(xs_idx_));
142 0 : agent_->controller()->StartEndOfRibTxTimer();
143 0 : end_of_config_timer()->end_of_config_processed_time_ = UTCTimestampUsec();
144 0 : }
145 :
146 19 : EndOfConfigData::EndOfConfigData(AgentIfMapXmppChannel *ch) :
147 19 : ControllerWorkQueueData(), channel_(ch) {
148 19 : }
149 :
150 : // Get active xmpp-peer
151 328 : static AgentXmppChannel *GetActivePeer(Agent *agent) {
152 328 : int active_index = agent->ifmap_active_xmpp_server_index();
153 328 : if (active_index == -1) {
154 136 : return NULL;
155 : }
156 :
157 192 : AgentXmppChannel *peer = agent->controller_xmpp_channel(active_index);
158 192 : if (peer == NULL)
159 0 : return NULL;
160 :
161 192 : if (AgentXmppChannel::IsBgpPeerActive(agent, peer) == false) {
162 8 : return NULL;
163 : }
164 :
165 184 : return peer;
166 : }
167 :
168 :
169 184 : static AgentIfMapXmppChannel *GetActiveChannel
170 : (Agent *agent, struct AgentIfMapVmExport::VmExportInfo *info) {
171 184 : int active_index = agent->ifmap_active_xmpp_server_index();
172 184 : AgentIfMapXmppChannel *ifmap = agent->ifmap_xmpp_channel(active_index);
173 : // Peer is valid, but channel may be down
174 184 : if (ifmap == NULL)
175 184 : return NULL;
176 :
177 0 : if (info && (info->seq_number_ != ifmap->GetSeqNumber())) {
178 0 : return NULL;
179 : }
180 :
181 0 : return ifmap;
182 : }
183 :
184 165 : void AgentIfMapVmExport::VmiDelete(const ControllerVmiSubscribeData *entry) {
185 165 : VmMap::iterator vm_it = vm_map_.find(entry->vm_uuid_);
186 165 : if (vm_map_.end() == vm_it) {
187 0 : return;
188 : }
189 :
190 165 : struct VmExportInfo *info = vm_it->second;
191 : //If delete already processed, neglect the delete
192 165 : if (!info)
193 0 : return;
194 :
195 : // Find VMI-UUID in the list of VMI
196 165 : UuidList::const_iterator vmi_it = std::find(info->vmi_list_.begin(),
197 : info->vmi_list_.end(),
198 330 : entry->vmi_uuid_);
199 165 : if (vmi_it == info->vmi_list_.end())
200 0 : return;
201 :
202 165 : CONTROLLER_TRACE(IFMapVmExportTrace, UuidToString(entry->vm_uuid_),
203 : UuidToString(entry->vmi_uuid_), "Delete");
204 165 : info->vmi_list_.remove(entry->vmi_uuid_);
205 : // Stop here if VM has more interfaces
206 165 : if (info->vmi_list_.size() != 0) {
207 2 : return;
208 : }
209 :
210 : // All interfaces deleted. Remove the VM entry
211 163 : vm_map_.erase(entry->vm_uuid_);
212 :
213 : // Unsubscribe from config if we have active channel
214 163 : AgentXmppChannel *peer = GetActivePeer(agent_);
215 163 : if (peer == NULL) {
216 0 : delete info;
217 0 : CONTROLLER_TRACE(IFMapVmExportTrace, UuidToString(entry->vm_uuid_), "",
218 : "Peer NULL skipped Unsubscribe ");
219 0 : return;
220 : }
221 :
222 163 : AgentIfMapXmppChannel *ifmap = GetActiveChannel(agent_, info);
223 163 : delete info;
224 163 : if (ifmap == NULL) {
225 163 : CONTROLLER_TRACE(IFMapVmExportTrace, UuidToString(entry->vm_uuid_), "",
226 : "Channel NULL skipped Unsubscribe ");
227 163 : return;
228 : }
229 :
230 0 : CONTROLLER_TRACE(IFMapVmExportTrace, UuidToString(entry->vm_uuid_), "",
231 : "Unsubscribe ");
232 0 : AgentXmppChannel::ControllerSendVmCfgSubscribe(peer, entry->vm_uuid_,
233 : false);
234 0 : return;
235 : }
236 :
237 165 : void AgentIfMapVmExport::VmiAdd(const ControllerVmiSubscribeData *entry) {
238 165 : struct VmExportInfo *info = NULL;
239 165 : VmMap::iterator vm_it = vm_map_.find(entry->vm_uuid_);
240 165 : if (vm_map_.end() != vm_it) {
241 2 : info = vm_it->second;
242 : } else {
243 : //If first VMI create the data
244 163 : info = new VmExportInfo(0);
245 163 : vm_map_[entry->vm_uuid_] = info;
246 : }
247 :
248 : //If VMI is not found, insert in the list
249 165 : UuidList::const_iterator vmi_it = std::find(info->vmi_list_.begin(),
250 : info->vmi_list_.end(),
251 330 : entry->vmi_uuid_);
252 165 : if (vmi_it == info->vmi_list_.end()) {
253 165 : CONTROLLER_TRACE(IFMapVmExportTrace, UuidToString(entry->vm_uuid_),
254 : UuidToString(entry->vmi_uuid_), "Add");
255 165 : info->vmi_list_.push_back(entry->vmi_uuid_);
256 : }
257 :
258 : // Ensure that peer exists and is in active state
259 165 : AgentXmppChannel *peer = GetActivePeer(agent_);
260 165 : if (peer == NULL) {
261 165 : return;
262 : }
263 :
264 : // Ensure that channel is valid
265 21 : AgentIfMapXmppChannel *ifmap = GetActiveChannel(agent_, NULL);
266 21 : if (ifmap == NULL) {
267 21 : return;
268 : }
269 :
270 : //We have already sent the subscribe
271 0 : if (info->seq_number_ == ifmap->GetSeqNumber()) {
272 0 : return;
273 : }
274 :
275 0 : CONTROLLER_TRACE(IFMapVmExportTrace, UuidToString(entry->vm_uuid_),
276 : UuidToString(entry->vmi_uuid_), "Subscribe");
277 0 : AgentXmppChannel::ControllerSendVmCfgSubscribe(peer, entry->vm_uuid_, true);
278 :
279 : //Update the sequence number
280 0 : info->seq_number_ = ifmap->GetSeqNumber();
281 : }
282 :
283 330 : void AgentIfMapVmExport::VmiEvent(const ControllerVmiSubscribeData *entry) {
284 330 : if (entry->del_) {
285 165 : VmiDelete(entry);
286 165 : return;
287 : }
288 :
289 165 : VmiAdd(entry);
290 165 : return;
291 : }
292 :
293 19 : void AgentIfMapVmExport::NotifyAll(AgentXmppChannel *peer) {
294 19 : VmMap::iterator vm_it;
295 19 : AgentIfMapXmppChannel *ifmap = NULL;
296 19 : struct VmExportInfo *info = NULL;
297 19 : Agent *agent = peer->agent();
298 :
299 19 : if (!AgentXmppChannel::IsBgpPeerActive(agent, peer)) {
300 0 : return;
301 : }
302 :
303 19 : ifmap = agent->ifmap_xmpp_channel(agent->ifmap_active_xmpp_server_index());
304 19 : if (!ifmap) {
305 19 : return;
306 : }
307 :
308 : //We have all the required data. Send config subscribe for all VM's
309 : AgentIfMapVmExport *agent_ifmap_vm_export = peer->agent()->controller()->
310 0 : agent_ifmap_vm_export();
311 0 : vm_it = agent_ifmap_vm_export->vm_map_.begin();
312 0 : for(; vm_it != agent_ifmap_vm_export->vm_map_.end(); vm_it++) {
313 0 : info = vm_it->second;
314 0 : if (info->seq_number_ == ifmap->GetSeqNumber()) {
315 0 : continue;
316 : }
317 :
318 0 : std::stringstream vmid;
319 0 : vmid << vm_it->first;
320 0 : CONTROLLER_TRACE(IFMapVmExportTrace, vmid.str(), "",
321 : "Subscribe");
322 0 : AgentXmppChannel::ControllerSendVmCfgSubscribe(peer,
323 0 : vm_it->first, true);
324 :
325 : //Update the sequence number so that we dont send duplicate
326 : //subscribe
327 0 : info->seq_number_ = ifmap->GetSeqNumber();
328 0 : }
329 :
330 0 : return;
331 : }
332 :
333 2 : AgentIfMapVmExport::AgentIfMapVmExport(Agent *agent) :
334 2 : agent_(agent) {
335 2 : }
336 :
337 2 : AgentIfMapVmExport::~AgentIfMapVmExport() {
338 2 : VmMap::iterator vm_it;
339 2 : struct VmExportInfo *info = NULL;
340 :
341 : //Destroy the vm_map and vmi list
342 2 : for(vm_it = vm_map_.begin(); vm_it != vm_map_.end(); vm_it++) {
343 0 : info = vm_it->second;
344 0 : delete info;
345 : }
346 :
347 2 : vm_map_.clear();
348 2 : }
|
