/
ovsdb_client_idl.cc
425 lines (369 loc) · 13.8 KB
/
ovsdb_client_idl.cc
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
/*
* Copyright (c) 2014 Juniper Networks, Inc. All rights reserved.
*/
#include <assert.h>
#include <cstddef>
#include <string.h>
#include <stdlib.h>
extern "C" {
#include <ovsdb_wrapper.h>
};
#include <oper/agent_sandesh.h>
#include <ovsdb_types.h>
#include <ovsdb_client_connection_state.h>
#include <ovsdb_client_idl.h>
#include <ovsdb_client_session.h>
#include <ovsdb_route_peer.h>
#include <ovsdb_entry.h>
#include <physical_switch_ovsdb.h>
#include <logical_switch_ovsdb.h>
#include <physical_port_ovsdb.h>
#include <physical_locator_ovsdb.h>
#include <vlan_port_binding_ovsdb.h>
#include <unicast_mac_local_ovsdb.h>
#include <unicast_mac_remote_ovsdb.h>
#include <multicast_mac_local_ovsdb.h>
#include <vm_interface_ksync.h>
#include <vn_ovsdb.h>
#include <vrf_ovsdb.h>
SandeshTraceBufferPtr OvsdbTraceBuf(SandeshTraceBufferCreate("Ovsdb", 5000));
SandeshTraceBufferPtr OvsdbPktTraceBuf(SandeshTraceBufferCreate("Ovsdb Pkt", 5000));
class PhysicalDeviceTable;
class InterfaceTable;
class PhysicalDeviceVnTable;
using OVSDB::OvsdbClientIdl;
using OVSDB::OvsdbClientSession;
using OVSDB::OvsdbEntryBase;
using OVSDB::VMInterfaceKSyncObject;
using OVSDB::PhysicalSwitchTable;
using OVSDB::LogicalSwitchTable;
using OVSDB::PhysicalPortTable;
using OVSDB::PhysicalLocatorTable;
using OVSDB::VlanPortBindingTable;
using OVSDB::UnicastMacLocalOvsdb;
using OVSDB::MulticastMacLocalOvsdb;
using OVSDB::VrfOvsdbObject;
using OVSDB::VnOvsdbObject;
using OVSDB::ConnectionStateTable;
namespace OVSDB {
void ovsdb_wrapper_idl_callback(void *idl_base, int op,
struct ovsdb_idl_row *row) {
OvsdbClientIdl *client_idl = (OvsdbClientIdl *) idl_base;
int i = ovsdb_wrapper_row_type(row);
if (i >= OvsdbClientIdl::OVSDB_TYPE_COUNT)
return;
if (client_idl->callback_[i] != NULL)
client_idl->callback_[i]((OvsdbClientIdl::Op)op, row);
}
void ovsdb_wrapper_idl_txn_ack(void *idl_base, struct ovsdb_idl_txn *txn) {
OvsdbClientIdl *client_idl = (OvsdbClientIdl *) idl_base;
OvsdbEntryBase *entry = client_idl->pending_txn_[txn];
bool success = ovsdb_wrapper_is_txn_success(txn);
if (!success) {
OVSDB_TRACE(Error, "Transaction failed: " +
std::string(ovsdb_wrapper_txn_get_error(txn)));
// we don't handle the case where txn fails, when entry is not present
// case of unicast_mac_remote entry.
assert(entry != NULL);
}
client_idl->DeleteTxn(txn);
if (entry)
entry->Ack(success);
}
void intrusive_ptr_add_ref(OvsdbClientIdl *p) {
assert(p->deleted_ == false);
p->refcount_++;
}
void intrusive_ptr_release(OvsdbClientIdl *p) {
int count = --p->refcount_;
switch (count) {
case 1:
// intrusive pointer for IDL is always taken first by session while
// creating new object, and the last reference remaining is always
// with the session object which on cleanup release idl object.
OVSDB_TRACE(Trace, "Triggered Session Cleanup on Close");
// intrusive pointer reference to idl is removed only when ksync
// object is empty, with this assumption trigger delete for KsyncDb
// Objects in KSync Context.
KSyncObjectManager::Unregister(p->vm_interface_table_.release());
KSyncObjectManager::Unregister(p->logical_switch_table_.release());
KSyncObjectManager::Unregister(p->vlan_port_table_.release());
KSyncObjectManager::Unregister(p->vn_ovsdb_.release());
p->session_->OnCleanup();
break;
case 0:
OVSDB_TRACE(Trace, "Deleted IDL associated to Closed Session");
delete p;
break;
default:
break;
}
}
};
OvsdbClientIdl::OvsdbClientIdl(OvsdbClientSession *session, Agent *agent,
OvsPeerManager *manager) : idl_(ovsdb_wrapper_idl_create()),
session_(session), agent_(agent), pending_txn_(), deleted_(false),
manager_(manager), connection_state_(OvsdbSessionRcvWait),
keepalive_timer_(TimerManager::CreateTimer(
*(agent->event_manager())->io_service(),
"OVSDB Client Keep Alive Timer",
TaskScheduler::GetInstance()->GetTaskId("Agent::KSync"), 0)),
monitor_request_id_(NULL) {
refcount_ = 0;
vtep_global_= ovsdb_wrapper_vteprec_global_first(idl_);
ovsdb_wrapper_idl_set_callback(idl_, (void *)this,
ovsdb_wrapper_idl_callback, ovsdb_wrapper_idl_txn_ack);
receive_queue_ = new WorkQueue<OvsdbMsg *>(
TaskScheduler::GetInstance()->GetTaskId("Agent::KSync"), 0,
boost::bind(&OvsdbClientIdl::ProcessMessage, this, _1));
for (int i = 0; i < OVSDB_TYPE_COUNT; i++) {
callback_[i] = NULL;
}
route_peer_.reset(manager->Allocate(session_->remote_ip()));
vm_interface_table_.reset(new VMInterfaceKSyncObject(this,
(DBTable *)agent->interface_table()));
physical_switch_table_.reset(new PhysicalSwitchTable(this));
logical_switch_table_.reset(new LogicalSwitchTable(this));
physical_port_table_.reset(new PhysicalPortTable(this));
physical_locator_table_.reset(new PhysicalLocatorTable(this));
vlan_port_table_.reset(new VlanPortBindingTable(this));
unicast_mac_local_ovsdb_.reset(new UnicastMacLocalOvsdb(this,
route_peer()));
multicast_mac_local_ovsdb_.reset(new MulticastMacLocalOvsdb(this,
route_peer()));
vrf_ovsdb_.reset(new VrfOvsdbObject(this));
vn_ovsdb_.reset(new VnOvsdbObject(this, (DBTable *)agent->vn_table()));
}
OvsdbClientIdl::~OvsdbClientIdl() {
if (monitor_request_id_ != NULL) {
ovsdb_wrapper_json_destroy(monitor_request_id_);
monitor_request_id_ = NULL;
}
TimerManager::DeleteTimer(keepalive_timer_);
receive_queue_->Shutdown();
delete receive_queue_;
manager_->Free(route_peer_.release());
ovsdb_wrapper_idl_destroy(idl_);
}
OvsdbClientIdl::OvsdbMsg::OvsdbMsg(struct jsonrpc_msg *m) : msg(m) {
}
OvsdbClientIdl::OvsdbMsg::~OvsdbMsg() {
if (this->msg != NULL) {
ovsdb_wrapper_jsonrpc_msg_destroy(this->msg);
this->msg = NULL;
}
}
void OvsdbClientIdl::OnEstablish() {
if (deleted_) {
OVSDB_TRACE(Trace, "IDL deleted skipping Monitor Request");
return;
}
struct jsonrpc_msg *monitor_request =
ovsdb_wrapper_idl_encode_monitor_request(idl_);
assert(monitor_request_id_ == NULL);
// clone and save json for monitor request
monitor_request_id_ = ovsdb_wrapper_jsonrpc_clone_id(monitor_request);
OVSDB_TRACE(Trace, "Sending Monitor Request");
SendJsonRpc(monitor_request);
int keepalive_intv = session_->keepalive_interval();
if (keepalive_intv < 0) {
// timer not configured, use default timer
keepalive_intv = OVSDBKeepAliveTimer;
} else if (keepalive_intv == 0) {
// timer configured not to run, return from here.
return;
}
if (keepalive_intv < OVSDBMinKeepAliveTimer) {
// keepalive interval is not supposed to be less than min value.
keepalive_intv = OVSDBMinKeepAliveTimer;
}
// Start the Keep Alives
keepalive_timer_->Start(keepalive_intv,
boost::bind(&OvsdbClientIdl::KeepAliveTimerCb, this));
}
void OvsdbClientIdl::SendJsonRpc(struct jsonrpc_msg *msg) {
session_->SendJsonRpc(msg);
}
bool OvsdbClientIdl::ProcessMessage(OvsdbMsg *msg) {
if (!deleted_) {
// echo req and reply messages are just enqueued to identify
// session activity, since they need no further processing
// skip and delete the message
if (!ovsdb_wrapper_msg_echo_req(msg->msg) &&
!ovsdb_wrapper_msg_echo_reply(msg->msg)) {
bool connect_oper_db = false;
if (ovsdb_wrapper_idl_msg_is_monitor_response(monitor_request_id_,
msg->msg)) {
// destroy saved monitor request json message
ovsdb_wrapper_json_destroy(monitor_request_id_);
monitor_request_id_ = NULL;
connect_oper_db = true;
}
ovsdb_wrapper_idl_msg_process(idl_, msg->msg);
// msg->msg is freed by process method above
msg->msg = NULL;
// after processing the response to monitor request
// connect to oper db.
if (connect_oper_db) {
ConnectOperDB();
}
}
connection_state_ = OvsdbSessionActive;
}
delete msg;
return true;
}
struct ovsdb_idl_txn *OvsdbClientIdl::CreateTxn(OvsdbEntryBase *entry,
KSyncEntry::KSyncEvent ack_event) {
if (deleted_) {
// Don't create new transactions for deleted idl.
return NULL;
}
struct ovsdb_idl_txn *txn = ovsdb_wrapper_idl_txn_create(idl_);
pending_txn_[txn] = entry;
if (entry != NULL) {
// if entry is available store the ack_event in entry
entry->ack_event_ = ack_event;
}
return txn;
}
void OvsdbClientIdl::DeleteTxn(struct ovsdb_idl_txn *txn) {
pending_txn_.erase(txn);
ovsdb_wrapper_idl_txn_destroy(txn);
}
// API to trigger ovs row del followed by add
// used by OvsdbEntry on catastrophic change event, which
// results in emulating a delete followed by add
void OvsdbClientIdl::NotifyDelAdd(struct ovsdb_idl_row *row) {
int i = ovsdb_wrapper_row_type(row);
if (i >= OvsdbClientIdl::OVSDB_TYPE_COUNT)
return;
if (callback_[i] != NULL) {
callback_[i](OvsdbClientIdl::OVSDB_DEL, row);
callback_[i](OvsdbClientIdl::OVSDB_ADD, row);
}
}
Ip4Address OvsdbClientIdl::tsn_ip() {
return session_->tsn_ip();
}
void OvsdbClientIdl::MessageProcess(struct jsonrpc_msg *msg) {
// Enqueue all received messages in receive queue running KSync task
// context, to assure only one thread is writting data to OVSDB client.
OvsdbMsg *ovs_msg = new OvsdbMsg(msg);
receive_queue_->Enqueue(ovs_msg);
}
Ip4Address OvsdbClientIdl::remote_ip() const {
return session_->remote_ip();
}
uint16_t OvsdbClientIdl::remote_port() const {
return session_->remote_port();
}
ConnectionStateTable *OvsdbClientIdl::connection_table() {
return session_->connection_table();
}
KSyncObjectManager *OvsdbClientIdl::ksync_obj_manager() {
return session_->ksync_obj_manager();
}
OvsPeer *OvsdbClientIdl::route_peer() {
return route_peer_.get();
}
VMInterfaceKSyncObject *OvsdbClientIdl::vm_interface_table() {
return vm_interface_table_.get();
}
PhysicalSwitchTable *OvsdbClientIdl::physical_switch_table() {
return physical_switch_table_.get();
}
LogicalSwitchTable *OvsdbClientIdl::logical_switch_table() {
return logical_switch_table_.get();
}
PhysicalPortTable *OvsdbClientIdl::physical_port_table() {
return physical_port_table_.get();
}
PhysicalLocatorTable *OvsdbClientIdl::physical_locator_table() {
return physical_locator_table_.get();
}
VlanPortBindingTable *OvsdbClientIdl::vlan_port_table() {
return vlan_port_table_.get();
}
UnicastMacLocalOvsdb *OvsdbClientIdl::unicast_mac_local_ovsdb() {
return unicast_mac_local_ovsdb_.get();
}
MulticastMacLocalOvsdb *OvsdbClientIdl::multicast_mac_local_ovsdb() {
return multicast_mac_local_ovsdb_.get();
}
VrfOvsdbObject *OvsdbClientIdl::vrf_ovsdb() {
return vrf_ovsdb_.get();
}
VnOvsdbObject *OvsdbClientIdl::vn_ovsdb() {
return vn_ovsdb_.get();
}
bool OvsdbClientIdl::IsKeepAliveTimerActive() {
return !keepalive_timer_->cancelled();
}
bool OvsdbClientIdl::KeepAliveTimerCb() {
switch (connection_state_) {
case OvsdbSessionActive:
// session is active, move to Receive wait state to
// identify session activity.
connection_state_ = OvsdbSessionRcvWait;
return true;
case OvsdbSessionRcvWait:
{
// send echo request and restart the timer to wait for reply
struct jsonrpc_msg *req = ovsdb_wrapper_jsonrpc_create_echo_request();
connection_state_ = OvsdbSessionEchoWait;
SendJsonRpc(req);
}
return true;
case OvsdbSessionEchoWait:
// echo reply not recevied ovsdb-server is not responding,
// close the session
OVSDB_TRACE(Error, "KeepAlive failed, Closing Session");
session_->TriggerClose();
// Connection is closed, timer doesn't need restart
return false;
}
return true;
}
void OvsdbClientIdl::TriggerDeletion() {
// if idl is already marked for delete, return from here
if (deleted_) {
return;
}
// mark idl being set for deletion, so we don't create further txn
deleted_ = true;
// Since IDL is scheduled for deletion cancel keepalive timer
keepalive_timer_->Cancel();
// trigger txn failure for pending transcations
PendingTxnMap::iterator it = pending_txn_.begin();
while (it != pending_txn_.end()) {
OvsdbEntryBase *entry = it->second;
DeleteTxn(it->first);
// Ack failure, if entry is available.
if (entry)
entry->Ack(false);
it = pending_txn_.begin();
}
// trigger KSync Object delete for all objects.
vm_interface_table_->DeleteTable();
physical_switch_table_->DeleteTable();
logical_switch_table_->DeleteTable();
physical_port_table_->DeleteTable();
physical_locator_table_->DeleteTable();
vlan_port_table_->DeleteTable();
unicast_mac_local_ovsdb_->DeleteTable();
multicast_mac_local_ovsdb_->DeleteTable();
vn_ovsdb_->DeleteTable();
// trigger delete table for vrf table, which internally handles
// deletion of route table.
vrf_ovsdb_->DeleteTable();
}
void OvsdbClientIdl::ConnectOperDB() {
OVSDB_TRACE(Trace, "Received Monitor Response connecting to OperDb");
logical_switch_table_->OvsdbRegisterDBTable(
(DBTable *)agent_->physical_device_vn_table());
vlan_port_table_->OvsdbRegisterDBTable(
(DBTable *)agent_->interface_table());
vrf_ovsdb_->OvsdbRegisterDBTable(
(DBTable *)agent_->vrf_table());
}