You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by gm...@apache.org on 2019/12/11 21:07:40 UTC
[qpid-dispatch] branch master updated: DISPATCH-1514 - Dynamically
turn on proton frame trace on existing connections. This closes #641
This is an automated email from the ASF dual-hosted git repository.
gmurthy 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 2417e17 DISPATCH-1514 - Dynamically turn on proton frame trace on existing connections. This closes #641
2417e17 is described below
commit 2417e17ddd5c5fc9667c9d9f5170f9bcb38c67a2
Author: Ganesh Murthy <gm...@apache.org>
AuthorDate: Tue Dec 10 15:42:30 2019 -0500
DISPATCH-1514 - Dynamically turn on proton frame trace on existing connections. This closes #641
---
src/dispatch.c | 9 +-
src/dispatch_private.h | 4 +
src/log.c | 15 +++
src/server.c | 22 ++++
src/server_private.h | 5 +
tests/CMakeLists.txt | 1 +
tests/system_test.py | 10 ++
tests/system_tests_log_level_update.py | 196 +++++++++++++++++++++++++++++++++
8 files changed, 261 insertions(+), 1 deletion(-)
diff --git a/src/dispatch.c b/src/dispatch.c
index 505db4c..c3f75c5 100644
--- a/src/dispatch.c
+++ b/src/dispatch.c
@@ -60,6 +60,13 @@ const char *MULTICAST_DISTRIBUTION = "multicast";
const char *BALANCED_DISTRIBUTION = "balanced";
const char *UNAVAILABLE_DISTRIBUTION = "unavailable";
+qd_dispatch_t *qd = 0;
+
+qd_dispatch_t *qd_dispatch_get_dispatch()
+{
+ return qd;
+}
+
qd_dispatch_t *qd_dispatch(const char *python_pkgdir, bool test_hooks)
{
//
@@ -69,7 +76,7 @@ qd_dispatch_t *qd_dispatch(const char *python_pkgdir, bool test_hooks)
gettimeofday(&time, NULL);
srandom((unsigned int)time.tv_sec + ((unsigned int)time.tv_usec << 11));
- qd_dispatch_t *qd = NEW(qd_dispatch_t);
+ qd = NEW(qd_dispatch_t);
ZERO(qd);
qd_entity_cache_initialize(); /* Must be first */
diff --git a/src/dispatch_private.h b/src/dispatch_private.h
index 741d34d..440e633 100644
--- a/src/dispatch_private.h
+++ b/src/dispatch_private.h
@@ -63,6 +63,10 @@ struct qd_dispatch_t {
char *metadata;
};
+qd_dispatch_t *qd_dispatch_get_dispatch();
+
+
+
/**
* Configure the router node from a configuration entity.
* If this is not called, the router will run in ENDPOINT mode.
diff --git a/src/log.c b/src/log.c
index 7ecea72..41e36ad 100644
--- a/src/log.c
+++ b/src/log.c
@@ -20,6 +20,7 @@
#include "python_private.h" // must be first!
#include "log_private.h"
+#include "server_private.h"
#include "entity.h"
#include "entity_cache.h"
#include "aprintf.h"
@@ -514,6 +515,7 @@ qd_error_t qd_log_entity(qd_entity_t *entity) {
char* module = 0;
char *output = 0;
char *enable = 0;
+ bool trace_enabled = false;
do {
@@ -541,7 +543,12 @@ qd_error_t qd_log_entity(qd_entity_t *entity) {
if (qd_entity_has(entity, "enable")) {
enable = qd_entity_get_string(entity, "enable");
QD_ERROR_BREAK();
+
src->mask = enable_mask(enable);
+
+ if (qd_log_enabled(src, QD_LOG_TRACE)) {
+ trace_enabled = true;
+ }
}
QD_ERROR_BREAK();
@@ -563,6 +570,14 @@ qd_error_t qd_log_entity(qd_entity_t *entity) {
sys_mutex_unlock(log_source_lock);
+ //
+ // If trace logging is enabled, loop thru all connections in the router and call the pn_transport_set_tracer callback
+ // so proton frame trace can be output as part of the router trace log.
+ //
+ if (trace_enabled) {
+ qd_server_trace_all_connections();
+ }
+
return qd_error_code();
}
diff --git a/src/server.c b/src/server.c
index 69ed73f..dcb61b1 100644
--- a/src/server.c
+++ b/src/server.c
@@ -18,6 +18,7 @@
*/
#include "python_private.h" // must be first!
+#include "dispatch_private.h"
#include <qpid/dispatch/python_embedded.h>
#include <qpid/dispatch/ctools.h>
@@ -1276,6 +1277,27 @@ void qd_server_set_container(qd_dispatch_t *qd, qd_container_t *container)
qd->server->container = container;
}
+void qd_server_trace_all_connections()
+{
+ qd_dispatch_t *qd = qd_dispatch_get_dispatch();
+ if (qd->server) {
+ sys_mutex_lock(qd->server->lock);
+ qd_connection_list_t conn_list = qd->server->conn_list;
+ qd_connection_t *conn = DEQ_HEAD(conn_list);
+ while(conn) {
+ //
+ // If there is already a tracer on the transport, nothing to do, move on to the next connection.
+ //
+ pn_transport_t *tport = pn_connection_transport(conn->pn_conn);
+ if (! pn_transport_get_tracer(tport)) {
+ pn_transport_trace(tport, PN_TRACE_FRM);
+ pn_transport_set_tracer(tport, transport_tracer);
+ }
+ conn = DEQ_NEXT(conn);
+ }
+ sys_mutex_unlock(qd->server->lock);
+ }
+}
void qd_server_run(qd_dispatch_t *qd)
{
diff --git a/src/server_private.h b/src/server_private.h
index 7ce6dc6..3fe836d 100644
--- a/src/server_private.h
+++ b/src/server_private.h
@@ -182,4 +182,9 @@ ALLOC_DECLARE(qd_connector_t);
ALLOC_DECLARE(qd_connection_t);
ALLOC_DECLARE(qd_pn_free_link_session_t);
+/**
+ * For every connection on the server's connection list, call pn_transport_set_tracer and enable trace logging
+ */
+void qd_server_trace_all_connections();
+
#endif
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index 049c520..18e6e5b 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -138,6 +138,7 @@ foreach(py_test_module
system_tests_fallback_dest
system_tests_router_mesh
system_tests_stuck_deliveries
+ system_tests_log_level_update
)
add_test(${py_test_module} ${TEST_WRAP} ${PYTHON_TEST_COMMAND} -v ${py_test_module})
diff --git a/tests/system_test.py b/tests/system_test.py
index 6626ce1..8944cea 100755
--- a/tests/system_test.py
+++ b/tests/system_test.py
@@ -1003,6 +1003,16 @@ class QdManager(object):
cmd += " %s=%s" % (k, v)
return json.loads(self(cmd))
+ def update(self, long_type, kwargs, name=None, identity=None):
+ cmd = 'UPDATE --type=%s' % long_type
+ if identity is not None:
+ cmd += " --identity=%s" % identity
+ elif name is not None:
+ cmd += " --name=%s" % name
+ for k, v in kwargs.items():
+ cmd += " %s=%s" % (k, v)
+ return json.loads(self(cmd))
+
def delete(self, long_type, name=None, identity=None):
cmd = 'DELETE --type=%s' % long_type
if identity is not None:
diff --git a/tests/system_tests_log_level_update.py b/tests/system_tests_log_level_update.py
new file mode 100644
index 0000000..b1ae588
--- /dev/null
+++ b/tests/system_tests_log_level_update.py
@@ -0,0 +1,196 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+import os
+
+from system_test import TestCase, Qdrouterd, main_module, TIMEOUT, DIR, Process
+from system_test import QdManager
+from proton.utils import BlockingConnection
+from proton import Message
+from proton.reactor import AtMostOnce
+
+apply_options = AtMostOnce()
+
+class LogLevelUpdateTest(TestCase):
+ @classmethod
+ def setUpClass(cls):
+ super(LogLevelUpdateTest, cls).setUpClass()
+ name = "test-router"
+ LogLevelUpdateTest.listen_port = cls.tester.get_port()
+ config = Qdrouterd.Config([
+ ('router', {'mode': 'standalone', 'id': 'QDR'}),
+ ('listener', {'port': LogLevelUpdateTest.listen_port}),
+
+ ('address', {'prefix': 'closest', 'distribution': 'closest'}),
+ ('address', {'prefix': 'balanced', 'distribution': 'balanced'}),
+ ('address', {'prefix': 'multicast', 'distribution': 'multicast'})
+ ])
+ cls.router = cls.tester.qdrouterd(name, config)
+ cls.router.wait_ready()
+ cls.address = cls.router.addresses[0]
+ cls.closest_count = 1
+
+ def create_sender_receiver(self, test_address, test_msg, blocking_connection=None):
+ if not blocking_connection:
+ blocking_connection = BlockingConnection(self.address)
+ blocking_receiver = blocking_connection.create_receiver(address=test_address)
+ blocking_sender = blocking_connection.create_sender(address=test_address, options=apply_options)
+ msg = Message(body=test_msg)
+ blocking_sender.send(msg)
+ received_message = blocking_receiver.receive()
+ self.assertEqual(test_msg, received_message.body)
+
+ def test_01_toggle_default_trace_logging(self):
+ hello_world_1 = "Hello World_1!"
+ hello_world_2 = "Hello World_2!"
+ hello_world_3 = "Hello World_3!"
+ hello_world_4 = "Hello World_4!"
+ qd_manager = QdManager(self, self.address)
+
+ blocking_connection = BlockingConnection(self.address)
+ TEST_ADDR = "apachetest1"
+ self.create_sender_receiver(TEST_ADDR, hello_world_1, blocking_connection)
+
+ # STEP 1: Make sure that proton trace logging is turned on already.
+ # Search for attach frames in the log for address TEST_ADDR. There should be 4 attaches
+ num_attaches = 0
+ logs = qd_manager.get_log()
+ for log in logs:
+ if u'SERVER' in log[0]:
+ if "@attach" in log[2] and TEST_ADDR in log[2]:
+ num_attaches += 1
+ # num_attaches for address TEST_ADDR must be 4, two attaches to/from sender and receiver
+ self.assertTrue(num_attaches == 4)
+
+ # STEP 2: Turn off trace logging using qdmanage
+ qd_manager.update("org.apache.qpid.dispatch.log", {"enable": "info+"}, name="log/DEFAULT")
+
+ # Step 3: Now, router trace logging is turned off (has been set to info+)
+ # Create the sender and receiver again on a different address and make
+ # sure that the attaches are NOT showing in the log for that address.
+
+ TEST_ADDR = "apachetest2"
+ self.create_sender_receiver(TEST_ADDR, hello_world_2, blocking_connection)
+
+ # STEP 3: Count the nimber of attaches for address TEST_ADDR, there should be none
+ num_attaches = 0
+ logs = qd_manager.get_log()
+ for log in logs:
+ if u'SERVER' in log[0]:
+ if "@attach" in log[2] and TEST_ADDR in log[2]:
+ num_attaches += 1
+ # There should be no attach frames with address TEST_ADDR
+ # because we turned of trace logging.
+ self.assertTrue(num_attaches == 0)
+
+
+ # STEP 4: Tuen trace logging back on again and make sure num_attaches = 4
+ TEST_ADDR = "apachetest3"
+ qd_manager.update("org.apache.qpid.dispatch.log", {"enable": "trace+"}, name="log/DEFAULT")
+ self.create_sender_receiver(TEST_ADDR, hello_world_3, blocking_connection)
+
+ # STEP 3: Count the number of attaches for address TEST_ADDR, there should be 4
+ num_attaches = 0
+ logs = qd_manager.get_log()
+ for log in logs:
+ if u'SERVER' in log[0]:
+ if "@attach" in log[2] and TEST_ADDR in log[2]:
+ num_attaches += 1
+ # There should be 4 attach frames with address TEST_ADDR
+ # because we turned on trace logging.
+ self.assertTrue(num_attaches == 4)
+
+ # Create a brand new blocking connection and make sure that connection
+ # is logging at trace level as well.
+ num_attaches = 0
+ TEST_ADDR = "apachetest4"
+ self.create_sender_receiver(TEST_ADDR, hello_world_4)
+ num_attaches = 0
+ logs = qd_manager.get_log()
+ for log in logs:
+ if u'SERVER' in log[0]:
+ if "@attach" in log[2] and TEST_ADDR in log[2]:
+ num_attaches += 1
+
+ self.assertTrue(num_attaches == 4)
+
+
+ def test_02_toggle_server_trace_logging(self):
+ """
+ This test is similar to test_01_toggle_default_trace_logging but it tests the
+ SERVER log level.
+ """
+ hello_world_5 = "Hello World_5!"
+ hello_world_6 = "Hello World_6!"
+ hello_world_7 = "Hello World_7!"
+ TEST_ADDR = "apachetest5"
+
+ # Step 1. Turn off trace logging for module DEFAULT and enable trace logging
+ # for the SERVER module and make sure it works.
+ qd_manager = QdManager(self, self.address)
+ # Set log level to info+ on the DEFAULT module
+ qd_manager.update("org.apache.qpid.dispatch.log", {"enable": "info+"}, name="log/DEFAULT")
+ # Set log level to trace+ on the SERVER module
+ qd_manager.update("org.apache.qpid.dispatch.log", {"enable": "trace+"}, name="log/SERVER")
+ blocking_connection = BlockingConnection(self.address)
+
+ self.create_sender_receiver(TEST_ADDR, hello_world_5,
+ blocking_connection)
+ # Count the nimber of attaches for address TEST_ADDR, there should be 4
+ num_attaches = 0
+ logs = qd_manager.get_log()
+ for log in logs:
+ if u'SERVER' in log[0]:
+ if "@attach" in log[2] and TEST_ADDR in log[2]:
+ num_attaches += 1
+ # There should be 4 attach frames with address TEST_ADDR
+ # because we turned on trace logging.
+ self.assertTrue(num_attaches == 4)
+
+ TEST_ADDR = "apachetest6"
+ qd_manager.update("org.apache.qpid.dispatch.log", {"enable": "info+"}, name="log/SERVER")
+
+ self.create_sender_receiver(TEST_ADDR, hello_world_6, blocking_connection)
+
+ # Count the nimber of attaches for address TEST_ADDR, there should be 0
+ num_attaches = 0
+ logs = qd_manager.get_log()
+ for log in logs:
+ if u'SERVER' in log[0]:
+ if "@attach" in log[2] and TEST_ADDR in log[2]:
+ num_attaches += 1
+ self.assertTrue(num_attaches == 0)
+
+ # Create a brand new blocking connection and make sure that connection
+ # is logging at info level as well.
+ TEST_ADDR = "apachetest7"
+ self.create_sender_receiver(TEST_ADDR, hello_world_7)
+ num_attaches = 0
+ logs = qd_manager.get_log()
+ for log in logs:
+ if u'SERVER' in log[0]:
+ if "@attach" in log[2] and TEST_ADDR in log[2]:
+ num_attaches += 1
+
+ self.assertTrue(num_attaches == 0)
+
+
+
+
+
+
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org