This is an automated email from the ASF dual-hosted git repository.
hulk pushed a commit to branch unstable
in repository https://gitbox.apache.org/repos/asf/kvrocks.git
The following commit(s) were added to refs/heads/unstable by this push:
new 5b54f9cd Record and export the keyspace hit/miss count to INFO command
(#1971)
5b54f9cd is described below
commit 5b54f9cdb1aa3b7b789ef2f05bb1b198e09c9da6
Author: hulk <[email protected]>
AuthorDate: Wed Dec 27 09:45:47 2023 +0800
Record and export the keyspace hit/miss count to INFO command (#1971)
Currently, Redis supports inspecting the keyspace hit/miss count via
INFO command to see if the hit ratio is expected. So we also export
the same metric to align with Redis metrics.
---
src/commands/cmd_replication.cc | 6 ++--
src/server/redis_connection.cc | 2 +-
src/server/redis_request.cc | 6 ++--
src/server/server.cc | 22 +++++++++------
src/stats/stats.h | 16 +++++------
src/storage/event_listener.cc | 4 +--
src/storage/redis_db.cc | 48 ++++++++------------------------
src/storage/redis_db.h | 4 +--
src/storage/storage.cc | 55 ++++++++++++++++++++++++++++++-------
src/storage/storage.h | 27 +++++++++++++-----
src/types/redis_string.cc | 30 ++++----------------
tests/gocase/unit/info/info_test.go | 41 +++++++++++++++++++++++++++
12 files changed, 154 insertions(+), 107 deletions(-)
diff --git a/src/commands/cmd_replication.cc b/src/commands/cmd_replication.cc
index 2a4f074a..0a86a9cc 100644
--- a/src/commands/cmd_replication.cc
+++ b/src/commands/cmd_replication.cc
@@ -84,7 +84,7 @@ class CommandPSync : public Commander {
}
if (need_full_sync) {
- srv->stats.IncrPSyncErrCounter();
+ srv->stats.IncrPSyncErrCount();
return {Status::RedisExecErr, *output};
}
@@ -98,7 +98,7 @@ class CommandPSync : public Commander {
return s.Prefixed("failed to set blocking mode on socket");
}
- srv->stats.IncrPSyncOKCounter();
+ srv->stats.IncrPSyncOKCount();
s = srv->AddSlave(conn, next_repl_seq_);
if (!s.IsOK()) {
std::string err = "-ERR " + s.Msg() + "\r\n";
@@ -216,7 +216,7 @@ class CommandFetchMeta : public Commander {
conn->NeedNotFreeBufferEvent();
conn->EnableFlag(redis::Connection::kCloseAsync);
- srv->stats.IncrFullSyncCounter();
+ srv->stats.IncrFullSyncCount();
// Feed-replica-meta thread
auto t = GET_OR_RET(util::CreateThread("feed-repl-info", [srv, repl_fd,
ip, bev = conn->GetBufferEvent()] {
diff --git a/src/server/redis_connection.cc b/src/server/redis_connection.cc
index 73ed3a80..ae80e950 100644
--- a/src/server/redis_connection.cc
+++ b/src/server/redis_connection.cc
@@ -126,7 +126,7 @@ void Connection::OnEvent(bufferevent *bev, int16_t events) {
}
void Connection::Reply(const std::string &msg) {
- owner_->srv->stats.IncrOutbondBytes(msg.size());
+ owner_->srv->stats.IncrOutboundBytes(msg.size());
redis::Reply(bufferevent_get_output(bev_), msg);
}
diff --git a/src/server/redis_request.cc b/src/server/redis_request.cc
index f6faa7b1..4c796b41 100644
--- a/src/server/redis_request.cc
+++ b/src/server/redis_request.cc
@@ -67,7 +67,7 @@ Status Request::Tokenize(evbuffer *input) {
}
pipeline_size++;
- srv_->stats.IncrInbondBytes(line.length);
+ srv_->stats.IncrInboundBytes(line.length);
if (line[0] == '*') {
auto parse_result = ParseInt<int64_t>(std::string(line.get() + 1,
line.length - 1), 10);
if (!parse_result) {
@@ -101,7 +101,7 @@ Status Request::Tokenize(evbuffer *input) {
UniqueEvbufReadln line(input, EVBUFFER_EOL_CRLF_STRICT);
if (!line || line.length <= 0) return Status::OK();
- srv_->stats.IncrInbondBytes(line.length);
+ srv_->stats.IncrInboundBytes(line.length);
if (line[0] != '$') {
return {Status::NotOK, "Protocol error: expected '$'"};
}
@@ -125,7 +125,7 @@ Status Request::Tokenize(evbuffer *input) {
char *data = reinterpret_cast<char *>(evbuffer_pullup(input,
static_cast<ssize_t>(bulk_len_ + 2)));
tokens_.emplace_back(data, bulk_len_);
evbuffer_drain(input, bulk_len_ + 2);
- srv_->stats.IncrInbondBytes(bulk_len_ + 2);
+ srv_->stats.IncrInboundBytes(bulk_len_ + 2);
--multi_bulk_len_;
if (multi_bulk_len_ == 0) {
state_ = ArrayLen;
diff --git a/src/server/server.cc b/src/server/server.cc
index ab7d311f..231d532d 100644
--- a/src/server/server.cc
+++ b/src/server/server.cc
@@ -801,8 +801,8 @@ void Server::GetRocksDBInfo(std::string *info) {
<< "]:" << cf_stats_map["memtable-limit-stops"] << "\r\n";
}
- auto db_stats = storage->GetDB()->GetDBOptions().statistics;
- if (db_stats) {
+ auto rocksdb_stats = storage->GetDB()->GetDBOptions().statistics;
+ if (rocksdb_stats) {
std::map<std::string, uint32_t> block_cache_stats = {
{"block_cache_hit", rocksdb::Tickers::BLOCK_CACHE_HIT},
{"block_cache_index_hit", rocksdb::Tickers::BLOCK_CACHE_INDEX_HIT},
@@ -814,7 +814,7 @@ void Server::GetRocksDBInfo(std::string *info) {
{"block_cache_data_miss", rocksdb::Tickers::BLOCK_CACHE_DATA_MISS},
};
for (const auto &iter : block_cache_stats) {
- string_stream << iter.first << ":" <<
db_stats->getTickerCount(iter.second) << "\r\n";
+ string_stream << iter.first << ":" <<
rocksdb_stats->getTickerCount(iter.second) << "\r\n";
}
}
@@ -829,8 +829,9 @@ void Server::GetRocksDBInfo(std::string *info) {
string_stream << "num_live_versions:" << num_live_versions << "\r\n";
string_stream << "num_super_version:" << num_super_version << "\r\n";
string_stream << "num_background_errors:" << num_background_errors << "\r\n";
- string_stream << "flush_count:" << storage->GetFlushCount() << "\r\n";
- string_stream << "compaction_count:" << storage->GetCompactionCount() <<
"\r\n";
+ auto db_stats = storage->GetDBStats();
+ string_stream << "flush_count:" << db_stats->flush_count << "\r\n";
+ string_stream << "compaction_count:" << db_stats->compaction_count << "\r\n";
string_stream << "put_per_sec:" <<
stats.GetInstantaneousMetric(STATS_METRIC_ROCKSDB_PUT) << "\r\n";
string_stream << "get_per_sec:"
<< stats.GetInstantaneousMetric(STATS_METRIC_ROCKSDB_GET) +
@@ -1027,9 +1028,14 @@ void Server::GetStatsInfo(std::string *info) {
<<
static_cast<float>(stats.GetInstantaneousMetric(STATS_METRIC_NET_INPUT) / 1024)
<< "\r\n";
string_stream << "instantaneous_output_kbps:"
<<
static_cast<float>(stats.GetInstantaneousMetric(STATS_METRIC_NET_OUTPUT) /
1024) << "\r\n";
- string_stream << "sync_full:" << stats.fullsync_counter << "\r\n";
- string_stream << "sync_partial_ok:" << stats.psync_ok_counter << "\r\n";
- string_stream << "sync_partial_err:" << stats.psync_err_counter << "\r\n";
+ string_stream << "sync_full:" << stats.fullsync_count << "\r\n";
+ string_stream << "sync_partial_ok:" << stats.psync_ok_count << "\r\n";
+ string_stream << "sync_partial_err:" << stats.psync_err_count << "\r\n";
+
+ auto db_stats = storage->GetDBStats();
+ string_stream << "keyspace_hits:" << db_stats->keyspace_hits << "\r\n";
+ string_stream << "keyspace_misses:" << db_stats->keyspace_misses << "\r\n";
+
{
std::lock_guard<std::mutex> lg(pubsub_channels_mu_);
string_stream << "pubsub_channels:" << pubsub_channels_.size() << "\r\n";
diff --git a/src/stats/stats.h b/src/stats/stats.h
index 114ac667..88ab2108 100644
--- a/src/stats/stats.h
+++ b/src/stats/stats.h
@@ -64,19 +64,19 @@ class Stats {
mutable std::shared_mutex inst_metrics_mutex;
std::vector<InstMetric> inst_metrics;
- std::atomic<uint64_t> fullsync_counter = {0};
- std::atomic<uint64_t> psync_err_counter = {0};
- std::atomic<uint64_t> psync_ok_counter = {0};
+ std::atomic<uint64_t> fullsync_count = {0};
+ std::atomic<uint64_t> psync_err_count = {0};
+ std::atomic<uint64_t> psync_ok_count = {0};
std::map<std::string, CommandStat> commands_stats;
Stats();
void IncrCalls(const std::string &command_name);
void IncrLatency(uint64_t latency, const std::string &command_name);
- void IncrInbondBytes(uint64_t bytes) { in_bytes.fetch_add(bytes,
std::memory_order_relaxed); }
- void IncrOutbondBytes(uint64_t bytes) { out_bytes.fetch_add(bytes,
std::memory_order_relaxed); }
- void IncrFullSyncCounter() { fullsync_counter.fetch_add(1,
std::memory_order_relaxed); }
- void IncrPSyncErrCounter() { psync_err_counter.fetch_add(1,
std::memory_order_relaxed); }
- void IncrPSyncOKCounter() { psync_ok_counter.fetch_add(1,
std::memory_order_relaxed); }
+ void IncrInboundBytes(uint64_t bytes) { in_bytes.fetch_add(bytes,
std::memory_order_relaxed); }
+ void IncrOutboundBytes(uint64_t bytes) { out_bytes.fetch_add(bytes,
std::memory_order_relaxed); }
+ void IncrFullSyncCount() { fullsync_count.fetch_add(1,
std::memory_order_relaxed); }
+ void IncrPSyncErrCount() { psync_err_count.fetch_add(1,
std::memory_order_relaxed); }
+ void IncrPSyncOKCount() { psync_ok_count.fetch_add(1,
std::memory_order_relaxed); }
static int64_t GetMemoryRSS();
void TrackInstantaneousMetric(int metric, uint64_t current_reading);
uint64_t GetInstantaneousMetric(int metric) const;
diff --git a/src/storage/event_listener.cc b/src/storage/event_listener.cc
index 6e054945..8bc204f7 100644
--- a/src/storage/event_listener.cc
+++ b/src/storage/event_listener.cc
@@ -84,7 +84,7 @@ void EventListener::OnCompactionCompleted(rocksdb::DB *db,
const rocksdb::Compac
<< ", input bytes: " << ci.stats.total_input_bytes << ", output
bytes:" << ci.stats.total_output_bytes
<< ", is_manual_compaction:" << (ci.stats.is_manual_compaction ?
"yes" : "no")
<< ", elapsed(micro): " << ci.stats.elapsed_micros;
- storage_->IncrCompactionCount(1);
+ storage_->RecordStat(engine::StatType::CompactionCount, 1);
storage_->CheckDBSizeLimit();
}
@@ -94,7 +94,7 @@ void EventListener::OnFlushBegin(rocksdb::DB *db, const
rocksdb::FlushJobInfo &f
}
void EventListener::OnFlushCompleted(rocksdb::DB *db, const
rocksdb::FlushJobInfo &fi) {
- storage_->IncrFlushCount(1);
+ storage_->RecordStat(engine::StatType::FlushCount, 1);
storage_->CheckDBSizeLimit();
LOG(INFO) << "[event_listener/flush_completed] column family: " <<
fi.cf_name << ", thread_id: " << fi.thread_id
<< ", job_id: " << fi.job_id << ", file: " << fi.file_path
diff --git a/src/storage/redis_db.cc b/src/storage/redis_db.cc
index a3734730..bdbd9789 100644
--- a/src/storage/redis_db.cc
+++ b/src/storage/redis_db.cc
@@ -25,6 +25,7 @@
#include <utility>
#include "cluster/redis_slot.h"
+#include "common/scope_exit.h"
#include "db_util.h"
#include "parse_util.h"
#include "rocksdb/iterator.h"
@@ -44,6 +45,15 @@ rocksdb::Status Database::ParseMetadata(RedisTypes types,
Slice *bytes, Metadata
std::string old_metadata;
metadata->Encode(&old_metadata);
+ bool is_keyspace_hit = false;
+ ScopeExit se([this, &is_keyspace_hit] {
+ if (is_keyspace_hit) {
+ storage_->RecordStat(engine::StatType::KeyspaceHits, 1);
+ } else {
+ storage_->RecordStat(engine::StatType::KeyspaceMisses, 1);
+ }
+ });
+
auto s = metadata->Decode(bytes);
if (!s.ok()) return s;
@@ -64,6 +74,7 @@ rocksdb::Status Database::ParseMetadata(RedisTypes types,
Slice *bytes, Metadata
auto _ [[maybe_unused]] = metadata->Decode(old_metadata);
return rocksdb::Status::NotFound("no element found");
}
+ is_keyspace_hit = true;
return s;
}
@@ -597,43 +608,6 @@ rocksdb::Status Database::ClearKeysOfSlot(const
rocksdb::Slice &ns, int slot) {
return rocksdb::Status::OK();
}
-rocksdb::Status Database::GetSlotKeysInfo(int slot, std::map<int, uint64_t>
*slotskeys, std::vector<std::string> *keys,
- int count) {
- LatestSnapShot ss(storage_);
- rocksdb::ReadOptions read_options = storage_->DefaultScanOptions();
- read_options.snapshot = ss.GetSnapShot();
-
- auto iter = util::UniqueIterator(storage_, read_options,
metadata_cf_handle_);
- bool end = false;
- for (int i = 0; i < HASH_SLOTS_SIZE; i++) {
- std::string prefix = ComposeSlotKeyPrefix(namespace_, i);
- uint64_t total = 0;
- int cnt = 0;
- if (slot != -1 && i != slot) {
- (*slotskeys)[i] = total;
- continue;
- }
- for (iter->Seek(prefix); iter->Valid(); iter->Next()) {
- if (!iter->key().starts_with(prefix)) {
- break;
- }
- total++;
- if (slot != -1 && count > 0 && !end) {
- // Get user key
- if (cnt < count) {
- auto [_, user_key] = ExtractNamespaceKey(iter->key(), true);
- keys->emplace_back(user_key.ToString());
- cnt++;
- }
- }
- }
- // Maybe cnt < count
- if (cnt > 0) end = true;
- (*slotskeys)[i] = total;
- }
- return rocksdb::Status::OK();
-}
-
rocksdb::Status Database::KeyExist(const std::string &key) {
int cnt = 0;
std::vector<rocksdb::Slice> keys;
diff --git a/src/storage/redis_db.h b/src/storage/redis_db.h
index 7258923c..65804441 100644
--- a/src/storage/redis_db.h
+++ b/src/storage/redis_db.h
@@ -34,7 +34,7 @@ class Database {
static constexpr uint64_t RANDOM_KEY_SCAN_LIMIT = 60;
explicit Database(engine::Storage *storage, std::string ns = "");
- [[nodiscard]] static rocksdb::Status ParseMetadata(RedisTypes types, Slice
*bytes, Metadata *metadata);
+ [[nodiscard]] rocksdb::Status ParseMetadata(RedisTypes types, Slice *bytes,
Metadata *metadata);
[[nodiscard]] rocksdb::Status GetMetadata(RedisTypes types, const Slice
&ns_key, Metadata *metadata);
[[nodiscard]] rocksdb::Status GetMetadata(RedisTypes types, const Slice
&ns_key, std::string *raw_value,
Metadata *metadata, Slice *rest);
@@ -60,8 +60,6 @@ class Database {
std::string *begin,
std::string *end,
rocksdb::ColumnFamilyHandle *cf_handle = nullptr);
[[nodiscard]] rocksdb::Status ClearKeysOfSlot(const rocksdb::Slice &ns, int
slot);
- [[nodiscard]] rocksdb::Status GetSlotKeysInfo(int slot, std::map<int,
uint64_t> *slotskeys,
- std::vector<std::string>
*keys, int count);
[[nodiscard]] rocksdb::Status KeyExist(const std::string &key);
protected:
diff --git a/src/storage/storage.cc b/src/storage/storage.cc
index a0e53c0b..c176db1b 100644
--- a/src/storage/storage.cc
+++ b/src/storage/storage.cc
@@ -200,12 +200,6 @@ Status Storage::SetOptionForAllColumnFamilies(const
std::string &key, const std:
return Status::OK();
}
-Status Storage::SetOption(const std::string &key, const std::string &value) {
- auto s = db_->SetOptions({{key, value}});
- if (!s.ok()) return {Status::NotOK, s.ToString()};
- return Status::OK();
-}
-
Status Storage::SetDBOption(const std::string &key, const std::string &value) {
auto s = db_->SetDBOptions({{key, value}});
if (!s.ok()) return {Status::NotOK, s.ToString()};
@@ -525,10 +519,15 @@ rocksdb::Status Storage::Get(const rocksdb::ReadOptions
&options, const rocksdb:
rocksdb::Status Storage::Get(const rocksdb::ReadOptions &options,
rocksdb::ColumnFamilyHandle *column_family,
const rocksdb::Slice &key, std::string *value) {
+ rocksdb::Status s;
if (is_txn_mode_ && txn_write_batch_->GetWriteBatch()->Count() > 0) {
- return txn_write_batch_->GetFromBatchAndDB(db_.get(), options,
column_family, key, value);
+ s = txn_write_batch_->GetFromBatchAndDB(db_.get(), options, column_family,
key, value);
+ } else {
+ s = db_->Get(options, column_family, key, value);
}
- return db_->Get(options, column_family, key, value);
+
+ recordKeyspaceStat(column_family, s);
+ return s;
}
rocksdb::Status Storage::Get(const rocksdb::ReadOptions &options, const
rocksdb::Slice &key,
@@ -538,16 +537,31 @@ rocksdb::Status Storage::Get(const rocksdb::ReadOptions
&options, const rocksdb:
rocksdb::Status Storage::Get(const rocksdb::ReadOptions &options,
rocksdb::ColumnFamilyHandle *column_family,
const rocksdb::Slice &key, rocksdb::PinnableSlice
*value) {
+ rocksdb::Status s;
if (is_txn_mode_ && txn_write_batch_->GetWriteBatch()->Count() > 0) {
- return txn_write_batch_->GetFromBatchAndDB(db_.get(), options,
column_family, key, value);
+ s = txn_write_batch_->GetFromBatchAndDB(db_.get(), options, column_family,
key, value);
+ } else {
+ s = db_->Get(options, column_family, key, value);
}
- return db_->Get(options, column_family, key, value);
+
+ recordKeyspaceStat(column_family, s);
+ return s;
}
rocksdb::Iterator *Storage::NewIterator(const rocksdb::ReadOptions &options) {
return NewIterator(options, db_->DefaultColumnFamily());
}
+void Storage::recordKeyspaceStat(const rocksdb::ColumnFamilyHandle
*column_family, const rocksdb::Status &s) {
+ if (column_family->GetName() != kMetadataColumnFamilyName) return;
+
+ // Don't record keyspace hits here because we cannot tell
+ // if the key was expired or not. So we record it when parsing the metadata.
+ if (s.IsNotFound() || s.IsInvalidArgument()) {
+ RecordStat(StatType::KeyspaceMisses, 1);
+ }
+}
+
rocksdb::Iterator *Storage::NewIterator(const rocksdb::ReadOptions &options,
rocksdb::ColumnFamilyHandle
*column_family) {
auto iter = db_->NewIterator(options, column_family);
@@ -566,6 +580,10 @@ void Storage::MultiGet(const rocksdb::ReadOptions
&options, rocksdb::ColumnFamil
} else {
db_->MultiGet(options, column_family, num_keys, keys, values, statuses,
false);
}
+
+ for (size_t i = 0; i < num_keys; i++) {
+ recordKeyspaceStat(column_family, statuses[i]);
+ }
}
rocksdb::Status Storage::Write(const rocksdb::WriteOptions &options,
rocksdb::WriteBatch *updates) {
@@ -641,6 +659,23 @@ Status Storage::ReplicaApplyWriteBatch(std::string
&&raw_batch) {
return Status::OK();
}
+void Storage::RecordStat(StatType type, uint64_t v) {
+ switch (type) {
+ case StatType::FlushCount:
+ db_stats_.flush_count += v;
+ break;
+ case StatType::CompactionCount:
+ db_stats_.compaction_count += v;
+ break;
+ case StatType::KeyspaceHits:
+ db_stats_.keyspace_hits += v;
+ break;
+ case StatType::KeyspaceMisses:
+ db_stats_.keyspace_misses += v;
+ break;
+ }
+}
+
rocksdb::ColumnFamilyHandle *Storage::GetCFHandle(const std::string &name) {
if (name == kMetadataColumnFamilyName) {
return cf_handles_[1];
diff --git a/src/storage/storage.h b/src/storage/storage.h
index 43c3cf1c..b0de2dd0 100644
--- a/src/storage/storage.h
+++ b/src/storage/storage.h
@@ -80,6 +80,20 @@ inline const std::vector<CompressionOption>
CompressionOptions = {
{rocksdb::kZSTD, "zstd", "kZSTD"},
};
+enum class StatType {
+ CompactionCount,
+ FlushCount,
+ KeyspaceHits,
+ KeyspaceMisses,
+};
+
+struct DBStats {
+ std::atomic<uint64_t> compaction_count = 0;
+ std::atomic<uint64_t> flush_count = 0;
+ std::atomic<uint64_t> keyspace_hits = 0;
+ std::atomic<uint64_t> keyspace_misses = 0;
+};
+
class Storage {
public:
explicit Storage(Config *config);
@@ -93,7 +107,6 @@ class Storage {
void SetBlobDB(rocksdb::ColumnFamilyOptions *cf_options);
rocksdb::Options InitRocksDBOptions();
Status SetOptionForAllColumnFamilies(const std::string &key, const
std::string &value);
- Status SetOption(const std::string &key, const std::string &value);
Status SetDBOption(const std::string &key, const std::string &value);
Status CreateColumnFamilies(const rocksdb::Options &options);
Status CreateBackup();
@@ -145,13 +158,12 @@ class Storage {
std::shared_lock<std::shared_mutex> ReadLockGuard();
std::unique_lock<std::shared_mutex> WriteLockGuard();
- uint64_t GetFlushCount() const { return flush_count_; }
- void IncrFlushCount(uint64_t n) { flush_count_.fetch_add(n); }
- uint64_t GetCompactionCount() const { return compaction_count_; }
- void IncrCompactionCount(uint64_t n) { compaction_count_.fetch_add(n); }
bool IsSlotIdEncoded() const { return config_->slot_id_encoded; }
Config *GetConfig() const { return config_; }
+ const DBStats *GetDBStats() const { return &db_stats_; }
+ void RecordStat(StatType type, uint64_t v);
+
Status BeginTxn();
Status CommitTxn();
ObserverOrUniquePtr<rocksdb::WriteBatchBase> GetWriteBatchBase();
@@ -214,8 +226,8 @@ class Storage {
std::vector<rocksdb::ColumnFamilyHandle *> cf_handles_;
LockManager lock_mgr_;
bool db_size_limit_reached_ = false;
- std::atomic<uint64_t> flush_count_{0};
- std::atomic<uint64_t> compaction_count_{0};
+
+ DBStats db_stats_;
std::shared_mutex db_rw_lock_;
bool db_closing_ = true;
@@ -234,6 +246,7 @@ class Storage {
rocksdb::WriteOptions write_opts_ = rocksdb::WriteOptions();
rocksdb::Status writeToDB(const rocksdb::WriteOptions &options,
rocksdb::WriteBatch *updates);
+ void recordKeyspaceStat(const rocksdb::ColumnFamilyHandle *column_family,
const rocksdb::Status &s);
};
} // namespace engine
diff --git a/src/types/redis_string.cc b/src/types/redis_string.cc
index cc851b0b..7178311a 100644
--- a/src/types/redis_string.cc
+++ b/src/types/redis_string.cc
@@ -47,20 +47,11 @@ std::vector<rocksdb::Status> String::getRawValues(const
std::vector<Slice> &keys
if (!statuses[i].ok()) continue;
(*raw_values)[i].assign(pin_values[i].data(), pin_values[i].size());
Metadata metadata(kRedisNone, false);
- auto s = metadata.Decode((*raw_values)[i]);
+ Slice slice = (*raw_values)[i];
+ auto s = ParseMetadata({kRedisString}, &slice, &metadata);
if (!s.ok()) {
- (*raw_values)[i].clear();
statuses[i] = s;
- continue;
- }
- if (metadata.Expired()) {
- (*raw_values)[i].clear();
- statuses[i] = rocksdb::Status::NotFound(kErrMsgKeyExpired);
- continue;
- }
- if (metadata.Type() != kRedisString && metadata.size > 0) {
(*raw_values)[i].clear();
- statuses[i] = rocksdb::Status::InvalidArgument(kErrMsgWrongType);
continue;
}
}
@@ -70,23 +61,12 @@ std::vector<rocksdb::Status> String::getRawValues(const
std::vector<Slice> &keys
rocksdb::Status String::getRawValue(const std::string &ns_key, std::string
*raw_value) {
raw_value->clear();
- rocksdb::ReadOptions read_options;
- LatestSnapShot ss(storage_);
- read_options.snapshot = ss.GetSnapShot();
- rocksdb::Status s = storage_->Get(read_options, metadata_cf_handle_, ns_key,
raw_value);
+ auto s = GetRawMetadata(ns_key, raw_value);
if (!s.ok()) return s;
Metadata metadata(kRedisNone, false);
- s = metadata.Decode(*raw_value);
- if (!s.ok()) return s;
- if (metadata.Expired()) {
- raw_value->clear();
- return rocksdb::Status::NotFound(kErrMsgKeyExpired);
- }
- if (metadata.Type() != kRedisString && metadata.size > 0) {
- return rocksdb::Status::InvalidArgument(kErrMsgWrongType);
- }
- return rocksdb::Status::OK();
+ Slice slice = *raw_value;
+ return ParseMetadata({kRedisString}, &slice, &metadata);
}
rocksdb::Status String::getValueAndExpire(const std::string &ns_key,
std::string *value, uint64_t *expire) {
diff --git a/tests/gocase/unit/info/info_test.go
b/tests/gocase/unit/info/info_test.go
index 136be197..dd4b0cdc 100644
--- a/tests/gocase/unit/info/info_test.go
+++ b/tests/gocase/unit/info/info_test.go
@@ -26,6 +26,8 @@ import (
"testing"
"time"
+ "github.com/redis/go-redis/v9"
+
"github.com/apache/kvrocks/tests/gocase/util"
"github.com/stretchr/testify/require"
)
@@ -101,3 +103,42 @@ func TestInfo(t *testing.T) {
require.Equal(t, "1", util.FindInfoEntry(rdb0,
"cluster_enabled", "cluster"))
})
}
+
+func TestKeyspaceHitMiss(t *testing.T) {
+ srv0 := util.StartServer(t, map[string]string{})
+ defer func() { srv0.Close() }()
+ rdb0 := srv0.NewClient()
+ defer func() { require.NoError(t, rdb0.Close()) }()
+
+ srv := util.StartServer(t, map[string]string{})
+ defer srv.Close()
+
+ ctx := context.Background()
+ rdb := srv.NewClient()
+ defer func() { require.NoError(t, rdb.Close()) }()
+
+ require.Equal(t, "0", util.FindInfoEntry(rdb0, "keyspace_hits",
"stats"))
+ require.Equal(t, "0", util.FindInfoEntry(rdb0, "keyspace_misses",
"stats"))
+ require.NoError(t, rdb0.Set(ctx, "foo", "bar", 0).Err())
+
+ require.NoError(t, rdb0.Get(ctx, "foo").Err())
+ require.Equal(t, "1", util.FindInfoEntry(rdb0, "keyspace_hits",
"stats"))
+ require.Equal(t, "0", util.FindInfoEntry(rdb0, "keyspace_misses",
"stats"))
+
+ require.EqualError(t, rdb0.Get(ctx, "no_exists_key").Err(),
redis.Nil.Error())
+ require.Equal(t, "1", util.FindInfoEntry(rdb0, "keyspace_hits",
"stats"))
+ require.Equal(t, "1", util.FindInfoEntry(rdb0, "keyspace_misses",
"stats"))
+
+ require.NoError(t, rdb0.HSet(ctx, "hash", "f1", "v1").Err())
+ require.Equal(t, "1", util.FindInfoEntry(rdb0, "keyspace_hits",
"stats"))
+ // increase by 2 because of the previous HSet will try to get the key
first
+ require.Equal(t, "2", util.FindInfoEntry(rdb0, "keyspace_misses",
"stats"))
+
+ require.NoError(t, rdb0.HGet(ctx, "hash", "f1").Err())
+ require.Equal(t, "2", util.FindInfoEntry(rdb0, "keyspace_hits",
"stats"))
+ require.Equal(t, "2", util.FindInfoEntry(rdb0, "keyspace_misses",
"stats"))
+
+ require.EqualError(t, rdb0.HGet(ctx, "no_exists_hash", "f1").Err(),
redis.Nil.Error())
+ require.Equal(t, "2", util.FindInfoEntry(rdb0, "keyspace_hits",
"stats"))
+ require.Equal(t, "3", util.FindInfoEntry(rdb0, "keyspace_misses",
"stats"))
+}
