From 8b28cff96f5b3f8b6a3b31a67036b3e0b23dcfca Mon Sep 17 00:00:00 2001 From: Simon Pichugin Date: Sun, 19 Jul 2026 22:15:12 -0700 Subject: [PATCH] Issue 7633 - RFE - Add offline diagnostics for thread pool saturation (#7634) Description: When the worker pool is fully saturated, cn=monitor cannot be used for diagnostics because the monitor search itself needs a worker thread. Add offline thread-pool status reporting by publishing pool gauges and per-worker activity into a hardened memory-mapped file under the instance run directory. The new dsctl thread-pool-status command reads this file directly, without an LDAP connection, so admins can inspect the pool even when worker threads are exhausted. cn=monitor also exposes a sanitized threadpoolworker attribute backed by the same data source. The feature is enabled by default and can be disabled with the new nsslapd-thread-pool-stats cn=config attribute. Changing this setting requires a restart. Fixes: https://github.com/389ds/389-ds-base/issues/7633 Reviewed by: @jchapma, @tbordaz, @mreynolds389 (Thanks!!!) (cherry picked from commit 05a17d0c6ca6bf22aab89b560b73d4c366ef2bbc) --- Makefile.am | 2 + .../suites/monitor/threadpool_status_test.py | 800 ++++++++++++++++++ ldap/schema/01core389.ldif | 1 + ldap/servers/slapd/configdse.c | 1 + ldap/servers/slapd/connection.c | 15 +- ldap/servers/slapd/daemon.c | 9 +- ldap/servers/slapd/fe.h | 2 +- ldap/servers/slapd/libglobs.c | 36 + ldap/servers/slapd/monitor.c | 2 + ldap/servers/slapd/proto-slap.h | 2 + ldap/servers/slapd/slap.h | 2 + ldap/servers/slapd/threadpool_stats.c | 762 +++++++++++++++++ ldap/servers/slapd/threadpool_stats.h | 94 ++ src/lib389/cli/dsctl | 2 + src/lib389/lib389/cli_ctl/threadpool.py | 468 ++++++++++ src/lib389/lib389/monitor.py | 8 + 16 files changed, 2202 insertions(+), 4 deletions(-) create mode 100644 dirsrvtests/tests/suites/monitor/threadpool_status_test.py create mode 100644 ldap/servers/slapd/threadpool_stats.c create mode 100644 ldap/servers/slapd/threadpool_stats.h create mode 100644 src/lib389/lib389/cli_ctl/threadpool.py diff --git a/Makefile.am b/Makefile.am index ab4fad5b2..1986efc3b 100644 --- a/Makefile.am +++ b/Makefile.am @@ -513,6 +513,7 @@ dist_noinst_HEADERS = \ ldap/servers/slapd/snmp_collator.h \ ldap/servers/slapd/sslerrstrs.h \ ldap/servers/slapd/statechange.h \ + ldap/servers/slapd/threadpool_stats.h \ ldap/servers/slapd/uuid.h \ ldap/servers/slapd/vattr_spi.h \ ldap/servers/slapd/views.h \ @@ -1953,6 +1954,7 @@ ns_slapd_SOURCES = ldap/servers/slapd/abandon.c \ ldap/servers/slapd/strdup.c \ ldap/servers/slapd/stubs.c \ ldap/servers/slapd/tempnam.c \ + ldap/servers/slapd/threadpool_stats.c \ ldap/servers/slapd/unbind.c \ ldap/servers/slapd/subentries.c diff --git a/dirsrvtests/tests/suites/monitor/threadpool_status_test.py b/dirsrvtests/tests/suites/monitor/threadpool_status_test.py new file mode 100644 index 000000000..d74a54d9c --- /dev/null +++ b/dirsrvtests/tests/suites/monitor/threadpool_status_test.py @@ -0,0 +1,800 @@ +# --- BEGIN COPYRIGHT BLOCK --- +# Copyright (C) 2026 Red Hat, Inc. +# All rights reserved. +# +# License: GPL (version 3 or any later version). +# See LICENSE for details. +# --- END COPYRIGHT BLOCK --- +# + +import json +import logging +import os +import re +import shutil +import signal +import stat +import struct +import subprocess +import threading +import time + +import ldap +import pytest + +from lib389._constants import DEFAULT_SUFFIX, DN_CONFIG, DN_DM, PW_DM +from lib389.cli_ctl.threadpool import (HEADER_FORMAT, TP_STATS_HEADER_SIZE, + TP_STATS_MAGIC, TP_STATS_WORKER_SLOT_SIZE) +from lib389.dseldif import DSEldif +from lib389.idm.account import Anonymous +from lib389.idm.user import UserAccounts +from lib389.monitor import Monitor +from test389.topologies import topology_st as topo + + +pytestmark = pytest.mark.tier1 + +DEBUGGING = os.getenv("DEBUGGING", default=False) +if DEBUGGING: + logging.getLogger(__name__).setLevel(logging.DEBUG) +else: + logging.getLogger(__name__).setLevel(logging.INFO) +log = logging.getLogger(__name__) + + +def _threadpool_path(inst): + dse = DSEldif(inst) + rundir = dse.get(DN_CONFIG, "nsslapd-rundir", single=True, lower=True) + if rundir is None: + rundir = inst.ds_paths.run_dir + prefix = inst.serverid if inst.serverid.startswith("slapd-") else f"slapd-{inst.serverid}" + return os.path.join(rundir, f"{prefix}.monitor", "threadpool") + + +def _monitor_dir(inst): + return os.path.dirname(_threadpool_path(inst)) + + +def _ensure_monitor_dir(inst): + """Create the monitor dir owned like the run dir, for planting files while stopped""" + dirname = _monitor_dir(inst) + rundir_st = os.stat(os.path.dirname(dirname)) + os.makedirs(dirname, exist_ok=True) + os.chown(dirname, rundir_st.st_uid, rundir_st.st_gid) + return dirname + + +def _wait_threadpool_file(inst, timeout=5): + path = _threadpool_path(inst) + deadline = time.time() + timeout + while time.time() < deadline: + if os.path.exists(path): + return path + time.sleep(0.1) + raise AssertionError(f"{path} was not created within {timeout}s") + + +def _archive_paths(inst): + path = _threadpool_path(inst) + dirname, base = os.path.split(path) + pattern = re.compile(re.escape(base) + r"\.\d{8}-\d{6}$") + try: + names = sorted(name for name in os.listdir(dirname) if pattern.match(name)) + except OSError: + return [] + return [os.path.join(dirname, name) for name in names] + + +def _purge_archives(inst): + for archive in _archive_paths(inst): + try: + os.unlink(archive) + except OSError: + pass + + +def _run_dsctl_threadpool(inst, json_output=False, timeout=10, extra_args=None): + cmd = ["dsctl"] + if json_output: + cmd.append("-j") + cmd.extend([inst.serverid, "thread-pool", "status"]) + if extra_args: + cmd.extend(extra_args) + return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) + + +def _cmd_output(result): + return f"{result.stdout}\n{result.stderr}" + + +def _safe_unbind(conn): + try: + conn.unbind_s() + except ldap.LDAPError: + pass + + +def _json_result(result): + assert result.returncode == 0, _cmd_output(result) + return json.loads(result.stdout) + + +def _assert_monitor_values_sanitized(values): + assert values + pattern = re.compile(r"^worker=\d+ state=\w+ op=\w* duration_ns=\d+$") + for value in values: + assert pattern.match(value) + assert "conn=" not in value + assert "op_id=" not in value + + +def test_file_created_mode(topo): + """The thread-pool mmap file is created with the expected mode and owner + + :id: 2bcb2478-b0e7-40e9-acdb-9a94f6039d00 + :setup: Standalone instance + :steps: + 1. Resolve the thread-pool mmap path from dse.ldif. + 2. Inspect the file and monitor directory metadata. + 3. Compare modes and owners with the runtime directory. + :expectedresults: + 1. The file exists. + 2. The file mode is 0640 and the monitor directory mode is 0750. + 3. The file and directory owners match the runtime directory owner. + """ + inst = topo.standalone + path = _wait_threadpool_file(inst) + + st = os.stat(path) + dir_st = os.stat(os.path.dirname(path)) + rundir_st = os.stat(os.path.dirname(os.path.dirname(path))) + assert stat.S_IMODE(st.st_mode) == 0o640 + assert stat.S_IMODE(dir_st.st_mode) == 0o750 + assert st.st_uid == rundir_st.st_uid + assert dir_st.st_uid == rundir_st.st_uid + + +def test_file_unlinked_on_stop(topo): + """The thread-pool mmap file is removed during clean shutdown + + :id: 7ffa8107-2403-4e97-ac83-cf448cb01463 + :setup: Standalone instance + :steps: + 1. Resolve the thread-pool mmap path. + 2. Stop the instance. + 3. Run dsctl thread-pool status. + 4. Start the instance again. + :expectedresults: + 1. The file exists while running. + 2. The file is absent after stop. + 3. dsctl reports that the instance is not running. + 4. The instance is restored for later tests. + """ + inst = topo.standalone + path = _wait_threadpool_file(inst) + + inst.stop() + try: + assert not os.path.exists(path) + result = _run_dsctl_threadpool(inst) + assert result.returncode != 0 + assert "instance is not running" in _cmd_output(result).lower() + finally: + inst.start() + + +def test_dsctl_basic_output(topo): + """dsctl thread-pool status reports text and JSON status + + :id: 3ba509a3-b347-4e81-a92c-04a1aa6ed17e + :setup: Standalone instance + :steps: + 1. Run dsctl thread-pool status. + 2. Run dsctl -j thread-pool status. + 3. Parse the JSON output. + :expectedresults: + 1. Text output includes pool gauges and a worker table. + 2. JSON output is valid. + 3. JSON output includes pool, worker, and warning fields. + """ + inst = topo.standalone + _wait_threadpool_file(inst) + + result = _run_dsctl_threadpool(inst) + assert result.returncode == 0, _cmd_output(result) + output = result.stdout + for expected in ("Instance:", "PID:", "Heartbeat age:", "Workers:", "Queue:", "Operations:", "IDX"): + assert expected in output + + data = _json_result(_run_dsctl_threadpool(inst, json_output=True)) + assert data["type"] == "result" + assert data["instance"] == inst.serverid + assert data["pool"]["max_workers"] >= 1 + assert isinstance(data["workers"], list) + assert isinstance(data["warnings"], list) + + +def test_dsctl_under_saturation(topo): + """dsctl remains available while the worker pool is busy + + :id: 04f806c0-4a4f-4cce-ac58-a84297d5b04c + :setup: Standalone instance + :steps: + 1. Restart the instance with two worker threads. + 2. Add enough test users to make subtree searches non-trivial. + 3. Run concurrent searches: authenticated persistent-connection + loops plus fresh anonymous connections whose search is the + first operation (op 0) on each connection. + 4. Poll dsctl -j thread-pool status while searches are active. + 5. Restore the original thread count. + :expectedresults: + 1. The instance restarts. + 2. Test users are created. + 3. Searches run concurrently. + 4. dsctl reports a busy worker with operation detail, including + a worker running a first operation with op_id 0 and a duration. + 5. The instance is restored for later tests. + """ + inst = topo.standalone + original_threadnumber = inst.config.get_attr_val_utf8("nsslapd-threadnumber") + created = [] + stop_event = threading.Event() + search_threads = [] + search_errors = [] + + def _new_conn(): + conn = ldap.initialize(inst.ldapuri) + # Restart syscalls interrupted by stray signals (subprocess + # reaping, harness timers) instead of failing with SERVER_DOWN + conn.set_option(ldap.OPT_RESTART, ldap.OPT_ON) + return conn + + def search_worker(): + conn = None + try: + conn = _new_conn() + conn.simple_bind_s(DN_DM, PW_DM) + while not stop_event.is_set(): + conn.search_s(DEFAULT_SUFFIX, ldap.SCOPE_SUBTREE, "(objectclass=*)", ["1.1"]) + except ldap.LDAPError as e: + if not stop_event.is_set(): + search_errors.append(str(e)) + finally: + if conn is not None: + _safe_unbind(conn) + + def first_op_search_worker(): + # No bind: the search is the first operation (op 0) on each connection + try: + while not stop_event.is_set(): + conn = _new_conn() + try: + conn.search_s(DEFAULT_SUFFIX, ldap.SCOPE_SUBTREE, "(objectclass=*)", ["1.1"]) + finally: + _safe_unbind(conn) + except ldap.LDAPError as e: + if not stop_event.is_set(): + search_errors.append(str(e)) + + try: + inst.config.replace("nsslapd-threadnumber", "2") + inst.restart() + _wait_threadpool_file(inst) + + users = UserAccounts(inst, DEFAULT_SUFFIX) + base_uid = 700000 + (int(time.time()) % 100000) + for uid in range(base_uid, base_uid + 50): + created.append(users.create_test_user(uid=uid, gid=uid)) + + for target in [search_worker] * 4 + [first_op_search_worker] * 4: + thread = threading.Thread(target=target) + thread.daemon = True + thread.start() + search_threads.append(thread) + + busy_worker = None + op0_worker = None + last_data = None + # Generous deadline: each poll pays a full dsctl python startup, + # which can take seconds on slow or sanitizer builds. + deadline = time.time() + 30 + while time.time() < deadline and (busy_worker is None or op0_worker is None): + last_data = _json_result(_run_dsctl_threadpool(inst, json_output=True, timeout=15)) + for worker in last_data["workers"]: + if worker["state"] != "busy" or worker["op_id"] is None: + continue + if not worker["op"] or not worker["conn"]: + continue + if busy_worker is None: + busy_worker = worker + if op0_worker is None and worker["op_id"] == 0: + op0_worker = worker + time.sleep(0.25) + + assert not search_errors, search_errors + assert busy_worker is not None, last_data + assert busy_worker["duration_ns"] >= 0 + assert op0_worker is not None, last_data + assert op0_worker["op"] + assert op0_worker["duration_ns"] >= 0 + finally: + stop_event.set() + for thread in search_threads: + thread.join(timeout=2) + for user in created: + try: + user.delete() + except ldap.NO_SUCH_OBJECT: + pass + inst.config.replace("nsslapd-threadnumber", original_threadnumber) + inst.restart() + + +def test_stale_file_after_kill(topo): + """dsctl reports a stale file after an unclean server exit + + :id: 4a865d53-3350-4f4a-8910-d30552f462b6 + :setup: Standalone instance + :steps: + 1. Read the server pid from dsctl JSON output. + 2. Kill the server process. + 3. Run dsctl thread-pool status against the leftover mmap file. + 4. Restart the instance. + :expectedresults: + 1. The pid is available. + 2. The server exits without clean mmap unlink. + 3. dsctl returns data with a stale-pid warning. + 4. The instance is restored for later tests. + """ + inst = topo.standalone + path = _wait_threadpool_file(inst) + data = _json_result(_run_dsctl_threadpool(inst, json_output=True)) + pid = data["pid"] + + os.kill(pid, signal.SIGKILL) + try: + deadline = time.time() + 10 + while time.time() < deadline and inst.status(): + time.sleep(0.2) + + assert os.path.exists(path) + result = _run_dsctl_threadpool(inst) + assert result.returncode == 0, _cmd_output(result) + assert "not running" in _cmd_output(result).lower() + finally: + if not inst.status(): + inst.start() + _purge_archives(inst) + + +def test_symlink_rejected(topo): + """Symlinks at the mmap path are never followed by the writer or the reader + + :id: d833f584-7cbf-460d-8079-0687a00fd483 + :setup: Standalone instance + :steps: + 1. Stop the instance and place a symlink at the thread-pool mmap path. + 2. Start the instance. + 3. Check the outcome of the startup symlink handling. + 4. Stop the instance and place another symlink at the same path. + 5. Run dsctl thread-pool status. + 6. Clean up and restart the instance. + :expectedresults: + 1. The symlink is in place before startup. + 2. The instance starts. + 3. Either startup replaced the symlink with a regular mmap file, or + (with SELinux denying the unlink of a foreign-labeled symlink) + the feature failed safe: the symlink was not followed, a warning + was logged, and dsctl refuses the path. + 4. The symlink is in place for the reader. + 5. dsctl refuses the symlink. + 6. The instance is restored for later tests. + """ + inst = topo.standalone + path = _threadpool_path(inst) + decoy = os.path.join(_monitor_dir(inst), "threadpool-decoy") + + inst.stop() + try: + _ensure_monitor_dir(inst) + with open(decoy, "wb") as decoy_file: + decoy_file.truncate(4096) + + if os.path.lexists(path): + os.unlink(path) + os.symlink(decoy, path) + inst.start() + if os.path.islink(path): + # SELinux may deny ns-slapd unlinking a foreign-labeled symlink + # (tclass=lnk_file). The server must fail safe: never follow the + # symlink, disable the feature, and log a warning. + result = _run_dsctl_threadpool(inst) + assert result.returncode != 0 + assert "symlink" in _cmd_output(result).lower() + assert inst.ds_error_log.match(".*Could not remove stale thread-pool status.*") + else: + assert stat.S_ISREG(os.stat(path).st_mode) + + inst.stop() + if os.path.lexists(path): + os.unlink(path) + os.symlink(decoy, path) + result = _run_dsctl_threadpool(inst) + assert result.returncode != 0 + assert "symlink" in _cmd_output(result).lower() + finally: + if os.path.lexists(path): + os.unlink(path) + if os.path.exists(decoy): + os.unlink(decoy) + # Full restart either way: a failed-safe startup leaves the feature + # disabled and would leak into the following tests. + if inst.status(): + inst.restart() + else: + inst.start() + + +def test_symlink_monitor_dir_rejected(topo): + """A symlink at the monitor directory path is refused at startup + + :id: 5b15d3e7-72e4-4d4c-ad79-3905fea1d0b7 + :setup: Standalone instance + :steps: + 1. Stop the instance and replace the monitor directory with a symlink + to a decoy directory. + 2. Start the instance. + 3. Check the errors log, the decoy directory, and dsctl output. + 4. Clean up and restart the instance. + :expectedresults: + 1. The symlink is in place before startup. + 2. The instance starts. + 3. The feature failed safe: the unsafe-directory warning is logged, + no status file was written into the decoy, and dsctl reports the + missing file. + 4. The instance is restored for later tests. + """ + inst = topo.standalone + monitor_dir = _monitor_dir(inst) + decoy_dir = os.path.join(os.path.dirname(monitor_dir), "monitor-decoy") + + inst.stop() + try: + os.makedirs(decoy_dir, exist_ok=True) + if os.path.islink(monitor_dir): + os.unlink(monitor_dir) + elif os.path.isdir(monitor_dir): + shutil.rmtree(monitor_dir) + os.symlink(decoy_dir, monitor_dir) + inst.start() + + assert os.path.islink(monitor_dir) + assert inst.ds_error_log.match(".*Refusing unsafe thread-pool monitor directory.*") + assert not os.path.exists(os.path.join(decoy_dir, "threadpool")) + result = _run_dsctl_threadpool(inst) + assert result.returncode != 0 + finally: + if os.path.islink(monitor_dir): + os.unlink(monitor_dir) + if os.path.isdir(decoy_dir): + shutil.rmtree(decoy_dir) + # A failed-safe startup leaves the feature disabled and would leak + # into the following tests. + if inst.status(): + inst.restart() + else: + inst.start() + + +def test_monitor_attr_present(topo): + """cn=monitor exposes sanitized threadpoolworker values + + :id: 5090ac9b-e865-48ee-a185-2f8f047f82ca + :setup: Standalone instance + :steps: + 1. Read threadpoolworker through lib389 Monitor. + 2. Validate the key=value format. + :expectedresults: + 1. At least one value is returned. + 2. Values contain only worker, state, op, and duration_ns tokens. + """ + values = Monitor(topo.standalone).get_thread_pool_workers() + _assert_monitor_values_sanitized(values) + + +def test_monitor_attr_sanitized(topo): + """Anonymous cn=monitor access exposes no connection or operation ids + + :id: a088e0b6-dcb5-43b0-ac1d-3b41bb393e95 + :setup: Standalone instance + :steps: + 1. Bind anonymously. + 2. Read threadpoolworker from cn=monitor. + 3. Validate the sanitized format. + :expectedresults: + 1. Anonymous bind succeeds. + 2. Values are returned. + 3. Values omit conn and op_id tokens. + """ + anon = Anonymous(topo.standalone).bind() + try: + values = Monitor(anon).get_thread_pool_workers() + _assert_monitor_values_sanitized(values) + finally: + anon.close() + + +def test_feature_disabled_by_config(topo): + """nsslapd-thread-pool-stats: off disables the diagnostics after a restart + + :id: 867c1bae-6dd6-49d6-92dd-75804bc84510 + :setup: Standalone instance + :steps: + 1. Set nsslapd-thread-pool-stats to an invalid value. + 2. Set nsslapd-thread-pool-stats to off and run dsctl before restarting. + 3. Restart and check the mmap file, dsctl output, and cn=monitor. + 4. Set nsslapd-thread-pool-stats back to on and run dsctl before + restarting. + 5. Restart. + :expectedresults: + 1. The invalid value is rejected. + 2. dsctl still reports data with a restart-pending warning. + 3. The file is absent, dsctl explains why, and threadpoolworker is gone. + 4. dsctl fails with a message mentioning the missing restart. + 5. The feature is active again. + """ + inst = topo.standalone + path = _threadpool_path(inst) + + with pytest.raises(ldap.OPERATIONS_ERROR): + inst.config.replace("nsslapd-thread-pool-stats", "maybe") + + try: + inst.config.replace("nsslapd-thread-pool-stats", "off") + # The running server keeps publishing until it is restarted + data = _json_result(_run_dsctl_threadpool(inst, json_output=True)) + assert any("until it is restarted" in warning for warning in data["warnings"]) + inst.restart() + + assert not os.path.exists(path) + result = _run_dsctl_threadpool(inst) + assert result.returncode != 0 + assert "disabled by nsslapd-thread-pool-stats" in _cmd_output(result) + assert not Monitor(inst).get_thread_pool_workers() + + inst.config.replace("nsslapd-thread-pool-stats", "on") + # Enabled in cn=config, but the running server has no file yet + result = _run_dsctl_threadpool(inst) + assert result.returncode != 0 + assert "without a restart" in _cmd_output(result) + finally: + inst.config.replace("nsslapd-thread-pool-stats", "on") + inst.restart() + + _wait_threadpool_file(inst) + assert Monitor(inst).get_thread_pool_workers() + + +def test_invalid_file_rejected(topo): + """dsctl refuses truncated and corrupted thread-pool status files + + :id: 421d1614-f57a-4775-afd8-45d3587cc923 + :setup: Standalone instance + :steps: + 1. Stop the instance. + 2. Place a truncated file at the thread-pool mmap path and run dsctl. + 3. Place a file with a corrupted magic and run dsctl. + 4. Clean up and start the instance. + :expectedresults: + 1. The instance is stopped. + 2. dsctl rejects the truncated file. + 3. dsctl rejects the corrupted magic. + 4. The instance is restored for later tests. + """ + inst = topo.standalone + path = _threadpool_path(inst) + + inst.stop() + try: + _ensure_monitor_dir(inst) + with open(path, "wb") as f: + f.write(b"\x00" * 100) + result = _run_dsctl_threadpool(inst) + assert result.returncode != 0 + assert "too short" in _cmd_output(result).lower() + + with open(path, "wb") as f: + f.write(b"\xff" * 8192) + result = _run_dsctl_threadpool(inst) + assert result.returncode != 0 + assert "magic" in _cmd_output(result).lower() + finally: + if os.path.exists(path): + os.unlink(path) + inst.start() + + +def test_stale_heartbeat_warning(topo): + """dsctl warns when a live server stops updating the heartbeat + + :id: dcc1281d-c63a-4bbb-ab1d-ed5349e92858 + :setup: Standalone instance + :steps: + 1. Read the server pid from dsctl JSON output. + 2. Stop the process with SIGSTOP and wait past the staleness threshold. + 3. Run dsctl thread-pool status. + 4. Resume the process with SIGCONT. + :expectedresults: + 1. The pid is available. + 2. The heartbeat stops updating while the process stays alive. + 3. dsctl reports data with a stalled-server warning. + 4. The instance keeps running for later tests. + """ + inst = topo.standalone + _wait_threadpool_file(inst) + data = _json_result(_run_dsctl_threadpool(inst, json_output=True)) + pid = data["pid"] + + os.kill(pid, signal.SIGSTOP) + try: + time.sleep(31) + data = _json_result(_run_dsctl_threadpool(inst, json_output=True)) + assert any("may be stalled" in warning for warning in data["warnings"]) + assert data["heartbeat_age_sec"] > 30 + finally: + os.kill(pid, signal.SIGCONT) + + +def test_crash_archive_created_after_kill(topo): + """A crash leftover is preserved as a timestamped archive on the next start + + :id: e7259a2b-6286-4e71-8c05-7bce3c8c9ab2 + :setup: Standalone instance + :steps: + 1. Remove existing archives and read the server pid from dsctl JSON output. + 2. Kill the server process and start the instance again. + 3. Check the live file, the archive count, and the errors log. + 4. Run dsctl thread-pool status against the running instance. + 5. Read the archive with dsctl thread-pool status --file. + :expectedresults: + 1. The pid is available. + 2. The instance starts. + 3. The live file is recreated, one archive exists, and the preserved + message is logged. + 4. The output warns that a crash archive is present. + 5. The archive reports the killed pid with a stale-file warning. + """ + inst = topo.standalone + _purge_archives(inst) + _wait_threadpool_file(inst) + data = _json_result(_run_dsctl_threadpool(inst, json_output=True)) + pid = data["pid"] + + try: + os.kill(pid, signal.SIGKILL) + deadline = time.time() + 10 + while time.time() < deadline and inst.status(): + time.sleep(0.2) + inst.start() + + _wait_threadpool_file(inst) + archives = _archive_paths(inst) + assert len(archives) == 1 + assert inst.ds_error_log.match(".*thread-pool status preserved as.*") + + data = _json_result(_run_dsctl_threadpool(inst, json_output=True)) + assert any("crash archive" in warning for warning in data["warnings"]) + + archive_data = _json_result( + _run_dsctl_threadpool(inst, json_output=True, extra_args=["--file", archives[0]]) + ) + assert archive_data["pid"] == pid + assert any("stale file" in warning for warning in archive_data["warnings"]) + finally: + if not inst.status(): + inst.start() + _purge_archives(inst) + + +def test_no_archive_after_clean_restart(topo): + """A clean restart does not create a crash archive + + :id: c2bbb29f-1669-4cc6-9269-6a4e04559658 + :setup: Standalone instance + :steps: + 1. Remove existing archives. + 2. Restart the instance. + 3. Check for archives. + :expectedresults: + 1. No archives remain. + 2. The instance restarts. + 3. No archive was created. + """ + inst = topo.standalone + _purge_archives(inst) + inst.restart() + _wait_threadpool_file(inst) + assert _archive_paths(inst) == [] + + +def test_archive_pruned_to_five(topo): + """Startup keeps at most five crash archives + + :id: d0a601d5-13ce-4114-8bb3-776bb20e65ff + :setup: Standalone instance + :steps: + 1. Stop the instance and remove existing archives. + 2. Plant seven dummy archives and a fabricated crash leftover at the + live path, owned by the server user. + 3. Start the instance. + 4. Count the archives. + :expectedresults: + 1. The instance is stopped. + 2. The files are in place. + 3. The instance starts and archives the leftover. + 4. Five archives remain: the four newest dummies plus the new one. + """ + inst = topo.standalone + path = _threadpool_path(inst) + dummies = [f"{path}.20250101-00000{i}" for i in range(7)] + + inst.stop() + _purge_archives(inst) + try: + _ensure_monitor_dir(inst) + dir_st = os.stat(os.path.dirname(path)) + for dummy in dummies: + with open(dummy, "wb") as f: + f.write(b"\x00") + + # A crash leftover the server will archive: valid magic, unclean shutdown + header = struct.pack(HEADER_FORMAT, TP_STATS_MAGIC, 1, 0, + TP_STATS_HEADER_SIZE, TP_STATS_WORKER_SLOT_SIZE, 1, + 0, 0, 0, 0, 0, + 0, 0, 0, 0, 0, 0, 0) + with open(path, "wb") as f: + f.write(header.ljust(TP_STATS_HEADER_SIZE + TP_STATS_WORKER_SLOT_SIZE, b"\x00")) + os.chown(path, dir_st.st_uid, dir_st.st_gid) + + inst.start() + + archives = [os.path.basename(archive) for archive in _archive_paths(inst)] + assert len(archives) == 5 + for dummy in dummies[:3]: + assert os.path.basename(dummy) not in archives + for dummy in dummies[3:]: + assert os.path.basename(dummy) in archives + finally: + if not inst.status(): + inst.start() + _purge_archives(inst) + + +def test_dsctl_file_option(topo): + """dsctl thread-pool status --file reads an explicit status file path + + :id: c68f4cf1-8e82-487c-84cf-2ca2108ed898 + :setup: Standalone instance + :steps: + 1. Run dsctl thread-pool status --file with a nonexistent path. + 2. Run it with the live file path of the running instance. + :expectedresults: + 1. The command fails with a not-found error. + 2. The command succeeds and reports the pool. + """ + inst = topo.standalone + path = _wait_threadpool_file(inst) + + result = _run_dsctl_threadpool(inst, extra_args=["--file", "/nonexistent/threadpool"]) + assert result.returncode != 0 + assert "not found" in _cmd_output(result).lower() + + data = _json_result( + _run_dsctl_threadpool(inst, json_output=True, extra_args=["--file", path]) + ) + assert data["pool"]["max_workers"] >= 1 + + +if __name__ == "__main__": + CURRENT_FILE = os.path.realpath(__file__) + pytest.main("-s %s" % CURRENT_FILE) diff --git a/ldap/schema/01core389.ldif b/ldap/schema/01core389.ldif index 7e2d1ac44..91dc25e3d 100644 --- a/ldap/schema/01core389.ldif +++ b/ldap/schema/01core389.ldif @@ -334,6 +334,7 @@ attributeTypes: ( 2.16.840.1.113730.3.1.2393 NAME 'nsslapd-auditlog-display-attr attributeTypes: ( 2.16.840.1.113730.3.1.2398 NAME 'nsslapd-haproxy-trusted-ip' DESC '389 Directory Server defined attribute type' SYNTAX 1.3.6.1.4.1.1466.115.121.1.15 X-ORIGIN '389 Directory Server' ) attributeTypes: ( 2.16.840.1.113730.3.1.2400 NAME 'nsslapd-pwdPBKDF2NumIterations' DESC '389 Directory Server defined attribute type' SYNTAX 1.3.6.1.4.1.1466.115.121.1.27 SINGLE-VALUE X-ORIGIN 'Directory Server' ) attributeTypes: ( 2.16.840.1.113730.3.1.2402 NAME 'nsslapd-maxcontrolsperop' DESC '389 Directory Server defined attribute type' SYNTAX 1.3.6.1.4.1.1466.115.121.1.27 SINGLE-VALUE X-ORIGIN '389 Directory Server' ) +attributeTypes: ( 2.16.840.1.113730.3.1.2404 NAME 'nsslapd-thread-pool-stats' DESC '389 Directory Server defined attribute type' SYNTAX 1.3.6.1.4.1.1466.115.121.1.15 SINGLE-VALUE X-ORIGIN '389 Directory Server' ) # # objectclasses # diff --git a/ldap/servers/slapd/configdse.c b/ldap/servers/slapd/configdse.c index eed046f2a..2be6036f2 100644 --- a/ldap/servers/slapd/configdse.c +++ b/ldap/servers/slapd/configdse.c @@ -49,6 +49,7 @@ static const char *requires_restart[] = { "cn=config:nsslapd-numlisteners", "cn=config:" CONFIG_RETURN_EXACT_CASE_ATTRIBUTE, "cn=config:" CONFIG_SCHEMA_IGNORE_TRAILING_SPACES, + "cn=config:" CONFIG_THREAD_POOL_STATS_ATTRIBUTE, "cn=config,cn=ldbm:nsslapd-idlistscanlimit", "cn=config,cn=ldbm:nsslapd-parentcheck", "cn=config,cn=ldbm:nsslapd-dbcachesize", diff --git a/ldap/servers/slapd/connection.c b/ldap/servers/slapd/connection.c index ff6d72372..fd8826b66 100644 --- a/ldap/servers/slapd/connection.c +++ b/ldap/servers/slapd/connection.c @@ -22,6 +22,7 @@ #include "prcvar.h" #include "prlog.h" /* for PR_ASSERT */ #include "fe.h" +#include "threadpool_stats.h" #include #include #if defined(LINUX) @@ -434,7 +435,7 @@ connection_reset(Connection *conn, int ns, PRNetAddr *from, int fromLen __attrib /* Create a pool of threads for handling the operations */ void -init_op_threads() +init_op_threads(int32_t threadnumber) { pthread_condattr_t condAttr; int32_t rc; @@ -464,7 +465,7 @@ init_op_threads() } pthread_condattr_destroy(&condAttr); /* no longer needed */ - max_threads = config_get_threadnumber(); + max_threads = threadnumber; work_q_stack = PR_CreateStack("connection_work_q"); op_stack = PR_CreateStack("connection_operation"); alloc_per_thread_snmp_vars(max_threads); @@ -1710,6 +1711,7 @@ connection_threadmain(void *arg) char tname[16]; snprintf(tname, sizeof(tname), "worker-%d", *snmp_vars_idx); slapi_set_thread_name(tname); + tp_stats_worker_idle((uint32_t)*snmp_vars_idx); /* wait forever for new pb until one is available or shutdown */ int32_t interval = 0; /* used be 10 seconds */ Connection *conn = NULL; @@ -1739,6 +1741,7 @@ connection_threadmain(void *arg) if (is_busy) { slapi_atomic_decr_32(¤t_busy_workers, __ATOMIC_ACQ_REL); } + tp_stats_worker_exited((uint32_t)*snmp_vars_idx); slapi_pblock_destroy(pb); g_decr_active_threadcnt(); return; @@ -1752,6 +1755,7 @@ connection_threadmain(void *arg) is_busy = false; slapi_atomic_decr_32(¤t_busy_workers, __ATOMIC_ACQ_REL); } + tp_stats_worker_idle((uint32_t)*snmp_vars_idx); /* If more data is left from the previous connection_read_operation, we should finish the op now. Client might be thinking it's @@ -1769,6 +1773,7 @@ connection_threadmain(void *arg) if (is_busy) { slapi_atomic_decr_32(¤t_busy_workers, __ATOMIC_ACQ_REL); } + tp_stats_worker_exited((uint32_t)*snmp_vars_idx); slapi_pblock_destroy(pb); g_decr_active_threadcnt(); return; @@ -1784,6 +1789,7 @@ connection_threadmain(void *arg) if (is_busy) { slapi_atomic_decr_32(¤t_busy_workers, __ATOMIC_ACQ_REL); } + tp_stats_worker_exited((uint32_t)*snmp_vars_idx); slapi_pblock_destroy(pb); g_decr_active_threadcnt(); return; @@ -1860,6 +1866,7 @@ connection_threadmain(void *arg) while (val > slapi_atomic_load_32(&max_busy_workers, __ATOMIC_RELAXED)) { slapi_atomic_store_32(&max_busy_workers, val, __ATOMIC_RELAXED); } + tp_stats_worker_busy((uint32_t)*snmp_vars_idx); } slapi_pblock_get(pb, SLAPI_CONNECTION, &conn); slapi_pblock_get(pb, SLAPI_OPERATION, &op); @@ -1868,6 +1875,7 @@ connection_threadmain(void *arg) if (is_busy) { slapi_atomic_decr_32(¤t_busy_workers, __ATOMIC_ACQ_REL); } + tp_stats_worker_exited((uint32_t)*snmp_vars_idx); slapi_pblock_destroy(pb); g_decr_active_threadcnt(); return; @@ -2067,6 +2075,7 @@ connection_threadmain(void *arg) /* * Call the do_ function to process this request. */ + tp_stats_worker_operation_start((uint32_t)*snmp_vars_idx, conn->c_connid, (uint64_t)op->o_opid, (uint32_t)op->o_tag); connection_dispatch_operation(conn, op, pb); done: @@ -2074,6 +2083,7 @@ connection_threadmain(void *arg) if (is_busy) { slapi_atomic_decr_32(¤t_busy_workers, __ATOMIC_ACQ_REL); } + tp_stats_worker_exited((uint32_t)*snmp_vars_idx); pthread_mutex_lock(&(conn->c_mutex)); connection_remove_operation_ext(pb, conn, op); connection_make_readable_nolock(conn); @@ -2097,6 +2107,7 @@ connection_threadmain(void *arg) PR_AtomicIncrement(&conn->c_opscompleted); /* total number of ops for the server */ slapi_counter_increment(g_get_per_thread_snmp_vars()->server_tbl.dsOpCompleted); + tp_stats_worker_operation_done((uint32_t)*snmp_vars_idx); /* If this op isn't a persistent search, remove it */ if (op->o_flags & OP_FLAG_PS) { /* Release the connection (i.e. decrease refcnt) at the condition diff --git a/ldap/servers/slapd/daemon.c b/ldap/servers/slapd/daemon.c index 42aea3df4..c99674896 100644 --- a/ldap/servers/slapd/daemon.c +++ b/ldap/servers/slapd/daemon.c @@ -59,6 +59,7 @@ #include "slap.h" #include "slapi-plugin.h" #include "snmp_collator.h" +#include "threadpool_stats.h" #include #include #include "fe.h" @@ -1181,6 +1182,7 @@ slapd_daemon(daemon_ports_t *ports) PRFileDesc **i_unix = NULL; PRFileDesc **fdesp = NULL; uint64_t threads; + int32_t threadnumber = config_get_threadnumber(); int in_referral_mode = config_check_referral_mode(); int connection_table_size = get_connection_table_size(); the_connection_table = connection_table_new(connection_table_size); @@ -1229,7 +1231,11 @@ slapd_daemon(daemon_ports_t *ports) } init_ct_list_threads(); - init_op_threads(); + tp_stats_init(threadnumber > 0 ? (uint32_t)threadnumber : 0); + init_op_threads(threadnumber); + /* Heartbeat must not start before init_op_threads: its callback reads + * per_thread_snmp_vars, which alloc_per_thread_snmp_vars reallocates. */ + tp_stats_start_heartbeat(); /* Start the SNMP collator if counters are enabled. */ if (config_get_slapi_counters()) { @@ -1479,6 +1485,7 @@ slapd_daemon(daemon_ports_t *ports) pageresult_lock_cleanup(); eq_stop(); /* deprecated */ eq_stop_rel(); + tp_stats_close(); if (!in_referral_mode) { task_shutdown(); uniqueIDGenCleanup(); diff --git a/ldap/servers/slapd/fe.h b/ldap/servers/slapd/fe.h index ad90dba2e..716a913bf 100644 --- a/ldap/servers/slapd/fe.h +++ b/ldap/servers/slapd/fe.h @@ -58,7 +58,7 @@ void connection_post_shutdown_cleanup(void); */ void connection_abandon_operations(Connection *conn); int connection_activity(Connection *conn, int maxthreads); -void init_op_threads(void); +void init_op_threads(int32_t threadnumber); int connection_new_private(Connection *conn); void connection_remove_operation(Connection *conn, Operation *op); void connection_remove_operation_ext(Slapi_PBlock *pb, Connection *conn, Operation *op); diff --git a/ldap/servers/slapd/libglobs.c b/ldap/servers/slapd/libglobs.c index 3b031052f..1456ad21c 100644 --- a/ldap/servers/slapd/libglobs.c +++ b/ldap/servers/slapd/libglobs.c @@ -245,6 +245,7 @@ slapi_onoff_t init_close_on_failed_bind; slapi_onoff_t init_minssf_exclude_rootdse; slapi_onoff_t init_force_sasl_external; slapi_onoff_t init_slapi_counters; +slapi_onoff_t init_thread_pool_stats; slapi_onoff_t init_entryusn_global; slapi_onoff_t init_disk_monitoring; slapi_onoff_t init_disk_threshold_readonly; @@ -956,6 +957,11 @@ static struct config_get_and_set (void **)&global_slapdFrontendConfig.slapi_counters, CONFIG_ON_OFF, (ConfigGetFunc)config_get_slapi_counters, &init_slapi_counters, NULL}, + {CONFIG_THREAD_POOL_STATS_ATTRIBUTE, config_set_thread_pool_stats, + NULL, 0, + (void **)&global_slapdFrontendConfig.thread_pool_stats, + CONFIG_ON_OFF, (ConfigGetFunc)config_get_thread_pool_stats, + &init_thread_pool_stats, NULL}, {CONFIG_ACCESSLOG_MINFREEDISKSPACE_ATTRIBUTE, NULL, log_set_mindiskspace, SLAPD_ACCESS_LOG, (void **)&global_slapdFrontendConfig.accesslog_minfreespace, @@ -1867,6 +1873,7 @@ FrontendConfig_init(void) init_close_on_failed_bind = cfg->close_on_failed_bind = LDAP_OFF; cfg->allow_anon_access = SLAPD_DEFAULT_ALLOW_ANON_ACCESS; init_slapi_counters = cfg->slapi_counters = LDAP_ON; + init_thread_pool_stats = cfg->thread_pool_stats = LDAP_ON; cfg->threadnumber = util_get_hardware_threads(); cfg->maxthreadsperconn = SLAPD_DEFAULT_MAX_THREADS_PER_CONN; cfg->reservedescriptors = SLAPD_DEFAULT_RESERVE_FDS; @@ -2239,6 +2246,11 @@ alloc_global_snmp_vars() /* Allocated the next slots of the arrays of counters * with a slot per worker thread + * + * Must complete before any reader of per_thread_snmp_vars starts (worker + * threads, snmp collator, thread-pool stats heartbeat): the slot count and + * the array pointer are published without synchronization, and the realloc + * frees the old array under a concurrent reader. */ void alloc_per_thread_snmp_vars(int32_t maxthread) @@ -3448,6 +3460,23 @@ config_set_slapi_counters(const char *attrname, char *value, char *errorbuf, int return retVal; } +/* + * Enable/disable the thread-pool status diagnostics (mmap file, "dsctl + * thread-pool status", threadpoolworker on cn=monitor). Read once at + * startup; changing it requires a restart. + */ +int32_t +config_set_thread_pool_stats(const char *attrname, char *value, char *errorbuf, int apply) +{ + int32_t retVal = LDAP_SUCCESS; + slapdFrontendConfig_t *slapdFrontendConfig = getFrontendConfig(); + + retVal = config_set_onoff(attrname, value, + &(slapdFrontendConfig->thread_pool_stats), errorbuf, apply); + + return retVal; +} + int config_set_securelistenhost(const char *attrname __attribute__((unused)), char *value, char *errorbuf __attribute__((unused)), int apply) { @@ -6345,6 +6374,13 @@ config_get_slapi_counters() } +int32_t +config_get_thread_pool_stats(void) +{ + slapdFrontendConfig_t *slapdFrontendConfig = getFrontendConfig(); + return slapi_atomic_load_32(&(slapdFrontendConfig->thread_pool_stats), __ATOMIC_ACQUIRE); +} + char * config_get_workingdir(void) { diff --git a/ldap/servers/slapd/monitor.c b/ldap/servers/slapd/monitor.c index f9a85cfbb..2f024557e 100644 --- a/ldap/servers/slapd/monitor.c +++ b/ldap/servers/slapd/monitor.c @@ -30,6 +30,7 @@ #include #include "slap.h" #include "fe.h" +#include "threadpool_stats.h" int32_t monitor_info(Slapi_PBlock *pb __attribute__((unused)), @@ -61,6 +62,7 @@ monitor_info(Slapi_PBlock *pb __attribute__((unused)), attrlist_replace(&e->e_attrs, "threads", vals); connection_table_as_entry(the_connection_table, e); + tp_stats_as_entry(e); val.bv_len = snprintf(buf, sizeof(buf), "%" PRIu64, g_get_num_ops_initiated()); val.bv_val = buf; diff --git a/ldap/servers/slapd/proto-slap.h b/ldap/servers/slapd/proto-slap.h index 9c82eabf8..0d0f98cc8 100644 --- a/ldap/servers/slapd/proto-slap.h +++ b/ldap/servers/slapd/proto-slap.h @@ -261,6 +261,7 @@ int config_set_ldapi_auto_dn_suffix(const char *attrname, char *value, char *err #endif int config_set_anon_limits_dn(const char *attrname, char *value, char *errorbuf, int apply); int config_set_slapi_counters(const char *attrname, char *value, char *errorbuf, int apply); +int32_t config_set_thread_pool_stats(const char *attrname, char *value, char *errorbuf, int apply); int config_set_srvtab(const char *attrname, char *value, char *errorbuf, int apply); int config_set_sizelimit(const char *attrname, char *value, char *errorbuf, int apply); int config_set_pagedsizelimit(const char *attrname, char *value, char *errorbuf, int apply); @@ -453,6 +454,7 @@ char *config_get_ldapi_auto_dn_suffix(void); #endif char *config_get_anon_limits_dn(void); int config_get_slapi_counters(void); +int32_t config_get_thread_pool_stats(void); char *config_get_srvtab(void); int config_get_sizelimit(void); int config_get_pagedsizelimit(void); diff --git a/ldap/servers/slapd/slap.h b/ldap/servers/slapd/slap.h index c968d1898..0a209921c 100644 --- a/ldap/servers/slapd/slap.h +++ b/ldap/servers/slapd/slap.h @@ -2298,6 +2298,7 @@ typedef struct _slapdEntryPoints #define CONFIG_LDAPI_AUTH_DN_ATTRIBUTE "nsslapd-authenticateAsDN" #define CONFIG_ANON_LIMITS_DN_ATTRIBUTE "nsslapd-anonlimitsdn" #define CONFIG_SLAPI_COUNTER_ATTRIBUTE "nsslapd-counters" +#define CONFIG_THREAD_POOL_STATS_ATTRIBUTE "nsslapd-thread-pool-stats" #define CONFIG_SECURITY_ATTRIBUTE "nsslapd-security" #define CONFIG_SSL3CIPHERS_ATTRIBUTE "nsslapd-SSL3ciphers" #define CONFIG_ACCESSLOG_ATTRIBUTE "nsslapd-accesslog" @@ -2726,6 +2727,7 @@ typedef struct _slapdFrontendConfig char *ldapi_auto_dn_suffix; /* suffix to be appended to auto gen DNs */ char *ldapi_auto_mapping_base; /* suffix/subtree containing LDAPI mapping entries */ slapi_onoff_t slapi_counters; /* switch to turn slapi_counters on/off */ + slapi_onoff_t thread_pool_stats; /* switch to turn thread-pool status diagnostics on/off */ slapi_onoff_t allow_unauth_binds; /* switch to enable/disable unauthenticated binds */ slapi_onoff_t require_secure_binds; /* switch to require simple binds to use a secure channel */ slapi_onoff_t allow_anon_access; /* switch to enable/disable anonymous access */ diff --git a/ldap/servers/slapd/threadpool_stats.c b/ldap/servers/slapd/threadpool_stats.c new file mode 100644 index 000000000..4e8aad1c5 --- /dev/null +++ b/ldap/servers/slapd/threadpool_stats.c @@ -0,0 +1,762 @@ +/** BEGIN COPYRIGHT BLOCK + * Copyright (C) 2026 Red Hat, Inc. + * All rights reserved. + * + * License: GPL (version 3 or any later version). + * See LICENSE for details. + * END COPYRIGHT BLOCK **/ + +#ifdef HAVE_CONFIG_H +#include +#endif + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "slap.h" +#include "plstr.h" +#include "threadpool_stats.h" + +#define TP_STATS_COMPONENT "threadpool_stats" +#define TP_STATS_DIR_SUFFIX ".monitor" +#define TP_STATS_FILENAME "threadpool" +#define TP_STATS_HEARTBEAT_INTERVAL_MS 1000 +#define TP_STATS_ARCHIVE_KEEP 5 + +/* + * Crash archives are named .YYYYMMDD-HHMMSS; the suffix is produced + * by strftime(TP_STATS_ARCHIVE_TIME_FMT). The lib389 reader matches the + * same pattern (\.\d{8}-\d{6}$). + */ +#define TP_STATS_ARCHIVE_TIME_FMT "%Y%m%d-%H%M%S" +#define TP_STATS_ARCHIVE_DATE_LEN (sizeof("YYYYMMDD") - 1) +#define TP_STATS_ARCHIVE_STAMP_LEN (sizeof("YYYYMMDD-HHMMSS") - 1) + +static tp_stats_header_t *tp_stats_header = NULL; +static int tp_stats_fd = -1; +static size_t tp_stats_len = 0; +static uint32_t tp_stats_max_workers = 0; +static char *tp_stats_path = NULL; +static Slapi_Eq_Context tp_stats_eq_ctx = NULL; + +static uint64_t +tp_stats_mono_ns(void) +{ + struct timespec ts = slapi_current_rel_time_hr(); + return ((uint64_t)ts.tv_sec * 1000000000ULL) + (uint64_t)ts.tv_nsec; +} + +static void +tp_store_32(uint32_t *ptr, uint32_t val, int memorder) +{ + slapi_atomic_store_32((int32_t *)ptr, (int32_t)val, memorder); +} + +static uint32_t +tp_load_32(uint32_t *ptr, int memorder) +{ + return (uint32_t)slapi_atomic_load_32((int32_t *)ptr, memorder); +} + +static tp_worker_slot_t * +tp_stats_slot_at(tp_stats_header_t *header, size_t idx) +{ + return (tp_worker_slot_t *)((uint8_t *)header + + TP_STATS_HEADER_SIZE + + (idx * TP_STATS_WORKER_SLOT_SIZE)); +} + +static tp_worker_slot_t * +tp_stats_get_slot(uint32_t worker_idx) +{ + if (tp_stats_header == NULL || worker_idx == 0 || worker_idx > tp_stats_max_workers) { + return NULL; + } + + return tp_stats_slot_at(tp_stats_header, worker_idx - 1); +} + +static const char * +tp_stats_state_name(uint32_t state) +{ + switch (state) { + case TP_WORKER_STATE_UNUSED: + return "unused"; + case TP_WORKER_STATE_IDLE: + return "idle"; + case TP_WORKER_STATE_BUSY: + return "busy"; + case TP_WORKER_STATE_EXITED: + return "exited"; + default: + return "unknown"; + } +} + +static const char * +tp_stats_op_name(uint32_t op_tag, char *buf, size_t buflen) +{ + switch ((ber_tag_t)op_tag) { + case 0: + return ""; + case LDAP_REQ_BIND: + return "bind"; + case LDAP_REQ_UNBIND: + return "unbind"; + case LDAP_REQ_SEARCH: + return "search"; + case LDAP_REQ_MODIFY: + return "modify"; + case LDAP_REQ_ADD: + return "add"; + case LDAP_REQ_DELETE: + return "delete"; + case LDAP_REQ_MODRDN: + return "modrdn"; + case LDAP_REQ_COMPARE: + return "compare"; + case LDAP_REQ_ABANDON: + return "abandon"; + case LDAP_REQ_EXTENDED: + return "extended"; + default: + snprintf(buf, buflen, "%" PRIu32, op_tag); + return buf; + } +} + +/* + * /slapd-.monitor/threadpool, matching what the dsctl + * reader derives from the instance config. Returns NULL when rundir or + * the slapd- name cannot be resolved: a file at any other path + * would be unreachable for the reader, so the caller disables the + * feature instead. + */ +static char * +tp_stats_make_path(void) +{ + char *rundir = config_get_rundir(); + char *configdir = config_get_configdir(); + char *instname = NULL; + char *path = NULL; + + if (configdir != NULL) { + instname = PL_strrstr(configdir, "slapd-"); + } + + if (rundir != NULL && instname != NULL) { + path = slapi_ch_smprintf("%s/%s%s/%s", rundir, instname, + TP_STATS_DIR_SUFFIX, TP_STATS_FILENAME); + } + slapi_ch_free_string(&rundir); + slapi_ch_free_string(&configdir); + return path; +} + +/* + * Create the per-instance monitor directory the status file lives in. + * A pre-existing entry is accepted only when lstat says it is a real + * directory owned by the server (a planted symlink fails the check). + * The directory is never removed at shutdown: crash archives stay in it. + */ +static int +tp_stats_prepare_dir(const char *path) +{ + const char *slash = strrchr(path, '/'); + struct stat st = {0}; + char *dir = NULL; + int rc = -1; + + if (slash == NULL) { + return -1; + } + dir = slapi_ch_smprintf("%.*s", (int)(slash - path), path); + + if (mkdir(dir, 0750) != 0) { + if (errno != EEXIST) { + int err = errno; + slapi_log_err(SLAPI_LOG_WARNING, TP_STATS_COMPONENT, + "Could not create thread-pool monitor directory %s: %d (%s)\n", + dir, err, slapd_system_strerror(err)); + goto done; + } + if (lstat(dir, &st) != 0 || !S_ISDIR(st.st_mode) || st.st_uid != geteuid()) { + slapi_log_err(SLAPI_LOG_WARNING, TP_STATS_COMPONENT, + "Refusing unsafe thread-pool monitor directory %s (mode=%o uid=%ld)\n", + dir, (unsigned int)st.st_mode, (long)st.st_uid); + goto done; + } + } + + if (chmod(dir, 0750) != 0) { + int err = errno; + slapi_log_err(SLAPI_LOG_WARNING, TP_STATS_COMPONENT, + "Could not set permissions on thread-pool monitor directory %s: %d (%s)\n", + dir, err, slapd_system_strerror(err)); + goto done; + } + rc = 0; + +done: + slapi_ch_free_string(&dir); + return rc; +} + +static int +tp_stats_name_cmp(const void *a, const void *b) +{ + return strcmp(*(char *const *)a, *(char *const *)b); +} + +/* The expected suffix format is 'YYYYMMDD-HHMMSS' */ +static bool +tp_stats_is_archive_suffix(const char *suffix) +{ + if (suffix == NULL || strlen(suffix) != TP_STATS_ARCHIVE_STAMP_LEN || + suffix[TP_STATS_ARCHIVE_DATE_LEN] != '-') { + return false; + } + for (size_t i = 0; i < TP_STATS_ARCHIVE_STAMP_LEN; i++) { + /* Every character is a digit except the '-' after the date part */ + if (i != TP_STATS_ARCHIVE_DATE_LEN && !isdigit((unsigned char)suffix[i])) { + return false; + } + } + return true; +} + +/* Remove the oldest archives so at most TP_STATS_ARCHIVE_KEEP remain */ +static void +tp_stats_prune_archives(const char *path) +{ + const char *slash = strrchr(path, '/'); + const char *base = NULL; + size_t baselen; + char *dir = NULL; + DIR *dirp = NULL; + struct dirent *entry = NULL; + char **names = NULL; + size_t count = 0; + + if (slash == NULL) { + return; + } + base = slash + 1; + baselen = strlen(base); + dir = slapi_ch_smprintf("%.*s", (int)(slash - path), path); + dirp = opendir(dir); + if (dirp == NULL) { + slapi_ch_free_string(&dir); + return; + } + + while ((entry = readdir(dirp)) != NULL) { + const char *name = entry->d_name; + if (strncmp(name, base, baselen) != 0 || name[baselen] != '.' || + !tp_stats_is_archive_suffix(name + baselen + 1)) { + continue; + } + names = (char **)slapi_ch_realloc((char *)names, (count + 1) * sizeof(char *)); + names[count++] = slapi_ch_strdup(name); + } + closedir(dirp); + + if (count > TP_STATS_ARCHIVE_KEEP) { + /* The timestamp suffix sorts lexicographically in time order */ + qsort(names, count, sizeof(char *), tp_stats_name_cmp); + for (size_t i = 0; i < count - TP_STATS_ARCHIVE_KEEP; i++) { + char *victim = slapi_ch_smprintf("%s/%s", dir, names[i]); + if (unlink(victim) != 0) { + int err = errno; + slapi_log_err(SLAPI_LOG_WARNING, TP_STATS_COMPONENT, + "Could not remove old thread-pool status archive %s: %d (%s)\n", + victim, err, slapd_system_strerror(err)); + } else { + slapi_log_err(SLAPI_LOG_INFO, TP_STATS_COMPONENT, + "Removed old thread-pool status archive %s\n", victim); + } + slapi_ch_free_string(&victim); + } + } + + for (size_t i = 0; i < count; i++) { + slapi_ch_free_string(&names[i]); + } + slapi_ch_free((void **)&names); + slapi_ch_free_string(&dir); +} + +/* + * Preserve a leftover status file from a crashed previous run by renaming + * it to .YYYYMMDD-HHMMSS (the rotated-log naming). Never unlinks and + * never fails startup: on any failure it returns having done nothing and + * the caller's unlink handles the leftover as before. Only a genuine crash + * leftover is preserved: the magic must match and shutdown_clean must be + * unset (tp_stats_close sets it before attempting the unlink). + */ +static void +tp_stats_archive_crash_file(const char *path) +{ + tp_stats_header_t hdr = {0}; + struct stat st = {0}; + struct tm tms = {0}; + char tbuf[32] = {0}; + char *archive = NULL; + time_t now; + ssize_t nread; + int fd; + + fd = open(path, O_RDONLY | O_NOFOLLOW | O_CLOEXEC); + if (fd < 0) { + return; + } + if (fstat(fd, &st) != 0 || !S_ISREG(st.st_mode) || st.st_uid != geteuid() || + st.st_nlink != 1 || st.st_size < (off_t)TP_STATS_HEADER_SIZE) { + close(fd); + return; + } + nread = pread(fd, &hdr, sizeof(hdr), 0); + close(fd); + if (nread != (ssize_t)sizeof(hdr) || hdr.magic != TP_STATS_MAGIC || + hdr.shutdown_clean != 0) { + return; + } + + now = slapi_current_utc_time(); + if (localtime_r(&now, &tms) == NULL || + strftime(tbuf, sizeof(tbuf), TP_STATS_ARCHIVE_TIME_FMT, &tms) == 0) { + return; + } + + archive = slapi_ch_smprintf("%s.%s", path, tbuf); + if (rename(path, archive) != 0) { + int err = errno; + slapi_log_err(SLAPI_LOG_WARNING, TP_STATS_COMPONENT, + "Could not preserve thread-pool status file %s from unclean shutdown: " + "%d (%s); removing it\n", + path, err, slapd_system_strerror(err)); + slapi_ch_free_string(&archive); + return; + } + + slapi_log_err(SLAPI_LOG_NOTICE, TP_STATS_COMPONENT, + "Previous server run did not shut down cleanly; " + "thread-pool status preserved as %s\n", + archive); + slapi_ch_free_string(&archive); + tp_stats_prune_archives(path); +} + +static void +tp_stats_cleanup_open_failure(int fd, char **path) +{ + if (fd >= 0) { + close(fd); + } + if (path != NULL && *path != NULL) { + unlink(*path); + slapi_ch_free_string(path); + } +} + +void +tp_collect_gauges(tp_gauges_t *out) +{ + long cur_connections; + + if (out == NULL) { + return; + } + + out->cur_work_queue = (uint64_t)get_work_q_size(); + out->max_work_queue = (uint64_t)get_work_q_size_max(); + out->cur_busy_workers = (uint64_t)get_busy_worker_count(); + out->max_busy_workers = (uint64_t)get_max_busy_worker_count(); + out->ops_initiated = (uint64_t)g_get_num_ops_initiated(); + out->ops_completed = (uint64_t)g_get_num_ops_completed(); + + cur_connections = g_get_current_conn_count(); + out->cur_connections = cur_connections > 0 ? (uint64_t)cur_connections : 0; +} + +static void +tp_stats_publish_gauges(tp_stats_header_t *header, tp_gauges_t *gauges) +{ + slapi_atomic_store_64(&header->cur_work_queue, gauges->cur_work_queue, __ATOMIC_RELAXED); + slapi_atomic_store_64(&header->max_work_queue, gauges->max_work_queue, __ATOMIC_RELAXED); + slapi_atomic_store_64(&header->cur_busy_workers, gauges->cur_busy_workers, __ATOMIC_RELAXED); + slapi_atomic_store_64(&header->max_busy_workers, gauges->max_busy_workers, __ATOMIC_RELAXED); + slapi_atomic_store_64(&header->ops_initiated, gauges->ops_initiated, __ATOMIC_RELAXED); + slapi_atomic_store_64(&header->ops_completed, gauges->ops_completed, __ATOMIC_RELAXED); + slapi_atomic_store_64(&header->cur_connections, gauges->cur_connections, __ATOMIC_RELAXED); +} + +static void +tp_stats_heartbeat(time_t when __attribute__((unused)), void *arg __attribute__((unused))) +{ + tp_gauges_t gauges = {0}; + tp_stats_header_t *header = tp_stats_header; + + if (header == NULL) { + return; + } + + tp_collect_gauges(&gauges); + tp_stats_publish_gauges(header, &gauges); + slapi_atomic_store_64(&header->heartbeat_wall_sec, (uint64_t)slapi_current_utc_time(), __ATOMIC_RELAXED); + slapi_atomic_store_64(&header->heartbeat_mono_ns, tp_stats_mono_ns(), __ATOMIC_RELEASE); +} + +int +tp_stats_init(uint32_t max_workers) +{ + tp_stats_header_t *header = NULL; + void *mapping = MAP_FAILED; + struct stat st = {0}; + char *path = NULL; + int fd = -1; + int rc; + size_t len; + + if (!config_get_thread_pool_stats()) { + slapi_log_err(SLAPI_LOG_INFO, TP_STATS_COMPONENT, + "Thread-pool status diagnostics disabled by " CONFIG_THREAD_POOL_STATS_ATTRIBUTE "\n"); + return 0; + } + + if (max_workers == 0) { + slapi_log_err(SLAPI_LOG_WARNING, TP_STATS_COMPONENT, + "Thread-pool status mmap disabled: worker count is zero\n"); + return -1; + } + + if (tp_stats_header != NULL) { + return 0; + } + + len = TP_STATS_HEADER_SIZE + ((size_t)max_workers * TP_STATS_WORKER_SLOT_SIZE); + path = tp_stats_make_path(); + if (path == NULL) { + slapi_log_err(SLAPI_LOG_WARNING, TP_STATS_COMPONENT, + "Thread-pool status mmap disabled: could not resolve runtime path\n"); + return -1; + } + + if (tp_stats_prepare_dir(path) != 0) { + slapi_ch_free_string(&path); + return -1; + } + + /* Best effort: on any failure the leftover falls through to the unlink below */ + tp_stats_archive_crash_file(path); + + if (unlink(path) != 0 && errno != ENOENT) { + int err = errno; + struct stat lst = {0}; + const char *kind = "file"; + + if (lstat(path, &lst) == 0 && S_ISLNK(lst.st_mode)) { + kind = "symlink"; + } + slapi_log_err(SLAPI_LOG_WARNING, TP_STATS_COMPONENT, + "Could not remove stale thread-pool status %s %s: %d (%s). " + "Possible SELinux denial; thread-pool status diagnostics are disabled\n", + kind, path, err, slapd_system_strerror(err)); + slapi_ch_free_string(&path); + return -1; + } + + fd = open(path, O_RDWR | O_CREAT | O_EXCL | O_NOFOLLOW | O_CLOEXEC, 0640); + if (fd < 0) { + int err = errno; + slapi_log_err(SLAPI_LOG_WARNING, TP_STATS_COMPONENT, + "Could not create thread-pool status file %s: %d (%s)\n", + path, err, slapd_system_strerror(err)); + tp_stats_cleanup_open_failure(fd, &path); + return -1; + } + + if (fstat(fd, &st) != 0) { + int err = errno; + slapi_log_err(SLAPI_LOG_WARNING, TP_STATS_COMPONENT, + "Could not inspect thread-pool status file %s: %d (%s)\n", + path, err, slapd_system_strerror(err)); + tp_stats_cleanup_open_failure(fd, &path); + return -1; + } + + if (!S_ISREG(st.st_mode) || st.st_uid != geteuid() || st.st_nlink != 1) { + slapi_log_err(SLAPI_LOG_WARNING, TP_STATS_COMPONENT, + "Refusing unsafe thread-pool status file %s (mode=%o uid=%ld nlink=%ld)\n", + path, (unsigned int)st.st_mode, (long)st.st_uid, (long)st.st_nlink); + tp_stats_cleanup_open_failure(fd, &path); + return -1; + } + + if (fchmod(fd, 0640) != 0) { + int err = errno; + slapi_log_err(SLAPI_LOG_WARNING, TP_STATS_COMPONENT, + "Could not set permissions on thread-pool status file %s: %d (%s)\n", + path, err, slapd_system_strerror(err)); + tp_stats_cleanup_open_failure(fd, &path); + return -1; + } + + /* + * Reserve backing pages up front: ftruncate alone leaves a sparse file, + * and a store into an unbacked page takes SIGBUS when the filesystem is + * full. With the reservation, slot and heartbeat writes can never fault. + * posix_fallocate returns the error code instead of setting errno. + */ + rc = posix_fallocate(fd, 0, (off_t)len); + if (rc == EOPNOTSUPP || rc == EINVAL) { + /* Filesystem without fallocate support: fall back to a sparse file. */ + if (ftruncate(fd, (off_t)len) != 0) { + int err = errno; + slapi_log_err(SLAPI_LOG_WARNING, TP_STATS_COMPONENT, + "Could not size thread-pool status file %s: %d (%s)\n", + path, err, slapd_system_strerror(err)); + tp_stats_cleanup_open_failure(fd, &path); + return -1; + } + } else if (rc != 0) { + slapi_log_err(SLAPI_LOG_WARNING, TP_STATS_COMPONENT, + "Could not reserve space for thread-pool status file %s: %d (%s)\n", + path, rc, slapd_system_strerror(rc)); + tp_stats_cleanup_open_failure(fd, &path); + return -1; + } + + mapping = mmap(NULL, len, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0); + if (mapping == MAP_FAILED) { + int err = errno; + slapi_log_err(SLAPI_LOG_WARNING, TP_STATS_COMPONENT, + "Could not map thread-pool status file %s: %d (%s)\n", + path, err, slapd_system_strerror(err)); + tp_stats_cleanup_open_failure(fd, &path); + return -1; + } + + memset(mapping, 0, len); + header = (tp_stats_header_t *)mapping; + header->ver_major = TP_STATS_VER_MAJOR; + header->ver_minor = TP_STATS_VER_MINOR; + header->header_size = TP_STATS_HEADER_SIZE; + header->worker_slot_size = TP_STATS_WORKER_SLOT_SIZE; + header->max_workers = max_workers; + header->server_pid = (uint64_t)getpid(); + header->start_wall_sec = (uint64_t)slapi_current_utc_time(); + + tp_stats_header = header; + tp_stats_fd = fd; + tp_stats_len = len; + tp_stats_max_workers = max_workers; + tp_stats_path = path; + + tp_stats_heartbeat(0, NULL); + slapi_atomic_store_64(&header->magic, TP_STATS_MAGIC, __ATOMIC_RELEASE); + + slapi_log_err(SLAPI_LOG_INFO, TP_STATS_COMPONENT, + "Thread-pool status mmap ready at %s (%zu bytes, %" PRIu32 " workers)\n", + tp_stats_path, tp_stats_len, max_workers); + return 0; +} + +/* + * Register the periodic heartbeat. Must run only after init_op_threads(): + * the callback reads per_thread_snmp_vars through g_get_num_ops_initiated(), + * and alloc_per_thread_snmp_vars() reallocates that array with no + * synchronization against readers. + */ +void +tp_stats_start_heartbeat(void) +{ + if (tp_stats_header == NULL || tp_stats_eq_ctx != NULL) { + return; + } + + tp_stats_eq_ctx = slapi_eq_repeat_rel(tp_stats_heartbeat, NULL, + slapi_current_rel_time_t(), + TP_STATS_HEARTBEAT_INTERVAL_MS); + if (tp_stats_eq_ctx == NULL) { + slapi_log_err(SLAPI_LOG_WARNING, TP_STATS_COMPONENT, + "Thread-pool status file %s was created, but heartbeat registration failed\n", + tp_stats_path); + } +} + +void +tp_stats_close(void) +{ + tp_stats_header_t *header = tp_stats_header; + + if (header == NULL) { + return; + } + + if (tp_stats_eq_ctx != NULL) { + slapi_eq_cancel_rel(tp_stats_eq_ctx); + tp_stats_eq_ctx = NULL; + } + + tp_store_32(&header->shutdown_clean, 1, __ATOMIC_RELEASE); + + if (tp_stats_path != NULL) { + if (unlink(tp_stats_path) != 0 && errno != ENOENT) { + int err = errno; + slapi_log_err(SLAPI_LOG_WARNING, TP_STATS_COMPONENT, + "Could not remove thread-pool status file %s: %d (%s)\n", + tp_stats_path, err, slapd_system_strerror(err)); + } + slapi_ch_free_string(&tp_stats_path); + } + + if (tp_stats_fd >= 0) { + close(tp_stats_fd); + tp_stats_fd = -1; + } + + /* + * Do not munmap: worker threads are unjoinable and a late slot write into + * an unmapped region would crash shutdown. The mapping dies with ns-slapd. + */ + tp_stats_header = NULL; + tp_stats_len = 0; + tp_stats_max_workers = 0; +} + +void +tp_stats_worker_idle(uint32_t worker_idx) +{ + tp_worker_slot_t *slot = tp_stats_get_slot(worker_idx); + + if (slot == NULL) { + return; + } + + slapi_atomic_store_64(&slot->conn_id, 0, __ATOMIC_RELAXED); + slapi_atomic_store_64(&slot->op_id, 0, __ATOMIC_RELAXED); + tp_store_32(&slot->op_tag, 0, __ATOMIC_RELAXED); + slapi_atomic_store_64(&slot->start_ns, 0, __ATOMIC_RELAXED); + tp_store_32(&slot->state, TP_WORKER_STATE_IDLE, __ATOMIC_RELEASE); +} + +void +tp_stats_worker_busy(uint32_t worker_idx) +{ + tp_worker_slot_t *slot = tp_stats_get_slot(worker_idx); + + if (slot == NULL) { + return; + } + + tp_store_32(&slot->state, TP_WORKER_STATE_BUSY, __ATOMIC_RELEASE); +} + +void +tp_stats_worker_operation_start(uint32_t worker_idx, uint64_t conn_id, uint64_t op_id, uint32_t op_tag) +{ + tp_worker_slot_t *slot = tp_stats_get_slot(worker_idx); + + if (slot == NULL) { + return; + } + + slapi_atomic_store_64(&slot->conn_id, conn_id, __ATOMIC_RELAXED); + slapi_atomic_store_64(&slot->op_id, op_id, __ATOMIC_RELAXED); + tp_store_32(&slot->op_tag, op_tag, __ATOMIC_RELAXED); + slapi_atomic_store_64(&slot->start_ns, tp_stats_mono_ns(), __ATOMIC_RELAXED); + tp_store_32(&slot->state, TP_WORKER_STATE_BUSY, __ATOMIC_RELEASE); +} + +void +tp_stats_worker_operation_done(uint32_t worker_idx) +{ + tp_worker_slot_t *slot = tp_stats_get_slot(worker_idx); + + if (slot == NULL) { + return; + } + + /* + * start_ns is the in-flight sentinel (op_id 0 is a valid first op on a + * connection): clear it first so a reader that still sees it set finds + * the op fields intact. + */ + slapi_atomic_store_64(&slot->start_ns, 0, __ATOMIC_RELAXED); + tp_store_32(&slot->op_tag, 0, __ATOMIC_RELAXED); + slapi_atomic_store_64(&slot->op_id, 0, __ATOMIC_RELAXED); +} + +void +tp_stats_worker_exited(uint32_t worker_idx) +{ + tp_worker_slot_t *slot = tp_stats_get_slot(worker_idx); + + if (slot == NULL) { + return; + } + + tp_store_32(&slot->state, TP_WORKER_STATE_EXITED, __ATOMIC_RELEASE); +} + +void +tp_stats_as_entry(Slapi_Entry *e) +{ + tp_stats_header_t *header = tp_stats_header; + struct berval val; + struct berval *vals[2]; + uint64_t now_ns; + + vals[0] = &val; + vals[1] = NULL; + attrlist_delete(&e->e_attrs, TP_STATS_ATTR_THREADPOOL_WORKER); + + if (header == NULL) { + return; + } + + now_ns = tp_stats_mono_ns(); + for (size_t i = 0; i < header->max_workers; i++) { + char buf[256]; + char op_buf[32]; + uint32_t state; + uint32_t op_tag; + uint64_t start_ns; + uint64_t duration_ns = 0; + const char *op_name; + + tp_worker_slot_t *slot = tp_stats_slot_at(header, i); + + state = tp_load_32(&slot->state, __ATOMIC_ACQUIRE); + if (state == TP_WORKER_STATE_UNUSED) { + continue; + } + + op_tag = tp_load_32(&slot->op_tag, __ATOMIC_RELAXED); + start_ns = slapi_atomic_load_64(&slot->start_ns, __ATOMIC_RELAXED); + /* start_ns is the in-flight sentinel; op_id 0 is a valid first op */ + if (start_ns != 0 && now_ns >= start_ns) { + duration_ns = now_ns - start_ns; + } + + op_name = tp_stats_op_name(op_tag, op_buf, sizeof(op_buf)); + snprintf(buf, sizeof(buf), + "worker=%zu state=%s op=%s duration_ns=%" PRIu64, + i + 1, tp_stats_state_name(state), op_name, duration_ns); + val.bv_val = buf; + val.bv_len = strlen(buf); + attrlist_merge(&e->e_attrs, TP_STATS_ATTR_THREADPOOL_WORKER, vals); + } +} diff --git a/ldap/servers/slapd/threadpool_stats.h b/ldap/servers/slapd/threadpool_stats.h new file mode 100644 index 000000000..4f30ef098 --- /dev/null +++ b/ldap/servers/slapd/threadpool_stats.h @@ -0,0 +1,94 @@ +/** BEGIN COPYRIGHT BLOCK + * Copyright (C) 2026 Red Hat, Inc. + * All rights reserved. + * + * License: GPL (version 3 or any later version). + * See LICENSE for details. + * END COPYRIGHT BLOCK **/ + +#pragma once + +#include + +typedef struct slapi_entry Slapi_Entry; + +#define TP_STATS_MAGIC 0x54504f4f4c535431ULL /* "TPOOLST1" */ +#define TP_STATS_VER_MAJOR 1 +#define TP_STATS_VER_MINOR 0 +#define TP_STATS_HEADER_SIZE 4096 +#define TP_STATS_WORKER_SLOT_SIZE 64 +#define TP_STATS_ATTR_THREADPOOL_WORKER "threadpoolworker" + +typedef enum { + TP_WORKER_STATE_UNUSED = 0, + TP_WORKER_STATE_IDLE = 1, + TP_WORKER_STATE_BUSY = 2, + TP_WORKER_STATE_EXITED = 3, +} tp_worker_state_t; + +typedef struct { + uint64_t cur_work_queue; + uint64_t max_work_queue; + uint64_t cur_busy_workers; + uint64_t max_busy_workers; + uint64_t ops_initiated; + uint64_t ops_completed; + uint64_t cur_connections; +} tp_gauges_t; + +/* + * Thread-pool status mmap ABI. + * + * The file is machine-local only: all integers are host-endian, fixed width, + * and naturally aligned. It contains no time_t, pointers, strings, DNs, IPs, + * filters, or other request content. + * + * start_ns doubles as the operation-in-flight sentinel. op_id 0 is a valid + * value (the first operation on a connection) and must not be used as one. + */ +typedef struct __attribute__((aligned(64))) tp_worker_slot { + uint32_t state; + uint32_t op_tag; + uint64_t conn_id; + uint64_t op_id; + uint64_t start_ns; +} tp_worker_slot_t; + +typedef struct tp_stats_header { + uint64_t magic; + uint16_t ver_major; + uint16_t ver_minor; + uint32_t header_size; + uint32_t worker_slot_size; + uint32_t max_workers; + uint64_t server_pid; + uint64_t start_wall_sec; + uint64_t heartbeat_mono_ns; + uint64_t heartbeat_wall_sec; + uint32_t shutdown_clean; + uint32_t pad0; + uint64_t cur_work_queue; + uint64_t max_work_queue; + uint64_t cur_busy_workers; + uint64_t max_busy_workers; + uint64_t ops_initiated; + uint64_t ops_completed; + uint64_t cur_connections; + uint8_t reserved[3976]; +} tp_stats_header_t; + +_Static_assert(sizeof(tp_worker_slot_t) == TP_STATS_WORKER_SLOT_SIZE, + "tp_worker_slot_t size must remain ABI-stable"); +_Static_assert(sizeof(tp_stats_header_t) == TP_STATS_HEADER_SIZE, + "tp_stats_header_t size must remain ABI-stable"); + +void tp_collect_gauges(tp_gauges_t *out); +int tp_stats_init(uint32_t max_workers); +void tp_stats_start_heartbeat(void); +void tp_stats_close(void); +void tp_stats_worker_idle(uint32_t worker_idx); +void tp_stats_worker_busy(uint32_t worker_idx); +void tp_stats_worker_operation_start(uint32_t worker_idx, uint64_t conn_id, uint64_t op_id, uint32_t op_tag); +void tp_stats_worker_operation_done(uint32_t worker_idx); +void tp_stats_worker_exited(uint32_t worker_idx); +void tp_stats_as_entry(Slapi_Entry *e); diff --git a/src/lib389/cli/dsctl b/src/lib389/cli/dsctl index 9ab830c69..9028f4acd 100755 --- a/src/lib389/cli/dsctl +++ b/src/lib389/cli/dsctl @@ -26,6 +26,7 @@ from lib389.cli_ctl import dbgen as cli_dbgen from lib389.cli_ctl import dsrc as cli_dsrc from lib389.cli_ctl import cockpit as cli_cockpit from lib389.cli_ctl import dblib as cli_dblib +from lib389.cli_ctl import threadpool as cli_threadpool from lib389.cli_ctl.instance import instance_remove_all from lib389.cli_base import ( disconnect_instance, @@ -61,6 +62,7 @@ cli_dbgen.create_parser(subparsers) cli_dsrc.create_parser(subparsers) cli_cockpit.create_parser(subparsers) cli_dblib.create_parser(subparsers) +cli_threadpool.create_parser(subparsers) argcomplete.autocomplete(parser) diff --git a/src/lib389/lib389/cli_ctl/threadpool.py b/src/lib389/lib389/cli_ctl/threadpool.py new file mode 100644 index 000000000..009740a1c --- /dev/null +++ b/src/lib389/lib389/cli_ctl/threadpool.py @@ -0,0 +1,468 @@ +# --- BEGIN COPYRIGHT BLOCK --- +# Copyright (C) 2026 Red Hat, Inc. +# All rights reserved. +# +# License: GPL (version 3 or any later version). +# See LICENSE for details. +# --- END COPYRIGHT BLOCK --- + +import errno +import json +import mmap +import os +import re +import stat +import struct +import time + +import psutil + +from lib389._constants import DN_CONFIG +from lib389.cli_base import CustomHelpFormatter +from lib389.dseldif import DSEldif + + +# File format constants; they mirror ldap/servers/slapd/threadpool_stats.h +# and must stay in sync with it. +TP_STATS_MAGIC = 0x54504f4f4c535431 # "TPOOLST1" +TP_STATS_VER_MAJOR = 1 +TP_STATS_HEADER_SIZE = 4096 +TP_STATS_WORKER_SLOT_SIZE = 64 + +# Byte-for-byte mirror of tp_stats_header_t. Each entry is +# (field name, struct format char); "4x" skips the C struct's pad0 field. +HEADER_FIELDS = [ + ("magic", "Q"), + ("ver_major", "H"), + ("ver_minor", "H"), + ("header_size", "I"), + ("worker_slot_size", "I"), + ("max_workers", "I"), + ("server_pid", "Q"), + ("start_wall_sec", "Q"), + ("heartbeat_mono_ns", "Q"), + ("heartbeat_wall_sec", "Q"), + ("shutdown_clean", "I"), + (None, "4x"), + ("cur_work_queue", "Q"), + ("max_work_queue", "Q"), + ("cur_busy_workers", "Q"), + ("max_busy_workers", "Q"), + ("ops_initiated", "Q"), + ("ops_completed", "Q"), + ("cur_connections", "Q"), +] +HEADER_FORMAT = "@" + "".join(fmt for _, fmt in HEADER_FIELDS) +HEADER_NAMES = [name for name, _ in HEADER_FIELDS if name] + +# Byte-for-byte mirror of tp_worker_slot_t; the slot is padded to +# TP_STATS_WORKER_SLOT_SIZE by its alignment. +WORKER_FIELDS = [ + ("state", "I"), + ("op_tag", "I"), + ("conn_id", "Q"), + ("op_id", "Q"), + ("start_ns", "Q"), +] +WORKER_FORMAT = "@" + "".join(fmt for _, fmt in WORKER_FIELDS) + +# Python counterpart of the _Static_asserts in threadpool_stats.h +assert struct.calcsize(HEADER_FORMAT) <= TP_STATS_HEADER_SIZE +assert struct.calcsize(WORKER_FORMAT) <= TP_STATS_WORKER_SLOT_SIZE + +# Reader-side staleness heuristics, not part of the file format +NS_PER_SEC = 1_000_000_000 +STALE_HEARTBEAT_NS = 30 * NS_PER_SEC +IMPLAUSIBLE_HEARTBEAT_NS = 365 * 24 * 3600 * NS_PER_SEC + +# tp_worker_state_t values +STATE_NAMES = { + 0: "unused", + 1: "idle", + 2: "busy", + 3: "exited", +} + +# LDAP protocol request tags (LDAP_REQ_* in ldap.h) +OP_NAMES = { + 0x60: "bind", + 0x42: "unbind", + 0x63: "search", + 0x66: "modify", + 0x68: "add", + 0x4A: "delete", + 0x6C: "modrdn", + 0x6E: "compare", + 0x50: "abandon", + 0x77: "extended", +} + + +def _server_file_prefix(serverid): + if serverid.startswith("slapd-"): + return serverid + return f"slapd-{serverid}" + + +def _crash_archives(path): + """Crash archives preserved for the status file at path, oldest first""" + dirname, base = os.path.split(path) + pattern = re.compile(re.escape(base) + r"\.\d{8}-\d{6}$") + try: + names = sorted(name for name in os.listdir(dirname) if pattern.match(name)) + except OSError: + return [] + return [os.path.join(dirname, name) for name in names] + + +def _config_threadnumber(dse): + value = dse.get(DN_CONFIG, "nsslapd-threadnumber", single=True, lower=True) + if value is None: + return None + try: + parsed = int(value) + except ValueError: + return None + return parsed if parsed > 0 else None + + +def _config_tp_stats_enabled(dse): + value = dse.get(DN_CONFIG, "nsslapd-thread-pool-stats", single=True, lower=True) + if value is None: + return True + return value.lower() != "off" + + +def _open_threadpool_file(path, inst, tp_stats_enabled, explicit=False): + try: + fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW) + except FileNotFoundError: + if explicit: + raise ValueError(f"thread-pool status file not found: {path}") + if inst.status(): + if not tp_stats_enabled: + raise ValueError( + "thread-pool status is disabled by nsslapd-thread-pool-stats in cn=config" + ) + raise ValueError( + "server is running but the thread-pool status file is missing " + "(initialization may have failed - check the errors log; " + "nsslapd-thread-pool-stats was switched on without a restart; " + "or the server predates this feature, or nsslapd-rundir mismatch)" + ) + raise ValueError("instance is not running (status file is removed on clean shutdown)") + except PermissionError: + raise ValueError("permission denied; run as root or a member of the dirsrv group") + except OSError as e: + if e.errno == errno.ELOOP: + raise ValueError("refusing to read thread-pool status through a symlink") + raise + + return fd + + +def _validate_stat(path, st): + if not stat.S_ISREG(st.st_mode): + raise ValueError(f"refusing to read non-regular thread-pool status file: {path}") + if st.st_size < TP_STATS_HEADER_SIZE: + raise ValueError( + f"thread-pool status file is too short: {st.st_size} bytes " + f"(expected at least {TP_STATS_HEADER_SIZE})" + ) + + +def _unpack_header(mm): + # _validate_stat checked the fstat size, but the mapping is what we read: + # guard against the file shrinking between fstat and mmap + if len(mm) < TP_STATS_HEADER_SIZE: + raise ValueError("thread-pool status header is truncated") + + header = dict(zip(HEADER_NAMES, struct.unpack_from(HEADER_FORMAT, mm, 0))) + + if header["magic"] != TP_STATS_MAGIC: + raise ValueError("bad thread-pool status magic; refusing to parse file") + if header["ver_major"] != TP_STATS_VER_MAJOR: + raise ValueError( + f"unsupported thread-pool status version " + f"{header['ver_major']}.{header['ver_minor']}" + ) + if header["header_size"] != TP_STATS_HEADER_SIZE: + raise ValueError( + f"unsupported thread-pool status header size {header['header_size']}" + ) + if header["worker_slot_size"] != TP_STATS_WORKER_SLOT_SIZE: + raise ValueError( + f"unsupported thread-pool worker slot size {header['worker_slot_size']}" + ) + if header["max_workers"] < 1 or header["max_workers"] > 65535: + raise ValueError(f"invalid thread-pool worker count {header['max_workers']}") + + expected_size = header["header_size"] + (header["max_workers"] * header["worker_slot_size"]) + if expected_size > len(mm): + raise ValueError( + f"thread-pool status file is truncated: {len(mm)} bytes " + f"(expected at least {expected_size})" + ) + + return header + + +def _state_name(state): + return STATE_NAMES.get(state, f"unknown-{state}") + + +def _op_name(op_tag): + if op_tag == 0: + return "" + return OP_NAMES.get(op_tag, str(op_tag)) + + +def _duration_ns(now_ns, start_ns): + if start_ns == 0 or now_ns < start_ns: + return 0 + return now_ns - start_ns + + +def _unpack_workers(mm, header, now_ns): + workers = [] + for idx in range(header["max_workers"]): + offset = header["header_size"] + (idx * header["worker_slot_size"]) + state, op_tag, conn_id, op_id, start_ns = struct.unpack_from(WORKER_FORMAT, mm, offset) + if state == 0: + continue + # start_ns is the in-flight sentinel; op_id 0 is a valid first op + in_flight = start_ns != 0 + workers.append({ + "idx": idx + 1, + "state": _state_name(state), + "op": _op_name(op_tag), + "conn": conn_id if conn_id != 0 else None, + "op_id": op_id if in_flight else None, + "duration_ns": _duration_ns(now_ns, start_ns), + }) + return workers + + +def _pid_warnings(pid): + """Return (warnings, pid_alive); pid_alive means a live ns-slapd owns the pid""" + warnings = [] + if pid == 0: + warnings.append("status file does not contain a valid server pid") + return warnings, False + + try: + name = psutil.Process(pid).name() + except (psutil.NoSuchProcess, psutil.ZombieProcess): + warnings.append(f"stale file from a crashed or killed server (pid {pid} is not running)") + return warnings, False + except psutil.AccessDenied: + warnings.append(f"server pid {pid} exists but process name could not be inspected") + return warnings, True + + if name != "ns-slapd": + warnings.append(f"stale file: pid {pid} belongs to {name!r}, not 'ns-slapd'") + return warnings, False + return warnings, True + + +def _heartbeat_age(now_ns, heartbeat_ns): + if heartbeat_ns == 0: + return None + return now_ns - heartbeat_ns + + +def _heartbeat_warnings(pid_alive, age_ns): + warnings = [] + if age_ns is None: + warnings.append("thread-pool heartbeat has never been written") + elif age_ns < 0: + warnings.append("thread-pool heartbeat is from a different monotonic-clock domain") + elif age_ns > IMPLAUSIBLE_HEARTBEAT_NS: + warnings.append("thread-pool heartbeat age is implausible; the file may predate a reboot") + elif age_ns > STALE_HEARTBEAT_NS: + if pid_alive: + warnings.append("server process exists but diagnostics are stale; server may be stalled") + else: + warnings.append("thread-pool diagnostics are stale") + return warnings + + +def _read_threadpool_status(inst, file_path=None): + warnings = [] + archives = [] + if file_path is not None: + path = file_path + configured_threads = None + tp_stats_enabled = True + else: + dse = DSEldif(inst) + rundir = dse.get(DN_CONFIG, "nsslapd-rundir", single=True, lower=True) + if rundir is None: + rundir = inst.ds_paths.run_dir + warnings.append("nsslapd-rundir is missing from dse.ldif; using lib389 run_dir fallback") + path = os.path.join(rundir, f"{_server_file_prefix(inst.serverid)}.monitor", "threadpool") + archives = _crash_archives(path) + configured_threads = _config_threadnumber(dse) + tp_stats_enabled = _config_tp_stats_enabled(dse) + + try: + fd = _open_threadpool_file(path, inst, tp_stats_enabled, explicit=file_path is not None) + except ValueError as e: + if archives: + raise ValueError(f"{e}; {len(archives)} crash archive(s) present, newest: {archives[-1]}") + raise + try: + st = os.fstat(fd) + _validate_stat(path, st) + with mmap.mmap(fd, 0, access=mmap.ACCESS_READ) as mm: + header = _unpack_header(mm) + now_ns = time.monotonic_ns() + age_ns = _heartbeat_age(now_ns, header["heartbeat_mono_ns"]) + pid_warnings, pid_alive = _pid_warnings(header["server_pid"]) + warnings.extend(pid_warnings) + warnings.extend(_heartbeat_warnings(pid_alive, age_ns)) + + if header["shutdown_clean"] != 0: + warnings.append("clean shutdown leftover") + if configured_threads is not None and configured_threads != header["max_workers"]: + warnings.append( + f"dse.ldif nsslapd-threadnumber is {configured_threads}, " + f"but status file was sized for {header['max_workers']} workers" + ) + + workers = _unpack_workers(mm, header, now_ns) + finally: + os.close(fd) + + if not tp_stats_enabled: + warnings.append( + "nsslapd-thread-pool-stats is off in cn=config; the running server " + "keeps publishing diagnostics until it is restarted" + ) + if archives: + warnings.append( + f"{len(archives)} crash archive(s) in {os.path.dirname(path)}, " + f"newest: {os.path.basename(archives[-1])} (read with --file)" + ) + + age_sec = None if age_ns is None else age_ns / NS_PER_SEC + start_wall = header["start_wall_sec"] + uptime_sec = max(0, int(time.time()) - start_wall) if start_wall else None + + return { + "type": "result", + "instance": inst.serverid, + "path": path, + "pid": header["server_pid"], + "version": { + "major": header["ver_major"], + "minor": header["ver_minor"], + }, + "start_wall_sec": start_wall, + "uptime_sec": uptime_sec, + "heartbeat_age_sec": age_sec, + "heartbeat_wall_sec": header["heartbeat_wall_sec"], + "pool": { + "max_workers": header["max_workers"], + "cur_busy_workers": header["cur_busy_workers"], + "max_busy_workers": header["max_busy_workers"], + "cur_work_queue": header["cur_work_queue"], + "max_work_queue": header["max_work_queue"], + "ops_initiated": header["ops_initiated"], + "ops_completed": header["ops_completed"], + "cur_connections": header["cur_connections"], + }, + "workers": workers, + "warnings": warnings, + } + + +def _format_seconds(value): + if value is None: + return "unknown" + return f"{value:.3f}s" + + +def _format_duration_ns(duration_ns): + if duration_ns == 0: + return "-" + seconds = duration_ns / NS_PER_SEC + if seconds < 1: + return f"{seconds * 1000:.1f}ms" + return f"{seconds:.3f}s" + + +def _format_optional(value): + return "-" if value is None else str(value) + + +def _emit_text(log, status): + pool = status["pool"] + log.info(f"Instance: {status['instance']}") + log.info(f"Path: {status['path']}") + log.info(f"PID: {status['pid']}") + log.info(f"Uptime: {_format_seconds(status['uptime_sec'])}") + log.info(f"Heartbeat age: {_format_seconds(status['heartbeat_age_sec'])}") + log.info( + "Workers: " + f"{pool['cur_busy_workers']}/{pool['max_workers']} busy " + f"(max {pool['max_busy_workers']})" + ) + log.info( + "Queue: " + f"{pool['cur_work_queue']} current " + f"(max {pool['max_work_queue']})" + ) + log.info( + "Operations: " + f"{pool['ops_initiated']} initiated, " + f"{pool['ops_completed']} completed" + ) + log.info(f"Current connections: {pool['cur_connections']}") + + if status["warnings"]: + log.info("Warnings:") + for warning in status["warnings"]: + log.info(f" - {warning}") + + log.info("") + log.info(f"{'IDX':>5} {'STATE':<8} {'OP':<10} {'CONN':>12} {'OP-ID':>12} {'DURATION':>12}") + for worker in status["workers"]: + op = worker["op"].upper() if worker["op"] else "-" + log.info( + f"{worker['idx']:>5} " + f"{worker['state'].upper():<8} " + f"{op:<10} " + f"{_format_optional(worker['conn']):>12} " + f"{_format_optional(worker['op_id']):>12} " + f"{_format_duration_ns(worker['duration_ns']):>12}" + ) + + +def thread_pool_status(inst, log, args): + status = _read_threadpool_status(inst, file_path=args.file) + if args.json: + log.info(json.dumps(status, indent=4)) + else: + _emit_text(log, status) + + +def create_parser(subparsers): + thread_pool_parser = subparsers.add_parser( + "thread-pool", + help="Offline thread pool diagnostics read from the local mmap status file", + formatter_class=CustomHelpFormatter, + ) + subcommands = thread_pool_parser.add_subparsers(help="action") + + status_parser = subcommands.add_parser( + "status", + help="Display pool gauges and per-worker activity without an LDAP connection", + formatter_class=CustomHelpFormatter, + ) + status_parser.add_argument( + "--file", default=None, + help="Read this thread-pool status file instead of the instance's live file " + "(e.g. a crash file preserved as threadpool.YYYYMMDD-HHMMSS)", + ) + status_parser.set_defaults(func=thread_pool_status) diff --git a/src/lib389/lib389/monitor.py b/src/lib389/lib389/monitor.py index 8b4acd1db..d66039ee2 100644 --- a/src/lib389/lib389/monitor.py +++ b/src/lib389/lib389/monitor.py @@ -68,6 +68,13 @@ class Monitor(DSLdapObject): maxbusyworkers = self.get_attr_vals_utf8('maxbusyworkers') return (currentworkqueue, maxworkqueue, currentbusyworkers, maxbusyworkers) + def get_thread_pool_workers(self): + """Get sanitized per-worker thread pool status values from cn=monitor + + :returns: Values of threadpoolworker attribute of cn=monitor + """ + return self.get_attr_vals_utf8('threadpoolworker') + def get_backends(self): """Get backends related attributes value for cn=monitor @@ -207,6 +214,7 @@ class Monitor(DSLdapObject): 'maxworkqueue', 'currentbusyworkers', 'maxbusyworkers', + 'threadpoolworker', ]) status.update(stats) -- 2.54.0