diff --git a/builtin-functions/_functions.txt b/builtin-functions/_functions.txt
index 16d6f06a1f..b6a52aab20 100644
--- a/builtin-functions/_functions.txt
+++ b/builtin-functions/_functions.txt
@@ -14,6 +14,7 @@ if (0) {
 }
 
 require_once __DIR__ . '/kphp_internal.txt';
+require_once __DIR__ . '/kphp_toggles.txt';
 require_once __DIR__ . '/kphp_tracing.txt';
 require_once __DIR__ . '/uberh3.txt';
 require_once __DIR__ . '/spl.txt';
@@ -1428,16 +1429,6 @@ function zstd_uncompress(string $data) ::: string | false;
 function zstd_compress_dict(string $data, string $dict) ::: string | false;
 function zstd_uncompress_dict(string $data, string $dict) ::: string | false;
 
-function set_migration_php8_warning ($mask ::: int) ::: void;
-
-function set_detect_incorrect_encoding_names_warning(bool $show) ::: void;
-
-function set_json_log_on_timeout_mode(bool $enabled) ::: void;
-
-function set_json_log_demangle_stacktrace(bool $enable) ::: void;
-
-function set_use_updated_gmmktime(bool $enable) ::: void;
-
 // re-initialize given ArrayIterator with another array;
 // in KPHP it returns the same ArrayIterator that is ready to be used
 // in PHP (via polyfills) it returns a newly allocated object
diff --git a/builtin-functions/kphp_toggles.txt b/builtin-functions/kphp_toggles.txt
new file mode 100644
index 0000000000..6bfee561be
--- /dev/null
+++ b/builtin-functions/kphp_toggles.txt
@@ -0,0 +1,15 @@
+<?php
+
+// Functions declared in this file are actually toggles to enable or disable some functionality in KPHP runtime.
+
+function set_migration_php8_warning ($mask ::: int) ::: void;
+
+function set_detect_incorrect_encoding_names_warning(bool $show) ::: void;
+
+function set_json_log_on_timeout_mode(bool $enabled) ::: void;
+
+function set_json_log_demangle_stacktrace(bool $enable) ::: void;
+
+function set_use_updated_gmmktime(bool $enable) ::: void;
+
+function kphp_turn_on_host_tag_in_inner_statshouse_metrics_toggle();
diff --git a/runtime/interface.h b/runtime/interface.h
index 40f1697cf7..a1ed113e9d 100644
--- a/runtime/interface.h
+++ b/runtime/interface.h
@@ -12,6 +12,7 @@
 #include "runtime/kphp_core.h"
 #include "runtime/optional.h"
 #include "server/php-query-data.h"
+#include "server/statshouse/statshouse-manager.h"
 #include "server/workers-control.h"
 
 extern string_buffer *coub;//TODO static
@@ -238,6 +239,10 @@ inline void f$set_json_log_demangle_stacktrace(bool enable) {
   is_demangled_stacktrace_logs_enabled = enable;
 }
 
+inline void f$kphp_turn_on_host_tag_in_inner_statshouse_metrics_toggle() {
+  StatsHouseManager::get().turn_on_host_tag_toggle();
+}
+
 int64_t f$numa_get_bound_node();
 
 bool f$extension_loaded(const string &extension);
diff --git a/server/php-engine.cpp b/server/php-engine.cpp
index b94d81db15..f9c227a9cb 100644
--- a/server/php-engine.cpp
+++ b/server/php-engine.cpp
@@ -6,6 +6,7 @@
 
 #include <cassert>
 #include <cerrno>
+#include <chrono>
 #include <cstdlib>
 #include <cstring>
 #include <fstream>
@@ -61,7 +62,6 @@
 #include "runtime/profiler.h"
 #include "runtime/rpc.h"
 #include "runtime/thread-pool.h"
-#include "server/server-config.h"
 #include "server/confdata-binlog-replay.h"
 #include "server/database-drivers/adaptor.h"
 #include "server/database-drivers/connector.h"
@@ -89,7 +89,7 @@
 #include "server/server-stats.h"
 #include "server/shared-data-worker-cache.h"
 #include "server/signal-handlers.h"
-#include "server/statshouse/statshouse-client.h"
+#include "server/statshouse/statshouse-manager.h"
 #include "server/workers-control.h"
 
 using job_workers::JobWorkersContext;
@@ -1416,12 +1416,15 @@ static void sigusr1_handler(const int sig) {
   pending_signals = pending_signals | (1ll << sig);
 }
 
-void cron() {
+void worker_cron() {
   if (master_flag == -1 && getppid() == 1) {
     turn_sigterm_on();
   }
   vk::singleton<SharedDataWorkerCache>::get().on_worker_cron();
   vk::singleton<ServerStats>::get().update_this_worker_stats();
+  auto virtual_memory_stat = get_self_mem_stats();
+  StatsHouseManager::get().add_worker_memory_stats(virtual_memory_stat);
+  StatsHouseManager::get().generic_cron();
 }
 
 void reopen_json_log() {
@@ -1445,7 +1448,7 @@ void generic_event_loop(WorkerType worker_type, bool init_and_listen_rpc_port) n
   }
 
   int http_port, http_sfd = -1;
-  int prev_time = 0;
+  double last_cron_time = 0;
   double next_create_outbound = 0;
 
   switch (worker_type) {
@@ -1568,9 +1571,9 @@ void generic_event_loop(WorkerType worker_type, bool init_and_listen_rpc_port) n
       reopen_json_log();
     }
 
-    if (now != prev_time) {
-      prev_time = now;
-      cron();
+    if (precise_now - last_cron_time >= 1.0) {
+      last_cron_time = precise_now;
+      worker_cron();
     }
 
     if (worker_type == WorkerType::general_worker) {
@@ -1664,6 +1667,7 @@ void init_all() {
     }
     log_server_warning(deprecation_warning);
   }
+  StatsHouseManager::get().set_common_tags();
 
   global_init_runtime_libs();
   global_init_php_scripts();
@@ -2089,7 +2093,7 @@ int main_args_handler(int i, const char *long_option) {
         host = "127.0.0.1";
       }
 
-      StatsHouseClient::init(host, port);
+      StatsHouseManager::init(host, port);
       return 0;
     }
     case 2027: {
diff --git a/server/php-master.cpp b/server/php-master.cpp
index 3397443d44..b0efe754ba 100644
--- a/server/php-master.cpp
+++ b/server/php-master.cpp
@@ -64,7 +64,7 @@
 #include "server/server-stats.h"
 #include "server/shared-data-worker-cache.h"
 #include "server/shared-data.h"
-#include "server/statshouse/statshouse-client.h"
+#include "server/statshouse/statshouse-manager.h"
 #include "server/workers-control.h"
 
 #include "server/php-master-restart.h"
@@ -1394,18 +1394,17 @@ void check_and_instance_cache_try_swap_memory() {
   }
 }
 
-static void cron() {
+static void master_cron() {
   if (!other->is_alive || in_old_master_on_restart()) {
     // write stats at the beginning to avoid spikes in graphs
     send_data_to_statsd_with_prefix(vk::singleton<ServerConfig>::get().get_statsd_prefix(), stats_tag_kphp_server);
-    if (StatsHouseClient::has()) {
-      const auto cpu_stats = server_stats.cpu[1].get_stat();
-      StatsHouseClient::get().send_common_master_stats(workers_stats, instance_cache_get_memory_stats(), cpu_stats.cpu_s_usage, cpu_stats.cpu_u_usage,
-                                                       instance_cache_memory_swaps_ok, instance_cache_memory_swaps_fail);
-    }
+    const auto cpu_stats = server_stats.cpu[1].get_stat();
+    StatsHouseManager::get().add_common_master_stats(workers_stats, instance_cache_get_memory_stats(), cpu_stats.cpu_s_usage, cpu_stats.cpu_u_usage,
+                                                     instance_cache_memory_swaps_ok, instance_cache_memory_swaps_fail);
   }
   create_all_outbound_connections();
   vk::singleton<ServerStats>::get().aggregate_stats();
+  StatsHouseManager::get().generic_cron();
 
   unsigned long long cpu_total = 0;
   unsigned long long utime = 0;
@@ -1620,7 +1619,7 @@ WorkerType run_master() {
     const auto new_tp = get_steady_tp_ms_now();
     if (new_tp - prev_cron_start_tp >= 1s) {
       prev_cron_start_tp = new_tp;
-      cron();
+      master_cron();
     }
   }
 }
diff --git a/server/server-stats.cpp b/server/server-stats.cpp
index c3e8544a28..40cdd87ddf 100644
--- a/server/server-stats.cpp
+++ b/server/server-stats.cpp
@@ -19,7 +19,7 @@
 
 #include "server/json-logger.h"
 #include "server/server-stats.h"
-#include "server/statshouse/statshouse-client.h"
+#include "server/statshouse/statshouse-manager.h"
 
 namespace {
 
@@ -604,7 +604,6 @@ void ServerStats::after_fork(pid_t worker_pid, uint64_t active_connections, uint
   gen_->seed(worker_pid);
   shared_stats_->workers.reset_worker_stats(worker_pid, active_connections, max_connections, worker_process_id_);
   last_update_aggr_stats = std::chrono::steady_clock::now();
-  last_update_statshouse = std::chrono::steady_clock::now();
 }
 
 void ServerStats::add_request_stats(double script_time_sec, double net_time_sec, int64_t script_queries, int64_t long_script_queries, int64_t memory_used,
@@ -617,11 +616,7 @@ void ServerStats::add_request_stats(double script_time_sec, double net_time_sec,
   stats.add_request_stats(queries_stat, error, memory_used, real_memory_used, curl_total_allocated);
   shared_stats_->workers.add_worker_stats(queries_stat, worker_process_id_);
 
-  using namespace statshouse;
-  if (StatsHouseClient::has()) {
-    StatsHouseClient::get().send_request_stats(worker_type_, script_time.count(), net_time.count(), memory_used, real_memory_used, script_queries,
-                                               long_script_queries);
-  }
+  StatsHouseManager::get().add_request_stats(script_time.count(), net_time.count(), error, memory_used, real_memory_used, script_queries, long_script_queries);
 }
 
 void ServerStats::add_job_stats(double job_wait_time_sec, int64_t request_memory_used, int64_t request_real_memory_used, int64_t response_memory_used,
@@ -629,18 +624,13 @@ void ServerStats::add_job_stats(double job_wait_time_sec, int64_t request_memory
   const auto job_wait_time = std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::duration<double>(job_wait_time_sec));
   shared_stats_->job_workers.add_job_stats(job_wait_time.count(), request_memory_used, request_real_memory_used, response_memory_used, response_real_memory_used);
 
-  if (StatsHouseClient::has()) {
-    StatsHouseClient::get().send_job_stats(job_wait_time.count(), request_memory_used, request_real_memory_used, response_memory_used,
-                                           response_real_memory_used);
-  }
+  StatsHouseManager::get().add_job_stats(job_wait_time.count(), request_memory_used, request_real_memory_used, response_memory_used, response_real_memory_used);
 }
 
 void ServerStats::add_job_common_memory_stats(int64_t common_request_memory_used, int64_t common_request_real_memory_used) noexcept {
   shared_stats_->job_workers.add_job_common_memory_stats(common_request_memory_used, common_request_real_memory_used);
 
-  if (StatsHouseClient::has()) {
-    StatsHouseClient::get().send_job_common_memory_stats(common_request_memory_used, common_request_real_memory_used);
-  }
+  StatsHouseManager::get().add_job_common_memory_stats(common_request_memory_used, common_request_real_memory_used);
 }
 
 void ServerStats::update_this_worker_stats() noexcept {
@@ -649,12 +639,6 @@ void ServerStats::update_this_worker_stats() noexcept {
     shared_stats_->workers.update_worker_stats(worker_process_id_);
     last_update_aggr_stats = now_tp;
   }
-
-  if (StatsHouseClient::has() && (now_tp - last_update_statshouse >= std::chrono::seconds{1})) {
-    auto virtual_memory_stat = get_self_mem_stats();
-    StatsHouseClient::get().send_worker_memory_stats(worker_type_, virtual_memory_stat);
-    last_update_statshouse = now_tp;
-  }
 }
 
 void ServerStats::update_active_connections(uint64_t active_connections, uint64_t max_connections) noexcept {
diff --git a/server/server-stats.h b/server/server-stats.h
index 9c5198c367..218bde09cc 100644
--- a/server/server-stats.h
+++ b/server/server-stats.h
@@ -60,7 +60,6 @@ class ServerStats : vk::not_copyable {
   WorkerType worker_type_{WorkerType::general_worker};
   uint16_t worker_process_id_{0};
   std::chrono::steady_clock::time_point last_update_aggr_stats;
-  std::chrono::steady_clock::time_point last_update_statshouse;
 
   std::mt19937 *gen_{nullptr};
 
diff --git a/server/server.cmake b/server/server.cmake
index 729c6f3fcb..257c82bf07 100644
--- a/server/server.cmake
+++ b/server/server.cmake
@@ -29,8 +29,7 @@ prepend(KPHP_SERVER_SOURCES ${BASE_DIR}/server/
         slot-ids-factory.cpp
         workers-control.cpp
         shared-data-worker-cache.cpp
-        signal-handlers.cpp
-        statshouse/statshouse-client.cpp)
+        signal-handlers.cpp)
 
 prepend(KPHP_JOB_WORKERS_SOURCES ${BASE_DIR}/server/job-workers/
         job-stats.cpp
@@ -40,6 +39,10 @@ prepend(KPHP_JOB_WORKERS_SOURCES ${BASE_DIR}/server/job-workers/
         pipe-io.cpp
         shared-memory-manager.cpp)
 
+prepend(KPHP_STATSHOUSE_SOURCES ${BASE_DIR}/server/statshouse/
+        statshouse-client.cpp
+        statshouse-manager.cpp)
+
 prepend(KPHP_DATABASE_DRIVERS_SOURCES ${BASE_DIR}/server/database-drivers/
         adaptor.cpp
         connector.cpp)
@@ -65,6 +68,7 @@ endif()
 set(KPHP_SERVER_ALL_SOURCES
     ${KPHP_SERVER_SOURCES}
     ${KPHP_JOB_WORKERS_SOURCES}
+    ${KPHP_STATSHOUSE_SOURCES}
     ${KPHP_DATABASE_DRIVERS_SOURCES}
     ${KPHP_DATABASE_DRIVERS_MYSQL_SOURCES}
     ${KPHP_DATABASE_DRIVERS_PGSQL_SOURCES})
diff --git a/server/shared-data-worker-cache.cpp b/server/shared-data-worker-cache.cpp
index 84d35c4b6c..06e305dcf3 100644
--- a/server/shared-data-worker-cache.cpp
+++ b/server/shared-data-worker-cache.cpp
@@ -13,11 +13,7 @@ void SharedDataWorkerCache::init_defaults() noexcept {
 }
 
 void SharedDataWorkerCache::on_worker_cron() noexcept {
-  const auto now = std::chrono::steady_clock::now();
-  if (now - last_update_ >= std::chrono::seconds{1}) {
-    cached_worker_stats = vk::singleton<SharedData>::get().load_worker_stats();
-    last_update_ = now;
-  }
+  cached_worker_stats = vk::singleton<SharedData>::get().load_worker_stats();
 }
 
 const WorkersStats &SharedDataWorkerCache::get_cached_worker_stats() const noexcept {
diff --git a/server/shared-data-worker-cache.h b/server/shared-data-worker-cache.h
index 2771c8ab64..eafcac43d0 100644
--- a/server/shared-data-worker-cache.h
+++ b/server/shared-data-worker-cache.h
@@ -24,7 +24,6 @@ class SharedDataWorkerCache : vk::not_copyable {
   const WorkersStats &get_cached_worker_stats() const noexcept;
 
 private:
-  std::chrono::steady_clock::time_point last_update_;
   WorkersStats cached_worker_stats;
 
   SharedDataWorkerCache() = default;
diff --git a/server/shared-data.h b/server/shared-data.h
index 65efa2242b..8bd115f5b7 100644
--- a/server/shared-data.h
+++ b/server/shared-data.h
@@ -5,6 +5,7 @@
 #pragma once
 
 #include <atomic>
+#include <chrono>
 #include <tuple>
 
 #include "common/mixin/not_copyable.h"
@@ -28,9 +29,13 @@ struct WorkersStats {
 class SharedData : vk::not_copyable {
   struct Storage {
     std::atomic<WorkersStats::PackerRepr> workers_stats;
+    std::atomic<uint64_t> start_use_host_in_statshouse_metrics_timestamp{0};
   };
 
 public:
+  using clock = std::chrono::steady_clock;
+  using time_point = clock::time_point;
+
   void init();
 
   void store_worker_stats(const WorkersStats &workers_stats) noexcept {
@@ -42,6 +47,15 @@ class SharedData : vk::not_copyable {
     workers_stats.unpack(storage->workers_stats.load(std::memory_order_relaxed));
     return workers_stats;
   };
+
+  void store_start_use_host_in_statshouse_metrics_timestamp(const time_point &tp) noexcept {
+    storage->start_use_host_in_statshouse_metrics_timestamp.store(tp.time_since_epoch().count(), std::memory_order_release);
+  }
+
+  time_point load_start_use_host_in_statshouse_metrics_timestamp() noexcept {
+    auto t = storage->start_use_host_in_statshouse_metrics_timestamp.load(std::memory_order_acquire);
+    return time_point{clock::duration{t}};
+  }
 private:
   Storage *storage;
 
diff --git a/server/statshouse/statshouse-client.cpp b/server/statshouse/statshouse-client.cpp
index 04291aad78..1d29a88c10 100644
--- a/server/statshouse/statshouse-client.cpp
+++ b/server/statshouse/statshouse-client.cpp
@@ -4,206 +4,11 @@
 
 #include "server/statshouse/statshouse-client.h"
 
-#include "common/precise-time.h"
-#include "runtime/instance-cache.h"
-#include "server/job-workers/shared-memory-manager.h"
-#include "server/json-logger.h"
-#include "server/server-config.h"
-#include "server/server-stats.h"
-
-StatsHouseClient *StatsHouseClient::inner = nullptr;
-
-template<typename T>
-T unpack(const std::atomic<T> &value) {
-  return value.load(std::memory_order_relaxed);
-}
-
-inline size_t get_memory_used(size_t acquired, size_t released, size_t buffer_size) {
-  return acquired > released ? (acquired - released) * buffer_size : 0;
-}
-
-StatsHouseClient::StatsHouseClient(const std::string &ip, int port)
-  : transport(ip, port){};
-
-void StatsHouseClient::send_request_stats(WorkerType raw_worker_type, uint64_t script_time_ns, uint64_t net_time_ns, uint64_t memory_used,
-                                         uint64_t real_memory_used, uint64_t script_queries, uint64_t long_script_queries) {
-  const char *cluster_name = vk::singleton<ServerConfig>::get().get_cluster_name();
-  const char *worker_type = raw_worker_type == WorkerType::general_worker ? "general" : "job";
-  transport.metric("kphp_request_time").tag(cluster_name).tag("script").tag(worker_type).write_value(script_time_ns);
-  transport.metric("kphp_request_time").tag(cluster_name).tag("net").tag(worker_type).write_value(net_time_ns);
-
-  transport.metric("kphp_memory_script_usage").tag(cluster_name).tag("used").tag(worker_type).write_value(memory_used);
-  transport.metric("kphp_memory_script_usage").tag(cluster_name).tag("real_used").tag(worker_type).write_value(real_memory_used);
-
-  transport.metric("kphp_requests_outgoing_queries").tag(cluster_name).tag(worker_type).write_value(script_queries);
-  transport.metric("kphp_requests_outgoing_long_queries").tag(cluster_name).tag(worker_type).write_value(long_script_queries);
-}
-
-void StatsHouseClient::send_job_stats(uint64_t job_wait_ns, uint64_t request_memory_used, uint64_t request_real_memory_used, uint64_t response_memory_used,
-                                     uint64_t response_real_memory_used) {
-  const char *cluster_name = vk::singleton<ServerConfig>::get().get_cluster_name();
-  transport.metric("kphp_job_queue_time").tag(cluster_name).write_value(job_wait_ns);
-
-  transport.metric("kphp_job_request_memory_usage").tag(cluster_name).tag("used").write_value(request_memory_used);
-  transport.metric("kphp_job_request_memory_usage").tag(cluster_name).tag("real_used").write_value(request_real_memory_used);
-
-  transport.metric("kphp_job_response_memory_usage").tag(cluster_name).tag("used").write_value(response_memory_used);
-  transport.metric("kphp_job_response_memory_usage").tag(cluster_name).tag("real_used").write_value(response_real_memory_used);
-}
-
-void StatsHouseClient::send_job_common_memory_stats(uint64_t job_common_request_memory_used, uint64_t job_common_request_real_memory_used) {
-  const char *cluster_name = vk::singleton<ServerConfig>::get().get_cluster_name();
-  transport.metric("kphp_job_common_request_memory").tag(cluster_name).tag("used").write_value(job_common_request_memory_used);
-  transport.metric("kphp_job_common_request_memory").tag(cluster_name).tag("real_used").write_value(job_common_request_real_memory_used);
-}
-
-void StatsHouseClient::send_worker_memory_stats(WorkerType raw_worker_type, const mem_info_t &mem_stats) {
-  const char *cluster_name = vk::singleton<ServerConfig>::get().get_cluster_name();
-  const char *worker_type = raw_worker_type == WorkerType::general_worker ? "general" : "job";
-  transport.metric("kphp_workers_memory").tag(cluster_name).tag(worker_type).tag("vm_peak").write_value(mem_stats.vm_peak);
-  transport.metric("kphp_workers_memory").tag(cluster_name).tag(worker_type).tag("vm").write_value(mem_stats.vm);
-  transport.metric("kphp_workers_memory").tag(cluster_name).tag(worker_type).tag("rss").write_value(mem_stats.rss);
-  transport.metric("kphp_workers_memory").tag(cluster_name).tag(worker_type).tag("rss_peak").write_value(mem_stats.rss_peak);
-}
-
-void StatsHouseClient::send_common_master_stats(const workers_stats_t &workers_stats, const memory_resource::MemoryStats &memory_stats, double cpu_s_usage,
-                                               double cpu_u_usage, long long int instance_cache_memory_swaps_ok,
-                                               long long int instance_cache_memory_swaps_fail) {
-  const char *cluster_name = vk::singleton<ServerConfig>::get().get_cluster_name();
-  if (engine_tag) {
-    transport.metric("kphp_version").tag(cluster_name).write_value(atoll(engine_tag));
-  }
-
-  transport.metric("kphp_uptime").tag(cluster_name).write_value(get_uptime());
-
-  const auto general_worker_group = vk::singleton<ServerStats>::get().collect_workers_stat(WorkerType::general_worker);
-  transport.metric("kphp_workers_general_processes").tag(cluster_name).tag("working").write_value(general_worker_group.running_workers);
-  transport.metric("kphp_workers_general_processes").tag(cluster_name).tag("working_but_waiting").write_value(general_worker_group.waiting_workers);
-  transport.metric("kphp_workers_general_processes").tag(cluster_name).tag("ready_for_accept").write_value(general_worker_group.ready_for_accept_workers);
-
-  const auto job_worker_group = vk::singleton<ServerStats>::get().collect_workers_stat(WorkerType::job_worker);
-  transport.metric("kphp_workers_job_processes").tag(cluster_name).tag("working").write_value(job_worker_group.running_workers);
-  transport.metric("kphp_workers_job_processes").tag(cluster_name).tag("working_but_waiting").write_value(job_worker_group.waiting_workers);
-
-  transport.metric("kphp_server_workers").tag(cluster_name).tag("started").write_value(workers_stats.tot_workers_started);
-  transport.metric("kphp_server_workers").tag(cluster_name).tag("dead").write_value(workers_stats.tot_workers_dead);
-  transport.metric("kphp_server_workers").tag(cluster_name).tag("strange_dead").write_value(workers_stats.tot_workers_strange_dead);
-  transport.metric("kphp_server_workers").tag(cluster_name).tag("killed").write_value(workers_stats.workers_killed);
-  transport.metric("kphp_server_workers").tag(cluster_name).tag("hung").write_value(workers_stats.workers_hung);
-  transport.metric("kphp_server_workers").tag(cluster_name).tag("terminated").write_value(workers_stats.workers_terminated);
-  transport.metric("kphp_server_workers").tag(cluster_name).tag("failed").write_value(workers_stats.workers_failed);
-
-  transport.metric("kphp_cpu_usage").tag(cluster_name).tag("stime").write_value(cpu_s_usage);
-  transport.metric("kphp_cpu_usage").tag(cluster_name).tag("utime").write_value(cpu_u_usage);
-
-  auto total_workers_json_count = vk::singleton<ServerStats>::get().collect_json_count_stat();
-  uint64_t master_json_logs_count = vk::singleton<JsonLogger>::get().get_json_logs_count();
-  transport.metric("kphp_server_total_json_logs_count").tag(cluster_name).write_value(std::get<0>(total_workers_json_count) + master_json_logs_count);
-  transport.metric("kphp_server_total_json_traces_count").tag(cluster_name).write_value(std::get<1>(total_workers_json_count));
-
-  transport.metric("kphp_instance_cache_memory").tag(cluster_name).tag("limit").write_value(memory_stats.memory_limit);
-  transport.metric("kphp_instance_cache_memory").tag(cluster_name).tag("used").write_value(memory_stats.memory_used);
-  transport.metric("kphp_instance_cache_memory").tag(cluster_name).tag("real_used").write_value(memory_stats.real_memory_used);
-
-  transport.metric("kphp_instance_cache_memory_defragmentation_calls").tag(cluster_name).write_value(memory_stats.defragmentation_calls);
-
-  transport.metric("kphp_instance_cache_memory_pieces").tag(cluster_name).tag("huge").write_value(memory_stats.huge_memory_pieces);
-  transport.metric("kphp_instance_cache_memory_pieces").tag(cluster_name).tag("small").write_value(memory_stats.small_memory_pieces);
-
-  transport.metric("kphp_instance_cache_memory_buffer_swaps").tag(cluster_name).tag("ok").write_value(instance_cache_memory_swaps_ok);
-  transport.metric("kphp_instance_cache_memory_buffer_swaps").tag(cluster_name).tag("fail").write_value(instance_cache_memory_swaps_fail);
-
-  const auto &instance_cache_element_stats = instance_cache_get_stats();
-  transport.metric("kphp_instance_cache_elements").tag(cluster_name).tag("stored").write_value(unpack(instance_cache_element_stats.elements_stored));
-  transport.metric("kphp_instance_cache_elements")
-    .tag(cluster_name)
-    .tag("stored_with_delay")
-    .write_value(unpack(instance_cache_element_stats.elements_stored_with_delay));
-  transport.metric("kphp_instance_cache_elements")
-    .tag(cluster_name)
-    .tag("storing_skipped_due_recent_update")
-    .write_value(unpack(instance_cache_element_stats.elements_storing_skipped_due_recent_update));
-  transport.metric("kphp_instance_cache_elements")
-    .tag(cluster_name)
-    .tag("storing_delayed_due_mutex")
-    .write_value(unpack(instance_cache_element_stats.elements_storing_delayed_due_mutex));
-  transport.metric("kphp_instance_cache_elements").tag(cluster_name).tag("fetched").write_value(unpack(instance_cache_element_stats.elements_fetched));
-  transport.metric("kphp_instance_cache_elements").tag(cluster_name).tag("missed").write_value(unpack(instance_cache_element_stats.elements_missed));
-  transport.metric("kphp_instance_cache_elements")
-    .tag(cluster_name)
-    .tag("missed_earlier")
-    .write_value(unpack(instance_cache_element_stats.elements_missed_earlier));
-  transport.metric("kphp_instance_cache_elements").tag(cluster_name).tag("expired").write_value(unpack(instance_cache_element_stats.elements_expired));
-  transport.metric("kphp_instance_cache_elements").tag(cluster_name).tag("created").write_value(unpack(instance_cache_element_stats.elements_created));
-  transport.metric("kphp_instance_cache_elements").tag(cluster_name).tag("destroyed").write_value(unpack(instance_cache_element_stats.elements_destroyed));
-  transport.metric("kphp_instance_cache_elements").tag(cluster_name).tag("cached").write_value(unpack(instance_cache_element_stats.elements_cached));
-  transport.metric("kphp_instance_cache_elements")
-    .tag(cluster_name)
-    .tag("logically_expired_and_ignored")
-    .write_value(unpack(instance_cache_element_stats.elements_logically_expired_and_ignored));
-  transport.metric("kphp_instance_cache_elements")
-    .tag(cluster_name)
-    .tag("logically_expired_but_fetched")
-    .write_value(unpack(instance_cache_element_stats.elements_logically_expired_but_fetched));
-
-  using namespace job_workers;
-  if (vk::singleton<job_workers::SharedMemoryManager>::get().is_initialized()) {
-    const JobStats &job_stats = vk::singleton<SharedMemoryManager>::get().get_stats();
-    transport.metric("kphp_workers_jobs_queue_size").tag(cluster_name).write_value(unpack(job_stats.job_queue_size));
-    this->add_job_workers_shared_memory_stats(cluster_name, job_stats);
-  }
-}
-
-void StatsHouseClient::add_job_workers_shared_memory_stats(const char *cluster_name, const job_workers::JobStats &job_stats) {
-  using namespace job_workers;
-
-  size_t total_used = this->add_job_workers_shared_messages_stats(cluster_name, job_stats.messages, JOB_SHARED_MESSAGE_BYTES);
-
-  constexpr std::array<const char *, JOB_EXTRA_MEMORY_BUFFER_BUCKETS> extra_memory_prefixes{
-    "256kb", "512kb", "1mb", "2mb", "4mb", "8mb", "16mb", "32mb", "64mb",
-  };
-  for (size_t i = 0; i != JOB_EXTRA_MEMORY_BUFFER_BUCKETS; ++i) {
-    const size_t buffer_size = get_extra_shared_memory_buffer_size(i);
-    total_used += this->add_job_workers_shared_memory_buffers_stats(cluster_name, job_stats.extra_memory[i], extra_memory_prefixes[i], buffer_size);
+statshouse::TransportUDPBase::MetricBuilder StatsHouseClient::metric(std::string_view name, bool force_tag_host) {
+  auto builder = transport.metric(name);
+  builder.tag(tag_cluster);
+  if (host_enabled || force_tag_host) {
+    builder.tag("host", tag_host);
   }
-
-  transport.metric("kphp_job_workers_shared_memory").tag(cluster_name).tag("limit").write_value(job_stats.memory_limit);
-  transport.metric("kphp_job_workers_shared_memory").tag(cluster_name).tag("used").write_value(total_used);
-}
-
-size_t StatsHouseClient::add_job_workers_shared_messages_stats(const char *cluster_name, const job_workers::JobStats::MemoryBufferStats &memory_buffers_stats,
-                                                               size_t buffer_size) {
-  using namespace job_workers;
-
-  const size_t acquired_buffers = unpack(memory_buffers_stats.acquired);
-  const size_t released_buffers = unpack(memory_buffers_stats.released);
-  const size_t memory_used = get_memory_used(acquired_buffers, released_buffers, buffer_size);
-
-  transport.metric("kphp_job_workers_shared_messages").tag(cluster_name).tag("reserved").write_value(memory_buffers_stats.count);
-  transport.metric("kphp_job_workers_shared_messages").tag(cluster_name).tag("acquire_fails").write_value(unpack(memory_buffers_stats.acquire_fails));
-  transport.metric("kphp_job_workers_shared_messages").tag(cluster_name).tag("acquired").write_value(acquired_buffers);
-  transport.metric("kphp_job_workers_shared_messages").tag(cluster_name).tag("released").write_value(released_buffers);
-
-  return memory_used;
-}
-
-size_t StatsHouseClient::add_job_workers_shared_memory_buffers_stats(const char *cluster_name,
-                                                                     const job_workers::JobStats::MemoryBufferStats &memory_buffers_stats, const char *size_tag,
-                                                                     size_t buffer_size) {
-  using namespace job_workers;
-
-  const size_t acquired_buffers = unpack(memory_buffers_stats.acquired);
-  const size_t released_buffers = unpack(memory_buffers_stats.released);
-  const size_t memory_used = get_memory_used(acquired_buffers, released_buffers, buffer_size);
-
-  transport.metric("kphp_job_workers_shared_extra_buffers").tag(cluster_name).tag(size_tag).tag("reserved").write_value(memory_buffers_stats.count);
-  transport.metric("kphp_job_workers_shared_extra_buffers")
-    .tag(cluster_name)
-    .tag(size_tag)
-    .tag("acquire_fails")
-    .write_value(unpack(memory_buffers_stats.acquire_fails));
-  transport.metric("kphp_job_workers_shared_extra_buffers").tag(cluster_name).tag(size_tag).tag("acquired").write_value(acquired_buffers);
-  transport.metric("kphp_job_workers_shared_extra_buffers").tag(cluster_name).tag(size_tag).tag("released").write_value(released_buffers);
-
-  return memory_used;
+  return builder;
 }
diff --git a/server/statshouse/statshouse-client.h b/server/statshouse/statshouse-client.h
index aa7e44b816..e154cc037e 100644
--- a/server/statshouse/statshouse-client.h
+++ b/server/statshouse/statshouse-client.h
@@ -6,58 +6,34 @@
 
 #include "third-party/statshouse.h"
 
-#include <cassert>
+class StatsHouseClient {
+public:
+  // Safe to use dummy instance
+  StatsHouseClient() : transport({}, {}) {}
 
-#include "common/dl-utils-lite.h"
-#include "common/mixin/not_copyable.h"
-#include "runtime/memory_resource/memory_resource.h"
-#include "server/job-workers/job-stats.h"
-#include "server/workers-control.h"
-#include "server/workers-stats.h"
+  StatsHouseClient(const std::string &ip, int port) : transport(ip, port) {}
 
-class StatsHouseClient : vk::not_copyable {
-public:
-  static void init(std::string ip, int port) {
-    static StatsHouseClient client{ip, port};
-    inner = &client;
-  }
+  statshouse::TransportUDPBase::MetricBuilder metric(std::string_view name, bool force_tag_host = false);
 
-  static bool has() {
-    return inner != nullptr;
+  void set_tag_cluster(std::string_view cluster) {
+    tag_cluster = cluster;
   }
 
-  static StatsHouseClient &get() {
-    assert(inner);
-    return *inner;
+  void set_tag_host(std::string_view host) {
+    tag_host = host;
   }
 
-  void send_request_stats(WorkerType raw_worker_type, uint64_t script_time_ns, uint64_t net_time_ns, uint64_t memory_used, uint64_t real_memory_used,
-                         uint64_t script_queries, uint64_t long_script_queries);
-
-  void send_job_stats(uint64_t job_wait_ns, uint64_t request_memory_used, uint64_t request_real_memory_used, uint64_t response_memory_used,
-                     uint64_t response_real_memory_used);
-
-  void send_job_common_memory_stats(uint64_t job_common_request_memory_used, uint64_t job_common_request_real_memory_used);
-
-  void send_worker_memory_stats(WorkerType raw_worker_type, const mem_info_t &mem_stats);
+  void enable_tag_host() {
+    host_enabled = true;
+  }
 
-  /**
-   * Must be called from master process only
-   */
-  void send_common_master_stats(const workers_stats_t &workers_stats, const memory_resource::MemoryStats &memory_stats, double cpu_s_usage, double cpu_u_usage,
-                               long long int instance_cache_memory_swaps_ok, long long int instance_cache_memory_swaps_fail);
+  void disable_tag_host() {
+    host_enabled = false;
+  }
 
 private:
-  explicit StatsHouseClient(const std::string &ip, int port);
-
-  void add_job_workers_shared_memory_stats(const char *cluster_name, const job_workers::JobStats &job_stats);
-
-  size_t add_job_workers_shared_messages_stats(const char *cluster_name, const job_workers::JobStats::MemoryBufferStats &memory_buffers_stats,
-                                               size_t buffer_size);
-
-  size_t add_job_workers_shared_memory_buffers_stats(const char *cluster_name, const job_workers::JobStats::MemoryBufferStats &memory_buffers_stats,
-                                                     const char *size_tag, size_t buffer_size);
-
-  static StatsHouseClient *inner;
   statshouse::TransportUDP transport;
+  std::string tag_cluster;
+  std::string tag_host;
+  bool host_enabled{false};
 };
diff --git a/server/statshouse/statshouse-manager.cpp b/server/statshouse/statshouse-manager.cpp
new file mode 100644
index 0000000000..cb3002e52c
--- /dev/null
+++ b/server/statshouse/statshouse-manager.cpp
@@ -0,0 +1,270 @@
+// Compiler for PHP (aka KPHP)
+// Copyright (c) 2023 LLC «V Kontakte»
+// Distributed under the GPL v3 License, see LICENSE.notice.txt
+
+#include "server/statshouse/statshouse-manager.h"
+
+#include <chrono>
+
+#include "common/precise-time.h"
+#include "common/resolver.h"
+#include "runtime/instance-cache.h"
+#include "server/job-workers/shared-memory-manager.h"
+#include "server/json-logger.h"
+#include "server/php-engine-vars.h"
+#include "server/php-runner.h"
+#include "server/server-config.h"
+#include "server/server-stats.h"
+#include "server/shared-data.h"
+
+namespace {
+template<typename T>
+T unpack(const std::atomic<T> &value) {
+  return value.load(std::memory_order_relaxed);
+}
+
+inline size_t get_memory_used(size_t acquired, size_t released, size_t buffer_size) {
+  return acquired > released ? (acquired - released) * buffer_size : 0;
+}
+
+const char *get_current_worker_type() {
+  switch (process_type) {
+    case ProcessType::http_worker: return "http";
+    case ProcessType::rpc_worker:  return "rpc";
+    case ProcessType::job_worker:  return "job";
+    default: return "";
+  }
+}
+
+const char *script_error_to_str(script_error_t error) {
+  switch (error) {
+    case script_error_t::no_error:                return "ok";
+    case script_error_t::memory_limit:            return "memory_limit";
+    case script_error_t::timeout:                 return "timeout";
+    case script_error_t::exception:               return "exception";
+    case script_error_t::stack_overflow:          return "stack_overflow";
+    case script_error_t::php_assert:              return "php_assert";
+    case script_error_t::http_connection_close:   return "http_connection_close";
+    case script_error_t::rpc_connection_close:    return "rpc_connection_close";
+    case script_error_t::net_event_error:         return "net_event_error";
+    case script_error_t::post_data_loading_error: return "post_data_loading_error";
+    default:                                      return "unclassified_error";
+  }
+}
+} // namespace
+
+StatsHouseManager::StatsHouseManager(const std::string &ip, int port)
+  : client(ip, port){};
+
+void StatsHouseManager::set_common_tags() {
+  client.set_tag_cluster(vk::singleton<ServerConfig>::get().get_cluster_name());
+  client.set_tag_host(kdb_gethostname());
+}
+
+void StatsHouseManager::generic_cron_check_if_tag_host_needed() {
+  using namespace std::chrono_literals;
+
+  static SharedData::time_point last_check_tp;
+  auto now_tp = SharedData::clock::now();
+
+  // check is done every 5 sec
+  if (now_tp - last_check_tp < 5s) {
+    return;
+  }
+  last_check_tp = now_tp;
+
+  auto &shared_data = vk::singleton<SharedData>::get();
+
+  if (need_write_enable_tag_host) {
+    // if php script called kphp_turn_on_host_tag_in_inner_statshouse_metrics_toggle()
+    // we store current time in shared memory to make other processes enable host tag.
+    // NB: so far it never happens in master process
+    shared_data.store_start_use_host_in_statshouse_metrics_timestamp(now_tp);
+    need_write_enable_tag_host = false;
+    client.enable_tag_host();
+    return;
+  }
+
+  // then every process checks the timestamp from shared memory to decide whether to enable host tag or not
+  auto tp = shared_data.load_start_use_host_in_statshouse_metrics_timestamp();
+  if (now_tp - tp < 30s) {
+    client.enable_tag_host();
+  } else {
+    // if 30 secs has already passed, disable tag host
+    client.disable_tag_host();
+  }
+}
+
+void StatsHouseManager::add_request_stats(uint64_t script_time_ns, uint64_t net_time_ns, script_error_t error, uint64_t memory_used,
+                                         uint64_t real_memory_used, uint64_t script_queries, uint64_t long_script_queries) {
+  const char *worker_type = get_current_worker_type();
+  const char *status = script_error_to_str(error);
+
+  client.metric("kphp_request_time").tag("script").tag(worker_type).tag(status).write_value(script_time_ns);
+  client.metric("kphp_request_time").tag("net").tag(worker_type).tag(status).write_value(net_time_ns);
+
+  client.metric("kphp_by_host_request_time", true).tag("script").tag(worker_type).write_value(script_time_ns);
+  client.metric("kphp_by_host_request_time", true).tag("net").tag(worker_type).write_value(net_time_ns);
+
+  if (error != script_error_t::no_error) {
+    client.metric("kphp_request_errors").tag(status).tag(worker_type).write_count(1);
+    client.metric("kphp_by_host_request_errors", true).tag(status).tag(worker_type).write_count(1);
+  }
+
+  client.metric("kphp_memory_script_usage").tag("used").tag(worker_type).write_value(memory_used);
+  client.metric("kphp_memory_script_usage").tag("real_used").tag(worker_type).write_value(real_memory_used);
+
+  client.metric("kphp_requests_outgoing_queries").tag(worker_type).write_value(script_queries);
+  client.metric("kphp_requests_outgoing_long_queries").tag(worker_type).write_value(long_script_queries);
+}
+
+void StatsHouseManager::add_job_stats(uint64_t job_wait_ns, uint64_t request_memory_used, uint64_t request_real_memory_used, uint64_t response_memory_used,
+                                     uint64_t response_real_memory_used) {
+  client.metric("kphp_job_queue_time").write_value(job_wait_ns);
+
+  client.metric("kphp_job_request_memory_usage").tag("used").write_value(request_memory_used);
+  client.metric("kphp_job_request_memory_usage").tag("real_used").write_value(request_real_memory_used);
+
+  client.metric("kphp_job_response_memory_usage").tag("used").write_value(response_memory_used);
+  client.metric("kphp_job_response_memory_usage").tag("real_used").write_value(response_real_memory_used);
+}
+
+void StatsHouseManager::add_job_common_memory_stats(uint64_t job_common_request_memory_used, uint64_t job_common_request_real_memory_used) {
+  dl::CriticalSectionGuard guard; // It's called from script context, so we need to ensure SIGALRM won't interrupt us here
+  client.metric("kphp_job_common_request_memory").tag("used").write_value(job_common_request_memory_used);
+  client.metric("kphp_job_common_request_memory").tag("real_used").write_value(job_common_request_real_memory_used);
+}
+
+void StatsHouseManager::add_worker_memory_stats(const mem_info_t &mem_stats) {
+  const char *worker_type = get_current_worker_type();
+  client.metric("kphp_workers_memory").tag(worker_type).tag("vm_peak").write_value(mem_stats.vm_peak);
+  client.metric("kphp_workers_memory").tag(worker_type).tag("vm").write_value(mem_stats.vm);
+  client.metric("kphp_workers_memory").tag(worker_type).tag("rss").write_value(mem_stats.rss);
+  client.metric("kphp_workers_memory").tag(worker_type).tag("rss_peak").write_value(mem_stats.rss_peak);
+
+  client.metric("kphp_by_host_workers_memory", true).tag(worker_type).tag("vm_peak").write_value(mem_stats.vm_peak);
+  client.metric("kphp_by_host_workers_memory", true).tag(worker_type).tag("vm").write_value(mem_stats.vm);
+  client.metric("kphp_by_host_workers_memory", true).tag(worker_type).tag("rss").write_value(mem_stats.rss);
+  client.metric("kphp_by_host_workers_memory", true).tag(worker_type).tag("rss_peak").write_value(mem_stats.rss_peak);
+}
+
+void StatsHouseManager::add_common_master_stats(const workers_stats_t &workers_stats, const memory_resource::MemoryStats &memory_stats, double cpu_s_usage,
+                                               double cpu_u_usage, long long int instance_cache_memory_swaps_ok,
+                                               long long int instance_cache_memory_swaps_fail) {
+  if (engine_tag) {
+    client.metric("kphp_version").write_value(atoll(engine_tag));
+  }
+
+  client.metric("kphp_uptime").write_value(get_uptime());
+
+  const auto general_worker_group = vk::singleton<ServerStats>::get().collect_workers_stat(WorkerType::general_worker);
+  client.metric("kphp_workers_general_processes").tag("working").write_value(general_worker_group.running_workers);
+  client.metric("kphp_workers_general_processes").tag("working_but_waiting").write_value(general_worker_group.waiting_workers);
+  client.metric("kphp_workers_general_processes").tag("ready_for_accept").write_value(general_worker_group.ready_for_accept_workers);
+
+  const auto job_worker_group = vk::singleton<ServerStats>::get().collect_workers_stat(WorkerType::job_worker);
+  client.metric("kphp_workers_job_processes").tag("working").write_value(job_worker_group.running_workers);
+  client.metric("kphp_workers_job_processes").tag("working_but_waiting").write_value(job_worker_group.waiting_workers);
+
+  client.metric("kphp_server_workers").tag("started").write_value(workers_stats.tot_workers_started);
+  client.metric("kphp_server_workers").tag("dead").write_value(workers_stats.tot_workers_dead);
+  client.metric("kphp_server_workers").tag("strange_dead").write_value(workers_stats.tot_workers_strange_dead);
+  client.metric("kphp_server_workers").tag("killed").write_value(workers_stats.workers_killed);
+  client.metric("kphp_server_workers").tag("hung").write_value(workers_stats.workers_hung);
+  client.metric("kphp_server_workers").tag("terminated").write_value(workers_stats.workers_terminated);
+  client.metric("kphp_server_workers").tag("failed").write_value(workers_stats.workers_failed);
+
+  client.metric("kphp_cpu_usage").tag("sys").write_value(cpu_s_usage);
+  client.metric("kphp_cpu_usage").tag("user").write_value(cpu_u_usage);
+
+  client.metric("kphp_by_host_cpu_usage", true).tag("sys").write_value(cpu_s_usage);
+  client.metric("kphp_by_host_cpu_usage", true).tag("user").write_value(cpu_u_usage);
+
+  auto total_workers_json_count = vk::singleton<ServerStats>::get().collect_json_count_stat();
+  uint64_t master_json_logs_count = vk::singleton<JsonLogger>::get().get_json_logs_count();
+  client.metric("kphp_server_total_json_logs_count").write_value(std::get<0>(total_workers_json_count) + master_json_logs_count);
+  client.metric("kphp_server_total_json_traces_count").write_value(std::get<1>(total_workers_json_count));
+
+  client.metric("kphp_instance_cache_memory").tag("limit").write_value(memory_stats.memory_limit);
+  client.metric("kphp_instance_cache_memory").tag("used").write_value(memory_stats.memory_used);
+  client.metric("kphp_instance_cache_memory").tag("real_used").write_value(memory_stats.real_memory_used);
+
+  client.metric("kphp_instance_cache_memory_defragmentation_calls").write_value(memory_stats.defragmentation_calls);
+
+  client.metric("kphp_instance_cache_memory_pieces").tag("huge").write_value(memory_stats.huge_memory_pieces);
+  client.metric("kphp_instance_cache_memory_pieces").tag("small").write_value(memory_stats.small_memory_pieces);
+
+  client.metric("kphp_instance_cache_memory_buffer_swaps").tag("ok").write_value(instance_cache_memory_swaps_ok);
+  client.metric("kphp_instance_cache_memory_buffer_swaps").tag("fail").write_value(instance_cache_memory_swaps_fail);
+
+  const auto &instance_cache_element_stats = instance_cache_get_stats();
+  client.metric("kphp_instance_cache_elements").tag("stored").write_value(unpack(instance_cache_element_stats.elements_stored));
+  client.metric("kphp_instance_cache_elements").tag("stored_with_delay").write_value(unpack(instance_cache_element_stats.elements_stored_with_delay));
+  client.metric("kphp_instance_cache_elements").tag("storing_skipped_due_recent_update").write_value(unpack(instance_cache_element_stats.elements_storing_skipped_due_recent_update));
+  client.metric("kphp_instance_cache_elements").tag("storing_delayed_due_mutex").write_value(unpack(instance_cache_element_stats.elements_storing_delayed_due_mutex));
+  client.metric("kphp_instance_cache_elements").tag("fetched").write_value(unpack(instance_cache_element_stats.elements_fetched));
+  client.metric("kphp_instance_cache_elements").tag("missed").write_value(unpack(instance_cache_element_stats.elements_missed));
+  client.metric("kphp_instance_cache_elements").tag("missed_earlier").write_value(unpack(instance_cache_element_stats.elements_missed_earlier));
+  client.metric("kphp_instance_cache_elements").tag("expired").write_value(unpack(instance_cache_element_stats.elements_expired));
+  client.metric("kphp_instance_cache_elements").tag("created").write_value(unpack(instance_cache_element_stats.elements_created));
+  client.metric("kphp_instance_cache_elements").tag("destroyed").write_value(unpack(instance_cache_element_stats.elements_destroyed));
+  client.metric("kphp_instance_cache_elements").tag("cached").write_value(unpack(instance_cache_element_stats.elements_cached));
+  client.metric("kphp_instance_cache_elements").tag("logically_expired_and_ignored").write_value(unpack(instance_cache_element_stats.elements_logically_expired_and_ignored));
+  client.metric("kphp_instance_cache_elements").tag("logically_expired_but_fetched").write_value(unpack(instance_cache_element_stats.elements_logically_expired_but_fetched));
+
+  using namespace job_workers;
+  if (vk::singleton<job_workers::SharedMemoryManager>::get().is_initialized()) {
+    const JobStats &job_stats = vk::singleton<SharedMemoryManager>::get().get_stats();
+    client.metric("kphp_workers_jobs_queue_size").write_value(unpack(job_stats.job_queue_size));
+    this->add_job_workers_shared_memory_stats(job_stats);
+  }
+}
+
+void StatsHouseManager::add_job_workers_shared_memory_stats(const job_workers::JobStats &job_stats) {
+  using namespace job_workers;
+
+  size_t total_used = this->add_job_workers_shared_messages_stats(job_stats.messages, JOB_SHARED_MESSAGE_BYTES);
+
+  constexpr std::array<const char *, JOB_EXTRA_MEMORY_BUFFER_BUCKETS> extra_memory_prefixes{
+    "256kb", "512kb", "1mb", "2mb", "4mb", "8mb", "16mb", "32mb", "64mb",
+  };
+  for (size_t i = 0; i != JOB_EXTRA_MEMORY_BUFFER_BUCKETS; ++i) {
+    const size_t buffer_size = get_extra_shared_memory_buffer_size(i);
+    total_used += this->add_job_workers_shared_memory_buffers_stats(job_stats.extra_memory[i], extra_memory_prefixes[i], buffer_size);
+  }
+
+  client.metric("kphp_job_workers_shared_memory").tag("limit").write_value(job_stats.memory_limit);
+  client.metric("kphp_job_workers_shared_memory").tag("used").write_value(total_used);
+}
+
+size_t StatsHouseManager::add_job_workers_shared_messages_stats(const job_workers::JobStats::MemoryBufferStats &memory_buffers_stats,
+                                                               size_t buffer_size) {
+  using namespace job_workers;
+
+  const size_t acquired_buffers = unpack(memory_buffers_stats.acquired);
+  const size_t released_buffers = unpack(memory_buffers_stats.released);
+  const size_t memory_used = get_memory_used(acquired_buffers, released_buffers, buffer_size);
+
+  client.metric("kphp_job_workers_shared_messages").tag("reserved").write_value(memory_buffers_stats.count);
+  client.metric("kphp_job_workers_shared_messages").tag("acquire_fails").write_value(unpack(memory_buffers_stats.acquire_fails));
+  client.metric("kphp_job_workers_shared_messages").tag("acquired").write_value(acquired_buffers);
+  client.metric("kphp_job_workers_shared_messages").tag("released").write_value(released_buffers);
+
+  return memory_used;
+}
+
+size_t StatsHouseManager::add_job_workers_shared_memory_buffers_stats(const job_workers::JobStats::MemoryBufferStats &memory_buffers_stats, const char *size_tag,
+                                                                     size_t buffer_size) {
+  using namespace job_workers;
+
+  const size_t acquired_buffers = unpack(memory_buffers_stats.acquired);
+  const size_t released_buffers = unpack(memory_buffers_stats.released);
+  const size_t memory_used = get_memory_used(acquired_buffers, released_buffers, buffer_size);
+
+  client.metric("kphp_job_workers_shared_extra_buffers").tag(size_tag).tag("reserved").write_value(memory_buffers_stats.count);
+  client.metric("kphp_job_workers_shared_extra_buffers").tag(size_tag).tag("acquire_fails").write_value(unpack(memory_buffers_stats.acquire_fails));
+  client.metric("kphp_job_workers_shared_extra_buffers").tag(size_tag).tag("acquired").write_value(acquired_buffers);
+  client.metric("kphp_job_workers_shared_extra_buffers").tag(size_tag).tag("released").write_value(released_buffers);
+
+  return memory_used;
+}
diff --git a/server/statshouse/statshouse-manager.h b/server/statshouse/statshouse-manager.h
new file mode 100644
index 0000000000..eed094a712
--- /dev/null
+++ b/server/statshouse/statshouse-manager.h
@@ -0,0 +1,83 @@
+// Compiler for PHP (aka KPHP)
+// Copyright (c) 2023 LLC «V Kontakte»
+// Distributed under the GPL v3 License, see LICENSE.notice.txt
+
+#pragma once
+
+#include <cassert>
+
+#include "common/dl-utils-lite.h"
+#include "common/mixin/not_copyable.h"
+#include "runtime/memory_resource/memory_resource.h"
+#include "server/job-workers/job-stats.h"
+#include "server/statshouse/statshouse-client.h"
+#include "server/workers-control.h"
+#include "server/workers-stats.h"
+
+enum class script_error_t : uint8_t;
+
+class StatsHouseManager : vk::not_copyable {
+public:
+  static void init(const std::string &ip, int port) {
+    storage_impl(ip, port);
+  }
+
+  static StatsHouseManager &get() {
+    return storage_impl({}, {});
+  }
+
+  void set_common_tags();
+
+  // Runs in master and workers cron with 1 sec period
+  void generic_cron() {
+    generic_cron_check_if_tag_host_needed();
+    set_common_tags();
+  }
+
+  /**
+   * This toggle turns on host tag for all statshouse metrics in all workers and master processes during the next 30 sec.
+   * See generic_cron_check_if_tag_host_needed() for details.
+   */
+  void turn_on_host_tag_toggle() {
+    need_write_enable_tag_host = true;
+  }
+
+  void add_request_stats(uint64_t script_time_ns, uint64_t net_time_ns, script_error_t error, uint64_t memory_used, uint64_t real_memory_used,
+                         uint64_t script_queries, uint64_t long_script_queries);
+
+  void add_job_stats(uint64_t job_wait_ns, uint64_t request_memory_used, uint64_t request_real_memory_used, uint64_t response_memory_used,
+                     uint64_t response_real_memory_used);
+
+  void add_job_common_memory_stats(uint64_t job_common_request_memory_used, uint64_t job_common_request_real_memory_used);
+
+  void add_worker_memory_stats(const mem_info_t &mem_stats);
+
+  /**
+   * Must be called from master process only
+   */
+  void add_common_master_stats(const workers_stats_t &workers_stats, const memory_resource::MemoryStats &memory_stats, double cpu_s_usage, double cpu_u_usage,
+                               long long int instance_cache_memory_swaps_ok, long long int instance_cache_memory_swaps_fail);
+
+private:
+  StatsHouseClient client;
+  bool need_write_enable_tag_host = false;
+
+  StatsHouseManager() = default;
+  explicit StatsHouseManager(const std::string &ip, int port);
+
+  // returns safe to use dummy instance if wasn't initialized
+  static StatsHouseManager &storage_impl(const std::string &ip, int port) {
+    static StatsHouseManager client{ip, port};
+    return client;
+  }
+
+  void generic_cron_check_if_tag_host_needed();
+
+  void add_job_workers_shared_memory_stats(const job_workers::JobStats &job_stats);
+
+  size_t add_job_workers_shared_messages_stats(const job_workers::JobStats::MemoryBufferStats &memory_buffers_stats,
+                                               size_t buffer_size);
+
+  size_t add_job_workers_shared_memory_buffers_stats(const job_workers::JobStats::MemoryBufferStats &memory_buffers_stats,
+                                                     const char *size_tag, size_t buffer_size);
+};