feat core: catch signals before component system is up and running

commit_hash:41ef856353d591c431d57baaea86c4d0ac4a9378

Author: segoon

.mapping.json

@@ -751,6 +751,11 @@
   "core/functional_tests/metrics/tests/conftest.py":"taxi/uservices/userver/core/functional_tests/metrics/tests/conftest.py",
   "core/functional_tests/metrics/tests/static/metrics_values.txt":"taxi/uservices/userver/core/functional_tests/metrics/tests/static/metrics_values.txt",
   "core/functional_tests/metrics/tests/test_metrics.py":"taxi/uservices/userver/core/functional_tests/metrics/tests/test_metrics.py",
+  "core/functional_tests/signal_during_boot/CMakeLists.txt":"taxi/uservices/userver/core/functional_tests/signal_during_boot/CMakeLists.txt",
+  "core/functional_tests/signal_during_boot/service.cpp":"taxi/uservices/userver/core/functional_tests/signal_during_boot/service.cpp",
+  "core/functional_tests/signal_during_boot/static_config.yaml":"taxi/uservices/userver/core/functional_tests/signal_during_boot/static_config.yaml",
+  "core/functional_tests/signal_during_boot/tests/conftest.py":"taxi/uservices/userver/core/functional_tests/signal_during_boot/tests/conftest.py",
+  "core/functional_tests/signal_during_boot/tests/test_rotate_logs.py":"taxi/uservices/userver/core/functional_tests/signal_during_boot/tests/test_rotate_logs.py",
   "core/functional_tests/slow_start/CMakeLists.txt":"taxi/uservices/userver/core/functional_tests/slow_start/CMakeLists.txt",
   "core/functional_tests/slow_start/config_vars.yaml":"taxi/uservices/userver/core/functional_tests/slow_start/config_vars.yaml",
   "core/functional_tests/slow_start/secure_data.json":"taxi/uservices/userver/core/functional_tests/slow_start/secure_data.json",

core/functional_tests/CMakeLists.txt

@@ -32,6 +32,9 @@ add_dependencies(${PROJECT_NAME} ${PROJECT_NAME}-static-service)
 add_subdirectory(slow_start)
 add_dependencies(${PROJECT_NAME} ${PROJECT_NAME}-slow-start)
 
+add_subdirectory(signal_during_boot)
+add_dependencies(${PROJECT_NAME} ${PROJECT_NAME}-signal-during-boot)
+
 add_subdirectory(tracing)
 add_dependencies(${PROJECT_NAME} ${PROJECT_NAME}-tracing)
 

core/functional_tests/signal_during_boot/CMakeLists.txt

@@ -0,0 +1,6 @@
+project(userver-core-tests-signal-during-boot CXX)
+
+add_executable(${PROJECT_NAME} service.cpp)
+target_link_libraries(${PROJECT_NAME} userver::core)
+
+userver_chaos_testsuite_add()

core/functional_tests/signal_during_boot/service.cpp

@@ -0,0 +1,35 @@
+#include <userver/utils/assert.hpp>
+
+#include <userver/clients/dns/component.hpp>
+#include <userver/clients/http/component.hpp>
+#include <userver/components/component.hpp>
+#include <userver/components/minimal_server_component_list.hpp>
+#include <userver/engine/sleep.hpp>
+#include <userver/http/url.hpp>
+#include <userver/server/handlers/http_handler_base.hpp>
+#include <userver/server/handlers/tests_control.hpp>
+#include <userver/testsuite/testsuite_support.hpp>
+#include <userver/utils/daemon_run.hpp>
+#include <userver/yaml_config/merge_schemas.hpp>
+
+#include <userver/utest/using_namespace_userver.hpp>
+
+class SlowComponent final : public components::ComponentBase {
+public:
+    static constexpr std::string_view kName = "slow-component";
+
+    SlowComponent(const components::ComponentConfig& config, const components::ComponentContext& context)
+        : components::ComponentBase(config, context) {
+        // Testsuite test should finish until 120s end
+        engine::InterruptibleSleepFor(std::chrono::seconds(120));
+
+        if (!engine::current_task::ShouldCancel()) {
+            throw std::runtime_error("Seems like testsuite test is still running, a bug in test?");
+        }
+    }
+};
+
+int main(int argc, char* argv[]) {
+    const auto component_list = components::MinimalServerComponentList().Append<SlowComponent>();
+    return utils::DaemonMain(argc, argv, component_list);
+}

core/functional_tests/signal_during_boot/static_config.yaml

@@ -0,0 +1,21 @@
+components_manager:
+    components:
+        slow-component:
+        server:
+            listener:
+                port: 8089
+                task_processor: main-task-processor
+        logging:
+            fs-task-processor: main-task-processor
+            loggers:
+                default:
+                    file_path: '@stderr'
+                    level: debug
+                    overflow_behavior: discard
+
+    task_processors:
+        main-task-processor:
+            worker_threads: 4
+
+    default_task_processor: main-task-processor
+    fs_task_processor: main-task-processor

core/functional_tests/signal_during_boot/tests/conftest.py

@@ -0,0 +1 @@
+pytest_plugins = ['pytest_userver.plugins.core']

core/functional_tests/signal_during_boot/tests/test_rotate_logs.py

@@ -0,0 +1,58 @@
+import asyncio
+import os
+import signal
+
+TIMEOUT = 60
+
+
+async def wait_for_logfile(fname: str):
+    for i in range(TIMEOUT):
+        with open(fname) as ifile:
+            for line in ifile.readlines():
+                if 'Starting to catch signals' in line:
+                    # Yes, sleeping for sync. Yes, that's bad.
+                    # No, it is not possible to wait for a specific event.
+                    await asyncio.sleep(1)
+                    return
+
+        # not yet started, waiting
+        await asyncio.sleep(1)
+
+    assert False, f'log file {fname} does not contain start log entry, has the service started?'
+
+
+async def test_boot_logs(
+    service_binary,
+    service_config_path_temp,
+    service_env,
+    create_daemon_scope,
+    ensure_daemon_started,
+    _service_logfile_path,
+):
+    async def _checker(*, session, process) -> bool:
+        return True
+
+    async with create_daemon_scope(
+        args=[
+            str(service_binary),
+            '--config',
+            str(service_config_path_temp),
+        ],
+        health_check=_checker,
+        env=service_env,
+    ) as scope:
+        daemon = await ensure_daemon_started(scope)
+
+        await wait_for_logfile(_service_logfile_path)
+
+        os.remove(_service_logfile_path)
+        assert not os.path.exists(_service_logfile_path)
+
+        daemon.process.send_signal(signal.SIGUSR1)
+
+        for i in range(TIMEOUT):
+            if os.path.exists(_service_logfile_path):
+                break
+            await asyncio.sleep(1)
+        else:
+            assert False, f'log file {_service_logfile_path} is not reopened'

core/src/components/manager.cpp

@@ -152,11 +152,7 @@ void Manager::TaskProcessorsStorage::WaitForAllTasksBlocking() const noexcept {
     }
 }
 
-Manager::Manager(
-    std::unique_ptr<ManagerConfig>&& config,
-    std::chrono::steady_clock::time_point start_time,
-    const ComponentList& component_list
-)
+Manager::Manager(std::unique_ptr<ManagerConfig>&& config, std::chrono::steady_clock::time_point start_time)
     : config_(std::move(config)),
       task_processors_storage_(std::make_shared<
                                engine::impl::TaskProcessorPools>(config_->coro_pool, config_->event_thread_pool)),
@@ -219,15 +215,18 @@ Manager::Manager(
     }
 
     default_task_processor_ = default_task_processor_it->second.get();
-    RunInCoro(*default_task_processor_, [this, &component_list]() { CreateComponentContext(component_list); });
-
-    if (!config_->disable_phdr_cache) {
-        engine::impl::InitPhdrCache();
-    }
+}
 
-    LOG_INFO()
-        << "Started components manager. All the components have started "
-           "successfully.";
+engine::TaskWithResult<void> Manager::StartComponentSystem(const ComponentList& component_list) {
+    return engine::CriticalAsyncNoSpan(*default_task_processor_, [this, &component_list]() {
+        CreateComponentContext(component_list);
+        if (!config_->disable_phdr_cache) {
+            engine::impl::InitPhdrCache();
+        }
+        LOG_INFO()
+            << "Started components manager. All the components have started "
+               "successfully.";
+    });
 }
 
 Manager::~Manager() {
@@ -325,6 +324,10 @@ void Manager::CreateComponentContext(const ComponentList& component_list) {
     component_context_ = std::make_unique<impl::ComponentContextImpl>(*this, std::move(loading_components));
 
     AddComponents(component_list);
+
+    LOG_INFO()
+        << "Started components manager. All the components have started "
+           "successfully.";
 }
 
 components::ComponentConfigMap Manager::MakeComponentConfigMap(const ComponentList& component_list) {

core/src/components/manager.hpp

@@ -14,9 +14,14 @@
 
 USERVER_NAMESPACE_BEGIN
 
-namespace engine::impl {
+namespace engine {
+template <typename T>
+class TaskWithResult;
+
+namespace impl {
 class TaskProcessorPools;
-}  // namespace engine::impl
+}  // namespace impl
+}  // namespace engine
 
 namespace os_signals {
 class ProcessorComponent;
@@ -38,13 +43,10 @@ using TaskProcessorsMap = utils::impl::TransparentMap<std::string, std::unique_p
 // to the outside world.
 class Manager final {
 public:
-    Manager(
-        std::unique_ptr<ManagerConfig>&& config,
-        std::chrono::steady_clock::time_point start_time,
-        const ComponentList& component_list
-    );
+    Manager(std::unique_ptr<ManagerConfig>&& config, std::chrono::steady_clock::time_point start_time);
     ~Manager();
 
+    engine::TaskWithResult<void> StartComponentSystem(const ComponentList& component_list);
     const ManagerConfig& GetConfig() const;
     const std::shared_ptr<engine::impl::TaskProcessorPools>& GetTaskProcessorPools() const;
     const TaskProcessorsMap& GetTaskProcessorsMap() const;

core/src/components/run.cpp

@@ -13,6 +13,8 @@
 #include <userver/crypto/openssl.hpp>
 #include <userver/dynamic_config/snapshot.hpp>
 #include <userver/dynamic_config/value.hpp>
+#include <userver/engine/exception.hpp>
+#include <userver/engine/task/task_with_result.hpp>
 #include <userver/formats/json/serialize.hpp>
 #include <userver/fs/blocking/read.hpp>
 #include <userver/logging/impl/mem_logger.hpp>
@@ -183,6 +185,33 @@ ManagerConfig ParseManagerConfigAndSetupLogging(
     }
 }
 
+void CatchSignalsLoop(impl::Manager& manager, RunMode run_mode, utils::SignalCatcher& signal_catcher) noexcept {
+    if (run_mode == RunMode::kOnce) {
+        return;
+    }
+
+    LOG_INFO() << "Starting to catch signals";
+    for (;;) {
+        auto signum = signal_catcher.Catch();
+        if (signum == SIGTERM || signum == SIGQUIT) {
+            break;
+        } else if (signum == SIGINT) {
+            if (IsTraced()) {
+                // SIGINT is masked and cannot be used
+                std::raise(SIGTRAP);
+            } else {
+                break;
+            }
+        } else if (signum == SIGUSR1 || signum == SIGUSR2) {
+            LOG_INFO() << "Signal caught: " << utils::strsignal(signum);
+            manager.OnSignal(signum);
+        } else {
+            LOG_WARNING() << "Got unexpected signal: " << signum << " (" << utils::strsignal(signum) << ')';
+            UASSERT_MSG(false, "unexpected signal");
+        }
+    }
+}
+
 void DoRun(
     const PathOrConfig& config,
     const std::optional<std::string>& config_vars_path,
@@ -217,34 +246,28 @@ void DoRun(
             PreheatStacktraceCollector();
         }
 
-        manager.emplace(std::make_unique<ManagerConfig>(std::move(manager_config)), start_time, component_list);
-    } catch (const std::exception& ex) {
-        LOG_ERROR() << "Loading failed: " << ex;
-        throw;
-    }
+        manager.emplace(std::make_unique<ManagerConfig>(std::move(manager_config)), start_time);
 
-    if (run_mode == RunMode::kOnce) {
-        return;
-    }
+        // Start component system in background.
+        // POSIX signals can be already handled while component system is loading.
+        auto start_components_task = manager->StartComponentSystem(component_list);
 
-    for (;;) {
-        auto signum = signal_catcher.Catch();
-        if (signum == SIGTERM || signum == SIGQUIT) {
-            break;
-        } else if (signum == SIGINT) {
-            if (IsTraced()) {
-                // SIGINT is masked and cannot be used
-                std::raise(SIGTRAP);
-            } else {
-                break;
-            }
-        } else if (signum == SIGUSR1 || signum == SIGUSR2) {
-            LOG_INFO() << "Signal caught: " << utils::strsignal(signum);
-            manager->OnSignal(signum);
-        } else {
-            LOG_WARNING() << "Got unexpected signal: " << signum << " (" << utils::strsignal(signum) << ')';
-            UASSERT_MSG(false, "unexpected signal");
+        // The main event loop.
+        // Other threads handle coroutines.
+        CatchSignalsLoop(*manager, run_mode, signal_catcher);
+
+        if (run_mode == RunMode::kNormal) {
+            start_components_task.RequestCancel();
+        }
+
+        try {
+            start_components_task.BlockingWait();
+            start_components_task.Get();
+        } catch (const engine::WaitInterruptedException&) {
         }
+    } catch (const std::exception& ex) {
+        LOG_ERROR() << "Loading failed: " << ex;
+        throw;
     }
 }