Skip to content

Commit 945ae4b

Browse files
trondntrondn
committed
MB-35882: Add support for filters to stats dcp
The filter allows for filtering only DCP streams connected to a single port or by a given user (or both): { "filter" : { "user" : "@ns_server", "port" : 666, } } Change-Id: Iad7b1874bd15fc442bffe97edcb30a49a0245be2 Reviewed-on: http://review.couchbase.org/114601 Tested-by: Build Bot <[email protected]> Reviewed-by: Jim Walker <[email protected]>
1 parent a078232 commit 945ae4b

File tree

11 files changed

+137
-18
lines changed

11 files changed

+137
-18
lines changed

‎auditd/tests/testauditd.cc‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -113,6 +113,13 @@ class AuditMockServerCookieApi : public ServerCookieIface {
113113
gsl::not_null<const void*> cookie) override {
114114
throw std::runtime_error("Not implemented");
115115
}
116+
std::string get_authenticated_user(
117+
gsl::not_null<const void*> cookie) override {
118+
throw std::runtime_error("Not implemented");
119+
}
120+
in_port_t get_connected_port(gsl::not_null<const void*> cookie) override {
121+
throw std::runtime_error("Not implemented");
122+
}
116123
void set_error_context(gsl::not_null<void*> cookie,
117124
cb::const_char_buffer message) override {
118125
throw std::runtime_error("Not implemented");

‎daemon/memcached.cc‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1702,6 +1702,18 @@ struct ServerCookieApi : public ServerCookieIface {
17021702
cookie->getConnection().getDescription());
17031703
}
17041704

1705+
std::string get_authenticated_user(
1706+
gsl::not_null<const void*> void_cookie) override {
1707+
auto* cookie = reinterpret_cast<const Cookie*>(void_cookie.get());
1708+
return cookie->getConnection().getUsername();
1709+
}
1710+
1711+
in_port_t get_connected_port(
1712+
gsl::not_null<const void*> void_cookie) override {
1713+
auto* cookie = reinterpret_cast<const Cookie*>(void_cookie.get());
1714+
return cookie->getConnection().getParentPort();
1715+
}
1716+
17051717
void set_error_context(gsl::not_null<void*> void_cookie,
17061718
cb::const_char_buffer message) override {
17071719
auto* cookie = reinterpret_cast<Cookie*>(void_cookie.get());

‎engines/ep/src/connhandler.cc‎

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -43,22 +43,26 @@ std::string to_string(ConnHandler::PausedReason r) {
4343

4444
ConnHandler::ConnHandler(EventuallyPersistentEngine& e,
4545
const void* c,
46-
const std::string& n)
46+
std::string n)
4747
: engine_(e),
4848
stats(engine_.getEpStats()),
49-
name(n),
49+
name(std::move(n)),
5050
cookie(const_cast<void*>(c)),
5151
reserved(false),
5252
created(ep_current_time()),
5353
disconnect(false),
54-
paused(false) {
54+
paused(false),
55+
authenticatedUser(e.getServerApi()->cookie->get_authenticated_user(c)),
56+
connected_port(e.getServerApi()->cookie->get_connected_port(c)) {
5557
logger = BucketLogger::createBucketLogger(
5658
std::to_string(reinterpret_cast<uintptr_t>(this)));
5759

5860
auto connId = e.getServerApi()->cookie->get_log_info(c).first;
5961
logger->setConnectionId(connId);
6062
}
6163

64+
ConnHandler::~ConnHandler() = default;
65+
6266
ENGINE_ERROR_CODE ConnHandler::addStream(uint32_t opaque,
6367
Vbid,
6468
uint32_t flags) {

‎engines/ep/src/connhandler.h‎

Lines changed: 18 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -85,10 +85,11 @@ class ConnHandler {
8585
Unknown
8686
};
8787

88-
ConnHandler(EventuallyPersistentEngine& engine, const void* c,
89-
const std::string& name);
88+
ConnHandler(EventuallyPersistentEngine& engine,
89+
const void* c,
90+
std::string name);
9091

91-
virtual ~ConnHandler() {}
92+
virtual ~ConnHandler();
9293

9394
virtual ENGINE_ERROR_CODE addStream(uint32_t opaque,
9495
Vbid vbucket,
@@ -331,6 +332,14 @@ class ConnHandler {
331332
paused.store(false);
332333
}
333334

335+
const std::string& getAuthenticatedUser() const {
336+
return authenticatedUser;
337+
}
338+
339+
in_port_t getConnectedPort() const {
340+
return connected_port;
341+
}
342+
334343
protected:
335344
EventuallyPersistentEngine &engine_;
336345
EPStats &stats;
@@ -370,6 +379,12 @@ class ConnHandler {
370379

371380
//! Description of why the connection is paused.
372381
std::atomic<PausedReason> reason;
382+
383+
/// The authenticated user the connection
384+
const std::string authenticatedUser;
385+
386+
/// The port the connection is connected to
387+
const in_port_t connected_port;
373388
};
374389

375390
std::string to_string(ConnHandler::PausedReason r);

‎engines/ep/src/dcp/consumer.cc‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,8 @@
3232
#include <memcached/server_cookie_iface.h>
3333
#include <phosphor/phosphor.h>
3434

35+
#include <utility>
36+
3537
const std::string DcpConsumer::noopCtrlMsg = "enable_noop";
3638
const std::string DcpConsumer::noopIntervalCtrlMsg = "set_noop_interval";
3739
const std::string DcpConsumer::connBufferCtrlMsg = "connection_buffer_size";

‎engines/ep/src/ep_engine.cc‎

Lines changed: 63 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -3510,27 +3510,76 @@ ENGINE_ERROR_CODE EventuallyPersistentEngine::doDurabilityMonitorStats(
35103510
return ENGINE_SUCCESS;
35113511
}
35123512

3513+
class DcpStatsFilter {
3514+
public:
3515+
explicit DcpStatsFilter(cb::const_char_buffer value) {
3516+
if (!value.empty()) {
3517+
try {
3518+
auto attributes = nlohmann::json::parse(value);
3519+
auto filter = attributes.find("filter");
3520+
if (filter != attributes.end()) {
3521+
auto iter = filter->find("user");
3522+
if (iter != filter->end()) {
3523+
user.reset(cb::tagUserData(iter->get<std::string>()));
3524+
}
3525+
iter = filter->find("port");
3526+
if (iter != filter->end()) {
3527+
port.reset(iter->get<in_port_t>());
3528+
}
3529+
}
3530+
} catch (const std::exception& e) {
3531+
EP_LOG_ERR(
3532+
"Failed to decode provided DCP filter: {}. Filter:{}",
3533+
e.what(),
3534+
value);
3535+
}
3536+
}
3537+
}
3538+
3539+
bool include(const std::shared_ptr<ConnHandler>& tc) {
3540+
if ((user && *user != tc->getAuthenticatedUser()) ||
3541+
(port && *port != tc->getConnectedPort())) {
3542+
// Connection should not be part of this output
3543+
return false;
3544+
}
3545+
3546+
return true;
3547+
}
3548+
3549+
protected:
3550+
boost::optional<std::string> user;
3551+
boost::optional<in_port_t> port;
3552+
};
3553+
35133554
/**
35143555
* Function object to send stats for a single dcp connection.
35153556
*/
35163557
struct ConnStatBuilder {
3517-
ConnStatBuilder(const void* c, const AddStatFn& as, ConnCounter& tc)
3518-
: cookie(c), add_stat(as), aggregator(tc) {
3558+
ConnStatBuilder(const void* c,
3559+
AddStatFn as,
3560+
DcpStatsFilter filter,
3561+
ConnCounter& tc)
3562+
: cookie(c),
3563+
add_stat(std::move(as)),
3564+
filter(std::move(filter)),
3565+
aggregator(tc) {
35193566
}
35203567

35213568
void operator()(std::shared_ptr<ConnHandler> tc) {
35223569
++aggregator.totalConns;
3523-
tc->addStats(add_stat, cookie);
3524-
3525-
auto tp = std::dynamic_pointer_cast<DcpProducer>(tc);
3526-
if (tp) {
3527-
++aggregator.totalProducers;
3528-
tp->aggregateQueueStats(aggregator);
3570+
if (filter.include(tc)) {
3571+
tc->addStats(add_stat, cookie);
3572+
auto tp = std::dynamic_pointer_cast<DcpProducer>(tc);
3573+
if (tp) {
3574+
++aggregator.totalProducers;
3575+
tp->aggregateQueueStats(aggregator);
3576+
}
35293577
}
35303578
}
35313579

35323580
const void *cookie;
35333581
AddStatFn add_stat;
3582+
DcpStatsFilter filter;
35343583
ConnCounter& aggregator;
35353584
};
35363585

@@ -3670,9 +3719,12 @@ ENGINE_ERROR_CODE EventuallyPersistentEngine::doConnAggStats(
36703719
}
36713720

36723721
ENGINE_ERROR_CODE EventuallyPersistentEngine::doDcpStats(
3673-
const void* cookie, const AddStatFn& add_stat) {
3722+
const void* cookie,
3723+
const AddStatFn& add_stat,
3724+
cb::const_char_buffer value) {
36743725
ConnCounter aggregator;
3675-
ConnStatBuilder dcpVisitor(cookie, add_stat, aggregator);
3726+
ConnStatBuilder dcpVisitor(
3727+
cookie, add_stat, DcpStatsFilter{value}, aggregator);
36763728
dcpConnMap_->each(dcpVisitor);
36773729

36783730
add_casted_stat("ep_dcp_count", aggregator.totalConns, add_stat, cookie);
@@ -4343,7 +4395,7 @@ ENGINE_ERROR_CODE EventuallyPersistentEngine::getStats(
43434395
return doConnAggStats(cookie, add_stat, key.data() + 7, key.size() - 7);
43444396
}
43454397
if (key == "dcp"_ccb) {
4346-
return doDcpStats(cookie, add_stat);
4398+
return doDcpStats(cookie, add_stat, value);
43474399
}
43484400
if (key == "eviction"_ccb) {
43494401
return doEvictionStats(cookie, add_stat);

‎engines/ep/src/ep_engine.h‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -870,7 +870,9 @@ class EventuallyPersistentEngine : public EngineIface, public DcpIface {
870870
ENGINE_ERROR_CODE doDurabilityMonitorDump(const void* cookie,
871871
const AddStatFn& addStat,
872872
cb::const_char_buffer keyArgs);
873-
ENGINE_ERROR_CODE doDcpStats(const void* cookie, const AddStatFn& add_stat);
873+
ENGINE_ERROR_CODE doDcpStats(const void* cookie,
874+
const AddStatFn& add_stat,
875+
cb::const_char_buffer value);
874876
ENGINE_ERROR_CODE doEvictionStats(const void* cookie,
875877
const AddStatFn& add_stat);
876878
ENGINE_ERROR_CODE doConnAggStats(const void* cookie,

‎engines/ep/tests/module_tests/dcp_test.cc‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -140,6 +140,13 @@ class WrappedServerCookieIface : public ServerCookieIface {
140140
gsl::not_null<const void*> cookie) override {
141141
return wrapped->get_log_info(cookie);
142142
}
143+
std::string get_authenticated_user(
144+
gsl::not_null<const void*> cookie) override {
145+
return wrapped->get_authenticated_user(cookie);
146+
}
147+
in_port_t get_connected_port(gsl::not_null<const void*> cookie) override {
148+
return wrapped->get_connected_port(cookie);
149+
}
143150
void set_error_context(gsl::not_null<void*> cookie,
144151
cb::const_char_buffer message) override {
145152
wrapped->set_error_context(cookie, message);

‎include/memcached/server_cookie_iface.h‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -221,6 +221,11 @@ struct ServerCookieIface {
221221
virtual std::pair<uint32_t, std::string> get_log_info(
222222
gsl::not_null<const void*> cookie) = 0;
223223

224+
virtual std::string get_authenticated_user(
225+
gsl::not_null<const void*> cookie) = 0;
226+
227+
virtual in_port_t get_connected_port(gsl::not_null<const void*> cookie) = 0;
228+
224229
/**
225230
* Set the error context string to be sent in response. This should not
226231
* contain security sensitive information. If sensitive information needs to

‎programs/engine_testapp/mock_server.cc‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -361,6 +361,17 @@ struct MockServerCookieApi : public ServerCookieIface {
361361
return std::make_pair(uint32_t(0xdead), std::string{"[you - me]"});
362362
}
363363

364+
std::string get_authenticated_user(
365+
gsl::not_null<const void*> cookie) override {
366+
auto* c = cookie_to_mock_object(cookie.get());
367+
return c->authenticatedUser;
368+
}
369+
370+
in_port_t get_connected_port(gsl::not_null<const void*> cookie) override {
371+
auto* c = cookie_to_mock_object(cookie.get());
372+
return c->parent_port;
373+
}
374+
364375
void set_error_context(gsl::not_null<void*> cookie,
365376
cb::const_char_buffer message) override {
366377
}

0 commit comments

Comments
 (0)