From c762e2247d3248556996b8c262e5d790f9123c65 Mon Sep 17 00:00:00 2001 From: Alexey Milovidov Date: Sun, 4 Jul 2021 00:00:50 +0300 Subject: [PATCH 1/2] More instrumentation for network interaction: add counters for recv/send bytes; add gauges for recvs/sends. --- src/Common/CurrentMetrics.cpp | 2 ++ src/Common/ProfileEvents.cpp | 6 ++++-- src/IO/ReadBufferFromPocoSocket.cpp | 11 +++++++++++ src/IO/WriteBufferFromPocoSocket.cpp | 10 ++++++++++ 4 files changed, 27 insertions(+), 2 deletions(-) diff --git a/src/Common/CurrentMetrics.cpp b/src/Common/CurrentMetrics.cpp index 1e482361f85..e9fa13e11e6 100644 --- a/src/Common/CurrentMetrics.cpp +++ b/src/Common/CurrentMetrics.cpp @@ -30,6 +30,8 @@ M(OpenFileForWrite, "Number of files open for writing") \ M(Read, "Number of read (read, pread, io_getevents, etc.) syscalls in fly") \ M(Write, "Number of write (write, pwrite, io_getevents, etc.) syscalls in fly") \ + M(NetworkReceive, "Number of threads receiving data from network. Only ClickHouse-related network interaction is included, not by 3rd party libraries.") \ + M(NetworkSend, "Number of threads sending data to network. Only ClickHouse-related network interaction is included, not by 3rd party libraries.") \ M(SendScalars, "Number of connections that are sending data for scalars to remote servers.") \ M(SendExternalTables, "Number of connections that are sending data for external tables to remote servers. External tables are used to implement GLOBAL IN and GLOBAL JOIN operators with distributed subqueries.") \ M(QueryThread, "Number of query processing threads") \ diff --git a/src/Common/ProfileEvents.cpp b/src/Common/ProfileEvents.cpp index 86f06f27455..dffe2239e62 100644 --- a/src/Common/ProfileEvents.cpp +++ b/src/Common/ProfileEvents.cpp @@ -49,8 +49,10 @@ M(CreatedReadBufferMMapFailed, "") \ M(DiskReadElapsedMicroseconds, "Total time spent waiting for read syscall. This include reads from page cache.") \ M(DiskWriteElapsedMicroseconds, "Total time spent waiting for write syscall. This include writes to page cache.") \ - M(NetworkReceiveElapsedMicroseconds, "") \ - M(NetworkSendElapsedMicroseconds, "") \ + M(NetworkReceiveElapsedMicroseconds, "Total time spent waiting for data to receive or receiving data from network. Only ClickHouse-related network interaction is included, not by 3rd party libraries.") \ + M(NetworkSendElapsedMicroseconds, "Total time spent waiting for data to send to network or sending data to network. Only ClickHouse-related network interaction is included, not by 3rd party libraries..") \ + M(NetworkReceiveBytes, "Total number of bytes received from network. Only ClickHouse-related network interaction is included, not by 3rd party libraries.") \ + M(NetworkSendBytes, "Total number of bytes send to network. Only ClickHouse-related network interaction is included, not by 3rd party libraries.") \ M(ThrottlerSleepMicroseconds, "Total time a query was sleeping to conform the 'max_network_bandwidth' setting.") \ \ M(QueryMaskingRulesMatch, "Number of times query masking rules was successfully matched.") \ diff --git a/src/IO/ReadBufferFromPocoSocket.cpp b/src/IO/ReadBufferFromPocoSocket.cpp index e043764d280..5e8e41d0c3e 100644 --- a/src/IO/ReadBufferFromPocoSocket.cpp +++ b/src/IO/ReadBufferFromPocoSocket.cpp @@ -5,11 +5,19 @@ #include #include #include +#include +#include namespace ProfileEvents { extern const Event NetworkReceiveElapsedMicroseconds; + extern const Event NetworkReceiveBytes; +} + +namespace CurrentMetrics +{ + extern const Metric NetworkReceive; } @@ -31,6 +39,8 @@ bool ReadBufferFromPocoSocket::nextImpl() /// Add more details to exceptions. try { + CurrentMetrics::Increment metric_increment(CurrentMetrics::NetworkReceive); + /// If async_callback is specified, and read will block, run async_callback and try again later. /// It is expected that file descriptor may be polled externally. /// Note that receive timeout is not checked here. External code should check it while polling. @@ -57,6 +67,7 @@ bool ReadBufferFromPocoSocket::nextImpl() /// NOTE: it is quite inaccurate on high loads since the thread could be replaced by another one ProfileEvents::increment(ProfileEvents::NetworkReceiveElapsedMicroseconds, watch.elapsedMicroseconds()); + ProfileEvents::increment(ProfileEvents::NetworkReceiveBytes, bytes_read); if (bytes_read) working_buffer.resize(bytes_read); diff --git a/src/IO/WriteBufferFromPocoSocket.cpp b/src/IO/WriteBufferFromPocoSocket.cpp index 78705857ec4..a0e4de4c831 100644 --- a/src/IO/WriteBufferFromPocoSocket.cpp +++ b/src/IO/WriteBufferFromPocoSocket.cpp @@ -6,11 +6,19 @@ #include #include #include +#include +#include namespace ProfileEvents { extern const Event NetworkSendElapsedMicroseconds; + extern const Event NetworkSendBytes; +} + +namespace CurrentMetrics +{ + extern const Metric NetworkSend; } @@ -40,6 +48,7 @@ void WriteBufferFromPocoSocket::nextImpl() /// Add more details to exceptions. try { + CurrentMetrics::Increment metric_increment(CurrentMetrics::NetworkSend); res = socket.impl()->sendBytes(working_buffer.begin() + bytes_written, offset() - bytes_written); } catch (const Poco::Net::NetException & e) @@ -62,6 +71,7 @@ void WriteBufferFromPocoSocket::nextImpl() } ProfileEvents::increment(ProfileEvents::NetworkSendElapsedMicroseconds, watch.elapsedMicroseconds()); + ProfileEvents::increment(ProfileEvents::NetworkSendBytes, bytes_written); } WriteBufferFromPocoSocket::WriteBufferFromPocoSocket(Poco::Net::Socket & socket_, size_t buf_size) From 1960c717ed9bfe9288cfae0180f32112e900fe48 Mon Sep 17 00:00:00 2001 From: Alexey Milovidov Date: Sun, 4 Jul 2021 01:59:28 +0300 Subject: [PATCH 2/2] A couple of tests --- ...01939_network_receive_bytes_metrics.reference | 1 + .../01939_network_receive_bytes_metrics.sh | 16 ++++++++++++++++ .../01939_network_send_bytes_metrics.reference | 1 + .../01939_network_send_bytes_metrics.sh | 16 ++++++++++++++++ 4 files changed, 34 insertions(+) create mode 100644 tests/queries/0_stateless/01939_network_receive_bytes_metrics.reference create mode 100755 tests/queries/0_stateless/01939_network_receive_bytes_metrics.sh create mode 100644 tests/queries/0_stateless/01939_network_send_bytes_metrics.reference create mode 100755 tests/queries/0_stateless/01939_network_send_bytes_metrics.sh diff --git a/tests/queries/0_stateless/01939_network_receive_bytes_metrics.reference b/tests/queries/0_stateless/01939_network_receive_bytes_metrics.reference new file mode 100644 index 00000000000..d00491fd7e5 --- /dev/null +++ b/tests/queries/0_stateless/01939_network_receive_bytes_metrics.reference @@ -0,0 +1 @@ +1 diff --git a/tests/queries/0_stateless/01939_network_receive_bytes_metrics.sh b/tests/queries/0_stateless/01939_network_receive_bytes_metrics.sh new file mode 100755 index 00000000000..03babad40f3 --- /dev/null +++ b/tests/queries/0_stateless/01939_network_receive_bytes_metrics.sh @@ -0,0 +1,16 @@ +#!/usr/bin/env bash + +CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) +# shellcheck source=../shell_config.sh +. "$CURDIR"/../shell_config.sh + +${CLICKHOUSE_CLIENT} --multiquery --query "DROP TABLE IF EXISTS t; CREATE TABLE t (x UInt64) ENGINE = Memory;" + +seq 1 1000 | ${CLICKHOUSE_CLIENT} --query "INSERT INTO t FORMAT TSV" + +${CLICKHOUSE_CLIENT} --multiquery --query "SYSTEM FLUSH LOGS; + WITH ProfileEvents['NetworkReceiveBytes'] AS bytes + SELECT bytes >= 8000 AND bytes < 9000 ? 1 : bytes FROM system.query_log + WHERE current_database = currentDatabase() AND query_kind = 'Insert' AND event_date >= yesterday() AND type = 2 ORDER BY event_time DESC LIMIT 1;" + +${CLICKHOUSE_CLIENT} --query "DROP TABLE t" diff --git a/tests/queries/0_stateless/01939_network_send_bytes_metrics.reference b/tests/queries/0_stateless/01939_network_send_bytes_metrics.reference new file mode 100644 index 00000000000..d00491fd7e5 --- /dev/null +++ b/tests/queries/0_stateless/01939_network_send_bytes_metrics.reference @@ -0,0 +1 @@ +1 diff --git a/tests/queries/0_stateless/01939_network_send_bytes_metrics.sh b/tests/queries/0_stateless/01939_network_send_bytes_metrics.sh new file mode 100755 index 00000000000..e862a273de4 --- /dev/null +++ b/tests/queries/0_stateless/01939_network_send_bytes_metrics.sh @@ -0,0 +1,16 @@ +#!/usr/bin/env bash + +CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) +# shellcheck source=../shell_config.sh +. "$CURDIR"/../shell_config.sh + +${CLICKHOUSE_CLIENT} --multiquery --query "DROP TABLE IF EXISTS t; CREATE TABLE t (x UInt64) ENGINE = Memory;" + +${CLICKHOUSE_CLIENT} --query "SELECT number FROM numbers(1000)" > /dev/null + +${CLICKHOUSE_CLIENT} --multiquery --query "SYSTEM FLUSH LOGS; + WITH ProfileEvents['NetworkSendBytes'] AS bytes + SELECT bytes >= 8000 AND bytes < 9000 ? 1 : bytes FROM system.query_log + WHERE current_database = currentDatabase() AND query_kind = 'Select' AND event_date >= yesterday() AND type = 2 ORDER BY event_time DESC LIMIT 1;" + +${CLICKHOUSE_CLIENT} --query "DROP TABLE t"