This is an automated email from the ASF dual-hosted git repository.
chug pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/qpid-dispatch.git
The following commit(s) were added to refs/heads/master by this push:
new 69868bf DISPATCH-1422: Add aggregated policy link and total denial
counts
69868bf is described below
commit 69868bff4a867ec6ae7cd082e0246f85067744d0
Author: Chuck Rolke <[email protected]>
AuthorDate: Fri Oct 11 11:46:44 2019 -0400
DISPATCH-1422: Add aggregated policy link and total denial counts
Policy status creates two global counts that get aggregated for
all links and connections:
* linksDenied - sum of all vhost sender and receiver denials
* totalDenials - sum of linksDenied, connectionsDenied, and
sessionsDenied
Implementation adds locks around changing shared static and
generating statistic reports for management.
Self test is added on to the PolicyVhostOverride as that test puts
its statistics into a known vhost. Otherwise stats could go into
system-specific vhosts like '0.0.0.0', '::0', 'localhost', etc.
adding unneeded complexity to the analysis.
---
python/qpid_dispatch/management/qdrouter.json | 12 +++-
.../qpid_dispatch_internal/policy/policy_local.py | 2 +
src/policy.c | 54 ++++++++++++++---
tests/system_tests_policy.py | 69 +++++++++++++++++++---
4 files changed, 120 insertions(+), 17 deletions(-)
diff --git a/python/qpid_dispatch/management/qdrouter.json
b/python/qpid_dispatch/management/qdrouter.json
index 3cd092f..961106a 100644
--- a/python/qpid_dispatch/management/qdrouter.json
+++ b/python/qpid_dispatch/management/qdrouter.json
@@ -1837,7 +1837,17 @@
},
"connectionsProcessed": {"type": "integer", "graph": true},
"connectionsDenied": {"type": "integer", "graph": true},
- "connectionsCurrent": {"type": "integer", "graph": true}
+ "connectionsCurrent": {"type": "integer", "graph": true},
+ "linksDenied": {
+ "type": "integer",
+ "graph": true,
+ "description": "The sum of all vhost sender and receiver
denials."
+ },
+ "totalDenials": {
+ "type": "integer",
+ "graph": true,
+ "description": "The total number of connection and link
denials."
+ }
}
},
diff --git a/python/qpid_dispatch_internal/policy/policy_local.py
b/python/qpid_dispatch_internal/policy/policy_local.py
index ecf717c..cff8df3 100644
--- a/python/qpid_dispatch_internal/policy/policy_local.py
+++ b/python/qpid_dispatch_internal/policy/policy_local.py
@@ -83,6 +83,8 @@ class PolicyKeys(object):
KW_CONNECTIONS_APPROVED = "connectionsApproved"
KW_CONNECTIONS_DENIED = "connectionsDenied"
KW_CONNECTIONS_CURRENT = "connectionsCurrent"
+ KW_LINKS_DENIED = "linksDenied"
+ KW_TOTAL_DENIALS = "totalDenials"
KW_PER_USER_STATE = "perUserState"
KW_PER_HOST_STATE = "perHostState"
diff --git a/src/policy.c b/src/policy.c
index 6019c7c..b2768f3 100644
--- a/src/policy.c
+++ b/src/policy.c
@@ -40,9 +40,13 @@
// The current statistics maintained globally through multiple
// reconfiguration of policy settings.
//
+static sys_mutex_t *stats_lock = 0;
+
static int n_connections = 0;
static int n_denied = 0;
static int n_processed = 0;
+static int n_links_denied = 0;
+static int n_total_denials = 0;
//
// error conditions signaled to effect denial
@@ -109,6 +113,7 @@ qd_policy_t *qd_policy(qd_dispatch_t *qd)
policy->max_connection_limit = 65535;
policy->tree_lock = sys_mutex();
policy->hostname_tree = qd_parse_tree_new(QD_PARSE_TREE_ADDRESS);
+ stats_lock = sys_mutex();
qd_log(policy->log_source, QD_LOG_TRACE, "Policy Initialized");
return policy;
@@ -127,6 +132,8 @@ void qd_policy_free(qd_policy_t *policy)
hostname_tree_free(policy->hostname_tree);
Py_XDECREF(module);
free(policy);
+ if (stats_lock)
+ sys_mutex_free(stats_lock);
}
//
@@ -209,9 +216,21 @@ qd_error_t qd_policy_c_counts_refresh(long ccounts,
qd_entity_t *entity)
**/
qd_error_t qd_entity_refresh_policy(qd_entity_t* entity, void *unused) {
// Return global stats
- if (!qd_entity_set_long(entity, "connectionsProcessed", n_processed) &&
- !qd_entity_set_long(entity, "connectionsDenied", n_denied) &&
- !qd_entity_set_long(entity, "connectionsCurrent", n_connections)
+ int np, nd, nc, nl, nt;
+ sys_mutex_lock(stats_lock);
+ {
+ np = n_processed;
+ nd = n_denied;
+ nc = n_connections;
+ nl = n_links_denied;
+ nt = n_total_denials;
+ }
+ sys_mutex_unlock(stats_lock);
+ if (!qd_entity_set_long(entity, "connectionsProcessed", np) &&
+ !qd_entity_set_long(entity, "connectionsDenied", nd) &&
+ !qd_entity_set_long(entity, "connectionsCurrent", nc) &&
+ !qd_entity_set_long(entity, "linksDenied", nl) &&
+ !qd_entity_set_long(entity, "totalDenials", nt)
)
return QD_ERROR_NONE;
return qd_error_code();
@@ -229,17 +248,25 @@ qd_error_t qd_entity_refresh_policy(qd_entity_t* entity,
void *unused) {
bool qd_policy_socket_accept(qd_policy_t *policy, const char *hostname)
{
bool result = true;
+ int nc;
+ sys_mutex_lock(stats_lock);
if (n_connections < policy->max_connection_limit) {
// connection counted and allowed
- n_connections += 1;
- qd_log(policy->log_source, QD_LOG_TRACE, "ALLOW Connection '%s' based
on global connection count. nConnections= %d", hostname, n_connections);
+ n_connections++;
+ n_processed++;
+ nc = n_connections;
+ sys_mutex_unlock(stats_lock);
+ qd_log(policy->log_source, QD_LOG_TRACE, "ALLOW Connection '%s' based
on global connection count. nConnections= %d", hostname, nc);
} else {
// connection denied
result = false;
- n_denied += 1;
- qd_log(policy->log_source, QD_LOG_INFO, "DENY Connection '%s' based on
global connection count. nConnections= %d", hostname, n_connections);
+ n_denied++;
+ n_total_denials++;
+ n_processed++;
+ nc = n_connections;
+ sys_mutex_unlock(stats_lock);
+ qd_log(policy->log_source, QD_LOG_INFO, "DENY Connection '%s' based on
global connection count. nConnections= %d", hostname, nc);
}
- n_processed += 1;
return result;
}
@@ -248,8 +275,10 @@ bool qd_policy_socket_accept(qd_policy_t *policy, const
char *hostname)
//
void qd_policy_socket_close(qd_policy_t *policy, const qd_connection_t *conn)
{
- n_connections -= 1;
+ sys_mutex_lock(stats_lock);
+ n_connections--;
assert (n_connections >= 0);
+ sys_mutex_unlock(stats_lock);
if (policy->enableVhostPolicy) {
// HACK ALERT: TODO: This should be deferred to a Python thread
qd_python_lock_state_t lock_state = qd_python_lock();
@@ -530,6 +559,9 @@ void qd_policy_deny_amqp_session(pn_session_t *ssn,
qd_connection_t *qd_conn)
(void) pn_condition_set_name( cond,
QD_AMQP_COND_RESOURCE_LIMIT_EXCEEDED);
(void) pn_condition_set_description(cond, SESSION_DISALLOWED);
pn_session_close(ssn);
+ sys_mutex_lock(stats_lock);
+ n_total_denials++;
+ sys_mutex_unlock(stats_lock);
if (qd_conn->policy_settings->denialCounts) {
qd_conn->policy_settings->denialCounts->sessionDenied++;
}
@@ -592,6 +624,10 @@ void _qd_policy_deny_amqp_link(pn_link_t *link,
qd_connection_t *qd_conn, const
(void) pn_condition_set_name( cond, condition);
(void) pn_condition_set_description(cond, LINK_DISALLOWED);
pn_link_close(link);
+ sys_mutex_lock(stats_lock);
+ n_links_denied++;
+ n_total_denials++;
+ sys_mutex_unlock(stats_lock);
}
diff --git a/tests/system_tests_policy.py b/tests/system_tests_policy.py
index 982ca75..d5d6159 100644
--- a/tests/system_tests_policy.py
+++ b/tests/system_tests_policy.py
@@ -38,7 +38,7 @@ from test_broker import FakeBroker
class AbsoluteConnectionCountLimit(TestCase):
"""
- Verify that connections beyond the absolute limit are denied
+ Verify that connections beyond the absolute limit are denied and counted
"""
@classmethod
def setUpClass(cls):
@@ -55,6 +55,18 @@ class AbsoluteConnectionCountLimit(TestCase):
def address(self):
return self.router.addresses[0]
+ def run_qdmanage(self, cmd, input=None, expect=Process.EXIT_OK):
+ p = self.popen(
+ ['qdmanage'] + cmd.split(' ') + ['--bus', re.sub(r'amqp://',
'amqp://u1:password@', self.address()), '--indent=-1', '--timeout',
str(TIMEOUT)],
+ stdin=PIPE, stdout=PIPE, stderr=STDOUT, expect=expect,
+ universal_newlines=True)
+ out = p.communicate(input)[0]
+ try:
+ p.teardown()
+ except Exception as e:
+ raise Exception("%s\n%s" % (e, out))
+ return out
+
def test_verify_maximum_connections(self):
addr = self.address()
@@ -80,6 +92,10 @@ class AbsoluteConnectionCountLimit(TestCase):
bc1.close()
bc2.close()
+ policystats = json.loads(self.run_qdmanage('query --type=policy'))
+ self.assertTrue(policystats[0]["connectionsDenied"] == 1)
+ self.assertTrue(policystats[0]["totalDenials"] == 1)
+
class LoadPolicyFromFolder(TestCase):
"""
Verify that specifying a policy folder from the router conf file
@@ -263,9 +279,7 @@ class LoadPolicyFromFolder(TestCase):
class SenderReceiverLimits(TestCase):
"""
- Verify that specifying a policy folder from the router conf file
- effects loading the policies in that folder.
- This test relies on qdmanage utility.
+ Verify that policy can limit senders and receivers by count.
"""
@classmethod
def setUpClass(cls):
@@ -312,11 +326,12 @@ class SenderReceiverLimits(TestCase):
bs1.close()
-
class PolicyVhostOverride(TestCase):
"""
- Verify that specifying a policy folder from the router conf file
- effects loading the policies in that folder.
+ Verify that listener policyVhost can override normally discovered vhost.
+ Verify that specific vhost and global denial counts are propagated.
+ This test conveniently forces the vhost denial statistics to be
+ on a named vhost and we know where to find them.
This test relies on qdmanage utility.
"""
@classmethod
@@ -335,6 +350,18 @@ class PolicyVhostOverride(TestCase):
def address(self):
return self.router.addresses[0]
+ def run_qdmanage(self, cmd, input=None, expect=Process.EXIT_OK):
+ p = self.popen(
+ ['qdmanage'] + cmd.split(' ') + ['--bus', re.sub(r'amqp://',
'amqp://u1:password@', self.address()), '--indent=-1', '--timeout',
str(TIMEOUT)],
+ stdin=PIPE, stdout=PIPE, stderr=STDOUT, expect=expect,
+ universal_newlines=True)
+ out = p.communicate(input)[0]
+ try:
+ p.teardown()
+ except Exception as e:
+ raise Exception("%s\n%s" % (e, out))
+ return out
+
def test_verify_n_receivers(self):
n = 4
addr = self.address()
@@ -352,6 +379,20 @@ class PolicyVhostOverride(TestCase):
br1.close()
+ vhoststats = json.loads(self.run_qdmanage('query --type=vhostStats'))
+ foundStat = False
+ for vhs in vhoststats:
+ if vhs["id"] == "override.host.com":
+ foundStat = True
+ self.assertTrue(vhs["senderDenied"] == 0)
+ self.assertTrue(vhs["receiverDenied"] == 1)
+ break
+ self.assertTrue(foundStat, msg="did not find virtual host id
'override.host.com' in stats")
+
+ policystats = json.loads(self.run_qdmanage('query --type=policy'))
+ self.assertTrue(policystats[0]["linksDenied"] == 1)
+ self.assertTrue(policystats[0]["totalDenials"] == 1)
+
def test_verify_n_senders(self):
n = 2
addr = self.address()
@@ -366,6 +407,20 @@ class PolicyVhostOverride(TestCase):
bs1.close()
+ vhoststats = json.loads(self.run_qdmanage('query --type=vhostStats'))
+ foundStat = False
+ for vhs in vhoststats:
+ if vhs["id"] == "override.host.com":
+ foundStat = True
+ self.assertTrue(vhs["senderDenied"] == 1)
+ self.assertTrue(vhs["receiverDenied"] == 1)
+ break
+ self.assertTrue(foundStat, msg="did not find virtual host id
'override.host.com' in stats")
+
+ policystats = json.loads(self.run_qdmanage('query --type=policy'))
+ self.assertTrue(policystats[0]["linksDenied"] == 2)
+ self.assertTrue(policystats[0]["totalDenials"] == 2)
+
class Capabilities(ReceiverOption):
def __init__(self, value):
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
