Merge pull request ceph#59780 from irq0/wip/osd-asok-messenger-dump

Add Asok Command: Dump Messenger Status (Connections, TCP stats, ..)

Author: irq0

PendingReleaseNotes

@@ -98,6 +98,8 @@
   an empty string can be assigned to it. Additionally, commands
   `ceph fs subvolume earmark set`, `ceph fs subvolume earmark get` and
   `ceph fs subvolume earmark rm` have been added to set, get and remove earmark from a given subvolume.
+* RADOS: Add ``messenger dump`` command to retrieve runtime information
+  on connections, sockets, and kernel TCP stats from the messenger.
 
 * RADOS: A performance botteneck in the balancer mgr module has been fixed.
   Related Tracker: https://tracker.ceph.com/issues/68657

doc/rados/operations/monitoring.rst

@@ -655,3 +655,86 @@ Runtime`_.
 
 .. _Viewing a Configuration at Runtime: ../../configuration/ceph-conf#viewing-a-configuration-at-runtime
 .. _Storage Capacity: ../../configuration/mon-config-ref#storage-capacity
+
+Messenger Status
+=================
+
+Ceph daemons and librados clients support an admin socket command
+``messenger dump`` that surfaces a snapshot of runtime information
+about connections, sockets, bound addresses, and kernel TCP stats (via
+tcp(7) TCP_INFO).
+
+.. note:: The queried messenger needs to lock the connection data
+  structures for the time it takes to create the snapshot. This lock's
+  duration is in the order of tens of milliseconds. This might
+  interfere with normal operation. Use the ``dumpcontents`` argument
+  to limit data structures dumped.
+
+Examples
+---------
+
+When a command is issued without specifying a messenger to dump, the
+list of available messengers is returned:
+
+.. prompt:: bash $
+
+   ceph tell osd.0 messenger dump
+
+.. code-block:: javascript
+
+ {
+    "messengers": [
+        "client",
+        "cluster",
+        "hb_back_client",
+        "hb_back_server",
+        "hb_front_client",
+        "hb_front_server",
+        "ms_objecter",
+        "temp_mon_client"
+    ]
+  }
+
+The ``client`` and ``cluster`` messengers correspond to the configured
+client / cluster network (see :doc:`/rados/configuration/network-config-ref`). Messengers
+with ``hb_`` prefix are part of the heartbeat system.
+
+List all current connections on the client messenger:
+
+.. code-block:: bash
+
+	    ceph tell osd.0 messenger dump client \
+	       | jq -r '.messenger.connections[].async_connection |
+	                   [.conn_id, .socket_fd, .worker_id,
+			    if .status.connected then "connected" else "disconnected" end,
+			    .state,
+			    "\(.peer.type).\(.peer.entity_name.id).\(.peer.id)",
+			    .protocol.v2.con_mode, .protocol.v2.crypto.rx, .protocol.v2.compression.rx] |
+			   @tsv'
+
+.. code-block:: bash
+
+  249     102     0       connected       STATE_CONNECTION_ESTABLISHED    client.admin.6407       crc    PLAIN   UNCOMPRESSED
+  242     99      1       connected       STATE_CONNECTION_ESTABLISHED    client.rgw.8000.4473    crc     PLAIN   UNCOMPRESSED
+  248     89      1       connected       STATE_CONNECTION_ESTABLISHED    mgr..-1 secure  AES-128-GCM     UNCOMPRESSED
+  32      101     2       connected       STATE_CONNECTION_ESTABLISHED    client.rgw.8000.4483    crc     PLAIN   UNCOMPRESSED
+  3       86      2       connected       STATE_CONNECTION_ESTABLISHED    mon..-1 secure  AES-128-GCM     UNCOMPRESSED
+  244     102     0       connected       STATE_CONNECTION_ESTABLISHED    client.admin.6383       crc     PLAIN   UNCOMPRESSED
+
+
+Print active connections and their TCP round trip time and retransmission counters:
+
+.. code-block:: bash
+
+	    ceph tell osd.0 messenger dump client --tcp-info \
+	       | jq -r '.messenger.connections[].async_connection |
+	                   select(.status.connected) |
+			   select(.peer.type != "client") |
+			   [.conn_id, .socket_fd, .worker_id,
+			    "\(.peer.type).\(.peer.global_id)",
+			    .tcp_info.tcpi_rtt_us, .tcp_info.tcpi_rttvar_us, .tcp_info.tcpi_total_retrans] |
+			    @tsv'
+.. code-block:: bash
+
+	248     89      1       mgr.0   863     1677    0
+	3       86      2       mon.0   230     278     0

qa/workunits/cephtool/test.sh

@@ -2900,6 +2900,51 @@ function test_per_pool_scrub_status()
   ceph osd pool rm noscrub_pool2 noscrub_pool2 --yes-i-really-really-mean-it
 }
 
+function do_messenger_dump_basics_test()
+{
+  local target="$1"
+  ceph tell "$target" messenger dump | expect_true jq --exit-status '.messengers | length > 0'
+  ceph tell "$target" messenger dump | jq -r '.messengers[]' | while read messenger; do
+    dump="$(ceph tell "$target" messenger dump "$messenger" all)"
+    expect_true jq --exit-status 'has("messenger")' <<< "$dump"
+    expect_true jq --exit-status 'has("name")' <<< "$dump"
+    expect_true jq --arg expected_messenger "$messenger" --exit-status '
+	 .name == $expected_messenger' <<< "$dump"
+    expect_true jq --exit-status '.messenger | type == "object"' <<< "$dump"
+    expect_true jq --exit-status '.messenger |
+        all([.connections,
+             .listen_sockets,
+	     .anon_conns,
+	     .accepting_conns,
+	     .deleted_conns][];
+            type == "array")' \
+		<<< "$dump"
+  done
+}
+
+function test_osd_messenger_dump()
+{
+  do_messenger_dump_basics_test osd.0
+}
+function test_mon_messenger_dump()
+{
+  do_messenger_dump_basics_test mon.a
+  # Testing the tcp_info feature requires at lease one messenger TCP
+  # conneciton. Test only the mon as it is very unlikely that it
+  # doesn't have an active connection. Also only test for one
+  # connection, as disconnected connections don't set tcp_info
+  expect_true ceph tell "$target" messenger dump mon --tcp-info \
+    | jq 'any(.messenger.connections[].async_connection.tcp_info; has("tcpi_state"))'
+}
+function test_mgr_messenger_dump()
+{
+  do_messenger_dump_basics_test mgr
+}
+function test_mds_messenger_dump()
+{
+  do_messenger_dump_basics_test mds.a
+}
+
 #
 # New tests should be added to the TESTS array below
 #
@@ -2944,6 +2989,7 @@ MON_TESTS+=" mon_caps"
 MON_TESTS+=" mon_cephdf_commands"
 MON_TESTS+=" mon_tell_help_command"
 MON_TESTS+=" mon_stdin_stdout"
+MON_TESTS+=" mon_messenger_dump"
 
 OSD_TESTS+=" osd_bench"
 OSD_TESTS+=" osd_negative_filestore_merge_threshold"
@@ -2952,14 +2998,17 @@ OSD_TESTS+=" admin_heap_profiler"
 OSD_TESTS+=" osd_tell_help_command"
 OSD_TESTS+=" osd_compact"
 OSD_TESTS+=" per_pool_scrub_status"
+OSD_TESTS+=" osd_messenger_dump"
 
 MDS_TESTS+=" mds_tell"
 MDS_TESTS+=" mon_mds"
 MDS_TESTS+=" mon_mds_metadata"
 MDS_TESTS+=" mds_tell_help_command"
+MDS_TESTS+=" mds_messenger_dump"
 
 MGR_TESTS+=" mgr_tell"
 MGR_TESTS+=" mgr_devices"
+MGR_TESTS+=" mgr_messenger_dump"
 
 TESTS+=$MON_TESTS
 TESTS+=$OSD_TESTS

src/common/CMakeLists.txt

@@ -104,7 +104,8 @@ set(common_srcs
   pretty_binary.cc
   utf8.c
   util.cc
-  version.cc)
+  version.cc
+  tcp_info.cc)
 
 if(WITH_SYSTEMD)
   list(APPEND common_srcs

src/common/ceph_context.cc

@@ -713,6 +713,7 @@ CephContext::CephContext(uint32_t module_type_,
 #ifdef CEPH_DEBUG_MUTEX
     _lockdep_obs(NULL),
 #endif
+    _msgr_hook(nullptr),
     crush_location(this)
 {
   if (options.create_log) {
@@ -775,6 +776,17 @@ CephContext::CephContext(uint32_t module_type_,
   lookup_or_create_singleton_object<MempoolObs>("mempool_obs", false, this);
 }
 
+void CephContext::modify_msgr_hook(
+    std::function<AdminSocketHook*(void)> create,
+    std::function<void(AdminSocketHook*)> add) {
+  std::lock_guard l{_msgr_hook_lock};
+  if (_msgr_hook) {
+    add(_msgr_hook.get());
+  } else {
+    _msgr_hook.reset(create());
+  }
+}
+
 CephContext::~CephContext()
 {
   associated_objs.clear();
@@ -788,6 +800,9 @@ CephContext::~CephContext()
 
   delete _plugin_registry;
 
+  if (_msgr_hook) {
+    _admin_socket->unregister_commands(_msgr_hook.get());
+  }
   _admin_socket->unregister_commands(_admin_hook);
   delete _admin_hook;
   delete _admin_socket;

src/common/ceph_context.h

@@ -47,6 +47,7 @@
 #include "crush/CrushLocation.h"
 
 class AdminSocket;
+class AdminSocketHook;
 class CryptoHandler;
 class CryptoRandom;
 class MonMap;
@@ -381,8 +382,13 @@ class CephContext {
 #ifdef CEPH_DEBUG_MUTEX
   md_config_obs_t *_lockdep_obs;
 #endif
+
+  std::unique_ptr<AdminSocketHook> _msgr_hook;
+  ceph::mutex _msgr_hook_lock = ceph::make_mutex("CephContext::msgr_hook");
 public:
   TOPNSPC::crush::CrushLocation crush_location;
+  void modify_msgr_hook(std::function<AdminSocketHook*(void)> create,
+			std::function<void(AdminSocketHook*)> add);
 private:
 
   enum {

src/common/tcp_info.cc

@@ -0,0 +1,121 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2024 Clyso GmbH
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#include "common/tcp_info.h"
+
+#include "common/Formatter.h"
+
+namespace ceph {
+
+#ifdef _WIN32
+struct tcp_info {};
+
+bool tcp_info(int fd, struct tcp_info& info) {
+  return false;
+}
+bool dump_tcp_info(int fd, Formatter* f) {
+  return false;
+}
+
+#else
+
+bool tcp_info(int fd, struct tcp_info& info) {
+  socklen_t info_len = sizeof(info);
+  return (getsockopt(fd, SOL_TCP, TCP_INFO, &info, &info_len) == 0);
+}
+
+static const char* get_tcpi_state_name(uint8_t state) {
+  switch (state) {
+    case TCP_ESTABLISHED:
+      return "established";
+    case TCP_SYN_SENT:
+      return "syn sent";
+    case TCP_SYN_RECV:
+      return "syn recv";
+    case TCP_FIN_WAIT1:
+      return "fin wait1";
+    case TCP_FIN_WAIT2:
+      return "fin wait2";
+    case TCP_TIME_WAIT:
+      return "time wait";
+    case TCP_CLOSE:
+      return "close";
+    case TCP_CLOSE_WAIT:
+      return "close wait";
+    case TCP_LAST_ACK:
+      return "last ack";
+    case TCP_LISTEN:
+      return "listen";
+    case TCP_CLOSING:
+      return "closing";
+    default:
+      return "UNKNOWN";
+  }
+}
+
+bool dump_tcp_info(int fd, Formatter* f) {
+  struct tcp_info info;
+  if (!tcp_info(fd, info)) {
+    return false;
+  }
+
+  f->open_object_section("tcp_info");
+  f->dump_string("tcpi_state", get_tcpi_state_name(info.tcpi_state));
+  f->dump_unsigned("tcpi_retransmits", info.tcpi_retransmits);
+  f->dump_unsigned("tcpi_probes", info.tcpi_probes);
+  f->dump_unsigned("tcpi_backoff", info.tcpi_backoff);
+  f->dump_unsigned("tcpi_rto_us", info.tcpi_rto);
+  f->dump_unsigned("tcpi_ato_us", info.tcpi_ato);
+  f->dump_unsigned("tcpi_snd_mss", info.tcpi_snd_mss);
+  f->dump_unsigned("tcpi_rcv_mss", info.tcpi_rcv_mss);
+  f->dump_unsigned("tcpi_unacked", info.tcpi_unacked);
+  f->dump_unsigned("tcpi_lost", info.tcpi_lost);
+  f->dump_unsigned("tcpi_retrans", info.tcpi_retrans);
+  f->dump_unsigned("tcpi_pmtu", info.tcpi_pmtu);
+  f->dump_unsigned("tcpi_rtt_us", info.tcpi_rtt);
+  f->dump_unsigned("tcpi_rttvar_us", info.tcpi_rttvar);
+  f->dump_unsigned("tcpi_total_retrans", info.tcpi_total_retrans);
+  f->dump_unsigned("tcpi_last_data_sent_ms", info.tcpi_last_data_sent);
+  f->dump_unsigned("tcpi_last_ack_sent_ms", info.tcpi_last_ack_sent);
+  f->dump_unsigned("tcpi_last_data_recv_ms", info.tcpi_last_data_recv);
+  f->dump_unsigned("tcpi_last_ack_recv_ms", info.tcpi_last_ack_recv);
+
+  f->open_array_section("tcpi_options");
+  if (info.tcpi_options & TCPI_OPT_TIMESTAMPS) {
+    f->dump_string("option", "timestamps");
+  }
+  if (info.tcpi_options & TCPI_OPT_SACK) {
+    f->dump_string("option", "sack");
+  }
+  if (info.tcpi_options & TCPI_OPT_WSCALE) {
+    f->dump_string("option", "wscale");
+  }
+  if (info.tcpi_options & TCPI_OPT_ECN) {
+    f->dump_string("option", "ecn");
+  }
+  if (info.tcpi_options & TCPI_OPT_ECN_SEEN) {
+    f->dump_string("option", "ecn seen");
+  }
+  if (info.tcpi_options & TCPI_OPT_SYN_DATA) {
+    f->dump_string("option", "syn data");
+  }
+  f->close_section();
+
+  f->close_section();
+  return true;
+}
+
+#endif
+
+}  // namespace ceph

src/common/tcp_info.h

@@ -0,0 +1,30 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2024 Clyso GmbH
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#pragma once
+
+#include <netinet/tcp.h>
+#include <sys/socket.h>
+
+#include "Formatter.h"
+
+namespace ceph {
+
+/// Return TCP_INFO socket stats (see tcp(7)). Return true on success.
+bool tcp_info(int fd, struct tcp_info& info);
+/// Dump TCP_INFO socket stats to formatter. Use struct tcp_info variables
+/// names as keys. Returns true on success.
+bool dump_tcp_info(int fd, Formatter* f);
+
+}  // namespace ceph