diff --git a/src/realtime/expose/stats.cpp b/src/realtime/expose/stats.cpp index 34b8aa3..0bd480a 100644 --- a/src/realtime/expose/stats.cpp +++ b/src/realtime/expose/stats.cpp @@ -80,6 +80,8 @@ std::string EnvJsonValue(const std::string& key) { return s.substr(p, e == std::string::npos ? e : e - p); } +} // namespace + std::string DatabaseUrl() { std::string v = EnvJsonValue("DATABASE_URL"); if (!v.empty()) return v; @@ -88,6 +90,7 @@ std::string DatabaseUrl() { class SqliteStore : public StatsStore { public: + const char* Backend() const override { return "sqlite"; } explicit SqliteStore(std::string path) : path_(std::move(path)) {} ~SqliteStore() override { if (db_) sqlite3_close(db_); @@ -188,6 +191,7 @@ class SqliteStore : public StatsStore { class PgStore : public StatsStore { public: + const char* Backend() const override { return "postgres"; } explicit PgStore(std::string url) : url_(std::move(url)) {} ~PgStore() override { if (db_) PQfinish(db_); @@ -269,8 +273,6 @@ class PgStore : public StatsStore { PGconn* db_ = nullptr; }; -} // namespace - std::unique_ptr MakeStatsStore(const std::string& database_url) { std::string u = database_url.empty() ? DatabaseUrl() : database_url; if (u.rfind("postgres", 0) == 0 || u.rfind("postgresql", 0) == 0) { @@ -284,6 +286,9 @@ std::unique_ptr MakeStatsStore(const std::string& database_url) { return std::make_unique(path); } +} // namespace molo::realtime::expose + +namespace molo::realtime::expose { static int FlushIntervalSecs() { std::string v = EnvJsonValue("STATS_FLUSH_SECS"); if (v.empty()) v = EnvOr("STATS_FLUSH_SECS", "5"); @@ -314,56 +319,56 @@ StatsCollector& StatsCollector::Global() { return g; } -size_t StatsCollector::CellFor(const std::string& method) { - { - std::lock_guard l(idx_mu_); - auto it = method_idx_.find(method); - if (it != method_idx_.end()) return it->second; - size_t i = cells_.size(); - cells_.push_back(std::make_unique()); - cell_names_.push_back(method); - method_idx_[method] = i; - return i; - } +std::shared_ptr StatsCollector::CellForPtr(const std::string& method) { + std::lock_guard l(idx_mu_); + auto it = cells_by_name_.find(method); + if (it != cells_by_name_.end()) return it->second; + auto c = std::make_shared(); + cells_by_name_[method] = c; + return c; } void StatsCollector::Note(const std::string& method, std::uint64_t bytes_in, std::uint64_t bytes_out, std::uint64_t latency_us) { if (!enabled_.load(std::memory_order_relaxed)) return; // single atomic load thread_local std::string last_method; - thread_local size_t last_idx = static_cast(-1); - thread_local size_t last_total = static_cast(-1); - size_t idx; - if (last_idx != static_cast(-1) && last_method == method) { - idx = last_idx; - } else { - idx = CellFor(method); + thread_local std::shared_ptr last_cell; + if (!last_cell || last_method != method) { + last_cell = CellForPtr(method); last_method = method; - last_idx = idx; - last_total = CellFor("_total"); } - cells_[idx]->count.fetch_add(1, std::memory_order_relaxed); - cells_[idx]->bytes_in.fetch_add(bytes_in, std::memory_order_relaxed); - cells_[idx]->bytes_out.fetch_add(bytes_out, std::memory_order_relaxed); - cells_[idx]->latency_us_sum.fetch_add(latency_us, std::memory_order_relaxed); - cells_[last_total]->count.fetch_add(1, std::memory_order_relaxed); - cells_[last_total]->bytes_in.fetch_add(bytes_in, std::memory_order_relaxed); - cells_[last_total]->bytes_out.fetch_add(bytes_out, std::memory_order_relaxed); - cells_[last_total]->latency_us_sum.fetch_add(latency_us, std::memory_order_relaxed); + last_cell->count.fetch_add(1, std::memory_order_relaxed); + last_cell->bytes_in.fetch_add(bytes_in, std::memory_order_relaxed); + last_cell->bytes_out.fetch_add(bytes_out, std::memory_order_relaxed); + last_cell->latency_us_sum.fetch_add(latency_us, std::memory_order_relaxed); + total_cell_->count.fetch_add(1, std::memory_order_relaxed); + total_cell_->bytes_in.fetch_add(bytes_in, std::memory_order_relaxed); + total_cell_->bytes_out.fetch_add(bytes_out, std::memory_order_relaxed); + total_cell_->latency_us_sum.fetch_add(latency_us, std::memory_order_relaxed); } bool StatsCollector::Flush(std::string& err) { std::map snap; { std::lock_guard l(idx_mu_); - for (size_t i = 0; i < cells_.size(); ++i) { - std::uint64_t c = cells_[i]->count.exchange(0); + for (auto& [name, cell] : cells_by_name_) { + std::uint64_t c = cell->count.exchange(0); if (!c) continue; - MethodStat& m = snap[cell_names_[i]]; + MethodStat& m = snap[name]; m.count = c; - m.bytes_in = cells_[i]->bytes_in.exchange(0); - m.bytes_out = cells_[i]->bytes_out.exchange(0); - m.latency_us_sum = cells_[i]->latency_us_sum.exchange(0); + m.bytes_in = cell->bytes_in.exchange(0); + m.bytes_out = cell->bytes_out.exchange(0); + m.latency_us_sum = cell->latency_us_sum.exchange(0); + } + { + std::uint64_t c = total_cell_->count.exchange(0); + if (c) { + MethodStat& m = snap["_total"]; + m.count = c; + m.bytes_in = total_cell_->bytes_in.exchange(0); + m.bytes_out = total_cell_->bytes_out.exchange(0); + m.latency_us_sum = total_cell_->latency_us_sum.exchange(0); + } } } auto now = std::chrono::steady_clock::now(); diff --git a/src/realtime/expose/stats.h b/src/realtime/expose/stats.h index efdfbe8..103d9be 100644 --- a/src/realtime/expose/stats.h +++ b/src/realtime/expose/stats.h @@ -34,6 +34,7 @@ struct StatsSnapshot { class StatsStore { public: virtual ~StatsStore() = default; + virtual const char* Backend() const = 0; virtual bool Open(std::string& err) = 0; // One transaction per Flush call. virtual bool Flush(const StatsSnapshot& snap, std::string& err) = 0; @@ -41,6 +42,7 @@ class StatsStore { }; std::unique_ptr MakeStatsStore(const std::string& database_url); +std::string DatabaseUrl(); // .env.json DATABASE_URL else env else sqlite://system.db // Zero-cost fast path: Note() is lock-free atomics after first-seen // method registration (thread-local index cache; mutex only on new-method @@ -65,6 +67,7 @@ class StatsCollector { void StartAutoFlush(std::chrono::seconds interval = std::chrono::seconds(5)); void StopAutoFlush(); + std::shared_ptr CellForPtr(const std::string& method); void Note(const std::string& method, std::uint64_t bytes_in, std::uint64_t bytes_out, std::uint64_t latency_us); // Swap accumulator and persist in one transaction. No-op without a store. @@ -82,9 +85,8 @@ class StatsCollector { std::mutex mu_; std::unique_ptr store_; std::map pending_; // drained via cells on Flush - std::map method_idx_; - std::vector> cells_; - std::vector cell_names_; + std::map> cells_by_name_; + std::shared_ptr total_cell_ = std::make_shared(); std::mutex idx_mu_; std::uint64_t flush_bytes_in_ = 0, flush_bytes_out_ = 0; std::chrono::steady_clock::time_point last_flush_tp_ = std::chrono::steady_clock::now(); diff --git a/tests/src/realtime/expose/expose_test.cpp b/tests/src/realtime/expose/expose_test.cpp index 1cbebb4..71ce43b 100644 --- a/tests/src/realtime/expose/expose_test.cpp +++ b/tests/src/realtime/expose/expose_test.cpp @@ -117,6 +117,43 @@ int main() { } std::remove(db.c_str()); } + { // many concurrent upstreams per tunnel: 16 threads x 1k Notes, counts exact + std::string db2 = "/tmp/molo_stats_conc.db"; + std::remove(db2.c_str()); + StatsCollector cc; + cc.Configure(MakeStatsStore("sqlite://" + db2)); + const int kT = 16, kN = 1000; + std::vector ts; + for (int t = 0; t < kT; ++t) + ts.emplace_back([&cc, t] { + for (int i = 0; i < kN; ++i) cc.Note("M" + std::to_string(t), 64, 64, 10); + }); + for (auto& t : ts) t.join(); + std::string e3, o3; + assert(cc.Flush(e3)); + assert(cc.Query(o3, false, e3)); + for (int t = 0; t < kT; ++t) { + assert(o3.find("M" + std::to_string(t)) != std::string::npos); // padded columns + size_t hits = 0, pos = 0; + std::string needle = "count=" + std::to_string(kN); + while ((pos = o3.find(needle, pos)) != std::string::npos) { ++hits; pos += needle.size(); } + assert(hits == (size_t)kT); // each of 16 upstreams flushed exactly 1000 + (void)pos; + break; + } + // verify each upstream label present (column-padded) exactly once in rows + for (int t = 0; t < kT; ++t) + assert(o3.find("M" + std::to_string(t)) != std::string::npos); + std::remove(db2.c_str()); + } + { // store backends: sqlite default, postgres factory by URL scheme + auto s1 = MakeStatsStore("sqlite://system.db"); + auto s2 = MakeStatsStore("postgres://u:p@h/db"); + auto s3 = MakeStatsStore("postgresql://u@h/db"); + assert(s1 && s2 && s3); + assert(std::string(s1->Backend()) == "sqlite" && std::string(s2->Backend()) == "postgres" && std::string(s3->Backend()) == "postgres"); + assert(DatabaseUrl() == "sqlite://system.db" || !DatabaseUrl().empty()); // .env.json default + } printf("expose_test PASS\n"); return 0; }