Skip to content

Commit 2ed834c

Browse files
tomflynntomflynn
committed
ERSPAN support updates
Signed-off-by: Tom Flynn <tom.flynn@gmail.com> Change-Id: I0c3286f3f109c9e98baf62bb1553f7db64a90462
1 parent 03ee31b commit 2ed834c

File tree

6 files changed

+31
-45
lines changed

6 files changed

+31
-45
lines changed

‎agent-ovs/lib/SpanManager.cpp

Lines changed: 14 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -330,7 +330,7 @@ namespace opflexagent {
330330

331331
void SpanManager::SpanUniverseListener::addEndpoint(
332332
const shared_ptr<LocalEp>& lEp, const shared_ptr<L2Ep>& l2Ep,
333-
const URI& srcMemberUri, const char dir) {
333+
const URI& srcMemberUri, const unsigned char dir) {
334334
LOG(DEBUG) << "get parent lEp " << (lEp ? "set" : "null") << " l2Ep " << (l2Ep ? "set" : "null");
335335
optional<URI> parent = SpanManager::getSession(lEp);
336336
if (parent) {
@@ -435,28 +435,18 @@ namespace opflexagent {
435435
// if a match is found, add the L2Ep to the list of sources
436436
// of the mirror.
437437
URI egUri(l2Ep->getGroup().get());
438-
vector<shared_ptr<EpGroup>> epGrpVec;
439-
spanmanager.getSrcEpGroups(epGrpVec);
440-
vector<shared_ptr<EpGroup>> epgVec;
441-
LOG(DEBUG) << "Looking for uri " << egUri;
442-
for (auto& epg : epGrpVec) {
443-
LOG(DEBUG) << "EPG URI " << epg->getURI();
444-
if (epg->getURI() == egUri) {
445-
LOG(DEBUG) << "found Epg for L2Ep";
446-
epgVec.push_back(epg);
447-
}
448-
}
449-
for (auto& pEpg : epgVec) {
438+
boost::optional<shared_ptr<EpGroup>> epgOpt = spanmanager.getEpgIfPartOfSession(egUri);
439+
if (epgOpt) {
450440
std::vector<OF_SHARED_PTR<modelgbp::gbp::EpGroupToSpanSessionRSrc>> vGrpToSess;
451-
pEpg->resolveGbpEpGroupToSpanSessionRSrc(vGrpToSess);
441+
epgOpt->get()->resolveGbpEpGroupToSpanSessionRSrc(vGrpToSess);
452442
for (auto& sesRsrc : vGrpToSess) {
453443
auto it = spanmanager.sess_map.find(sesRsrc->getTargetURI().get());
454444
if (it != spanmanager.sess_map.end()) {
455445
LOG(DEBUG) << "found session " << sesRsrc->getTargetURI().get();
456-
optional<shared_ptr<SrcMember>> pSmem =
457-
spanmanager.findSrcMem(sesRsrc->getTargetURI().get(), pEpg->getURI());
458-
if (pSmem) {
459-
optional<const unsigned char> dir = pSmem.get()->getDir();
446+
optional<shared_ptr<SrcMember>> srcMem =
447+
spanmanager.findSrcMem(sesRsrc->getTargetURI().get(), egUri);
448+
if (srcMem) {
449+
optional<const unsigned char> dir = srcMem.get()->getDir();
460450
if (dir) {
461451
SourceEndpoint srcEp(l2Ep->getURI().toString(),
462452
l2Ep->getInterfaceName().get(),
@@ -471,13 +461,13 @@ namespace opflexagent {
471461
}
472462
}
473463

474-
void SpanManager::getSrcEpGroups(vector<shared_ptr<EpGroup>>& epGrpVec) {
464+
boost::optional<shared_ptr<EpGroup>> SpanManager::getEpgIfPartOfSession(const URI& epgUri) {
475465
for (const auto& sess : sess_map) {
476-
optional<shared_ptr<Session>> sPtr = Session::resolve(framework, sess.first);
477-
if (!sPtr)
466+
optional<shared_ptr<Session>> session = Session::resolve(framework, sess.first);
467+
if (!session)
478468
continue;
479469
vector <shared_ptr<SrcGrp>> srcGrpVec;
480-
sPtr.get()->resolveSpanSrcGrp(srcGrpVec);
470+
session.get()->resolveSpanSrcGrp(srcGrpVec);
481471
for (auto& srcGrp : srcGrpVec) {
482472
vector<shared_ptr<SrcMember>> srcMemVec;
483473
srcGrp->resolveSpanSrcMember(srcMemVec);
@@ -490,21 +480,14 @@ namespace opflexagent {
490480
class_id_t class_id = memRef->getTargetClass().get();
491481
if (class_id == modelgbp::gbp::EpGroup::CLASS_ID) {
492482
if (memRef->getTargetURI()) {
493-
URI epgUri = memRef->getTargetURI().get();
494-
LOG(DEBUG) << epgUri.toString();
495-
optional<shared_ptr<EpGroup>> pEpGrp =
496-
EpGroup::resolve(framework, epgUri);
497-
if (pEpGrp) {
498-
epGrpVec.push_back(pEpGrp.get());
499-
} else {
500-
LOG(DEBUG) << "Unable to resolve " << epgUri;
501-
}
483+
return EpGroup::resolve(framework, memRef->getTargetURI().get());
502484
}
503485
}
504486
}
505487
}
506488
}
507489
}
508490
}
491+
return boost::none;
509492
}
510493
}

‎agent-ovs/lib/include/opflexagent/SpanManager.h

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -177,7 +177,7 @@ class SpanManager {
177177
const shared_ptr<LocalEp>& lEp,
178178
const shared_ptr<L2Ep>& l2Ep,
179179
const URI& srcMemberUri,
180-
const char dir);
180+
const unsigned char dir);
181181

182182
/**
183183
* process EP group
@@ -208,7 +208,7 @@ class SpanManager {
208208
private:
209209

210210
optional<shared_ptr<SrcMember>> findSrcMem(const URI& sessUri, const URI& uri);
211-
void getSrcEpGroups(vector <shared_ptr<EpGroup>>& epGrpVec);
211+
boost::optional<shared_ptr<EpGroup>> getEpgIfPartOfSession(const URI& epgUri);
212212

213213
opflex::ofcore::OFFramework& framework;
214214
/**

‎agent-ovs/ovs/JsonRpc.cpp

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -299,7 +299,7 @@ bool JsonRpc::getOvsdbMirrorConfig(mirror& mir) {
299299
//replace port UUIDs with names in the mirror struct
300300
substituteSet(mir.src_ports, portMap);
301301
substituteSet(mir.dst_ports, portMap);
302-
auto itr = portMap.find(mir.out_port);
302+
auto itr = portMap.find(mir.out_port);
303303
if (itr != portMap.end()) {
304304
LOG(DEBUG) << "out_port name " << itr->second;
305305
mir.out_port = itr->second;
@@ -504,7 +504,6 @@ void JsonRpc::handleGetBridgeUuidResp(uint64_t reqId, const Document& payload, s
504504
bool JsonRpc::createMirror(const string& brUuid, const string& name, const set<string>& srcPorts,
505505
const set<string>& dstPorts) {
506506
map<string, string> portUuidMap;
507-
508507
set<string> ports;
509508
ports.insert(srcPorts.begin(), srcPorts.end());
510509
ports.insert(dstPorts.begin(), dstPorts.end());
@@ -571,7 +570,7 @@ bool JsonRpc::createMirror(const string& brUuid, const string& name, const set<s
571570
tdSet = TupleDataSet(tuples);
572571
msg2.rowData.emplace("mirrors", tdSet);
573572

574-
const list<JsonRpcTransactMessage> requests = {msg1};
573+
const list<JsonRpcTransactMessage> requests = {msg1, msg2};
575574
if (!sendRequestAndAwaitResponse(requests)) {
576575
LOG(DEBUG) << "Error sending message";
577576
return false;
@@ -704,7 +703,7 @@ void JsonRpc::connect() {
704703
}
705704

706705
bool JsonRpc::isConnected() {
707-
unique_lock<mutex> lock(conn->mtx);
706+
unique_lock<mutex> lock(OvsdbConnection::ovsdbMtx);
708707
if (!conn->ready.wait_for(lock, milliseconds(WAIT_TIMEOUT*1000),
709708
[=]{return conn->isConnected();})) {
710709
LOG(DEBUG) << "lock timed out, no connection";

‎agent-ovs/ovs/OvsdbConnection.cpp

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -23,16 +23,17 @@ extern "C" {
2323

2424
namespace opflexagent {
2525

26+
mutex OvsdbConnection::ovsdbMtx;
2627

2728
void OvsdbConnection::send_req_cb(uv_async_t* handle) {
29+
unique_lock<mutex> lock(OvsdbConnection::ovsdbMtx);
2830
auto* reqCbd = (req_cb_data*)handle->data;
29-
TransactReq* req = reqCbd->req;
31+
shared_ptr<TransactReq>& req = reqCbd->req;
3032
yajr::rpc::MethodName method(req->getMethod().c_str());
31-
opflex::jsonrpc::PayloadWrapper wrapper(req);
33+
opflex::jsonrpc::PayloadWrapper wrapper(req.get());
3234
yajr::rpc::OutboundRequest outr =
3335
yajr::rpc::OutboundRequest(wrapper, &method, req->getReqId(), reqCbd->peer);
3436
outr.send();
35-
delete(req);
3637
delete(reqCbd);
3738
}
3839

@@ -44,15 +45,15 @@ void OvsdbConnection::sendTransaction(const list<JsonRpcTransactMessage>& reques
4445
transactions[reqId] = trans;
4546
}
4647
auto* reqCbd = new req_cb_data();
47-
reqCbd->req = new TransactReq(requests, reqId);
48+
reqCbd->req = std::make_shared<TransactReq>(requests, reqId);
4849
reqCbd->peer = getPeer();
4950
send_req_async.data = (void*)reqCbd;
5051
uv_async_send(&send_req_async);
5152
}
5253

5354
void OvsdbConnection::start() {
5455
LOG(DEBUG) << "Starting .....";
55-
unique_lock<mutex> lock(mtx);
56+
unique_lock<mutex> lock(OvsdbConnection::ovsdbMtx);
5657
client_loop = threadManager.initTask("OvsdbConnection");
5758
yajr::initLoop(client_loop);
5859
uv_async_init(client_loop,&connect_async, connect_cb);
@@ -62,6 +63,7 @@ void OvsdbConnection::start() {
6263
}
6364

6465
void OvsdbConnection::connect_cb(uv_async_t* handle) {
66+
unique_lock<mutex> lock(OvsdbConnection::ovsdbMtx);
6567
OvsdbConnection* ocp = (OvsdbConnection*)handle->data;
6668
if (ocp->ovsdbUseLocalTcpPort) {
6769
ocp->peer = yajr::Peer::create("127.0.0.1",

‎agent-ovs/ovs/include/JsonRpc.h

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -391,7 +391,7 @@ class JsonRpc : public Transaction {
391391

392392
template <typename T>
393393
inline bool sendRequestAndAwaitResponse(const list<T> &tl) {
394-
unique_lock<mutex> lock(conn->mtx);
394+
unique_lock<mutex> lock(OvsdbConnection::ovsdbMtx);
395395
if (!conn->ready.wait_for(lock, milliseconds(WAIT_TIMEOUT*1000),
396396
[=]{return conn->isConnected();})) {
397397
LOG(DEBUG) << "lock timed out";

‎agent-ovs/ovs/include/OvsdbConnection.h

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -156,8 +156,10 @@ class OvsdbConnection : public opflex::jsonrpc::RpcConnection {
156156
/**
157157
* mutex used for synchronizing JSON/RPC
158158
* request and response
159+
*
160+
* static for now as we only have a single OVSDB connection
159161
*/
160-
mutex mtx;
162+
static mutex ovsdbMtx;
161163

162164
/**
163165
* set the next request ID
@@ -181,7 +183,7 @@ class OvsdbConnection : public opflex::jsonrpc::RpcConnection {
181183
yajr::Peer* peer;
182184

183185
typedef struct req_cb_data_ {
184-
TransactReq* req;
186+
shared_ptr<TransactReq> req;
185187
yajr::Peer* peer;
186188
} req_cb_data;
187189

0 commit comments

Comments
 (0)