Line data Source code
1 : /*
2 : * Copyright (c) 2017 Juniper Networks, Inc. All rights reserved.
3 : * Copyright Contributors to the OpenSDN project
4 : */
5 :
6 : #include <sandesh/sandesh_types.h>
7 : #include <sandesh/sandesh.h>
8 : #include <cmn/agent_cmn.h>
9 : #include <init/agent_init.h>
10 : #include <oper/interface.h>
11 : #include <oper/vm_interface.h>
12 : #include <oper/metadata_ip.h>
13 : #include <pkt/proto_handler.h>
14 : #include "pkt/pkt_init.h"
15 : #include <services/bfd_proto.h>
16 : #include <services/bfd_handler.h>
17 : #include <services/services_types.h>
18 : #include <services/services_init.h>
19 : #include "base/logging.h"
20 :
21 2 : BfdProto::BfdProto(Agent *agent, boost::asio::io_context &io) :
22 : Proto(agent, "Agent::Services", PktHandler::BFD, io),
23 2 : msg_(new PktInfo(agent, BFD_TX_BUFF_LEN, PktHandler::BFD, 0)),
24 2 : communicator_(this),
25 2 : server_(new BFD::Server(agent->event_manager(), &communicator_)),
26 4 : client_(new BFD::Client(&communicator_)), handler_(agent, msg_, io) {
27 :
28 : // limit the number of entries in the workqueue
29 2 : work_queue_.SetSize(agent->params()->services_queue_limit());
30 2 : work_queue_.SetBounded(true);
31 :
32 2 : agent->health_check_table()->RegisterHealthCheckCallback(
33 : boost::bind(&BfdProto::BfdHealthCheckSessionControl, this, _1, _2),
34 : HealthCheckService::BFD);
35 2 : }
36 :
37 4 : BfdProto::~BfdProto() {
38 4 : }
39 :
40 0 : ProtoHandler *BfdProto::AllocProtoHandler(boost::shared_ptr<PktInfo> info,
41 : boost::asio::io_context &io) {
42 0 : return new BfdHandler(agent(), info, io);
43 : }
44 :
45 0 : bool BfdSessionsKey::IsLess(const BfdSessionsKey &rhs) const {
46 0 : if (id_ != rhs.id_) {
47 0 : return id_ < rhs.id_;
48 : }
49 0 : return ip_ < rhs.ip_;
50 : }
51 :
52 0 : bool BfdSessionsKey::IsEqual(const BfdSessionsKey &rhs) const {
53 0 : if (id_ != rhs.id_) {
54 0 : return false;
55 : }
56 0 : if (ip_ != rhs.ip_) {
57 0 : return false;
58 : }
59 0 : return true;
60 : }
61 :
62 0 : bool BfdProto::BfdHealthCheckSessionControl(
63 : HealthCheckTable::HealthCheckServiceAction action,
64 : HealthCheckInstanceService *service) {
65 :
66 0 : uint16_t remote_port = service->is_multi_hop() ?
67 0 : BFD::kMultiHop : BFD::kSingleHop;
68 0 : IpAddress source_ip = service->update_source_ip();
69 0 : IpAddress destination_ip = service->destination_ip();
70 0 : if (source_ip.is_unspecified()) {
71 0 : BFD_TRACE(Trace, "Ignore",
72 : destination_ip.to_string(),
73 : source_ip.to_string(), service->interface()->id(),
74 : 0, 0, 0);
75 0 : return false;
76 : }
77 : BFD::SessionKey key(destination_ip,
78 0 : BFD::SessionIndex(service->interface()->id()),
79 : remote_port,
80 0 : source_ip);
81 :
82 0 : std::scoped_lock lock(mutex_);
83 0 : switch (action) {
84 0 : case HealthCheckTable::CREATE_SERVICE:
85 : case HealthCheckTable::UPDATE_SERVICE:
86 : {
87 0 : if (source_ip.is_v4() &&
88 0 : source_ip.to_v4() == Ip4Address(METADATA_IP_ADDR)) {
89 0 : return false;
90 : }
91 :
92 0 : uint32_t tx_interval = service->service()->delay() * 1000000 +
93 0 : service->service()->delay_usecs();
94 0 : if (!tx_interval) {
95 0 : tx_interval = kMinTxInterval;
96 : }
97 0 : uint32_t rx_interval = service->service()->timeout() * 1000000 +
98 0 : service->service()->timeout_usecs();
99 0 : if (!rx_interval) {
100 0 : rx_interval = kMinRxInterval;
101 : }
102 0 : uint32_t multiplier = service->service()->max_retries();
103 0 : if (!multiplier) {
104 0 : multiplier = kMultiplier;
105 : }
106 :
107 0 : BFD::SessionConfig session_config;
108 0 : session_config.desiredMinTxInterval =
109 0 : boost::posix_time::microseconds(tx_interval);
110 0 : session_config.requiredMinRxInterval =
111 0 : boost::posix_time::microseconds(rx_interval);
112 0 : session_config.detectionTimeMultiplier = multiplier;
113 :
114 0 : client_->AddSession(key, session_config);
115 0 : BfdSessionsKey skey(service->interface()->id(),
116 0 : service->destination_ip());
117 0 : sessions_[skey] = service;
118 0 : BFD_TRACE(Trace, "Add / Update",
119 : destination_ip.to_string(),
120 : source_ip.to_string(), service->interface()->id(),
121 : tx_interval, rx_interval, multiplier);
122 0 : break;
123 0 : }
124 :
125 0 : case HealthCheckTable::DELETE_SERVICE:
126 : {
127 0 : client_->DeleteSession(key);
128 0 : BfdSessionsKey skey(service->interface()->id(),
129 0 : service->destination_ip());
130 0 : Sessions::iterator it = sessions_.find(skey);
131 0 : if (it != sessions_.end()) {
132 0 : sessions_.erase(it);
133 0 : BFD_TRACE(Trace, "Delete",
134 : destination_ip.to_string(),
135 : source_ip.to_string(), service->interface()->id(),
136 : 0, 0, 0);
137 : } else {
138 : // TODO: replace log with trace
139 0 : std::string str("BFD delete session ");
140 0 : str += " interface-id: " + service->interface()->id();
141 0 : str += " source-ip: " + source_ip.to_string();
142 0 : str += " destination-ip: " + destination_ip.to_string();
143 :
144 0 : LOG(ERROR, "Duplicate request: " << str);
145 0 : }
146 0 : break;
147 0 : }
148 :
149 0 : case HealthCheckTable::RUN_SERVICE:
150 0 : break;
151 :
152 0 : case HealthCheckTable::STOP_SERVICE:
153 0 : break;
154 :
155 0 : default:
156 0 : assert(0);
157 : }
158 :
159 0 : return true;
160 0 : }
161 :
162 0 : void BfdProto::NotifyHealthCheckInstanceService(uint32_t interface,
163 : IpAddress address,
164 : std::string &data) {
165 0 : std::scoped_lock lock(mutex_);
166 0 : BfdSessionsKey key(interface, address);
167 0 : Sessions::iterator it = sessions_.find(key);
168 0 : if (it == sessions_.end()) {
169 0 : return;
170 : }
171 :
172 0 : const bool up = data.find("success") != std::string::npos;
173 0 : const bool down = data.find("failure") != std::string::npos;
174 0 : if (up || down) {
175 0 : it->second->BfdStateChange(up);
176 : }
177 :
178 0 : it->second->OnRead(data);
179 :
180 : // log the BFD up/down event
181 0 : std::string str("BFD session ");
182 0 : if (data.find("success") != std::string::npos) {
183 0 : str += "Up,";
184 : }
185 0 : if (data.find("failure") != std::string::npos) {
186 0 : str += "Down,";
187 : }
188 0 : if (it->second->service()) {
189 0 : str += " service-health-check: " + it->second->service()->name();
190 : } else {
191 0 : str += " service: null";
192 : }
193 0 : if (it->second->interface()) {
194 0 : str += " interface: " + it->second->interface()->name();
195 0 : if (it->second->interface()->vrf()) {
196 0 : str += " vrf: " + it->second->interface()->vrf()->GetName();
197 : } else {
198 0 : str += " vrf: null";
199 : }
200 : } else {
201 0 : str += " interface: null";
202 : }
203 0 : if (!it->second->destination_ip().is_unspecified()) {
204 0 : str += " destination-ip: " + it->second->destination_ip().to_string();
205 : } else {
206 0 : str += " destination-ip: null";
207 : }
208 0 : LOG(WARN, "SYS_NOTICE " << str);
209 :
210 0 : }
211 :
212 0 : void BfdProto::HandleReceiveSafe(
213 : boost::asio::const_buffer packet,
214 : const boost::asio::ip::udp::endpoint &local_endpoint,
215 : const boost::asio::ip::udp::endpoint &remote_endpoint,
216 : const BFD::SessionIndex &session_index,
217 : uint8_t len,
218 : boost::system::error_code ec) {
219 :
220 0 : std::scoped_lock lock(rx_mutex_);
221 0 : bfd_communicator().HandleReceive(
222 : packet, local_endpoint, remote_endpoint,
223 : session_index,
224 : len, ec);
225 :
226 0 : IncrementReceived();
227 0 : }
228 :
229 0 : void BfdProto::BfdCommunicator::SendPacket(
230 : const boost::asio::ip::udp::endpoint &local_endpoint,
231 : const boost::asio::ip::udp::endpoint &remote_endpoint,
232 : const BFD::SessionIndex &session_index,
233 : const boost::asio::mutable_buffer &packet, int pktSize) {
234 0 : bfd_proto_->handler_.SendPacket(local_endpoint, remote_endpoint,
235 0 : session_index.if_index, packet, pktSize);
236 0 : bfd_proto_->IncrementSent();
237 0 : }
238 :
239 0 : void BfdProto::BfdCommunicator::NotifyStateChange(const BFD::SessionKey &key,
240 : const bool &up) {
241 0 : std::string data = up ? "success" : "failure";
242 0 : bfd_proto_->NotifyHealthCheckInstanceService(key.index.if_index,
243 0 : key.remote_address ,data);
244 0 : }
245 :
246 0 : bool BfdProto::Enqueue(boost::shared_ptr<PktInfo> msg) {
247 :
248 0 : if (msg->is_bfd_keepalive) {
249 : // Handle BFD keepalive packets
250 0 : return ProcessBfdKeepAlive(msg);
251 : }
252 : // Handle BFD control packets
253 0 : return Proto::Enqueue(msg);
254 : }
255 :
256 : // Handle BFD packet stats
257 0 : void BfdProto::ProcessStats(PktStatsType::Type type) {
258 0 : switch(type) {
259 0 : case PktStatsType::PKT_RX_DROP_STATS:
260 0 : IncrementReceiveDropCount();
261 0 : break;
262 0 : case PktStatsType::PKT_RX_ENQUEUE:
263 0 : IncrementKaEnqueueCount();
264 0 : break;
265 0 : default:
266 0 : return;
267 : }
268 : }
269 :
270 0 : bool BfdProto::ProcessBfdKeepAlive(boost::shared_ptr<PktInfo> msg_info) {
271 0 : PktHandler *pkt_handler = Agent::GetInstance()->pkt()->pkt_handler();
272 :
273 0 : BfdHandler *handler = new BfdHandler(agent(), msg_info, get_io_service());
274 0 : if (handler->HandleReceive())
275 0 : delete handler;
276 :
277 0 : return true;;
278 : }
|
