mirror of
https://github.com/zeek/zeek.git
synced 2025-10-02 14:48:21 +00:00
Merge remote-tracking branch 'origin/topic/timw/storage-metrics'
* origin/topic/timw/storage-metrics: Add SQLite page_count and file_size metrics Add btests to cover storage metrics Add storage metrics for operations, expirations, data transferred Fix ordering of telemtry metrics when running under test Make RunPragma take an optional value parser to return data Make SQLite::Step take a callback function for parsing result data
This commit is contained in:
Tim Wojtulewicz
commit
cb2e193452
28 changed files with 686 additions and 97 deletions
14
CHANGES
14
CHANGES
|
|
@ -1,3 +1,17 @@
|
|||
8.0.0-dev.693 | 2025-07-18 14:28:29 -0700
|
||||
|
||||
* Add SQLite page_count and file_size metrics (Tim Wojtulewicz, Corelight)
|
||||
|
||||
* Add btests to cover storage metrics (Tim Wojtulewicz, Corelight)
|
||||
|
||||
* Add storage metrics for operations, expirations, data transferred (Tim Wojtulewicz, Corelight)
|
||||
|
||||
* Fix ordering of telemtry metrics when running under test (Tim Wojtulewicz, Corelight)
|
||||
|
||||
* Make RunPragma take an optional value parser to return data (Tim Wojtulewicz, Corelight)
|
||||
|
||||
* Make SQLite::Step take a callback function for parsing result data (Tim Wojtulewicz, Corelight)
|
||||
|
||||
8.0.0-dev.684 | 2025-07-17 12:58:57 -0700
|
||||
|
||||
* Update libkqueue submodule [nomail] (Tim Wojtulewicz, Corelight)
|
||||
|
|
|
|||
2
VERSION
2
VERSION
|
|
@ -1 +1 @@
|
|||
8.0.0-dev.684
|
||||
8.0.0-dev.693
|
||||
|
|
|
|||
|
|
@ -29,4 +29,7 @@ export {
|
|||
## backend.
|
||||
expire_time: interval &default=0sec;
|
||||
};
|
||||
|
||||
# The histogram buckets to use for operation latency metrics, in seconds.
|
||||
const latency_metric_bounds: vector of double = { 0.001, 0.01, 0.1, 1.0, } &redef;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -8,9 +8,30 @@
|
|||
#include "zeek/storage/Manager.h"
|
||||
#include "zeek/storage/ReturnCode.h"
|
||||
#include "zeek/storage/storage-events.bif.h"
|
||||
#include "zeek/telemetry/Counter.h"
|
||||
#include "zeek/telemetry/Histogram.h"
|
||||
#include "zeek/telemetry/Manager.h"
|
||||
|
||||
namespace zeek::storage {
|
||||
|
||||
namespace detail {
|
||||
|
||||
OperationMetrics::OperationMetrics(const telemetry::CounterFamilyPtr& results_family,
|
||||
const telemetry::HistogramFamilyPtr& latency_family, std::string_view operation_type,
|
||||
std::string_view backend_type, std::string_view backend_config)
|
||||
: success(results_family->GetOrAdd(
|
||||
{{"operation", operation_type}, {"type", backend_type}, {"config", backend_config}, {"result", "success"}})),
|
||||
fail(results_family->GetOrAdd(
|
||||
{{"operation", operation_type}, {"type", backend_type}, {"config", backend_config}, {"result", "fail"}})),
|
||||
error(results_family->GetOrAdd(
|
||||
{{"operation", operation_type}, {"type", backend_type}, {"config", backend_config}, {"result", "error"}})),
|
||||
timeouts(results_family->GetOrAdd(
|
||||
{{"operation", operation_type}, {"type", backend_type}, {"config", backend_config}, {"result", "timeout"}})),
|
||||
latency(latency_family->GetOrAdd(
|
||||
{{"operation", operation_type}, {"type", backend_type}, {"config", backend_config}})) {}
|
||||
|
||||
} // namespace detail
|
||||
|
||||
RecordValPtr OperationResult::BuildVal() { return MakeVal(code, err_str, value); }
|
||||
|
||||
RecordValPtr OperationResult::MakeVal(EnumValPtr code, std::string_view err_str, ValPtr value) {
|
||||
|
|
@ -32,9 +53,14 @@ ResultCallback::ResultCallback(zeek::detail::trigger::TriggerPtr trigger, const
|
|||
void ResultCallback::Timeout() {
|
||||
if ( ! IsSyncCallback() )
|
||||
trigger->Cache(assoc, OperationResult::MakeVal(ReturnCode::TIMEOUT).get());
|
||||
|
||||
if ( operation_metrics )
|
||||
operation_metrics->timeouts->Inc();
|
||||
}
|
||||
|
||||
void ResultCallback::Complete(OperationResult res) {
|
||||
UpdateOperationMetrics(res.code);
|
||||
|
||||
// If this is a sync callback, there isn't a trigger to process. Store the result and bail.
|
||||
if ( IsSyncCallback() ) {
|
||||
result = std::move(res);
|
||||
|
|
@ -46,6 +72,25 @@ void ResultCallback::Complete(OperationResult res) {
|
|||
trigger->Release();
|
||||
}
|
||||
|
||||
void ResultCallback::Init(detail::OperationMetrics* m) {
|
||||
operation_metrics = m;
|
||||
start_time = util::current_time(true);
|
||||
}
|
||||
|
||||
void ResultCallback::UpdateOperationMetrics(EnumValPtr c) {
|
||||
if ( operation_metrics ) {
|
||||
if ( c == ReturnCode::SUCCESS )
|
||||
operation_metrics->success->Inc();
|
||||
else if ( c == ReturnCode::KEY_EXISTS || c == ReturnCode::KEY_NOT_FOUND )
|
||||
operation_metrics->fail->Inc();
|
||||
else if ( c != ReturnCode::IN_PROGRESS )
|
||||
operation_metrics->error->Inc();
|
||||
|
||||
// Store the latency between start and completion in milliseconds.
|
||||
operation_metrics->latency->Observe(util::current_time(true) - start_time);
|
||||
}
|
||||
}
|
||||
|
||||
OpenResultCallback::OpenResultCallback(IntrusivePtr<detail::BackendHandleVal> backend)
|
||||
: ResultCallback(), backend(std::move(backend)) {}
|
||||
|
||||
|
|
@ -55,6 +100,7 @@ OpenResultCallback::OpenResultCallback(zeek::detail::trigger::TriggerPtr trigger
|
|||
|
||||
void OpenResultCallback::Complete(OperationResult res) {
|
||||
if ( res.code == ReturnCode::SUCCESS ) {
|
||||
backend->backend->InitMetrics();
|
||||
backend->backend->EnqueueBackendOpened();
|
||||
}
|
||||
|
||||
|
|
@ -68,6 +114,12 @@ void OpenResultCallback::Complete(OperationResult res) {
|
|||
Backend::Backend(uint8_t modes, std::string_view tag_name) : modes(modes) {
|
||||
tag = storage_mgr->BackendMgr().GetComponentTag(std::string{tag_name});
|
||||
tag_str = zeek::obj_desc_short(tag.AsVal().get());
|
||||
|
||||
// The rest of the metrics are initialized after the backend opens, but this one has
|
||||
// to be here because it's possible it gets used by the open callback before Open()
|
||||
// fully returns.
|
||||
backends_opened_metric =
|
||||
telemetry_mgr->CounterInstance("zeek", "storage_backends_opened", {}, "Number of backends opened", "");
|
||||
}
|
||||
|
||||
OperationResult Backend::Open(OpenResultCallback* cb, RecordValPtr options, TypePtr kt, TypePtr vt) {
|
||||
|
|
@ -88,12 +140,22 @@ OperationResult Backend::Open(OpenResultCallback* cb, RecordValPtr options, Type
|
|||
if ( ! ret.value )
|
||||
ret.value = cb->Backend();
|
||||
|
||||
if ( ret.code == ReturnCode::SUCCESS )
|
||||
InitMetrics();
|
||||
|
||||
// Complete sync callbacks to make sure the metrics get initialized plus that the
|
||||
// backend_opened event gets posted.
|
||||
if ( cb->IsSyncCallback() )
|
||||
CompleteCallback(cb, ret);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
OperationResult Backend::Close(ResultCallback* cb) { return DoClose(cb); }
|
||||
|
||||
OperationResult Backend::Put(ResultCallback* cb, ValPtr key, ValPtr value, bool overwrite, double expiration_time) {
|
||||
cb->Init(put_metrics.get());
|
||||
|
||||
// The intention for this method is to do some other heavy lifting in regard
|
||||
// to backends that need to pass data through the manager instead of directly
|
||||
// through the workers. For the first versions of the storage framework it
|
||||
|
|
@ -109,10 +171,16 @@ OperationResult Backend::Put(ResultCallback* cb, ValPtr key, ValPtr value, bool
|
|||
return ret;
|
||||
}
|
||||
|
||||
return DoPut(cb, std::move(key), std::move(value), overwrite, expiration_time);
|
||||
auto ret = DoPut(cb, std::move(key), std::move(value), overwrite, expiration_time);
|
||||
if ( cb->IsSyncCallback() )
|
||||
cb->UpdateOperationMetrics(ret.code);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
OperationResult Backend::Get(ResultCallback* cb, ValPtr key) {
|
||||
cb->Init(get_metrics.get());
|
||||
|
||||
// See the note in Put().
|
||||
if ( ! same_type(key->GetType(), key_type) ) {
|
||||
auto ret = OperationResult{ReturnCode::KEY_TYPE_MISMATCH};
|
||||
|
|
@ -120,10 +188,16 @@ OperationResult Backend::Get(ResultCallback* cb, ValPtr key) {
|
|||
return ret;
|
||||
}
|
||||
|
||||
return DoGet(cb, std::move(key));
|
||||
auto ret = DoGet(cb, std::move(key));
|
||||
if ( cb->IsSyncCallback() )
|
||||
cb->UpdateOperationMetrics(ret.code);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
OperationResult Backend::Erase(ResultCallback* cb, ValPtr key) {
|
||||
cb->Init(erase_metrics.get());
|
||||
|
||||
// See the note in Put().
|
||||
if ( ! same_type(key->GetType(), key_type) ) {
|
||||
auto ret = OperationResult{ReturnCode::KEY_TYPE_MISMATCH};
|
||||
|
|
@ -131,7 +205,11 @@ OperationResult Backend::Erase(ResultCallback* cb, ValPtr key) {
|
|||
return ret;
|
||||
}
|
||||
|
||||
return DoErase(cb, std::move(key));
|
||||
auto ret = DoErase(cb, std::move(key));
|
||||
if ( cb->IsSyncCallback() )
|
||||
cb->UpdateOperationMetrics(ret.code);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
void Backend::CompleteCallback(ResultCallback* cb, const OperationResult& data) const {
|
||||
|
|
@ -145,12 +223,57 @@ void Backend::CompleteCallback(ResultCallback* cb, const OperationResult& data)
|
|||
}
|
||||
}
|
||||
|
||||
void Backend::EnqueueBackendOpened() { event_mgr.Enqueue(Storage::backend_opened, tag.AsVal(), backend_options); }
|
||||
void Backend::EnqueueBackendOpened() {
|
||||
event_mgr.Enqueue(Storage::backend_opened, tag.AsVal(), backend_options);
|
||||
backends_opened_metric->Inc();
|
||||
}
|
||||
|
||||
void Backend::EnqueueBackendLost(std::string_view reason) {
|
||||
event_mgr.Enqueue(Storage::backend_lost, tag.AsVal(), backend_options, make_intrusive<StringVal>(reason));
|
||||
}
|
||||
|
||||
void Backend::InitMetrics() {
|
||||
if ( metrics_initialized )
|
||||
return;
|
||||
|
||||
metrics_initialized = true;
|
||||
auto results_family =
|
||||
telemetry_mgr->CounterFamily("zeek", "storage_backend_operation_results",
|
||||
{"operation", "type", "config", "result"}, "Storage operation results");
|
||||
|
||||
auto bounds_val = zeek::id::find_val<zeek::VectorVal>("Storage::latency_metric_bounds");
|
||||
std::vector<double> bounds(bounds_val->Size());
|
||||
for ( unsigned int i = 0; i < bounds_val->Size(); i++ )
|
||||
bounds[i] = bounds_val->DoubleAt(i);
|
||||
|
||||
auto latency_family =
|
||||
telemetry_mgr->HistogramFamily("zeek", "storage_backend_operation_latency", {"operation", "type", "config"},
|
||||
bounds, "Storage Operation Latency", "seconds");
|
||||
|
||||
std::string metrics_config = GetConfigMetricsLabel();
|
||||
put_metrics =
|
||||
std::make_unique<detail::OperationMetrics>(results_family, latency_family, "put", Tag(), metrics_config);
|
||||
get_metrics =
|
||||
std::make_unique<detail::OperationMetrics>(results_family, latency_family, "get", Tag(), metrics_config);
|
||||
erase_metrics =
|
||||
std::make_unique<detail::OperationMetrics>(results_family, latency_family, "erase", Tag(), metrics_config);
|
||||
|
||||
bytes_read_metric = telemetry_mgr->CounterInstance("zeek", "storage_backend_data_written",
|
||||
{{"type", Tag()}, {"config", metrics_config}},
|
||||
"Storage data written to backend", "bytes");
|
||||
bytes_written_metric = telemetry_mgr->CounterInstance("zeek", "storage_backend_data_read",
|
||||
{{"type", Tag()}, {"config", metrics_config}},
|
||||
"Storage data read from backend", "bytes");
|
||||
|
||||
expired_entries_metric = telemetry_mgr->CounterInstance("zeek", "storage_backend_expired_entries",
|
||||
{{"type", Tag()}, {"config", metrics_config}},
|
||||
"Storage expired entries removed by backend", "");
|
||||
}
|
||||
|
||||
void Backend::IncBytesWrittenMetric(size_t written) { bytes_written_metric->Inc(written); }
|
||||
void Backend::IncBytesReadMetric(size_t read) { bytes_read_metric->Inc(read); }
|
||||
void Backend::IncExpiredEntriesMetric(size_t expired) { expired_entries_metric->Inc(expired); }
|
||||
|
||||
zeek::OpaqueTypePtr detail::backend_opaque;
|
||||
IMPLEMENT_OPAQUE_VALUE(detail::BackendHandleVal)
|
||||
|
||||
|
|
|
|||
|
|
@ -2,6 +2,8 @@
|
|||
|
||||
#pragma once
|
||||
|
||||
#include <memory>
|
||||
|
||||
#include "zeek/OpaqueVal.h"
|
||||
#include "zeek/Tag.h"
|
||||
#include "zeek/Val.h"
|
||||
|
|
@ -12,8 +14,33 @@ class Trigger;
|
|||
using TriggerPtr = IntrusivePtr<Trigger>;
|
||||
} // namespace zeek::detail::trigger
|
||||
|
||||
namespace zeek::telemetry {
|
||||
class Counter;
|
||||
using CounterPtr = std::shared_ptr<Counter>;
|
||||
class CounterFamily;
|
||||
using CounterFamilyPtr = std::shared_ptr<CounterFamily>;
|
||||
class Histogram;
|
||||
using HistogramPtr = std::shared_ptr<Histogram>;
|
||||
class HistogramFamily;
|
||||
using HistogramFamilyPtr = std::shared_ptr<HistogramFamily>;
|
||||
} // namespace zeek::telemetry
|
||||
|
||||
namespace zeek::storage {
|
||||
|
||||
namespace detail {
|
||||
struct OperationMetrics {
|
||||
telemetry::CounterPtr success;
|
||||
telemetry::CounterPtr fail;
|
||||
telemetry::CounterPtr error;
|
||||
telemetry::CounterPtr timeouts;
|
||||
telemetry::HistogramPtr latency;
|
||||
|
||||
OperationMetrics(const telemetry::CounterFamilyPtr& results_family,
|
||||
const telemetry::HistogramFamilyPtr& latency_family, std::string_view operation_type,
|
||||
std::string_view backend_type, std::string_view backend_config);
|
||||
};
|
||||
} // namespace detail
|
||||
|
||||
class Manager;
|
||||
|
||||
/**
|
||||
|
|
@ -82,10 +109,37 @@ public:
|
|||
|
||||
OperationResult Result() const { return result; }
|
||||
|
||||
/**
|
||||
* Stores the collection of metrics instruments to update when the operation completes
|
||||
* and sets the start time for an operation to be used to update the latency metric
|
||||
* when the operation completes. This is unset for open/close callbacks.
|
||||
*/
|
||||
void Init(detail::OperationMetrics* m);
|
||||
|
||||
/**
|
||||
* Update the metrics based on a return value.
|
||||
*/
|
||||
void UpdateOperationMetrics(EnumValPtr c);
|
||||
|
||||
/**
|
||||
* Stores the amount of data transferred in the operation. This can be used by async
|
||||
* backends to set the amount transferred in Put operations so it can be added to the
|
||||
* metrics when the operation finishes.
|
||||
*/
|
||||
void AddDataTransferredSize(size_t size) { transferred_size += size; }
|
||||
|
||||
/**
|
||||
* Returns the amount of data transferred in this operation.
|
||||
*/
|
||||
size_t GetDataTransferredSize() const { return transferred_size; }
|
||||
|
||||
protected:
|
||||
zeek::detail::trigger::TriggerPtr trigger;
|
||||
const void* assoc = nullptr;
|
||||
OperationResult result;
|
||||
detail::OperationMetrics* operation_metrics = nullptr;
|
||||
double start_time = 0.0;
|
||||
size_t transferred_size = 0;
|
||||
};
|
||||
|
||||
class OpenResultCallback;
|
||||
|
|
@ -101,7 +155,7 @@ public:
|
|||
/**
|
||||
* Returns a descriptive tag representing the source for debugging.
|
||||
*/
|
||||
const char* Tag() { return tag_str.c_str(); }
|
||||
const char* Tag() const { return tag_str.c_str(); }
|
||||
|
||||
/**
|
||||
* Store a new key/value pair in the backend.
|
||||
|
|
@ -234,6 +288,34 @@ protected:
|
|||
*/
|
||||
void CompleteCallback(ResultCallback* cb, const OperationResult& data) const;
|
||||
|
||||
/**
|
||||
* Returns a string compatible with Prometheus that's used as a tag to differentiate
|
||||
* entries of backend instances.
|
||||
*/
|
||||
std::string GetConfigMetricsLabel() const { return DoGetConfigMetricsLabel(); }
|
||||
|
||||
/**
|
||||
* Utility method to increase the metrics for number of bytes written by a backend.
|
||||
*
|
||||
* @param written The number of bytes written by the last operation.
|
||||
*/
|
||||
void IncBytesWrittenMetric(size_t written);
|
||||
|
||||
/**
|
||||
* Utility method to increase the metrics for number of bytes read by a backend.
|
||||
*
|
||||
* @param read The number of bytes read by the last operation.
|
||||
*/
|
||||
void IncBytesReadMetric(size_t read);
|
||||
|
||||
/**
|
||||
* Utility method to increase the metrics for number of entries expired and removed
|
||||
* from the backend.
|
||||
*
|
||||
* @param expired The number of elements removed by the last operation.
|
||||
*/
|
||||
void IncExpiredEntriesMetric(size_t expired);
|
||||
|
||||
TypePtr key_type;
|
||||
TypePtr val_type;
|
||||
RecordValPtr backend_options;
|
||||
|
|
@ -291,7 +373,30 @@ private:
|
|||
*/
|
||||
virtual void DoExpire(double current_network_time) {}
|
||||
|
||||
/**
|
||||
* Returns a string compatible with Prometheus that's used as a tag to differentiate
|
||||
* entries of backend instances.
|
||||
*/
|
||||
virtual std::string DoGetConfigMetricsLabel() const = 0;
|
||||
|
||||
/**
|
||||
* Initializes the instruments for various storage metrics.
|
||||
*/
|
||||
void InitMetrics();
|
||||
|
||||
uint8_t modes;
|
||||
bool metrics_initialized = false;
|
||||
|
||||
// These are owned by the backend but are passed into the callbacks to be
|
||||
// updated when those complete/timeout.
|
||||
std::unique_ptr<detail::OperationMetrics> put_metrics;
|
||||
std::unique_ptr<detail::OperationMetrics> get_metrics;
|
||||
std::unique_ptr<detail::OperationMetrics> erase_metrics;
|
||||
|
||||
telemetry::CounterPtr bytes_written_metric;
|
||||
telemetry::CounterPtr bytes_read_metric;
|
||||
telemetry::CounterPtr backends_opened_metric;
|
||||
telemetry::CounterPtr expired_entries_metric;
|
||||
};
|
||||
|
||||
using BackendPtr = zeek::IntrusivePtr<Backend>;
|
||||
|
|
|
|||
|
|
@ -11,6 +11,7 @@
|
|||
#include "zeek/Val.h"
|
||||
#include "zeek/iosource/Manager.h"
|
||||
#include "zeek/storage/ReturnCode.h"
|
||||
#include "zeek/telemetry/Counter.h"
|
||||
|
||||
#include "hiredis/adapters/poll.h"
|
||||
#include "hiredis/async.h"
|
||||
|
|
@ -246,6 +247,17 @@ constexpr char REQUIRED_VERSION[] = "6.2.0";
|
|||
|
||||
storage::BackendPtr Redis::Instantiate() { return make_intrusive<Redis>(); }
|
||||
|
||||
std::string Redis::DoGetConfigMetricsLabel() const {
|
||||
static auto running_under_test = id::find_val("running_under_test")->AsBool();
|
||||
std::string tag;
|
||||
if ( running_under_test )
|
||||
tag = util::fmt("server_addr-%s", key_prefix.c_str());
|
||||
else
|
||||
tag = util::fmt("%s-%s", server_addr.c_str(), key_prefix.c_str());
|
||||
|
||||
return tag;
|
||||
}
|
||||
|
||||
/**
|
||||
* Called by the manager system to open the backend.
|
||||
*/
|
||||
|
|
@ -406,6 +418,8 @@ OperationResult Redis::DoPut(ResultCallback* cb, ValPtr key, ValPtr value, bool
|
|||
if ( connected && status == REDIS_ERR )
|
||||
return {ReturnCode::OPERATION_FAILED, util::fmt("Failed to queue put operation: %s", async_ctx->errstr)};
|
||||
|
||||
cb->AddDataTransferredSize(key_data->size() + val_data->size());
|
||||
|
||||
++active_ops;
|
||||
|
||||
// If reading pcaps insert into a secondary set that's ordered by expiration
|
||||
|
|
@ -537,6 +551,8 @@ void Redis::DoExpire(double current_network_time) {
|
|||
// TODO: do we care if this failed?
|
||||
}
|
||||
|
||||
IncExpiredEntriesMetric(elements.size());
|
||||
|
||||
freeReplyObject(reply);
|
||||
|
||||
// Remove all of the elements from the range-set that match the time range.
|
||||
|
|
@ -566,6 +582,8 @@ void Redis::HandlePutResult(redisReply* reply, ResultCallback* callback) {
|
|||
else if ( reply->type == REDIS_REPLY_ERROR )
|
||||
res = ParseReplyError("put", reply->str);
|
||||
|
||||
IncBytesWrittenMetric(callback->GetDataTransferredSize());
|
||||
|
||||
freeReplyObject(reply);
|
||||
CompleteCallback(callback, res);
|
||||
}
|
||||
|
|
@ -583,6 +601,7 @@ void Redis::HandleGetResult(redisReply* reply, ResultCallback* callback) {
|
|||
else if ( reply->type == REDIS_REPLY_ERROR )
|
||||
res = ParseReplyError("get", reply->str);
|
||||
else {
|
||||
IncBytesReadMetric(reply->len);
|
||||
auto val = serializer->Unserialize({(std::byte*)reply->str, reply->len}, val_type);
|
||||
if ( val )
|
||||
res = {ReturnCode::SUCCESS, "", val.value()};
|
||||
|
|
|
|||
|
|
@ -27,7 +27,7 @@ public:
|
|||
*
|
||||
* @return The debugging name.
|
||||
*/
|
||||
const char* Tag() override { return tag_str.c_str(); }
|
||||
const char* Tag() override { return Backend::Tag(); }
|
||||
|
||||
// IOSource interface
|
||||
double GetNextTimeout() override { return -1; }
|
||||
|
|
@ -61,6 +61,7 @@ private:
|
|||
OperationResult DoErase(ResultCallback* cb, ValPtr key) override;
|
||||
void DoExpire(double current_network_time) override;
|
||||
void DoPoll() override;
|
||||
std::string DoGetConfigMetricsLabel() const override;
|
||||
|
||||
OperationResult ParseReplyError(std::string_view op_str, std::string_view reply_err_str) const;
|
||||
OperationResult CheckServerVersion();
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@
|
|||
|
||||
#include "zeek/storage/backend/sqlite/SQLite.h"
|
||||
|
||||
#include <sys/stat.h>
|
||||
#include <filesystem>
|
||||
#include <thread>
|
||||
|
||||
|
|
@ -12,6 +13,7 @@
|
|||
#include "zeek/Func.h"
|
||||
#include "zeek/Val.h"
|
||||
#include "zeek/storage/ReturnCode.h"
|
||||
#include "zeek/telemetry/Manager.h"
|
||||
|
||||
#include "const.bif.netvar_h"
|
||||
|
||||
|
|
@ -19,7 +21,8 @@ using namespace std::chrono_literals;
|
|||
|
||||
namespace zeek::storage::backend::sqlite {
|
||||
|
||||
OperationResult SQLite::RunPragma(std::string_view name, std::optional<std::string_view> value) {
|
||||
OperationResult SQLite::RunPragma(std::string_view name, std::optional<std::string_view> value,
|
||||
StepResultParser value_parser) {
|
||||
char* errorMsg = nullptr;
|
||||
std::chrono::milliseconds time_spent = 0ms;
|
||||
|
||||
|
|
@ -29,12 +32,24 @@ OperationResult SQLite::RunPragma(std::string_view name, std::optional<std::stri
|
|||
|
||||
DBG_LOG(DBG_STORAGE, "Executing '%s' on %s", cmd.c_str(), full_path.c_str());
|
||||
|
||||
sqlite3_stmt* stmt;
|
||||
if ( auto check_res = CheckError(sqlite3_prepare_v2(db, cmd.c_str(), static_cast<int>(cmd.size()), &stmt, nullptr));
|
||||
check_res.code != ReturnCode::SUCCESS )
|
||||
return check_res;
|
||||
|
||||
OperationResult success_res;
|
||||
|
||||
// Make a unique_ptr out of the statement so we don't have to manually call sqlite3_finalize
|
||||
// one it when we return in various places.
|
||||
unique_stmt_ptr stmt_ptr{stmt, sqlite3_finalize};
|
||||
|
||||
while ( pragma_timeout == 0ms || time_spent < pragma_timeout ) {
|
||||
int res = sqlite3_exec(db, cmd.c_str(), nullptr, nullptr, &errorMsg);
|
||||
if ( res == SQLITE_OK ) {
|
||||
auto res = Step(stmt, value_parser, true);
|
||||
if ( res.code == ReturnCode::SUCCESS ) {
|
||||
success_res = res;
|
||||
break;
|
||||
}
|
||||
else if ( res == SQLITE_BUSY ) {
|
||||
else if ( res.code == ReturnCode::TIMEOUT ) {
|
||||
// If we got back that the database is busy, it likely means that another process is trying to
|
||||
// do their pragmas at startup too. Exponentially back off and try again after a sleep.
|
||||
sqlite3_free(errorMsg);
|
||||
|
|
@ -42,8 +57,8 @@ OperationResult SQLite::RunPragma(std::string_view name, std::optional<std::stri
|
|||
time_spent += pragma_wait_on_busy;
|
||||
}
|
||||
else {
|
||||
std::string err = util::fmt("Error while executing '%s': %s (%d)", cmd.c_str(), errorMsg, res);
|
||||
sqlite3_free(errorMsg);
|
||||
std::string err =
|
||||
util::fmt("Error while executing '%s': %s (%d)", cmd.c_str(), sqlite3_errmsg(db), sqlite3_errcode(db));
|
||||
DBG_LOG(DBG_STORAGE, "%s", err.c_str());
|
||||
return {ReturnCode::INITIALIZATION_FAILED, std::move(err)};
|
||||
}
|
||||
|
|
@ -57,11 +72,16 @@ OperationResult SQLite::RunPragma(std::string_view name, std::optional<std::stri
|
|||
|
||||
DBG_LOG(DBG_STORAGE, "'%s' successful", cmd.c_str());
|
||||
|
||||
return {ReturnCode::SUCCESS};
|
||||
return success_res;
|
||||
}
|
||||
|
||||
storage::BackendPtr SQLite::Instantiate() { return make_intrusive<SQLite>(); }
|
||||
|
||||
std::string SQLite::DoGetConfigMetricsLabel() const {
|
||||
std::string tag = util::fmt("%s-%s", full_path.c_str(), table_name.c_str());
|
||||
return tag;
|
||||
}
|
||||
|
||||
/**
|
||||
* Called by the manager system to open the backend.
|
||||
*/
|
||||
|
|
@ -241,6 +261,33 @@ OperationResult SQLite::DoOpen(OpenResultCallback* cb, RecordValPtr options) {
|
|||
get_expiry_last_run_stmt = std::move(stmt_ptrs[6]);
|
||||
update_expiry_last_run_stmt = std::move(stmt_ptrs[7]);
|
||||
|
||||
page_count_metric =
|
||||
telemetry_mgr->GaugeInstance("zeek", "storage_sqlite_database_size", {{"config", GetConfigMetricsLabel()}},
|
||||
"Storage sqlite backend value of page_count pragma", "pages", [this]() {
|
||||
static auto double_parser = [](sqlite3_stmt* stmt) -> OperationResult {
|
||||
double val = sqlite3_column_double(stmt, 0);
|
||||
return {ReturnCode::SUCCESS, "", make_intrusive<DoubleVal>(val)};
|
||||
};
|
||||
|
||||
auto res = RunPragma("page_count", std::nullopt, double_parser);
|
||||
if ( res.code == ReturnCode::SUCCESS ) {
|
||||
last_page_count_value = cast_intrusive<DoubleVal>(res.value)->Get();
|
||||
}
|
||||
|
||||
return last_page_count_value;
|
||||
});
|
||||
|
||||
file_size_metric =
|
||||
telemetry_mgr->GaugeInstance("zeek", "storage_sqlite_database_size", {{"config", GetConfigMetricsLabel()}},
|
||||
"Storage sqlite backend database file size on disk", "bytes", [this]() {
|
||||
struct stat s{0};
|
||||
if ( int ret = stat(full_path.c_str(), &s); ret == 0 ) {
|
||||
last_file_size_value = static_cast<double>(s.st_size);
|
||||
}
|
||||
|
||||
return last_file_size_value;
|
||||
});
|
||||
|
||||
return {ReturnCode::SUCCESS};
|
||||
}
|
||||
|
||||
|
|
@ -287,6 +334,12 @@ OperationResult SQLite::DoClose(ResultCallback* cb) {
|
|||
expire_db = nullptr;
|
||||
}
|
||||
|
||||
if ( page_count_metric )
|
||||
page_count_metric->RemoveCallback();
|
||||
|
||||
if ( file_size_metric )
|
||||
file_size_metric->RemoveCallback();
|
||||
|
||||
return op_res;
|
||||
}
|
||||
|
||||
|
|
@ -341,7 +394,7 @@ OperationResult SQLite::DoPut(ResultCallback* cb, ValPtr key, ValPtr value, bool
|
|||
return res;
|
||||
}
|
||||
|
||||
auto step_result = Step(stmt.get(), false);
|
||||
auto step_result = Step(stmt.get(), nullptr);
|
||||
if ( ! overwrite )
|
||||
if ( step_result.code == ReturnCode::SUCCESS ) {
|
||||
int changed = sqlite3_changes(db);
|
||||
|
|
@ -349,6 +402,8 @@ OperationResult SQLite::DoPut(ResultCallback* cb, ValPtr key, ValPtr value, bool
|
|||
step_result.code = ReturnCode::KEY_EXISTS;
|
||||
}
|
||||
|
||||
IncBytesWrittenMetric(val_data->size());
|
||||
|
||||
return step_result;
|
||||
}
|
||||
|
||||
|
|
@ -375,7 +430,20 @@ OperationResult SQLite::DoGet(ResultCallback* cb, ValPtr key) {
|
|||
return res;
|
||||
}
|
||||
|
||||
return Step(stmt.get(), true);
|
||||
auto value_parser = [this](sqlite3_stmt* stmt) -> OperationResult {
|
||||
auto blob = static_cast<const std::byte*>(sqlite3_column_blob(stmt, 0));
|
||||
size_t blob_size = sqlite3_column_bytes(stmt, 0);
|
||||
IncBytesReadMetric(blob_size);
|
||||
|
||||
auto val = serializer->Unserialize({blob, blob_size}, val_type);
|
||||
|
||||
if ( val )
|
||||
return {ReturnCode::SUCCESS, "", val.value()};
|
||||
|
||||
return {ReturnCode::OPERATION_FAILED, val.error()};
|
||||
};
|
||||
|
||||
return Step(stmt.get(), value_parser);
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -396,7 +464,7 @@ OperationResult SQLite::DoErase(ResultCallback* cb, ValPtr key) {
|
|||
return res;
|
||||
}
|
||||
|
||||
return Step(stmt.get(), false);
|
||||
return Step(stmt.get(), nullptr);
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -505,6 +573,11 @@ void SQLite::DoExpire(double current_network_time) {
|
|||
Error(err.c_str());
|
||||
}
|
||||
|
||||
// Get the number of changes from the delete statement. This should be identical to the num_to_expire
|
||||
// value earlier because we're under a transaction, but this should be the exact number that changed.
|
||||
int changes = sqlite3_changes(db);
|
||||
IncExpiredEntriesMetric(changes);
|
||||
|
||||
sqlite3_exec(expire_db, "commit transaction", nullptr, nullptr, &errMsg);
|
||||
sqlite3_free(errMsg);
|
||||
}
|
||||
|
|
@ -518,28 +591,22 @@ OperationResult SQLite::CheckError(int code) {
|
|||
return {ReturnCode::SUCCESS};
|
||||
}
|
||||
|
||||
OperationResult SQLite::Step(sqlite3_stmt* stmt, bool parse_value) {
|
||||
OperationResult SQLite::Step(sqlite3_stmt* stmt, StepResultParser parser, bool is_pragma) {
|
||||
OperationResult ret;
|
||||
|
||||
int step_status = sqlite3_step(stmt);
|
||||
if ( step_status == SQLITE_ROW ) {
|
||||
if ( parse_value ) {
|
||||
auto blob = static_cast<const std::byte*>(sqlite3_column_blob(stmt, 0));
|
||||
size_t blob_size = sqlite3_column_bytes(stmt, 0);
|
||||
|
||||
auto val = serializer->Unserialize({blob, blob_size}, val_type);
|
||||
|
||||
if ( val )
|
||||
ret = {ReturnCode::SUCCESS, "", val.value()};
|
||||
else
|
||||
ret = {ReturnCode::OPERATION_FAILED, val.error()};
|
||||
}
|
||||
else {
|
||||
if ( parser )
|
||||
ret = parser(stmt);
|
||||
else if ( ! is_pragma )
|
||||
ret = {ReturnCode::OPERATION_FAILED, "sqlite3_step should not have returned a value"};
|
||||
}
|
||||
else
|
||||
// For most pragmas we don't care about the output, so we don't pass a parser. We still
|
||||
// may get a row as a response, but we can discard it and just return SUCCESS.
|
||||
ret = {ReturnCode::SUCCESS};
|
||||
}
|
||||
else if ( step_status == SQLITE_DONE ) {
|
||||
if ( parse_value )
|
||||
if ( parser )
|
||||
ret = {ReturnCode::KEY_NOT_FOUND};
|
||||
else
|
||||
ret = {ReturnCode::SUCCESS};
|
||||
|
|
|
|||
|
|
@ -25,6 +25,8 @@ public:
|
|||
bool IsOpen() override { return db != nullptr; }
|
||||
|
||||
private:
|
||||
using StepResultParser = std::function<OperationResult(sqlite3_stmt*)>;
|
||||
|
||||
OperationResult DoOpen(OpenResultCallback* cb, RecordValPtr options) override;
|
||||
OperationResult DoClose(ResultCallback* cb) override;
|
||||
OperationResult DoPut(ResultCallback* cb, ValPtr key, ValPtr value, bool overwrite,
|
||||
|
|
@ -32,6 +34,7 @@ private:
|
|||
OperationResult DoGet(ResultCallback* cb, ValPtr key) override;
|
||||
OperationResult DoErase(ResultCallback* cb, ValPtr key) override;
|
||||
void DoExpire(double current_network_time) override;
|
||||
std::string DoGetConfigMetricsLabel() const override;
|
||||
|
||||
/**
|
||||
* Checks whether a status code returned by an sqlite call is a success.
|
||||
|
|
@ -45,12 +48,13 @@ private:
|
|||
* Abstracts calls to sqlite3_step to properly create an OperationResult
|
||||
* structure based on the result.
|
||||
*/
|
||||
OperationResult Step(sqlite3_stmt* stmt, bool parse_value = false);
|
||||
OperationResult Step(sqlite3_stmt* stmt, StepResultParser parser, bool is_pragma = false);
|
||||
|
||||
/**
|
||||
* Helper utility for running pragmas on the database.
|
||||
*/
|
||||
OperationResult RunPragma(std::string_view name, std::optional<std::string_view> value = std::nullopt);
|
||||
OperationResult RunPragma(std::string_view name, std::optional<std::string_view> value = std::nullopt,
|
||||
StepResultParser value_parser = nullptr);
|
||||
|
||||
sqlite3* db = nullptr;
|
||||
sqlite3* expire_db = nullptr;
|
||||
|
|
@ -72,6 +76,12 @@ private:
|
|||
std::string table_name;
|
||||
std::chrono::milliseconds pragma_timeout;
|
||||
std::chrono::milliseconds pragma_wait_on_busy;
|
||||
|
||||
telemetry::GaugePtr page_count_metric;
|
||||
telemetry::GaugePtr file_size_metric;
|
||||
|
||||
double last_page_count_value = 0.0;
|
||||
double last_file_size_value = 0.0;
|
||||
};
|
||||
|
||||
} // namespace zeek::storage::backend::sqlite
|
||||
|
|
|
|||
|
|
@ -62,6 +62,7 @@ public:
|
|||
|
||||
bool HasCallback() const noexcept { return callback != nullptr; }
|
||||
double RunCallback() const { return callback(); }
|
||||
void RemoveCallback() { callback = nullptr; }
|
||||
|
||||
private:
|
||||
friend class CounterFamily;
|
||||
|
|
|
|||
|
|
@ -84,6 +84,7 @@ public:
|
|||
|
||||
bool HasCallback() const noexcept { return callback != nullptr; }
|
||||
double RunCallback() const { return callback(); }
|
||||
void RemoveCallback() { callback = nullptr; }
|
||||
|
||||
private:
|
||||
FamilyType* family = nullptr;
|
||||
|
|
|
|||
|
|
@ -206,29 +206,6 @@ RecordValPtr Manager::GetMetricOptsRecord(const prometheus::MetricFamily& metric
|
|||
return record_val;
|
||||
}
|
||||
|
||||
static bool compare_string_vectors(const VectorValPtr& a, const VectorValPtr& b) {
|
||||
if ( a->Size() < b->Size() )
|
||||
return true;
|
||||
if ( a->Size() > b->Size() )
|
||||
return false;
|
||||
|
||||
auto a_v = a->RawVec();
|
||||
auto b_v = b->RawVec();
|
||||
|
||||
auto b_it = b_v.begin();
|
||||
for ( auto a_it = a_v.begin(); a_it != a_v.end(); ++a_it, ++b_it ) {
|
||||
if ( ! a_it->has_value() )
|
||||
return false;
|
||||
if ( ! b_it->has_value() )
|
||||
return true;
|
||||
|
||||
if ( (*a_it)->AsString()->ToStdStringView() < (*b_it)->AsString()->ToStdStringView() )
|
||||
return true;
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
static bool comparer(const std::optional<ZVal>& a, const std::optional<ZVal>& b, const RecordTypePtr& type) {
|
||||
if ( ! a )
|
||||
return false;
|
||||
|
|
@ -239,9 +216,40 @@ static bool comparer(const std::optional<ZVal>& a, const std::optional<ZVal>& b,
|
|||
auto a_r = a->ToVal(type)->AsRecordVal();
|
||||
auto b_r = b->ToVal(type)->AsRecordVal();
|
||||
|
||||
auto a_opts = a_r->GetField<RecordVal>("opts");
|
||||
auto b_opts = b_r->GetField<RecordVal>("opts");
|
||||
|
||||
auto a_name = a_opts->GetField<StringVal>("name");
|
||||
auto b_name = b_opts->GetField<StringVal>("name");
|
||||
if ( a_name->Len() > b_name->Len() )
|
||||
return false;
|
||||
if ( a_name->Len() < b_name->Len() )
|
||||
return true;
|
||||
if ( memcmp(a_name->Bytes(), b_name->Bytes(), a_name->Len()) < 0 )
|
||||
return true;
|
||||
|
||||
auto a_prefix = a_opts->GetField<StringVal>("prefix");
|
||||
auto b_prefix = b_opts->GetField<StringVal>("prefix");
|
||||
if ( a_prefix->Len() > b_prefix->Len() )
|
||||
return false;
|
||||
if ( a_prefix->Len() < b_prefix->Len() )
|
||||
return true;
|
||||
if ( memcmp(a_prefix->Bytes(), b_prefix->Bytes(), a_prefix->Len()) < 0 )
|
||||
return true;
|
||||
|
||||
auto a_labels = a_r->GetField<VectorVal>("label_values");
|
||||
std::vector<std::string> a_label_vec;
|
||||
a_label_vec.reserve(a_labels->Size());
|
||||
for ( const auto& sv : a_labels->RawVec() )
|
||||
a_label_vec.push_back(sv->AsString()->ToStdString());
|
||||
|
||||
auto b_labels = b_r->GetField<VectorVal>("label_values");
|
||||
return compare_string_vectors(a_labels, b_labels);
|
||||
std::vector<std::string> b_label_vec;
|
||||
b_label_vec.reserve(b_labels->Size());
|
||||
for ( const auto& sv : b_labels->RawVec() )
|
||||
b_label_vec.push_back(sv->AsString()->ToStdString());
|
||||
|
||||
return a_label_vec < b_label_vec;
|
||||
}
|
||||
|
||||
static bool compare_metrics(const std::optional<ZVal>& a, const std::optional<ZVal>& b) {
|
||||
|
|
@ -308,6 +316,8 @@ ValPtr Manager::CollectMetrics(std::string_view prefix_pattern, std::string_view
|
|||
full_pattern.append("*");
|
||||
|
||||
auto collected = prometheus_registry->Collect();
|
||||
ret_val->Reserve(collected.size());
|
||||
|
||||
for ( const auto& fam : collected ) {
|
||||
if ( fam.type == prometheus::MetricType::Histogram )
|
||||
continue;
|
||||
|
|
@ -350,7 +360,7 @@ ValPtr Manager::CollectMetrics(std::string_view prefix_pattern, std::string_view
|
|||
static auto running_under_test = id::find_val("running_under_test")->AsBool();
|
||||
if ( running_under_test ) {
|
||||
auto& vec = ret_val->RawVec();
|
||||
std::sort(vec.begin(), vec.end(), compare_histograms);
|
||||
std::sort(vec.begin(), vec.end(), compare_metrics);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -388,6 +398,8 @@ ValPtr Manager::CollectHistogramMetrics(std::string_view prefix_pattern, std::st
|
|||
full_pattern.append("*");
|
||||
|
||||
auto collected = prometheus_registry->Collect();
|
||||
ret_val->Reserve(collected.size());
|
||||
|
||||
for ( const auto& fam : collected ) {
|
||||
if ( fam.type != prometheus::MetricType::Histogram )
|
||||
continue;
|
||||
|
|
|
|||
|
|
@ -3,4 +3,23 @@ open result, [code=Storage::SUCCESS, error_str=<uninitialized>, value=<opaque of
|
|||
put result, [code=Storage::SUCCESS, error_str=<uninitialized>, value=<uninitialized>]
|
||||
get result, [code=Storage::SUCCESS, error_str=<uninitialized>, value=value5678]
|
||||
get result same as inserted, T
|
||||
|
||||
Post-operation metrics:
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backends_opened_total, [], [], 1.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_data_read_bytes_total, [config, type], [server_addr-testing, Storage::STORAGE_BACKEND_REDIS], 34.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_expired_entries_total, [config, type], [server_addr-testing, Storage::STORAGE_BACKEND_REDIS], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [server_addr-testing, erase, error, Storage::STORAGE_BACKEND_REDIS], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [server_addr-testing, erase, fail, Storage::STORAGE_BACKEND_REDIS], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [server_addr-testing, erase, success, Storage::STORAGE_BACKEND_REDIS], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [server_addr-testing, erase, timeout, Storage::STORAGE_BACKEND_REDIS], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [server_addr-testing, get, error, Storage::STORAGE_BACKEND_REDIS], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [server_addr-testing, get, fail, Storage::STORAGE_BACKEND_REDIS], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [server_addr-testing, get, success, Storage::STORAGE_BACKEND_REDIS], 1.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [server_addr-testing, get, timeout, Storage::STORAGE_BACKEND_REDIS], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [server_addr-testing, put, error, Storage::STORAGE_BACKEND_REDIS], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [server_addr-testing, put, fail, Storage::STORAGE_BACKEND_REDIS], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [server_addr-testing, put, success, Storage::STORAGE_BACKEND_REDIS], 1.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [server_addr-testing, put, timeout, Storage::STORAGE_BACKEND_REDIS], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_data_written_bytes_total, [config, type], [server_addr-testing, Storage::STORAGE_BACKEND_REDIS], 18.0
|
||||
|
||||
close result, [code=Storage::SUCCESS, error_str=<uninitialized>, value=<uninitialized>]
|
||||
|
|
|
|||
|
|
@ -9,5 +9,25 @@ get result same as originally inserted, T
|
|||
put result, [code=Storage::SUCCESS, error_str=<uninitialized>, value=<uninitialized>]
|
||||
get result, [code=Storage::SUCCESS, error_str=<uninitialized>, value=value2345]
|
||||
get result same as overwritten, T
|
||||
Storage::backend_opened, Storage::STORAGE_BACKEND_REDIS, [serializer=Storage::STORAGE_SERIALIZER_JSON, redis=[server_host=127.0.0.1, server_port=xxxx/tcp, server_unix_socket=<uninitialized>, key_prefix=testing, connect_timeout=5.0 secs, operation_timeout=5.0 secs, username=<uninitialized>, password=<uninitialized>]]
|
||||
Storage::backend_lost, Storage::STORAGE_BACKEND_REDIS, [serializer=Storage::STORAGE_SERIALIZER_JSON, redis=[server_host=127.0.0.1, server_port=xxxx/tcp, server_unix_socket=<uninitialized>, key_prefix=testing, connect_timeout=5.0 secs, operation_timeout=5.0 secs, username=<uninitialized>, password=<uninitialized>]], Client disconnected
|
||||
get result, [code=Storage::KEY_NOT_FOUND, error_str=<uninitialized>, value=<uninitialized>]
|
||||
Storage::backend_opened, Storage::STORAGE_BACKEND_REDIS, [serializer=Storage::STORAGE_SERIALIZER_JSON, redis=[server_host=127.0.0.1, server_port=XXXX/tcp, server_unix_socket=<uninitialized>, key_prefix=testing, connect_timeout=5.0 secs, operation_timeout=5.0 secs, username=<uninitialized>, password=<uninitialized>]]
|
||||
|
||||
Post-operation metrics:
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backends_opened_total, [], [], 1.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_data_read_bytes_total, [config, type], [server_addr-testing, Storage::STORAGE_BACKEND_REDIS], 102.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_expired_entries_total, [config, type], [server_addr-testing, Storage::STORAGE_BACKEND_REDIS], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [server_addr-testing, erase, error, Storage::STORAGE_BACKEND_REDIS], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [server_addr-testing, erase, fail, Storage::STORAGE_BACKEND_REDIS], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [server_addr-testing, erase, success, Storage::STORAGE_BACKEND_REDIS], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [server_addr-testing, erase, timeout, Storage::STORAGE_BACKEND_REDIS], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [server_addr-testing, get, error, Storage::STORAGE_BACKEND_REDIS], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [server_addr-testing, get, fail, Storage::STORAGE_BACKEND_REDIS], 1.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [server_addr-testing, get, success, Storage::STORAGE_BACKEND_REDIS], 3.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [server_addr-testing, get, timeout, Storage::STORAGE_BACKEND_REDIS], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [server_addr-testing, put, error, Storage::STORAGE_BACKEND_REDIS], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [server_addr-testing, put, fail, Storage::STORAGE_BACKEND_REDIS], 1.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [server_addr-testing, put, success, Storage::STORAGE_BACKEND_REDIS], 2.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [server_addr-testing, put, timeout, Storage::STORAGE_BACKEND_REDIS], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_data_written_bytes_total, [config, type], [server_addr-testing, Storage::STORAGE_BACKEND_REDIS], 54.0
|
||||
|
||||
Storage::backend_lost, Storage::STORAGE_BACKEND_REDIS, [serializer=Storage::STORAGE_SERIALIZER_JSON, redis=[server_host=127.0.0.1, server_port=XXXX/tcp, server_unix_socket=<uninitialized>, key_prefix=testing, connect_timeout=5.0 secs, operation_timeout=5.0 secs, username=<uninitialized>, password=<uninitialized>]], Client disconnected
|
||||
|
|
|
|||
|
|
@ -4,3 +4,9 @@ put result, [code=Storage::SUCCESS, error_str=<uninitialized>, value=<uninitiali
|
|||
get result, [code=Storage::SUCCESS, error_str=<uninitialized>, value=value5678]
|
||||
get result same as inserted, T
|
||||
closed succesfully
|
||||
Storage::backend_opened, Storage::STORAGE_BACKEND_SQLITE, [serializer=Storage::STORAGE_SERIALIZER_JSON, sqlite=[database_path=test.sqlite, table_name=testing, busy_timeout=5.0 secs, pragma_commands={
|
||||
[integrity_check] = ,
|
||||
[journal_mode] = WAL,
|
||||
[synchronous] = normal,
|
||||
[temp_store] = memory
|
||||
}, pragma_timeout=500.0 msecs, pragma_wait_on_busy=5.0 msecs]]
|
||||
|
|
|
|||
|
|
@ -9,4 +9,25 @@ open result, [code=Storage::SUCCESS, error_str=<uninitialized>, value=<opaque of
|
|||
put result, [code=Storage::SUCCESS, error_str=<uninitialized>, value=<uninitialized>]
|
||||
get result, [code=Storage::SUCCESS, error_str=<uninitialized>, value=value5678]
|
||||
get result same as inserted, T
|
||||
|
||||
Post-operation metrics:
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backends_opened_total, [], [], 1.0
|
||||
Telemetry::GAUGE, zeek, zeek_storage_sqlite_database_size_bytes, [config], [test.sqlite-testing], 4096.0
|
||||
Telemetry::GAUGE, zeek, zeek_storage_sqlite_database_size_pages, [config], [test.sqlite-testing], 5.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_data_read_bytes_total, [config, type], [test.sqlite-testing, Storage::STORAGE_BACKEND_SQLITE], 18.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_expired_entries_total, [config, type], [test.sqlite-testing, Storage::STORAGE_BACKEND_SQLITE], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [test.sqlite-testing, erase, error, Storage::STORAGE_BACKEND_SQLITE], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [test.sqlite-testing, erase, fail, Storage::STORAGE_BACKEND_SQLITE], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [test.sqlite-testing, erase, success, Storage::STORAGE_BACKEND_SQLITE], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [test.sqlite-testing, erase, timeout, Storage::STORAGE_BACKEND_SQLITE], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [test.sqlite-testing, get, error, Storage::STORAGE_BACKEND_SQLITE], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [test.sqlite-testing, get, fail, Storage::STORAGE_BACKEND_SQLITE], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [test.sqlite-testing, get, success, Storage::STORAGE_BACKEND_SQLITE], 1.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [test.sqlite-testing, get, timeout, Storage::STORAGE_BACKEND_SQLITE], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [test.sqlite-testing, put, error, Storage::STORAGE_BACKEND_SQLITE], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [test.sqlite-testing, put, fail, Storage::STORAGE_BACKEND_SQLITE], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [test.sqlite-testing, put, success, Storage::STORAGE_BACKEND_SQLITE], 1.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [test.sqlite-testing, put, timeout, Storage::STORAGE_BACKEND_SQLITE], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_data_written_bytes_total, [config, type], [test.sqlite-testing, Storage::STORAGE_BACKEND_SQLITE], 18.0
|
||||
|
||||
closed succesfully
|
||||
|
|
|
|||
|
|
@ -2,3 +2,9 @@
|
|||
open result, [code=Storage::SUCCESS, error_str=<uninitialized>, value=<opaque of BackendHandleVal>]
|
||||
erase result, [code=Storage::SUCCESS, error_str=<uninitialized>, value=<uninitialized>]
|
||||
get result, [code=Storage::KEY_NOT_FOUND, error_str=<uninitialized>, value=<uninitialized>]
|
||||
Storage::backend_opened, Storage::STORAGE_BACKEND_SQLITE, [serializer=Storage::STORAGE_SERIALIZER_JSON, sqlite=[database_path=storage-test.sqlite, table_name=testing, busy_timeout=5.0 secs, pragma_commands={
|
||||
[integrity_check] = ,
|
||||
[journal_mode] = WAL,
|
||||
[synchronous] = normal,
|
||||
[temp_store] = memory
|
||||
}, pragma_timeout=500.0 msecs, pragma_wait_on_busy=5.0 msecs]]
|
||||
|
|
|
|||
|
|
@ -0,0 +1 @@
|
|||
### BTest baseline data generated by btest-diff. Do not edit. Use "btest -U/-u" to update. Requires BTest >= 0.63.
|
||||
|
|
@ -0,0 +1,26 @@
|
|||
### BTest baseline data generated by btest-diff. Do not edit. Use "btest -U/-u" to update. Requires BTest >= 0.63.
|
||||
open result, [code=Storage::SUCCESS, error_str=<uninitialized>, value=<opaque of BackendHandleVal>]
|
||||
put result, [code=Storage::SUCCESS, error_str=<uninitialized>, value=<uninitialized>]
|
||||
get result 1, [code=Storage::SUCCESS, error_str=<uninitialized>, value=value5678]
|
||||
erase result, [code=Storage::SUCCESS, error_str=<uninitialized>, value=<uninitialized>]
|
||||
get result 2, [code=Storage::KEY_NOT_FOUND, error_str=<uninitialized>, value=<uninitialized>]
|
||||
|
||||
Post-operation metrics:
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backends_opened_total, [], [], 1.0
|
||||
Telemetry::GAUGE, zeek, zeek_storage_sqlite_database_size_bytes, [config], [test.sqlite-testing], 4096.0
|
||||
Telemetry::GAUGE, zeek, zeek_storage_sqlite_database_size_pages, [config], [test.sqlite-testing], 5.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_data_read_bytes_total, [config, type], [test.sqlite-testing, Storage::STORAGE_BACKEND_SQLITE], 18.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_expired_entries_total, [config, type], [test.sqlite-testing, Storage::STORAGE_BACKEND_SQLITE], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [test.sqlite-testing, erase, error, Storage::STORAGE_BACKEND_SQLITE], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [test.sqlite-testing, erase, fail, Storage::STORAGE_BACKEND_SQLITE], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [test.sqlite-testing, erase, success, Storage::STORAGE_BACKEND_SQLITE], 1.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [test.sqlite-testing, erase, timeout, Storage::STORAGE_BACKEND_SQLITE], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [test.sqlite-testing, get, error, Storage::STORAGE_BACKEND_SQLITE], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [test.sqlite-testing, get, fail, Storage::STORAGE_BACKEND_SQLITE], 1.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [test.sqlite-testing, get, success, Storage::STORAGE_BACKEND_SQLITE], 1.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [test.sqlite-testing, get, timeout, Storage::STORAGE_BACKEND_SQLITE], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [test.sqlite-testing, put, error, Storage::STORAGE_BACKEND_SQLITE], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [test.sqlite-testing, put, fail, Storage::STORAGE_BACKEND_SQLITE], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [test.sqlite-testing, put, success, Storage::STORAGE_BACKEND_SQLITE], 1.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_operation_results_total, [config, operation, result, type], [test.sqlite-testing, put, timeout, Storage::STORAGE_BACKEND_SQLITE], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_storage_backend_data_written_bytes_total, [config, type], [test.sqlite-testing, Storage::STORAGE_BACKEND_SQLITE], 18.0
|
||||
|
|
@ -1,8 +1,8 @@
|
|||
### BTest baseline data generated by btest-diff. Do not edit. Use "btest -U/-u" to update. Requires BTest >= 0.63.
|
||||
### zeek_session_metrics |3|
|
||||
Telemetry::GAUGE, zeek, zeek_active_sessions, [protocol], [tcp], 500.0
|
||||
Telemetry::COUNTER, zeek, zeek_ended_sessions_total, [reason], [inactivity], 0.0
|
||||
Telemetry::COUNTER, zeek, zeek_total_sessions_total, [protocol], [tcp], 500.0
|
||||
Telemetry::GAUGE, zeek, zeek_active_sessions, [protocol], [tcp], 500.0
|
||||
### bt* metrics |5|
|
||||
Telemetry::COUNTER, btest, btest_a_test_total, [x, y], [a, b], 1.0
|
||||
Telemetry::COUNTER, btest, btest_b_test_total, [x, y], [a, b], 10.0
|
||||
|
|
|
|||
|
|
@ -1,17 +1,17 @@
|
|||
### BTest baseline data generated by btest-diff. Do not edit. Use "btest -U/-u" to update. Requires BTest >= 0.63.
|
||||
### broker |7|
|
||||
Telemetry::GAUGE, broker, broker_connections, [type], [native], 0.0
|
||||
value, 0.0
|
||||
Telemetry::GAUGE, broker, broker_connections, [type], [web-socket], 0.0
|
||||
value, 0.0
|
||||
Telemetry::COUNTER, broker, broker_processed_messages_total, [type], [command], 0.0
|
||||
value, 0.0
|
||||
Telemetry::COUNTER, broker, broker_processed_messages_total, [type], [data], 0.0
|
||||
value, 0.0
|
||||
Telemetry::GAUGE, broker, broker_connections, [type], [native], 0.0
|
||||
value, 0.0
|
||||
Telemetry::COUNTER, broker, broker_processed_messages_total, [type], [ping], 0.0
|
||||
value, 0.0
|
||||
Telemetry::COUNTER, broker, broker_processed_messages_total, [type], [pong], 0.0
|
||||
value, 0.0
|
||||
Telemetry::COUNTER, broker, broker_processed_messages_total, [type], [routing-update], 0.0
|
||||
value, 0.0
|
||||
Telemetry::GAUGE, broker, broker_connections, [type], [web-socket], 0.0
|
||||
value, 0.0
|
||||
### broker |0|
|
||||
|
|
|
|||
|
|
@ -87,4 +87,6 @@ OperationResult StorageDummy::DoErase(ResultCallback* cb, ValPtr key) {
|
|||
return {ReturnCode::KEY_NOT_FOUND};
|
||||
}
|
||||
|
||||
std::string StorageDummy::DoGetConfigMetricsLabel() const { return "storage-dummy"; }
|
||||
|
||||
} // namespace btest::storage::backend
|
||||
|
|
|
|||
|
|
@ -49,6 +49,8 @@ public:
|
|||
*/
|
||||
zeek::storage::OperationResult DoErase(zeek::storage::ResultCallback* cb, zeek::ValPtr key) override;
|
||||
|
||||
std::string DoGetConfigMetricsLabel() const override;
|
||||
|
||||
private:
|
||||
std::map<zeek::byte_buffer, zeek::byte_buffer> data;
|
||||
bool open = false;
|
||||
|
|
|
|||
|
|
@ -4,16 +4,43 @@
|
|||
# @TEST-PORT: REDIS_PORT
|
||||
|
||||
# @TEST-EXEC: btest-bg-run redis-server run-redis-server ${REDIS_PORT%/tcp}
|
||||
# @TEST-EXEC: zeek -b %INPUT > out
|
||||
# @TEST-EXEC: zeek -b %INPUT | sed "s|-${REDIS_PORT%/tcp}-|-XXXX-|g" > out
|
||||
# @TEST-EXEC: btest-bg-wait -k 0
|
||||
|
||||
# @TEST-EXEC: btest-diff out
|
||||
|
||||
@load base/frameworks/storage/async
|
||||
@load base/frameworks/storage/sync
|
||||
@load policy/frameworks/storage/backend/redis
|
||||
@load base/frameworks/telemetry
|
||||
|
||||
# Make sure the telemetry output is in a fixed order.
|
||||
redef running_under_test = T;
|
||||
|
||||
redef exit_only_after_terminate = T;
|
||||
global b : opaque of Storage::BackendHandle;
|
||||
|
||||
event print_metrics_and_close()
|
||||
{
|
||||
print "";
|
||||
print "Post-operation metrics:";
|
||||
local storage_metrics = Telemetry::collect_metrics("zeek", "storage*");
|
||||
for (_, m in storage_metrics)
|
||||
{
|
||||
print m$opts$metric_type, m$opts$prefix, m$opts$name, m$label_names, m$label_values, m$value;
|
||||
}
|
||||
print "";
|
||||
|
||||
when [] ( local close_res = Storage::Async::close_backend(b) )
|
||||
{
|
||||
print "close result", close_res;
|
||||
terminate();
|
||||
}
|
||||
timeout 5sec
|
||||
{
|
||||
print "close result", close_res;
|
||||
terminate();
|
||||
}
|
||||
}
|
||||
|
||||
event zeek_init()
|
||||
{
|
||||
|
|
@ -28,29 +55,20 @@ event zeek_init()
|
|||
Storage::STORAGE_BACKEND_REDIS, opts, string, string) )
|
||||
{
|
||||
print "open result", open_res;
|
||||
local b = open_res$value;
|
||||
b = open_res$value;
|
||||
|
||||
when [b, key, value] ( local put_res = Storage::Async::put(b, [ $key=key,
|
||||
when [key, value] ( local put_res = Storage::Async::put(b, [ $key=key,
|
||||
$value=value ]) )
|
||||
{
|
||||
print "put result", put_res;
|
||||
|
||||
when [b, key, value] ( local get_res = Storage::Async::get(b, key) )
|
||||
when [key, value] ( local get_res = Storage::Async::get(b, key) )
|
||||
{
|
||||
print "get result", get_res;
|
||||
if ( get_res$code == Storage::SUCCESS && get_res?$value )
|
||||
print "get result same as inserted", value == ( get_res$value as string );
|
||||
|
||||
when [b] ( local close_res = Storage::Async::close_backend(b) )
|
||||
{
|
||||
print "close result", close_res;
|
||||
terminate();
|
||||
}
|
||||
timeout 5sec
|
||||
{
|
||||
print "close request timed out";
|
||||
terminate();
|
||||
}
|
||||
schedule 100 msec { print_metrics_and_close() };
|
||||
}
|
||||
timeout 5sec
|
||||
{
|
||||
|
|
|
|||
|
|
@ -4,13 +4,34 @@
|
|||
# @TEST-PORT: REDIS_PORT
|
||||
|
||||
# @TEST-EXEC: btest-bg-run redis-server run-redis-server ${REDIS_PORT%/tcp}
|
||||
# @TEST-EXEC: zeek -b %INPUT | sed 's|=[0-9]*/tcp|=xxxx/tcp|g' > out
|
||||
# @TEST-EXEC: zeek -b %INPUT | sed 's|=[0-9]*/tcp|=XXXX/tcp|g' | sed "s|-${REDIS_PORT%/tcp}-|-XXXX-|g" > out
|
||||
# @TEST-EXEC: btest-bg-wait -k 0
|
||||
|
||||
# @TEST-EXEC: btest-diff out
|
||||
|
||||
@load base/frameworks/storage/sync
|
||||
@load policy/frameworks/storage/backend/redis
|
||||
@load base/frameworks/telemetry
|
||||
|
||||
# Make sure the telemetry output is in a fixed order.
|
||||
redef running_under_test = T;
|
||||
|
||||
global b : opaque of Storage::BackendHandle;
|
||||
|
||||
event print_metrics_and_close()
|
||||
{
|
||||
print "";
|
||||
print "Post-operation metrics:";
|
||||
local storage_metrics = Telemetry::collect_metrics("zeek", "storage*");
|
||||
for (i in storage_metrics)
|
||||
{
|
||||
local m = storage_metrics[i];
|
||||
print m$opts$metric_type, m$opts$prefix, m$opts$name, m$label_names, m$label_values, m$value;
|
||||
}
|
||||
print "";
|
||||
|
||||
Storage::Sync::close_backend(b);
|
||||
}
|
||||
|
||||
event Storage::backend_opened(tag: Storage::Backend, config: any) {
|
||||
print "Storage::backend_opened", tag, config;
|
||||
|
|
@ -35,7 +56,7 @@ event zeek_init()
|
|||
local res = Storage::Sync::open_backend(Storage::STORAGE_BACKEND_REDIS, opts, string, string);
|
||||
print "open_result", res;
|
||||
|
||||
local b = res$value;
|
||||
b = res$value;
|
||||
|
||||
# Put a first value. This should return Storage::SUCCESS.
|
||||
res = Storage::Sync::put(b, [$key=key, $value=value]);
|
||||
|
|
@ -67,5 +88,9 @@ event zeek_init()
|
|||
if ( res$code == Storage::SUCCESS && res?$value )
|
||||
print "get result same as overwritten", value2 == (res$value as string);
|
||||
|
||||
Storage::Sync::close_backend(b);
|
||||
# Attempt to get a value that doesn't exist.
|
||||
res = Storage::Sync::get(b, "testing");
|
||||
print "get result", res;
|
||||
|
||||
schedule 100 msec { print_metrics_and_close() };
|
||||
}
|
||||
|
|
|
|||
|
|
@ -5,9 +5,39 @@
|
|||
|
||||
@load base/frameworks/storage/async
|
||||
@load policy/frameworks/storage/backend/sqlite
|
||||
@load base/frameworks/telemetry
|
||||
|
||||
# Make sure the telemetry output is in a fixed order.
|
||||
redef running_under_test = T;
|
||||
|
||||
redef exit_only_after_terminate = T;
|
||||
|
||||
global b : opaque of Storage::BackendHandle;
|
||||
|
||||
event print_metrics_and_close()
|
||||
{
|
||||
print "";
|
||||
print "Post-operation metrics:";
|
||||
local storage_metrics = Telemetry::collect_metrics("zeek", "storage*");
|
||||
for (i in storage_metrics)
|
||||
{
|
||||
local m = storage_metrics[i];
|
||||
print m$opts$metric_type, m$opts$prefix, m$opts$name, m$label_names, m$label_values, m$value;
|
||||
}
|
||||
print "";
|
||||
|
||||
when [] ( local close_res = Storage::Async::close_backend(b) )
|
||||
{
|
||||
print "closed succesfully";
|
||||
terminate();
|
||||
}
|
||||
timeout 5sec
|
||||
{
|
||||
print "close request timed out";
|
||||
terminate();
|
||||
}
|
||||
}
|
||||
|
||||
event Storage::backend_opened(tag: Storage::Backend, config: any) {
|
||||
print "Storage::backend_opened", tag, config;
|
||||
}
|
||||
|
|
@ -28,29 +58,20 @@ event zeek_init()
|
|||
Storage::STORAGE_BACKEND_SQLITE, opts, string, string) )
|
||||
{
|
||||
print "open result", open_res;
|
||||
local b = open_res$value;
|
||||
b = open_res$value;
|
||||
|
||||
when [b, key, value] ( local put_res = Storage::Async::put(b, [ $key=key,
|
||||
when [key, value] ( local put_res = Storage::Async::put(b, [ $key=key,
|
||||
$value=value ]) )
|
||||
{
|
||||
print "put result", put_res;
|
||||
|
||||
when [b, key, value] ( local get_res = Storage::Async::get(b, key) )
|
||||
when [key, value] ( local get_res = Storage::Async::get(b, key) )
|
||||
{
|
||||
print "get result", get_res;
|
||||
if ( get_res$code == Storage::SUCCESS && get_res?$value )
|
||||
print "get result same as inserted", value == ( get_res$value as string );
|
||||
|
||||
when [b] ( local close_res = Storage::Async::close_backend(b) )
|
||||
{
|
||||
print "closed succesfully";
|
||||
terminate();
|
||||
}
|
||||
timeout 5sec
|
||||
{
|
||||
print "close request timed out";
|
||||
terminate();
|
||||
}
|
||||
schedule 100 msec { print_metrics_and_close() };
|
||||
}
|
||||
timeout 5sec
|
||||
{
|
||||
|
|
|
|||
|
|
@ -7,6 +7,11 @@
|
|||
@load base/frameworks/storage/sync
|
||||
@load policy/frameworks/storage/backend/sqlite
|
||||
|
||||
event Storage::backend_opened(tag: Storage::Backend, config: any)
|
||||
{
|
||||
print "Storage::backend_opened", tag, config;
|
||||
}
|
||||
|
||||
event zeek_init()
|
||||
{
|
||||
# Create a database file in the .tmp directory with a 'testing' table
|
||||
|
|
|
|||
|
|
@ -0,0 +1,61 @@
|
|||
# @TEST-DOC: Test metrics for storage/sqlite in a sync context
|
||||
# @TEST-EXEC: zeek -b %INPUT > out
|
||||
# @TEST-EXEC: TEST_DIFF_CANONIFIER=$SCRIPTS/diff-remove-abspath btest-diff out
|
||||
# @TEST-EXEC: TEST_DIFF_CANONIFIER=$SCRIPTS/diff-remove-abspath btest-diff .stderr
|
||||
|
||||
@load base/frameworks/telemetry
|
||||
@load base/frameworks/storage/sync
|
||||
@load policy/frameworks/storage/backend/sqlite
|
||||
|
||||
# Make sure the telemetry output is in a fixed order.
|
||||
redef running_under_test = T;
|
||||
|
||||
global b : opaque of Storage::BackendHandle;
|
||||
|
||||
event print_metrics_and_close()
|
||||
{
|
||||
print "";
|
||||
print "Post-operation metrics:";
|
||||
local storage_metrics = Telemetry::collect_metrics("zeek", "storage*");
|
||||
for (i in storage_metrics)
|
||||
{
|
||||
local m = storage_metrics[i];
|
||||
print m$opts$metric_type, m$opts$prefix, m$opts$name, m$label_names, m$label_values, m$value;
|
||||
}
|
||||
|
||||
Storage::Sync::close_backend(b);
|
||||
}
|
||||
|
||||
event zeek_init()
|
||||
{
|
||||
# Create a database file in the .tmp directory with a 'testing' table
|
||||
local opts: Storage::BackendOptions;
|
||||
opts$sqlite = [ $database_path="test.sqlite", $table_name="testing" ];
|
||||
|
||||
local key = "key1234";
|
||||
local value = "value5678";
|
||||
|
||||
# Test inserting/retrieving a key/value pair that we know won't be in
|
||||
# the backend yet.
|
||||
local res = Storage::Sync::open_backend(Storage::STORAGE_BACKEND_SQLITE, opts, string, string);
|
||||
print "open result", res;
|
||||
b = res$value;
|
||||
|
||||
res = Storage::Sync::put(b, [$key=key, $value=value]);
|
||||
print "put result", res;
|
||||
|
||||
res = Storage::Sync::get(b, key);
|
||||
print "get result 1", res;
|
||||
|
||||
res = Storage::Sync::erase(b, key);
|
||||
print "erase result", res;
|
||||
|
||||
local res2 = Storage::Sync::get(b, key);
|
||||
print "get result 2", res2;
|
||||
|
||||
# Schedule this part for two reasons:
|
||||
# 1. To let the backend_opened event happen before we print metrics.
|
||||
# 2. To print the metrics before closing so that the page_count and file_size callbacks
|
||||
# run before closing.
|
||||
schedule 100 msecs { print_metrics_and_close() };
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue