diff --git a/src/commands/cmd_server.cc b/src/commands/cmd_server.cc index be5ff20a228..8eb9fe74807 100644 --- a/src/commands/cmd_server.cc +++ b/src/commands/cmd_server.cc @@ -1722,16 +1722,22 @@ class CommandLatency : public Commander { return Status::OK(); } + // Report the caller's namespace histogram; the admin/default namespace sees the aggregate. + auto stats_holder = conn->GetNamespace() == kDefaultNamespace + ? srv->AggregateNamespaceStats() + : srv->GetOrCreateNamespaceStats(conn->GetNamespace()); + const Stats &cmd_stats = *stats_holder; + std::vector *> target_histograms; if (args_.size() > 2) { for (size_t i = 2; i < args_.size(); i++) { - auto it = srv->stats.commands_histogram.find(util::ToLower(args_[i])); - if (it != srv->stats.commands_histogram.end() && it->second.calls > 0) { + auto it = cmd_stats.commands_histogram.find(util::ToLower(args_[i])); + if (it != cmd_stats.commands_histogram.end() && it->second.calls > 0) { target_histograms.push_back(&(*it)); } } } else { - for (const auto &iter : srv->stats.commands_histogram) { + for (const auto &iter : cmd_stats.commands_histogram) { if (iter.second.calls > 0) { target_histograms.push_back(&iter); } @@ -1750,8 +1756,8 @@ class CommandLatency : public Commander { if (cumulative == 0) continue; int64_t boundary = 0; - if (i < srv->stats.bucket_boundaries.size()) { - boundary = static_cast(srv->stats.bucket_boundaries[i]); + if (i < cmd_stats.bucket_boundaries.size()) { + boundary = static_cast(cmd_stats.bucket_boundaries[i]); } else { boundary = -1; } @@ -1824,5 +1830,5 @@ REDIS_REGISTER_COMMANDS( MakeCmdAttr("sst", -3, "write exclusive admin", 1, 1, 1), MakeCmdAttr("flushmemtable", -1, "exclusive write", NO_KEY), MakeCmdAttr("flushblockcache", 1, "exclusive write", NO_KEY), - MakeCmdAttr("latency", -2, "read-only admin", NO_KEY), ) + MakeCmdAttr("latency", -2, "read-only", NO_KEY), ) } // namespace redis diff --git a/src/server/redis_connection.cc b/src/server/redis_connection.cc index aa301b32ffa..b29fb2b780a 100644 --- a/src/server/redis_connection.cc +++ b/src/server/redis_connection.cc @@ -52,6 +52,12 @@ Connection::Connection(bufferevent *bev, Worker *owner) int64_t now = util::GetTimeStamp(); create_time_ = now; last_interaction_ = now; + cached_ns_stats_ = srv_->GetOrCreateNamespaceStats(kDefaultNamespace); +} + +void Connection::SetNamespace(std::string ns) { + ns_ = std::move(ns); + cached_ns_stats_ = srv_->GetOrCreateNamespaceStats(ns_); } Connection::~Connection() { @@ -399,7 +405,9 @@ void Connection::RecordProfilingSampleIfNeed(const std::string &cmd, uint64_t du Status Connection::ExecuteCommand(engine::Context &ctx, const std::string &cmd_name, const std::vector &cmd_tokens, Commander *current_cmd, std::string *reply) { - srv_->stats.IncrCalls(cmd_name); + // Local copy so calls and latency hit the same namespace even if the command changes it (e.g. AUTH). + auto ns_stats = cached_ns_stats_; + ns_stats->IncrCalls(cmd_name); auto start = std::chrono::high_resolution_clock::now(); bool is_profiling = IsProfilingEnabled(cmd_name); @@ -409,7 +417,7 @@ Status Connection::ExecuteCommand(engine::Context &ctx, const std::string &cmd_n if (is_profiling) RecordProfilingSampleIfNeed(cmd_name, duration); srv_->SlowlogPushEntryIfNeeded(&cmd_tokens, duration, this); - srv_->stats.IncrLatency(static_cast(duration), cmd_name); + ns_stats->IncrLatency(static_cast(duration), cmd_name); return s; } diff --git a/src/server/redis_connection.h b/src/server/redis_connection.h index 909ce8759e5..f07ea9a04f9 100644 --- a/src/server/redis_connection.h +++ b/src/server/redis_connection.h @@ -33,6 +33,7 @@ #include "event_util.h" #include "redis_request.h" #include "server/redis_reply.h" +#include "stats/stats.h" class Worker; @@ -163,7 +164,7 @@ class Connection : public EvbufCallbackBase { void BecomeAdmin() { is_admin_ = true; } void BecomeUser() { is_admin_ = false; } std::string GetNamespace() const { return ns_; } - void SetNamespace(std::string ns) { ns_ = std::move(ns); } + void SetNamespace(std::string ns); void NeedFreeBufferEvent(bool need_free = true) { need_free_bev_ = need_free; } void NeedNotFreeBufferEvent() { NeedFreeBufferEvent(false); } @@ -210,6 +211,8 @@ class Connection : public EvbufCallbackBase { uint64_t id_ = 0; std::atomic flags_ = 0; std::string ns_; + // Cache of this connection's per-namespace command stats, refreshed by SetNamespace. + std::shared_ptr cached_ns_stats_; std::string name_; SetInfo set_info_; std::string ip_; diff --git a/src/server/server.cc b/src/server/server.cc index 6ebc2c3c4ea..ca32a9e0754 100644 --- a/src/server/server.cc +++ b/src/server/server.cc @@ -66,21 +66,7 @@ Server::Server(engine::Storage *storage, Config *config) config_(config), namespace_(storage) { // init commands stats here to prevent concurrent insert, and cause core - auto commands = redis::CommandTable::GetOriginal(); - - for (const auto &iter : *commands) { - stats.commands_stats[iter.first].calls = 0; - stats.commands_stats[iter.first].latency = 0; - - if (stats.bucket_boundaries.size() > 0) { - // NB: Extra index for the last bucket (Inf) - for (std::size_t i{0}; i <= stats.bucket_boundaries.size(); ++i) { - stats.commands_histogram[iter.first].buckets.push_back(std::make_unique>(0)); - } - stats.commands_histogram[iter.first].calls = 0; - stats.commands_histogram[iter.first].sum = 0; - } - } + initCommandStats(&stats); // init cursor_dict_ cursor_dict_ = std::make_unique(); @@ -872,7 +858,18 @@ uint64_t Server::GetClientID() { return client_id_.fetch_add(1, std::memory_orde void Server::recordInstantaneousMetrics() { auto rocksdb_stats = storage->GetDB()->GetDBOptions().statistics; - stats.TrackInstantaneousMetric(STATS_METRIC_COMMAND, stats.total_calls); + // Sample each namespace's command metric, and feed the sum into the global metric so the + // admin/default view reports aggregate ops/sec without keeping a global command counter on the hot path. + uint64_t total_calls = 0; + { + std::shared_lock lock(ns_stats_mu_); + for (const auto &[ns, ns_stats] : ns_stats_) { + auto calls = ns_stats->total_calls.load(); + ns_stats->TrackInstantaneousMetric(STATS_METRIC_COMMAND, calls); + total_calls += calls; + } + } + stats.TrackInstantaneousMetric(STATS_METRIC_COMMAND, total_calls); stats.TrackInstantaneousMetric(STATS_METRIC_NET_INPUT, stats.in_bytes); stats.TrackInstantaneousMetric(STATS_METRIC_NET_OUTPUT, stats.out_bytes); stats.TrackInstantaneousMetric(STATS_METRIC_ROCKSDB_PUT, @@ -1392,11 +1389,75 @@ int64_t Server::GetLastBgsaveTime() { return last_bgsave_timestamp_secs_ == -1 ? start_time_secs_ : last_bgsave_timestamp_secs_; } -Server::InfoEntries Server::GetStatsInfo() { +void Server::initCommandStats(Stats *stats) { + auto commands = redis::CommandTable::GetOriginal(); + for (const auto &iter : *commands) { + stats->commands_stats[iter.first].calls = 0; + stats->commands_stats[iter.first].latency = 0; + + if (stats->bucket_boundaries.size() > 0) { + // NB: Extra index for the last bucket (Inf) + for (std::size_t i{0}; i <= stats->bucket_boundaries.size(); ++i) { + stats->commands_histogram[iter.first].buckets.push_back(std::make_unique>(0)); + } + stats->commands_histogram[iter.first].calls = 0; + stats->commands_histogram[iter.first].sum = 0; + } + } +} + +std::shared_ptr Server::GetOrCreateNamespaceStats(const std::string &ns) { + { + std::shared_lock lock(ns_stats_mu_); + if (auto it = ns_stats_.find(ns); it != ns_stats_.end()) { + return it->second; + } + } + + std::unique_lock lock(ns_stats_mu_); + if (auto it = ns_stats_.find(ns); it != ns_stats_.end()) { + return it->second; + } + auto ns_stats = std::make_shared(config_->histogram_bucket_boundaries); + initCommandStats(ns_stats.get()); + ns_stats_[ns] = ns_stats; + return ns_stats; +} + +std::shared_ptr Server::AggregateNamespaceStats() { + auto agg = std::make_shared(config_->histogram_bucket_boundaries); + initCommandStats(agg.get()); + + std::shared_lock lock(ns_stats_mu_); + for (const auto &[ns, ns_stats] : ns_stats_) { + agg->total_calls.fetch_add(ns_stats->total_calls.load(), std::memory_order_relaxed); + for (const auto &[cmd, stat] : ns_stats->commands_stats) { + agg->commands_stats[cmd].calls.fetch_add(stat.calls.load(), std::memory_order_relaxed); + agg->commands_stats[cmd].latency.fetch_add(stat.latency.load(), std::memory_order_relaxed); + } + for (const auto &[cmd, hist] : ns_stats->commands_histogram) { + auto &agg_hist = agg->commands_histogram[cmd]; + agg_hist.calls.fetch_add(hist.calls.load(), std::memory_order_relaxed); + agg_hist.sum.fetch_add(hist.sum.load(), std::memory_order_relaxed); + for (std::size_t i = 0; i < hist.buckets.size(); ++i) { + agg_hist.buckets[i]->fetch_add(hist.buckets[i]->load(), std::memory_order_relaxed); + } + } + } + return agg; +} + +Server::InfoEntries Server::GetStatsInfo(const std::string &ns) { + // Command stats are per namespace; the admin/default namespace sees the aggregate across all of them. + auto cmd_stats_ptr = ns == kDefaultNamespace ? AggregateNamespaceStats() : GetOrCreateNamespaceStats(ns); + const Stats &cmd_stats = *cmd_stats_ptr; + Server::InfoEntries entries; entries.emplace_back("total_connections_received", total_clients_.load()); - entries.emplace_back("total_commands_processed", stats.total_calls.load()); - entries.emplace_back("instantaneous_ops_per_sec", stats.GetInstantaneousMetric(STATS_METRIC_COMMAND)); + entries.emplace_back("total_commands_processed", cmd_stats.total_calls.load()); + auto ops_per_sec = ns == kDefaultNamespace ? stats.GetInstantaneousMetric(STATS_METRIC_COMMAND) + : cmd_stats.GetInstantaneousMetric(STATS_METRIC_COMMAND); + entries.emplace_back("instantaneous_ops_per_sec", ops_per_sec); entries.emplace_back("total_net_input_bytes", stats.in_bytes.load()); entries.emplace_back("total_net_output_bytes", stats.out_bytes.load()); entries.emplace_back("instantaneous_input_kbps", @@ -1420,10 +1481,13 @@ Server::InfoEntries Server::GetStatsInfo() { return entries; } -Server::InfoEntries Server::GetCommandsStatsInfo() { +Server::InfoEntries Server::GetCommandsStatsInfo(const std::string &ns) { + auto cmd_stats_ptr = ns == kDefaultNamespace ? AggregateNamespaceStats() : GetOrCreateNamespaceStats(ns); + const Stats &cmd_stats = *cmd_stats_ptr; + InfoEntries entries; - for (const auto &cmd_stat : stats.commands_stats) { + for (const auto &cmd_stat : cmd_stats.commands_stats) { auto calls = cmd_stat.second.calls.load(); if (calls == 0) continue; @@ -1433,18 +1497,18 @@ Server::InfoEntries Server::GetCommandsStatsInfo() { static_cast(latency) / static_cast(calls))); } - for (const auto &cmd_hist : stats.commands_histogram) { + for (const auto &cmd_hist : cmd_stats.commands_histogram) { auto command_name = cmd_hist.first; - auto calls = stats.commands_histogram[command_name].calls.load(); + auto calls = cmd_hist.second.calls.load(); if (calls == 0) continue; - auto sum = stats.commands_histogram[command_name].sum.load(); + auto sum = cmd_hist.second.sum.load(); std::string result; - for (std::size_t i{0}; i < stats.commands_histogram[command_name].buckets.size(); ++i) { - auto bucket_value = stats.commands_histogram[command_name].buckets[i]->load(); + for (std::size_t i{0}; i < cmd_hist.second.buckets.size(); ++i) { + auto bucket_value = cmd_hist.second.buckets[i]->load(); auto bucket_bound = std::numeric_limits::infinity(); - if (i < stats.bucket_boundaries.size()) { - bucket_bound = stats.bucket_boundaries[i]; + if (i < cmd_stats.bucket_boundaries.size()) { + bucket_bound = cmd_stats.bucket_boundaries[i]; } result.append(fmt::format("{}={},", bucket_bound, bucket_value)); @@ -1539,11 +1603,16 @@ Server::InfoEntries Server::GetKeyspaceInfo(const std::string &ns) { // this section can't be shown when loading(i.e. !is_loading_). std::string Server::GetInfo(const std::string &ns, const std::vector §ions, InfoFormat format) { std::vector>> info_funcs = { - {"Server", &Server::GetServerInfo}, {"Clients", &Server::GetClientsInfo}, - {"Memory", &Server::GetMemoryInfo}, {"Persistence", &Server::GetPersistenceInfo}, - {"Stats", &Server::GetStatsInfo}, {"Replication", &Server::GetReplicationInfo}, - {"CPU", &Server::GetCpuInfo}, {"CommandStats", &Server::GetCommandsStatsInfo}, - {"Cluster", &Server::GetClusterInfo}, {"Keyspace", [&ns](Server *srv) { return srv->GetKeyspaceInfo(ns); }}, + {"Server", &Server::GetServerInfo}, + {"Clients", &Server::GetClientsInfo}, + {"Memory", &Server::GetMemoryInfo}, + {"Persistence", &Server::GetPersistenceInfo}, + {"Stats", [&ns](Server *srv) { return srv->GetStatsInfo(ns); }}, + {"Replication", &Server::GetReplicationInfo}, + {"CPU", &Server::GetCpuInfo}, + {"CommandStats", [&ns](Server *srv) { return srv->GetCommandsStatsInfo(ns); }}, + {"Cluster", &Server::GetClusterInfo}, + {"Keyspace", [&ns](Server *srv) { return srv->GetKeyspaceInfo(ns); }}, {"RocksDB", &Server::GetRocksDBInfo}, }; diff --git a/src/server/server.h b/src/server/server.h index 4214cecc53b..212489b9b5a 100644 --- a/src/server/server.h +++ b/src/server/server.h @@ -301,18 +301,21 @@ class Server { }; using InfoEntries = std::vector; - InfoEntries GetStatsInfo(); + InfoEntries GetStatsInfo(const std::string &ns); InfoEntries GetServerInfo(); InfoEntries GetMemoryInfo(); InfoEntries GetRocksDBInfo(); InfoEntries GetClientsInfo(); InfoEntries GetReplicationInfo(); - InfoEntries GetCommandsStatsInfo(); + InfoEntries GetCommandsStatsInfo(const std::string &ns); InfoEntries GetClusterInfo(); InfoEntries GetPersistenceInfo(); InfoEntries GetCpuInfo(); InfoEntries GetKeyspaceInfo(const std::string &ns); + std::shared_ptr GetOrCreateNamespaceStats(const std::string &ns); + std::shared_ptr AggregateNamespaceStats(); + enum class InfoFormat { Text, Json }; std::string GetInfo(const std::string &ns, const std::vector §ions, InfoFormat format = InfoFormat::Text); @@ -444,6 +447,13 @@ class Server { std::map db_scan_infos_; + // Per-namespace command statistics (keyed by namespace name), guarded by ns_stats_mu_. The global + // `stats` keeps the non-namespaced counters (net bytes, replication) and the sampled aggregate ops/sec. + std::unordered_map> ns_stats_; + std::shared_mutex ns_stats_mu_; + // Pre-populate a Stats' per-command maps so no runtime map insertion (and thus no data race) happens. + static void initCommandStats(Stats *stats); + LogCollector slow_log_; LogCollector perf_log_; diff --git a/tests/gocase/unit/info/info_test.go b/tests/gocase/unit/info/info_test.go index 09e593e2d7a..d15c3763416 100644 --- a/tests/gocase/unit/info/info_test.go +++ b/tests/gocase/unit/info/info_test.go @@ -305,3 +305,81 @@ func TestInfoFormat(t *testing.T) { require.ErrorContains(t, rdb.Do(ctx, "INFO", "server", "FORMAT").Err(), "syntax error") }) } + +func TestNamespaceStats(t *testing.T) { + password := "adminpass" + srv := util.StartServer(t, map[string]string{ + "requirepass": password, + "histogram-bucket-boundaries": "10,20,30", + }) + defer srv.Close() + + ctx := context.Background() + // admin is scoped to the default namespace; user authenticates with a namespace token. + admin := srv.NewClientWithOption(&redis.Options{Password: password}) + defer func() { require.NoError(t, admin.Close()) }() + + require.NoError(t, admin.Do(ctx, "NAMESPACE", "ADD", "ns1", "tok1").Err()) + user := srv.NewClientWithOption(&redis.Options{Password: "tok1"}) + defer func() { require.NoError(t, user.Close()) }() + + // GET is the only command that increments cmdstat_get, so the count is exact per namespace. + // The keys don't exist, but a lookup still counts as a command call. + const nsGets = 5 + const adminGets = 2 + for i := 0; i < nsGets; i++ { + require.ErrorIs(t, user.Get(ctx, fmt.Sprintf("k%d", i)).Err(), redis.Nil) + } + for i := 0; i < adminGets; i++ { + require.ErrorIs(t, admin.Get(ctx, fmt.Sprintf("a%d", i)).Err(), redis.Nil) + } + + getCalls := func(rdb *redis.Client, section string) string { + v := util.FindInfoEntry(rdb, "cmdstat_get", section) + return v // e.g. "calls=5,usec=...,usec_per_call=..." + } + + t.Run("commandstats are scoped to the caller namespace", func(t *testing.T) { + require.True(t, strings.HasPrefix(getCalls(user, "commandstats"), fmt.Sprintf("calls=%d,", nsGets))) + }) + + t.Run("admin sees the aggregate across all namespaces", func(t *testing.T) { + require.True(t, strings.HasPrefix(getCalls(admin, "commandstats"), fmt.Sprintf("calls=%d,", nsGets+adminGets))) + }) + + t.Run("total_commands_processed is namespace-scoped, admin is the aggregate", func(t *testing.T) { + mustAtoi := func(s string) int { + n, err := strconv.Atoi(s) + require.NoError(t, err) + return n + } + nsTotal := mustAtoi(util.FindInfoEntry(user, "total_commands_processed", "stats")) + adminTotal := mustAtoi(util.FindInfoEntry(admin, "total_commands_processed", "stats")) + require.GreaterOrEqual(t, nsTotal, nsGets) + // the aggregate also includes the admin/default namespace's own commands + require.Greater(t, adminTotal, nsTotal) + }) + + t.Run("namespace user can read its own LATENCY HISTOGRAM", func(t *testing.T) { + // LATENCY is no longer admin-only; a namespace user gets its own histogram. + res, err := user.Do(ctx, "LATENCY", "HISTOGRAM", "get").Result() + require.NoError(t, err) + require.Contains(t, fmt.Sprintf("%v", res), "get") + }) + + t.Run("admin LATENCY HISTOGRAM reflects the aggregate", func(t *testing.T) { + res, err := admin.Do(ctx, "LATENCY", "HISTOGRAM", "get").Result() + require.NoError(t, err) + require.Contains(t, fmt.Sprintf("%v", res), "get") + }) + + t.Run("deleting a namespace keeps its stats and its connection working", func(t *testing.T) { + // Stats are not erased on NAMESPACE DEL, so the aggregate is unchanged and a connection still + // scoped to the deleted namespace keeps updating the same (still-aggregated) stats. + require.NoError(t, admin.Do(ctx, "NAMESPACE", "DEL", "ns1").Err()) + require.True(t, strings.HasPrefix(getCalls(admin, "commandstats"), fmt.Sprintf("calls=%d,", nsGets+adminGets))) + + require.ErrorIs(t, user.Get(ctx, "k0").Err(), redis.Nil) + require.True(t, strings.HasPrefix(getCalls(admin, "commandstats"), fmt.Sprintf("calls=%d,", nsGets+adminGets+1))) + }) +}