diff --git a/dockerfiles/dev.Dockerfile b/dockerfiles/dev.Dockerfile index b96befe..7c7b353 100644 --- a/dockerfiles/dev.Dockerfile +++ b/dockerfiles/dev.Dockerfile @@ -3,7 +3,6 @@ FROM debian:stable RUN apt-get update && apt-get install -y \ build-essential \ clang-format \ - cmake \ curl \ gcc \ gdb \ @@ -25,6 +24,8 @@ RUN apt-get update && apt-get install -y \ zip \ && rm -rf /var/lib/apt/lists/* +RUN pip3 install --break-system-packages --no-cache-dir "cmake>=4.3,<5" + ENV VCPKG_FORCE_SYSTEM_BINARIES=1 # clone and install vcpkg diff --git a/src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.cpp b/src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.cpp index 0c56872..76af8c6 100644 --- a/src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.cpp +++ b/src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.cpp @@ -35,6 +35,17 @@ BALL_LOG_SET_NAMESPACE_CATEGORY("RMQAMQP.HOSTHEALTHMONITOR") const bool RESPECT_HOST_HEALTH = true; +void syncReceiveChannelsToHostHealth(rmqamqp::Connection& connection, + HostHealthMonitor::HostHealth health) +{ + if (health == HostHealthMonitor::HEALTHY) { + connection.resumeReceiveChannels(RESPECT_HOST_HEALTH); + } + else if (health == HostHealthMonitor::UNHEALTHY) { + connection.pauseReceiveChannels(RESPECT_HOST_HEALTH); + } +} + } // namespace HostHealthMonitor::HostHealthMonitor( @@ -42,6 +53,9 @@ HostHealthMonitor::HostHealthMonitor( rmqp::MetricPublisher* metricPublisher) : d_hostHealthConfig(hostHealthConfig) , d_currentTries(0) +// Fail-safe: assume unhealthy until the first check, so connections that +// register beforehand start paused. +, d_latestHealthCheckResult(UNHEALTHY) , d_timer() , d_metricPublisher(metricPublisher) { @@ -65,7 +79,9 @@ void HostHealthMonitor::start( bdlf::BindUtil::bind(&HostHealthMonitor::handleTimerFired, weak_from_this(), bdlf::PlaceHolders::_1)); - scheduleNextCheck(); + // Fire the first check immediately; checkHealth() then reschedules at + // pollInterval. + d_timer->reset(bsls::TimeInterval(0)); } void HostHealthMonitor::stop() @@ -83,6 +99,18 @@ void HostHealthMonitor::registerConnection( d_metricPublisher->publishGauge(Metrics::HEALTH_AWARE_VHOSTS, static_cast(d_connections.size())); + + // Bring the connection into the current known state now, rather than + // waiting for the next poll. + bsl::shared_ptr connection = conn.lock(); + if (connection) { + BALL_LOG_INFO << "healthState=" << d_latestHealthCheckResult + << " action=" + << (d_latestHealthCheckResult == HEALTHY ? "resume" + : "pause") + << " reason=newly-registered-connection"; + syncReceiveChannelsToHostHealth(*connection, d_latestHealthCheckResult); + } } void HostHealthMonitor::handleTimerFired( @@ -121,9 +149,9 @@ void HostHealthMonitor::checkHealth() bsls::SystemTime::nowMonotonicClock(); const double durationMs = (endTime - startTime).totalMilliseconds(); - BALL_LOG_INFO << "Health check completed in " << durationMs - << " ms with result: " << healthCheckerResult - << ". Set health state to: " << result; + BALL_LOG_DEBUG << "event=health-check-completed durationMs=" + << durationMs << " result=" << healthCheckerResult + << " healthState=" << result; d_metricPublisher->publishSummary(Metrics::HEALTH_CHECK_DURATION_MS, durationMs); @@ -131,23 +159,24 @@ void HostHealthMonitor::checkHealth() const double HEALTHCHECK_DURATION_REPORTING_THRESHOLD = bsl::min(d_hostHealthConfig.pollInterval() * 1000.0 * 0.8, 1000.0); if (durationMs > HEALTHCHECK_DURATION_REPORTING_THRESHOLD) { - BALL_LOG_WARN << "Host health check took " << durationMs - << " ms which exceeds the threshold of " - << HEALTHCHECK_DURATION_REPORTING_THRESHOLD << " ms."; + BALL_LOG_WARN << "event=health-check-duration-exceeds-threshold " + "durationMs=" + << durationMs << " thresholdMs=" + << HEALTHCHECK_DURATION_REPORTING_THRESHOLD; d_metricPublisher->publishCounter( Metrics::HEALTH_CHECK_BLOCKED_EVENT_LOOP, 1.0); } } catch (const bsl::exception& e) { - BALL_LOG_ERROR << "Host health check failed with exception: " - << e.what(); + BALL_LOG_ERROR << "event=health-check-failed exception=\"" << e.what() + << "\""; result = RETRY; d_metricPublisher->publishCounter(Metrics::HEALTH_CHECK_FAILURES_TOTAL, 1.0); } catch (...) { - BALL_LOG_ERROR << "Host health check failed with unknown exception."; + BALL_LOG_ERROR << "event=health-check-failed exception=unknown"; result = RETRY; d_metricPublisher->publishCounter(Metrics::HEALTH_CHECK_FAILURES_TOTAL, @@ -156,9 +185,9 @@ void HostHealthMonitor::checkHealth() if (result == RETRY) { if (d_currentTries++ > d_hostHealthConfig.maxRetriesOnFailure()) { - BALL_LOG_ERROR << "Exceeded max retries on failure of " + BALL_LOG_ERROR << "event=max-retries-exceeded maxRetries=" << d_hostHealthConfig.maxRetriesOnFailure() - << ". Marking host as UNHEALTHY."; + << " action=mark-unhealthy"; result = UNHEALTHY; d_metricPublisher->publishGauge( @@ -166,12 +195,12 @@ void HostHealthMonitor::checkHealth() static_cast(d_currentTries)); } else { - BALL_LOG_WARN << "Current tries " << d_currentTries - << " do not exceed max retries on failure of " + BALL_LOG_WARN << "event=health-check-retry currentTries=" + << d_currentTries << " maxRetries=" << d_hostHealthConfig.maxRetriesOnFailure() - << ". Will retry after " + << " retryAfterSeconds=" << d_hostHealthConfig.pollInterval() - << " seconds. Will NOT pause the consumers yet."; + << " action=none"; if (d_metricPublisher) { d_metricPublisher->publishGauge( @@ -190,8 +219,18 @@ void HostHealthMonitor::processHealthResult(HostHealth result) { d_currentTries = 0; - BALL_LOG_DEBUG << (result == HEALTHY ? "Resuming" : "Pausing") - << " host health aware consumers."; + if (result != d_latestHealthCheckResult) { + BALL_LOG_INFO << "event=health-state-changed previousHealthState=" + << d_latestHealthCheckResult << " healthState=" << result; + } + + // Cache for registerConnection. RETRY returns early, so this only ever + // holds HEALTHY or UNHEALTHY. + d_latestHealthCheckResult = result; + + BALL_LOG_DEBUG << "healthState=" << result + << " action=" << (result == HEALTHY ? "resume" : "pause") + << " target=health-aware-consumers"; const double statusValue = (result == HEALTHY) ? 1.0 : 0.0; d_metricPublisher->publishGauge(Metrics::HEALTH_CHECK_STATUS, statusValue); @@ -225,12 +264,7 @@ void HostHealthMonitor::processHealthResult(HostHealth result) continue; } - if (result == HEALTHY) { - connection->resumeReceiveChannels(RESPECT_HOST_HEALTH); - } - else { - connection->pauseReceiveChannels(RESPECT_HOST_HEALTH); - } + syncReceiveChannelsToHostHealth(*connection, result); ++conn; } diff --git a/src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.h b/src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.h index d3d06f7..581c829 100644 --- a/src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.h +++ b/src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.h @@ -39,6 +39,12 @@ namespace rmqamqp { /// complete promptly, as it will delay AMQP heartbeats and message delivery /// for all connections by its execution time every \c pollInterval seconds. /// Each check schedules the next one via a timer (self-rescheduling pattern). +/// +/// One monitor exists per context: the checker runs once per \c pollInterval +/// regardless of connection or consumer count. The latest result is cached and +/// applied to every registered connection (and to connections registering +/// between checks, see \c registerConnection). Defaults to unhealthy until the +/// first check. class HostHealthMonitor : public bsl::enable_shared_from_this { public: @@ -64,16 +70,17 @@ class HostHealthMonitor ~HostHealthMonitor(); - /// Start the health monitoring timer. The first health check will fire - /// after one poll interval. + /// Start the health monitoring timer. The first health check fires + /// immediately (on the event loop); subsequent checks run every + /// \c pollInterval seconds. void start(const bsl::shared_ptr& timerFactory); /// Stop the health monitoring timer. void stop(); - /// Register a connection to be notified about host health changes. - /// When the host becomes unhealthy, the connection's receive channels - /// will be paused. When the host recovers, they will be resumed. + /// Register a connection to be paused/resumed as host health changes. + /// At registration the connection is brought into the current known state + /// immediately: resumed if the host is known healthy, paused otherwise. void registerConnection(const bsl::weak_ptr& connection); @@ -91,6 +98,7 @@ class HostHealthMonitor rmqt::HostHealthConfig d_hostHealthConfig; bsl::list > d_connections; unsigned int d_currentTries; + HostHealth d_latestHealthCheckResult; bsl::shared_ptr d_timer; rmqp::MetricPublisher* d_metricPublisher; }; diff --git a/src/tests/rmqamqp/rmqamqp_hosthealthmonitor.t.cpp b/src/tests/rmqamqp/rmqamqp_hosthealthmonitor.t.cpp index 5e172c5..ac9a6ae 100644 --- a/src/tests/rmqamqp/rmqamqp_hosthealthmonitor.t.cpp +++ b/src/tests/rmqamqp/rmqamqp_hosthealthmonitor.t.cpp @@ -236,6 +236,17 @@ TEST_F(HostHealthMonitorTests, HealthyHostResumesConnections) stepOnePollInterval(); } +TEST_F(HostHealthMonitorTests, FirstCheckFiresImmediately) +{ + d_configurableHealthChecker.d_nextResult = true; + + EXPECT_CALL(*d_connection, resumeReceiveChannels(true)).Times(1); + EXPECT_CALL(*d_connection, pauseReceiveChannels(_)).Times(0); + + // First check is scheduled with zero delay, so stepping by zero fires it. + d_timerFactory->step_time(bsls::TimeInterval(0)); +} + TEST_F(HostHealthMonitorTests, UnhealthyHostPausesConnections) { d_configurableHealthChecker.d_nextResult = false; @@ -246,6 +257,56 @@ TEST_F(HostHealthMonitorTests, UnhealthyHostPausesConnections) stepOnePollInterval(); } +TEST_F(HostHealthMonitorTests, RegisterOnUnhealthyHostPausesImmediately) +{ + // Drive one check that marks the host UNHEALTHY. + d_configurableHealthChecker.d_nextResult = false; + + EXPECT_CALL(*d_connection, pauseReceiveChannels(true)).Times(1); + stepAndClear(); + + bsl::shared_ptr lateConn = makeConnection("late-unhealthy"); + + EXPECT_CALL(*lateConn, pauseReceiveChannels(true)).Times(1); + EXPECT_CALL(*lateConn, resumeReceiveChannels(_)).Times(0); + + d_monitor->registerConnection(bsl::weak_ptr(lateConn)); +} + +TEST_F(HostHealthMonitorTests, RegisterOnHealthyHostResumesImmediately) +{ + // Drive one check that marks the host HEALTHY. + d_configurableHealthChecker.d_nextResult = true; + + EXPECT_CALL(*d_connection, resumeReceiveChannels(true)).Times(1); + stepAndClear(); + + bsl::shared_ptr lateConn = makeConnection("late-healthy"); + + EXPECT_CALL(*lateConn, resumeReceiveChannels(true)).Times(1); + EXPECT_CALL(*lateConn, pauseReceiveChannels(_)).Times(0); + + d_monitor->registerConnection(bsl::weak_ptr(lateConn)); +} + +TEST_F(HostHealthMonitorTests, RegisterBeforeFirstCheckPausesImmediately) +{ + // No check has run yet, so the monitor's fail-safe default (UNHEALTHY) + // applies. + bsl::shared_ptr monitor = + bsl::make_shared(d_config, d_metricPublisher.get()); + monitor->start(d_timerFactory); + + EXPECT_CALL(*d_metricPublisher, publishGauge(_, _, _)).Times(AtLeast(0)); + + bsl::shared_ptr earlyConn = makeConnection("early"); + + EXPECT_CALL(*earlyConn, pauseReceiveChannels(true)).Times(1); + EXPECT_CALL(*earlyConn, resumeReceiveChannels(_)).Times(0); + + monitor->registerConnection(bsl::weak_ptr(earlyConn)); +} + TEST_F(HostHealthMonitorTests, ExpiredConnectionIsRemovedAndNotUsed) { bsl::shared_ptr liveConn =