Mirror/zeek
1
0
Fork 0
mirror of https://github.com/zeek/zeek.git synced 2025-10-08 17:48:21 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Add JSON storage serializer, use with existing backends/tests

Browse source
This commit is contained in:
Tim Wojtulewicz 2025-04-03 12:38:18 -07:00
parent 201d4508e6
commit 88786a28a2
18 changed files with 161 additions and 59 deletions
Download patch file Download diff file Expand all files Collapse all files

42
src/storage/backend/redis/Redis.cc
View file

@ -254,24 +254,30 @@ OperationResult Redis::DoPut(ResultCallback* cb, ValPtr key, ValPtr value, bool
auto locked_scope = conditionally_lock(zeek::run_state::reading_traces, expire_mutex);
std::string format = "SET %s:%s %s";
std::string format = "SET %s:%b %b";
if ( ! overwrite )
format.append(" NX");
auto json_key = key->ToJSON()->ToStdString();
auto json_value = value->ToJSON()->ToStdString();
auto key_data = serializer->Serialize(key);
if ( ! key_data )
return {ReturnCode::SERIALIZATION_FAILED, "Failed to serialize key"};
auto val_data = serializer->Serialize(value);
if ( ! val_data )
return {ReturnCode::SERIALIZATION_FAILED, "Failed to serialize value"};
int status;
// Use built-in expiration if reading live data, since time will move
// forward consistently. If reading pcaps, we'll do something else.
if ( expiration_time > 0.0 && ! zeek::run_state::reading_traces ) {
format.append(" PXAT %" PRIu64);
status = redisAsyncCommand(async_ctx, redisPut, cb, format.c_str(), key_prefix.data(), json_key.data(),
json_value.data(), static_cast<uint64_t>(expiration_time * 1e3));
status = redisAsyncCommand(async_ctx, redisPut, cb, format.c_str(), key_prefix.data(), key_data->data(),
key_data->size(), val_data->data(), val_data->size(),
static_cast<uint64_t>(expiration_time * 1e3));
}
else
status = redisAsyncCommand(async_ctx, redisPut, cb, format.c_str(), key_prefix.data(), json_key.data(),
json_value.data());
status = redisAsyncCommand(async_ctx, redisPut, cb, format.c_str(), key_prefix.data(), key_data->data(),
key_data->size(), val_data->data(), val_data->size());
if ( connected && status == REDIS_ERR )
return {ReturnCode::OPERATION_FAILED, util::fmt("Failed to queue put operation: %s", async_ctx->errstr)};
@ -284,10 +290,10 @@ OperationResult Redis::DoPut(ResultCallback* cb, ValPtr key, ValPtr value, bool
format = "ZADD %s_expire";
if ( ! overwrite )
format.append(" NX");
format += " %f %s";
format += " %f %b";
status = redisAsyncCommand(async_ctx, redisZADD, NULL, format.c_str(), key_prefix.data(), expiration_time,
json_key.data());
key_data->data(), key_data->size());
if ( connected && status == REDIS_ERR )
return {ReturnCode::OPERATION_FAILED, util::fmt("ZADD operation failed: %s", async_ctx->errstr)};
@ -307,8 +313,12 @@ OperationResult Redis::DoGet(ResultCallback* cb, ValPtr key) {
auto locked_scope = conditionally_lock(zeek::run_state::reading_traces, expire_mutex);
int status = redisAsyncCommand(async_ctx, redisGet, cb, "GET %s:%s", key_prefix.data(),
key->ToJSON()->ToStdStringView().data());
auto key_data = serializer->Serialize(key);
if ( ! key_data )
return {ReturnCode::SERIALIZATION_FAILED, "Failed to serialize key"};
int status =
redisAsyncCommand(async_ctx, redisGet, cb, "GET %s:%b", key_prefix.data(), key_data->data(), key_data->size());
if ( connected && status == REDIS_ERR )
return {ReturnCode::OPERATION_FAILED, util::fmt("Failed to queue get operation: %s", async_ctx->errstr)};
@ -330,8 +340,12 @@ OperationResult Redis::DoErase(ResultCallback* cb, ValPtr key) {
auto locked_scope = conditionally_lock(zeek::run_state::reading_traces, expire_mutex);
int status = redisAsyncCommand(async_ctx, redisErase, cb, "DEL %s:%s", key_prefix.data(),
key->ToJSON()->ToStdStringView().data());
auto key_data = serializer->Serialize(key);
if ( ! key_data )
return {ReturnCode::SERIALIZATION_FAILED, "Failed to serialize key"};
int status = redisAsyncCommand(async_ctx, redisErase, cb, "DEL %s:%b", key_prefix.data(), key_data->data(),
key_data->size());
if ( connected && status == REDIS_ERR )
return {ReturnCode::OPERATION_FAILED, async_ctx->errstr};
@ -439,7 +453,7 @@ void Redis::HandleGetResult(redisReply* reply, ResultCallback* callback) {
else if ( reply->type == REDIS_REPLY_ERROR )
res = ParseReplyError("get", reply->str);
else {
auto val = zeek::detail::ValFromJSON(reply->str, val_type, Func::nil);
auto val = serializer->Unserialize({(std::byte*)reply->str, reply->len}, val_type);
if ( val )
res = {ReturnCode::SUCCESS, "", val.value()};
else

42
src/storage/backend/sqlite/SQLite.cc
View file

@ -44,7 +44,7 @@ OperationResult SQLite::DoOpen(OpenResultCallback* cb, RecordValPtr options) {
}
std::string create = "create table if not exists " + table_name + " (";
create.append("key_str text primary key, value_str text not null, expire_time real);");
create.append("key_str blob primary key, value_str blob not null, expire_time real);");
char* errorMsg = nullptr;
if ( int res = sqlite3_exec(db, create.c_str(), NULL, NULL, &errorMsg); res != SQLITE_OK ) {
@ -151,8 +151,9 @@ OperationResult SQLite::DoPut(ResultCallback* cb, ValPtr key, ValPtr value, bool
if ( ! db )
return {ReturnCode::NOT_CONNECTED};
auto json_key = key->ToJSON();
auto json_value = value->ToJSON();
auto key_data = serializer->Serialize(key);
if ( ! key_data )
return {ReturnCode::SERIALIZATION_FAILED, "Failed to serialize key"};
sqlite3_stmt* stmt;
if ( ! overwrite )
@ -160,15 +161,17 @@ OperationResult SQLite::DoPut(ResultCallback* cb, ValPtr key, ValPtr value, bool
else
stmt = put_update_stmt.get();
auto key_str = json_key->ToStdStringView();
if ( auto res = CheckError(sqlite3_bind_text(stmt, 1, key_str.data(), key_str.size(), SQLITE_STATIC));
if ( auto res = CheckError(sqlite3_bind_blob(stmt, 1, key_data->data(), key_data->size(), SQLITE_STATIC));
res.code != ReturnCode::SUCCESS ) {
sqlite3_reset(stmt);
return res;
}
auto value_str = json_value->ToStdStringView();
if ( auto res = CheckError(sqlite3_bind_text(stmt, 2, value_str.data(), value_str.size(), SQLITE_STATIC));
auto val_data = serializer->Serialize(value);
if ( ! val_data )
return {ReturnCode::SERIALIZATION_FAILED, "Failed to serialize value"};
if ( auto res = CheckError(sqlite3_bind_blob(stmt, 2, val_data->data(), val_data->size(), SQLITE_STATIC));
res.code != ReturnCode::SUCCESS ) {
sqlite3_reset(stmt);
return res;
@ -180,7 +183,7 @@ OperationResult SQLite::DoPut(ResultCallback* cb, ValPtr key, ValPtr value, bool
}
if ( overwrite ) {
if ( auto res = CheckError(sqlite3_bind_text(stmt, 4, value_str.data(), value_str.size(), SQLITE_STATIC));
if ( auto res = CheckError(sqlite3_bind_blob(stmt, 4, val_data->data(), val_data->size(), SQLITE_STATIC));
res.code != ReturnCode::SUCCESS ) {
sqlite3_reset(stmt);
return res;
@ -197,11 +200,13 @@ OperationResult SQLite::DoGet(ResultCallback* cb, ValPtr key) {
if ( ! db )
return {ReturnCode::NOT_CONNECTED};
auto json_key = key->ToJSON();
auto key_data = serializer->Serialize(key);
if ( ! key_data )
return {ReturnCode::SERIALIZATION_FAILED, "Failed to serialize key"};
auto stmt = get_stmt.get();
auto key_str = json_key->ToStdStringView();
if ( auto res = CheckError(sqlite3_bind_text(stmt, 1, key_str.data(), key_str.size(), SQLITE_STATIC));
if ( auto res = CheckError(sqlite3_bind_blob(stmt, 1, key_data->data(), key_data->size(), SQLITE_STATIC));
res.code != ReturnCode::SUCCESS ) {
sqlite3_reset(stmt);
return res;
@ -217,11 +222,13 @@ OperationResult SQLite::DoErase(ResultCallback* cb, ValPtr key) {
if ( ! db )
return {ReturnCode::NOT_CONNECTED};
auto json_key = key->ToJSON();
auto key_data = serializer->Serialize(key);
if ( ! key_data )
return {ReturnCode::SERIALIZATION_FAILED, "Failed to serialize key"};
auto stmt = erase_stmt.get();
auto key_str = json_key->ToStdStringView();
if ( auto res = CheckError(sqlite3_bind_text(stmt, 1, key_str.data(), key_str.size(), SQLITE_STATIC));
if ( auto res = CheckError(sqlite3_bind_blob(stmt, 1, key_data->data(), key_data->size(), SQLITE_STATIC));
res.code != ReturnCode::SUCCESS ) {
sqlite3_reset(stmt);
return res;
@ -266,9 +273,10 @@ OperationResult SQLite::Step(sqlite3_stmt* stmt, bool parse_value) {
int step_status = sqlite3_step(stmt);
if ( step_status == SQLITE_ROW ) {
if ( parse_value ) {
// Column 1 is the value
const char* text = (const char*)sqlite3_column_text(stmt, 0);
auto val = zeek::detail::ValFromJSON(text, val_type, Func::nil);
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);
sqlite3_reset(stmt);
if ( val )

1
src/storage/serializer/CMakeLists.txt
View file

@ -0,0 +1 @@
add_subdirectory(json)

3
src/storage/serializer/json/CMakeLists.txt Normal file
View file

@ -0,0 +1,3 @@
zeek_add_plugin(
Zeek Storage_Serializer_JSON
SOURCES JSON.cc Plugin.cc)

29
src/storage/serializer/json/JSON.cc Normal file
View file

@ -0,0 +1,29 @@
// See the file "COPYING" in the main distribution directory for copyright.
#include "zeek/storage/serializer/json/JSON.h"
#include "zeek/Func.h"
namespace zeek::storage::serializer::json {
std::unique_ptr<Serializer> JSON::Instantiate() { return std::make_unique<JSON>(); }
JSON::JSON() : Serializer("JSON") {}
std::optional<detail::byte_buffer> JSON::Serialize(ValPtr val) {
detail::byte_buffer buf;
auto json = val->ToJSON();
buf.reserve(json->Len());
std::transform(json->Bytes(), json->Bytes() + json->Len(), std::back_inserter(buf),
[](u_char c) { return std::byte(c); });
return buf;
}
zeek::expected<ValPtr, std::string> JSON::Unserialize(detail::byte_buffer_span buf, TypePtr type) {
std::string_view text{reinterpret_cast<std::string::const_pointer>(buf.data()), buf.size()};
return zeek::detail::ValFromJSON(text, type, Func::nil);
}
} // namespace zeek::storage::serializer::json

20
src/storage/serializer/json/JSON.h Normal file
View file

@ -0,0 +1,20 @@
// See the file "COPYING" in the main distribution directory for copyright.
#pragma once
#include "zeek/storage/Serializer.h"
namespace zeek::storage::serializer::json {
class JSON final : public Serializer {
public:
static std::unique_ptr<Serializer> Instantiate();
JSON();
~JSON() override = default;
std::optional<detail::byte_buffer> Serialize(ValPtr val) override;
zeek::expected<ValPtr, std::string> Unserialize(detail::byte_buffer_span buf, TypePtr type) override;
};
} // namespace zeek::storage::serializer::json

22
src/storage/serializer/json/Plugin.cc Normal file
View file

@ -0,0 +1,22 @@
// See the file "COPYING" in the main distribution directory for copyright.
#include "zeek/plugin/Plugin.h"
#include "zeek/storage/Component.h"
#include "zeek/storage/serializer/json/JSON.h"
namespace zeek::storage::serializer::json {
class Plugin final : public plugin::Plugin {
public:
plugin::Configuration Configure() override {
AddComponent(new storage::SerializerComponent("JSON", serializer::json::JSON::Instantiate));
plugin::Configuration config;
config.name = "Zeek::Storage_Serializer_JSON";
config.description = "JSON serializer for storage framework";
return config;
}
} plugin;
} // namespace zeek::storage::serializer::json

4
src/storage/storage-async.bif
View file

@ -80,9 +80,9 @@ function Storage::Async::__open_backend%(btype: Storage::Backend, options: any,
return nullptr;
auto btype_val = IntrusivePtr<EnumVal>{NewRef{}, btype->AsEnumVal()};
Tag btag{btype_val};
Tag tag{btype_val};
auto b = storage_mgr->InstantiateBackend(btag);
auto b = storage_mgr->InstantiateBackend(tag);
if ( ! b.has_value() ) {
trigger->Cache(

4
src/storage/storage-sync.bif
View file

@ -31,9 +31,9 @@ module Storage::Sync;
function Storage::Sync::__open_backend%(btype: Storage::Backend, options: any, key_type: any, val_type: any%): Storage::OperationResult
%{
auto btype_val = IntrusivePtr<EnumVal>{NewRef{}, btype->AsEnumVal()};
Tag btag{btype_val};
Tag tag{btype_val};
auto b = storage_mgr->InstantiateBackend(btag);
auto b = storage_mgr->InstantiateBackend(tag);
if ( ! b.has_value() ) {
emit_builtin_error(b.error().c_str());

4
testing/btest/Baseline/scripts.base.frameworks.storage.redis-disconnect/out
View file

@ -1,4 +1,4 @@
### 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>]
Storage::backend_opened, Storage::REDIS, [redis=[server_host=127.0.0.1, server_port=xxxx/tcp, server_unix_socket=<uninitialized>, key_prefix=testing]]
Storage::backend_lost, Storage::REDIS, [redis=[server_host=127.0.0.1, server_port=xxxx/tcp, server_unix_socket=<uninitialized>, key_prefix=testing]], Server closed the connection
Storage::backend_opened, Storage::REDIS, [serializer=Storage::JSON, redis=[server_host=127.0.0.1, server_port=xxxx/tcp, server_unix_socket=<uninitialized>, key_prefix=testing]]
Storage::backend_lost, Storage::REDIS, [serializer=Storage::JSON, redis=[server_host=127.0.0.1, server_port=xxxx/tcp, server_unix_socket=<uninitialized>, key_prefix=testing]], Server closed the connection

4
testing/btest/Baseline/scripts.base.frameworks.storage.redis-sync/out
View file

@ -9,5 +9,5 @@ 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::REDIS, [redis=[server_host=127.0.0.1, server_port=xxxx/tcp, server_unix_socket=<uninitialized>, key_prefix=testing]]
Storage::backend_lost, Storage::REDIS, [redis=[server_host=127.0.0.1, server_port=xxxx/tcp, server_unix_socket=<uninitialized>, key_prefix=testing]], Client disconnected
Storage::backend_opened, Storage::REDIS, [serializer=Storage::JSON, redis=[server_host=127.0.0.1, server_port=xxxx/tcp, server_unix_socket=<uninitialized>, key_prefix=testing]]
Storage::backend_lost, Storage::REDIS, [serializer=Storage::JSON, redis=[server_host=127.0.0.1, server_port=xxxx/tcp, server_unix_socket=<uninitialized>, key_prefix=testing]], Client disconnected

2
testing/btest/Baseline/scripts.base.frameworks.storage.sqlite-basic/out
View file

@ -1,5 +1,5 @@
### BTest baseline data generated by btest-diff. Do not edit. Use "btest -U/-u" to update. Requires BTest >= 0.63.
Storage::backend_opened, Storage::SQLITE, [sqlite=[database_path=test.sqlite, table_name=testing, tuning_params={
Storage::backend_opened, Storage::SQLITE, [serializer=Storage::JSON, sqlite=[database_path=test.sqlite, table_name=testing, tuning_params={
[synchronous] = normal,
[temp_store] = memory,
[journal_mode] = WAL

3
testing/btest/plugins/storage-plugin/src/Plugin.cc
View file

@ -11,7 +11,8 @@ Plugin plugin;
using namespace btest::plugin::Testing_StorageDummy;
zeek::plugin::Configuration Plugin::Configure() {
AddComponent(new zeek::storage::Component("StorageDummy", btest::storage::backend::StorageDummy::Instantiate));
AddComponent(
new zeek::storage::BackendComponent("StorageDummy", btest::storage::backend::StorageDummy::Instantiate));
zeek::plugin::Configuration config;
config.name = "Testing::StorageDummy";

31
testing/btest/plugins/storage-plugin/src/StorageDummy.cc
View file

@ -49,9 +49,10 @@ OperationResult StorageDummy::DoPut(ResultCallback* cb, ValPtr key, ValPtr value
if ( timeout_put )
return {ReturnCode::TIMEOUT};
auto json_key = key->ToJSON()->ToStdString();
auto json_value = value->ToJSON()->ToStdString();
data[json_key] = json_value;
auto key_data = serializer->Serialize(key);
auto val_data = serializer->Serialize(value);
data[*key_data] = *val_data;
return {ReturnCode::SUCCESS};
}
@ -59,31 +60,31 @@ OperationResult StorageDummy::DoPut(ResultCallback* cb, ValPtr key, ValPtr value
* The workhorse method for Get(). This must be implemented for plugins.
*/
OperationResult StorageDummy::DoGet(ResultCallback* cb, ValPtr key) {
auto json_key = key->ToJSON();
auto it = data.find(json_key->ToStdString());
auto key_data = serializer->Serialize(key);
auto it = data.find(*key_data);
if ( it == data.end() )
return {ReturnCode::KEY_NOT_FOUND};
auto val = zeek::detail::ValFromJSON(it->second.c_str(), val_type, Func::nil);
if ( std::holds_alternative<ValPtr>(val) ) {
ValPtr val_v = std::get<ValPtr>(val);
return {ReturnCode::SUCCESS, "", val_v};
}
auto val = serializer->Unserialize(it->second, val_type);
if ( val )
return {ReturnCode::SUCCESS, "", val.value()};
return {ReturnCode::OPERATION_FAILED, std::get<std::string>(val)};
return {ReturnCode::UNSERIALIZATION_FAILED, val.error()};
}
/**
* The workhorse method for Erase(). This must be implemented for plugins.
*/
OperationResult StorageDummy::DoErase(ResultCallback* cb, ValPtr key) {
auto json_key = key->ToJSON();
auto it = data.find(json_key->ToStdString());
if ( it == data.end() )
return {ReturnCode::KEY_NOT_FOUND};
auto key_data = serializer->Serialize(key);
if ( auto it = data.find(*key_data); it != data.end() ) {
data.erase(it);
return {ReturnCode::SUCCESS};
}
return {ReturnCode::KEY_NOT_FOUND};
}
} // namespace btest::storage::backend

2
testing/btest/plugins/storage-plugin/src/StorageDummy.h
View file

@ -50,7 +50,7 @@ public:
zeek::storage::OperationResult DoErase(zeek::storage::ResultCallback* cb, zeek::ValPtr key) override;
private:
std::map<std::string, std::string> data;
std::map<zeek::storage::detail::byte_buffer, zeek::storage::detail::byte_buffer> data;
bool open = false;
};

1
testing/btest/scripts/base/frameworks/storage/redis-disconnect.zeek
View file

@ -26,6 +26,7 @@ event Storage::backend_lost(tag: Storage::Backend, config: any, reason: string)
event zeek_init()
{
local opts: Storage::BackendOptions;
opts$serializer = Storage::JSON;
opts$redis = [ $server_host="127.0.0.1", $server_port=to_port(getenv(
"REDIS_PORT")), $key_prefix="testing" ];

1
testing/btest/scripts/base/frameworks/storage/redis-sync.zeek
View file

@ -24,6 +24,7 @@ event Storage::backend_lost(tag: Storage::Backend, config: any, reason: string)
event zeek_init()
{
local opts: Storage::BackendOptions;
opts$serializer = Storage::JSON;
opts$redis = [ $server_host="127.0.0.1", $server_port=to_port(getenv(
"REDIS_PORT")), $key_prefix="testing" ];

1
testing/btest/scripts/base/frameworks/storage/sqlite-basic.zeek
View file

@ -16,6 +16,7 @@ event zeek_init()
{
# Create a database file in the .tmp directory with a 'testing' table
local opts: Storage::BackendOptions;
opts$serializer = Storage::JSON;
opts$sqlite = [ $database_path="test.sqlite", $table_name="testing" ];
local key = "key1234";