mirror of
https://github.com/zeek/zeek.git
synced 2025-10-02 06:38:20 +00:00
e1f70164e0
Also use the generic metadata version for publishing, keep the ts-based API for now, but only add timestamps when EventMetadata::add_network_timestamp is T. I'm not sure what the right way forward here is, maybe deprecating Broker's publish event variations and funneling through cluster.
2203 lines
78 KiB
C++
2203 lines
78 KiB
C++
// See the file "COPYING" in the main distribution directory for copyright.
|
|
|
|
#include "zeek/broker/Manager.h"
|
|
|
|
#include <broker/config.hh>
|
|
#include <broker/configuration.hh>
|
|
#include <broker/endpoint.hh>
|
|
#include <broker/event.hh>
|
|
#include <broker/event_observer.hh>
|
|
#include <broker/logger.hh>
|
|
#include <broker/time.hh>
|
|
#include <broker/variant.hh>
|
|
#include <broker/zeek.hh>
|
|
#include <unistd.h>
|
|
#include <cstdio>
|
|
#include <cstring>
|
|
#include <string>
|
|
#include <string_view>
|
|
|
|
#include "zeek/DebugLogger.h"
|
|
#include "zeek/Desc.h"
|
|
#include "zeek/Event.h"
|
|
#include "zeek/Flare.h"
|
|
#include "zeek/Func.h"
|
|
#include "zeek/IntrusivePtr.h"
|
|
#include "zeek/Reporter.h"
|
|
#include "zeek/RunState.h"
|
|
#include "zeek/Scope.h"
|
|
#include "zeek/SerializationFormat.h"
|
|
#include "zeek/Type.h"
|
|
#include "zeek/Var.h"
|
|
#include "zeek/broker/Data.h"
|
|
#include "zeek/broker/Store.h"
|
|
#include "zeek/broker/comm.bif.h"
|
|
#include "zeek/broker/messaging.bif.h"
|
|
#include "zeek/broker/store.bif.h"
|
|
#include "zeek/cluster/serializer/broker/Serializer.h"
|
|
#include "zeek/iosource/Manager.h"
|
|
#include "zeek/logging/Manager.h"
|
|
#include "zeek/logging/Types.h"
|
|
#include "zeek/plugin/Manager.h"
|
|
#include "zeek/plugin/Plugin.h"
|
|
#include "zeek/telemetry/Manager.h"
|
|
#include "zeek/util.h"
|
|
|
|
#include "const.bif.netvar_h"
|
|
|
|
using namespace std;
|
|
|
|
namespace {
|
|
|
|
broker::vector broker_vector_from(const broker::variant& arg) {
|
|
auto tmp = arg.to_data();
|
|
return std::move(broker::get<broker::vector>(tmp));
|
|
}
|
|
|
|
void print_escaped(std::string& buf, std::string_view str) {
|
|
buf.push_back('"');
|
|
for ( auto c : str ) {
|
|
switch ( c ) {
|
|
default: buf.push_back(c); break;
|
|
case '\\':
|
|
buf.push_back('\\');
|
|
buf.push_back('\\');
|
|
break;
|
|
case '\b':
|
|
buf.push_back('\\');
|
|
buf.push_back('b');
|
|
break;
|
|
case '\f':
|
|
buf.push_back('\\');
|
|
buf.push_back('f');
|
|
break;
|
|
case '\n':
|
|
buf.push_back('\\');
|
|
buf.push_back('n');
|
|
break;
|
|
case '\r':
|
|
buf.push_back('\\');
|
|
buf.push_back('r');
|
|
break;
|
|
case '\t':
|
|
buf.push_back('\\');
|
|
buf.push_back('t');
|
|
break;
|
|
case '\v':
|
|
buf.push_back('\\');
|
|
buf.push_back('v');
|
|
break;
|
|
case '"':
|
|
buf.push_back('\\');
|
|
buf.push_back('"');
|
|
break;
|
|
}
|
|
}
|
|
buf.push_back('"');
|
|
}
|
|
|
|
// Track metrics for a given peering's send buffer.
|
|
class PeerBufferState {
|
|
public:
|
|
struct Stats {
|
|
// The rendered peer ID. Storing this here helps reuse.
|
|
// Note that we only ever touch this from Zeek's main thread, not
|
|
// any of Broker's.
|
|
zeek::StringValPtr peer_id;
|
|
|
|
// Whether Broker has removed the peer, and this instance still
|
|
// needs to be removed.
|
|
bool is_zombie = false;
|
|
|
|
// Number of messages queued locally in the send buffer.
|
|
uint32_t queued = 0;
|
|
|
|
// Maximum number queued in the last Broker::buffer_stats_reset_interval.
|
|
// This improces visibility into message bursts since instantaneous
|
|
// queueing (captured above) can be short-lived.
|
|
uint32_t max_queued_recently = 0;
|
|
|
|
// Number of times the buffer overflowed at send time. For the
|
|
// "disconnect" overflow policy (via Broker::peer_overflow_policy), this
|
|
// count will at most be 1 since Broker will remove the peering upon
|
|
// overflow. The existing Zeek-level metric for tracking disconnects
|
|
// (see frameworks/broker/broker-backpressure.zeek) covers this one more
|
|
// permanently. For the "drop_newest" and "drop_oldest" policies it
|
|
// equals a count of the number of messages lost, since the peering
|
|
// continues.
|
|
uint64_t overflows = 0;
|
|
|
|
// When we last started a stats-tracking interval for this peering.
|
|
double last_interval = 0;
|
|
};
|
|
|
|
// For per-peering tracking, map endpoint IDs to the above state.
|
|
using EndpointMetricMap = std::unordered_map<broker::endpoint_id, Stats>;
|
|
|
|
PeerBufferState(size_t a_buffer_size, double a_stats_reset_interval)
|
|
: buffer_size(a_buffer_size), stats_reset_interval(a_stats_reset_interval) {
|
|
stats_table =
|
|
zeek::make_intrusive<zeek::TableVal>(zeek::id::find_type<zeek::TableType>("BrokerPeeringStatsTable"));
|
|
stats_record_type = zeek::id::find_type<zeek::RecordType>("BrokerPeeringStats");
|
|
}
|
|
|
|
void SetEndpoint(const broker::endpoint* a_endpoint) { endpoint = a_endpoint; }
|
|
|
|
// Update the peering's stats. This runs in Broker's execution context.
|
|
// Broker does not expose send-buffer/queue state explicitly, so track
|
|
// arrivals (a push, is_push == true) and departures (a pull, is_push ==
|
|
// false) as they happen. Note that this must not touch Zeek-side Vals.
|
|
void Observe(const broker::endpoint_id& peer, bool is_push) {
|
|
std::lock_guard<std::mutex> lock(mutex);
|
|
auto it = stats_map.find(peer);
|
|
|
|
if ( it == stats_map.end() ) {
|
|
stats_map.emplace(peer, Stats());
|
|
it = stats_map.find(peer);
|
|
}
|
|
|
|
auto& stats = it->second;
|
|
|
|
// Stick to Broker's notion of time here.
|
|
double now{0};
|
|
if ( endpoint != nullptr )
|
|
broker::convert(endpoint->now(), now);
|
|
|
|
if ( now - stats.last_interval > stats_reset_interval ) {
|
|
stats.last_interval = now;
|
|
stats.max_queued_recently = stats.queued;
|
|
}
|
|
|
|
if ( stats.queued == 0 ) {
|
|
// Watch for underflows. We could report somehow. Note that this
|
|
// runs in the context of Broker's threads.
|
|
assert(is_push);
|
|
}
|
|
|
|
if ( is_push && stats.queued == buffer_size )
|
|
stats.overflows += 1;
|
|
else {
|
|
stats.queued += is_push ? 1 : -1;
|
|
if ( stats.queued > stats.max_queued_recently )
|
|
stats.max_queued_recently = stats.queued;
|
|
}
|
|
}
|
|
|
|
// Updates the internal table[string] of BrokerPeeringStats and returns it.
|
|
const zeek::TableValPtr& GetPeeringStatsTable() {
|
|
std::lock_guard<std::mutex> lock(mutex);
|
|
|
|
for ( auto it = stats_map.begin(); it != stats_map.end(); ) {
|
|
auto& peer = it->first;
|
|
auto& stats = it->second;
|
|
|
|
if ( stats.peer_id == nullptr )
|
|
stats.peer_id = PeerIdToStringVal(peer);
|
|
|
|
// Broker told us the peer is gone, in RemovePeer() below. Remove it
|
|
// now from both tables. We add/remove from stats_table only here,
|
|
// not in Observer() and/or RemovePeer(), to ensure we only touch
|
|
// the Zeek-side Table from Zeek's main thread.
|
|
if ( stats.is_zombie ) {
|
|
stats_table->Remove(*stats.peer_id);
|
|
it = stats_map.erase(it);
|
|
continue;
|
|
}
|
|
|
|
auto stats_v = stats_table->Find(stats.peer_id);
|
|
|
|
if ( stats_v == nullptr ) {
|
|
stats_v = zeek::make_intrusive<zeek::RecordVal>(stats_record_type);
|
|
stats_table->Assign(stats.peer_id, stats_v);
|
|
}
|
|
|
|
// We may get here more than stats_reset_interval after the last
|
|
// Observe(), in which case the max_queued_recently value is now
|
|
// stale. Update if so.
|
|
double now{0};
|
|
if ( endpoint != nullptr )
|
|
broker::convert(endpoint->now(), now);
|
|
|
|
if ( now - stats.last_interval > stats_reset_interval ) {
|
|
stats.last_interval = now;
|
|
stats.max_queued_recently = stats.queued;
|
|
}
|
|
|
|
int n = 0;
|
|
stats_v->AsRecordVal()->Assign(n++, zeek::val_mgr->Count(stats.queued));
|
|
stats_v->AsRecordVal()->Assign(n++, zeek::val_mgr->Count(stats.max_queued_recently));
|
|
stats_v->AsRecordVal()->Assign(n++, zeek::val_mgr->Count(stats.overflows));
|
|
|
|
++it;
|
|
}
|
|
|
|
return stats_table;
|
|
}
|
|
|
|
void RemovePeer(const broker::endpoint_id& peer) {
|
|
std::lock_guard<std::mutex> lock(mutex);
|
|
if ( auto it = stats_map.find(peer); it != stats_map.end() )
|
|
it->second.is_zombie = true;
|
|
}
|
|
|
|
private:
|
|
zeek::StringValPtr PeerIdToStringVal(const broker::endpoint_id& peer) const {
|
|
std::string peer_s;
|
|
broker::convert(peer, peer_s);
|
|
return zeek::make_intrusive<zeek::StringVal>(peer_s);
|
|
}
|
|
|
|
// The maximum number of messages queueable for transmission to a peer,
|
|
// see Broker::peer_buffer_size and Broker::web_socket_buffer_size.
|
|
size_t buffer_size;
|
|
|
|
// Seconds after which we reset stats tracked per time window.
|
|
double stats_reset_interval;
|
|
|
|
EndpointMetricMap stats_map;
|
|
zeek::TableValPtr stats_table;
|
|
zeek::RecordTypePtr stats_record_type;
|
|
|
|
mutable std::mutex mutex;
|
|
const broker::endpoint* endpoint = nullptr;
|
|
};
|
|
|
|
using PeerBufferStatePtr = std::shared_ptr<PeerBufferState>;
|
|
|
|
class LoggerQueue {
|
|
public:
|
|
void Push(broker::event_ptr event) {
|
|
std::list<broker::event_ptr> tmp;
|
|
tmp.emplace_back(std::move(event));
|
|
{
|
|
std::lock_guard<std::mutex> lock(mutex_);
|
|
queue_.splice(queue_.end(), tmp);
|
|
if ( queue_.size() == 1 ) {
|
|
flare_.Fire();
|
|
}
|
|
}
|
|
}
|
|
|
|
auto Drain() {
|
|
std::list<broker::event_ptr> events;
|
|
std::lock_guard<std::mutex> lock(mutex_);
|
|
if ( ! queue_.empty() ) {
|
|
queue_.swap(events);
|
|
flare_.Extinguish();
|
|
}
|
|
return events;
|
|
}
|
|
|
|
auto FlareFd() const noexcept { return flare_.FD(); }
|
|
|
|
private:
|
|
std::mutex mutex_;
|
|
zeek::detail::Flare flare_;
|
|
std::list<broker::event_ptr> queue_;
|
|
};
|
|
|
|
using LoggerQueuePtr = std::shared_ptr<LoggerQueue>;
|
|
|
|
using BrokerSeverityLevel = broker::event::severity_level;
|
|
|
|
class Observer : public broker::event_observer {
|
|
public:
|
|
using LogSeverityLevel = broker::event::severity_level;
|
|
|
|
explicit Observer(LogSeverityLevel severity, LoggerQueuePtr queue, PeerBufferStatePtr pbstate)
|
|
: severity_(severity), queue_(std::move(queue)), pbstate_(std::move(pbstate)) {}
|
|
|
|
void on_peer_buffer_push(const broker::endpoint_id& peer, const broker::node_message&) override {
|
|
pbstate_->Observe(peer, true);
|
|
}
|
|
|
|
void on_peer_buffer_pull(const broker::endpoint_id& peer, const broker::node_message&) override {
|
|
pbstate_->Observe(peer, false);
|
|
}
|
|
|
|
void on_peer_disconnect(const broker::endpoint_id& peer, const broker::error&) override {
|
|
pbstate_->RemovePeer(peer);
|
|
}
|
|
|
|
void observe(broker::event_ptr what) override { queue_->Push(std::move(what)); }
|
|
|
|
bool accepts(LogSeverityLevel severity, broker::event::component_type) const override {
|
|
return severity <= severity_;
|
|
}
|
|
|
|
private:
|
|
LogSeverityLevel severity_;
|
|
LoggerQueuePtr queue_;
|
|
PeerBufferStatePtr pbstate_;
|
|
};
|
|
|
|
} // namespace
|
|
|
|
namespace zeek::Broker {
|
|
static inline Val* get_option(const char* option) {
|
|
const auto& id = zeek::detail::global_scope()->Find(option);
|
|
|
|
if ( ! (id && id->GetVal()) )
|
|
reporter->FatalError("Unknown Broker option %s", option);
|
|
|
|
return id->GetVal().get();
|
|
}
|
|
|
|
template<class T>
|
|
static inline void set_option(const char* option, const T& value) {
|
|
const auto& id = zeek::detail::global_scope()->Find(option);
|
|
|
|
if ( ! id )
|
|
reporter->FatalError("Unknown Broker option %s", option);
|
|
|
|
if constexpr ( std::is_same_v<T, broker::port> ) {
|
|
switch ( value.type() ) {
|
|
case broker::port::protocol::tcp: id->SetVal(val_mgr->Port(value.number(), TRANSPORT_TCP)); break;
|
|
case broker::port::protocol::udp: id->SetVal(val_mgr->Port(value.number(), TRANSPORT_UDP)); break;
|
|
case broker::port::protocol::icmp: id->SetVal(val_mgr->Port(value.number(), TRANSPORT_ICMP)); break;
|
|
default: id->SetVal(val_mgr->Port(value.number(), TRANSPORT_UNKNOWN));
|
|
}
|
|
}
|
|
else if constexpr ( std::is_same_v<T, broker::timespan> ) {
|
|
using std::chrono::duration_cast;
|
|
auto ts = duration_cast<broker::fractional_seconds>(value);
|
|
id->SetVal(make_intrusive<IntervalVal>(ts.count()));
|
|
}
|
|
else if constexpr ( std::is_same_v<T, std::vector<std::string>> ) {
|
|
auto ptr = make_intrusive<VectorVal>(zeek::id::string_vec);
|
|
for ( const auto& str : value )
|
|
ptr->Append(make_intrusive<StringVal>(str));
|
|
id->SetVal(std::move(ptr));
|
|
}
|
|
else {
|
|
static_assert(std::is_same_v<T, std::string>);
|
|
id->SetVal(make_intrusive<StringVal>(value));
|
|
}
|
|
}
|
|
|
|
namespace {
|
|
|
|
struct opt_mapping {
|
|
broker::configuration* cfg;
|
|
std::string broker_name;
|
|
const char* zeek_name;
|
|
|
|
template<class T>
|
|
auto broker_read() {
|
|
return broker::get_as<T>(*cfg, broker_name);
|
|
}
|
|
|
|
template<class T>
|
|
auto broker_write(T&& val) {
|
|
cfg->set(broker_name, std::forward<T>(val));
|
|
}
|
|
|
|
auto zeek_read() { return get_option(zeek_name); }
|
|
|
|
template<class T>
|
|
auto zeek_write(const T& val) {
|
|
set_option(zeek_name, val);
|
|
}
|
|
};
|
|
|
|
#define WITH_OPT_MAPPING(broker_name, zeek_name) if ( auto opt = opt_mapping{&config, broker_name, zeek_name}; true )
|
|
|
|
} // namespace
|
|
|
|
class BrokerState {
|
|
public:
|
|
using LogSeverityLevel = Observer::LogSeverityLevel;
|
|
|
|
BrokerState(broker::configuration config, LoggerQueuePtr queue, PeerBufferStatePtr pbstate)
|
|
: endpoint(std::move(config), telemetry_mgr->GetRegistry()),
|
|
subscriber(endpoint.make_subscriber({broker::topic::statuses(), broker::topic::errors()})),
|
|
loggerQueue(std::move(queue)),
|
|
peerBufferState(std::move(pbstate)) {
|
|
peerBufferState->SetEndpoint(&endpoint);
|
|
}
|
|
|
|
broker::endpoint endpoint;
|
|
broker::subscriber subscriber;
|
|
LoggerQueuePtr loggerQueue;
|
|
PeerBufferStatePtr peerBufferState;
|
|
LogSeverityLevel logSeverity = LogSeverityLevel::critical;
|
|
LogSeverityLevel stderrSeverity = LogSeverityLevel::critical;
|
|
std::unordered_set<broker::network_info> outbound_peerings;
|
|
};
|
|
|
|
const broker::endpoint_info Manager::NoPeer{{}, {}};
|
|
|
|
int Manager::script_scope = 0;
|
|
|
|
struct scoped_reporter_location {
|
|
scoped_reporter_location(zeek::detail::Frame* frame) { reporter->PushLocation(frame->GetCallLocation()); }
|
|
|
|
~scoped_reporter_location() { reporter->PopLocation(); }
|
|
};
|
|
|
|
#ifdef DEBUG
|
|
namespace {
|
|
|
|
std::string RenderMessage(const broker::variant& d) { return util::json_escape_utf8(broker::to_string(d)); }
|
|
|
|
std::string RenderMessage(const broker::variant_list& d) { return util::json_escape_utf8(broker::to_string(d)); }
|
|
|
|
std::string RenderMessage(const broker::store::response& x) {
|
|
return util::fmt("%s [id %" PRIu64 "]", (x.answer ? broker::to_string(*x.answer).c_str() : "<no answer>"), x.id);
|
|
}
|
|
|
|
std::string RenderMessage(const broker::status& s) { return broker::to_string(s.code()); }
|
|
|
|
std::string RenderMessage(const broker::error& e) {
|
|
if ( auto ctx = e.context() )
|
|
return util::fmt("%s (%s)", broker::to_string(e.code()).c_str(), to_string(*ctx).c_str());
|
|
else
|
|
return util::fmt("%s (null)", broker::to_string(e.code()).c_str());
|
|
}
|
|
|
|
std::string RenderMessage(const std::string& topic, const broker::variant& x) {
|
|
return util::fmt("%s -> %s", RenderMessage(x).c_str(), topic.c_str());
|
|
}
|
|
|
|
template<class VariantOrList>
|
|
std::string RenderEvent(const std::string& topic, const std::string& name, const VariantOrList& args) {
|
|
return util::fmt("%s(%s) -> %s", name.c_str(), RenderMessage(args).c_str(), topic.c_str());
|
|
}
|
|
|
|
} // namespace
|
|
#endif
|
|
|
|
Manager::Manager(bool arg_use_real_time) : Backend("Broker", nullptr, nullptr, nullptr), iosource::IOSource(true) {
|
|
bound_port = 0;
|
|
use_real_time = arg_use_real_time;
|
|
peer_count = 0;
|
|
hub_count = 0;
|
|
log_batch_size = 0;
|
|
log_topic_func = nullptr;
|
|
log_id_type = nullptr;
|
|
writer_id_type = nullptr;
|
|
}
|
|
|
|
void Manager::DoInitPostScript() {
|
|
DBG_LOG(DBG_BROKER, "Initializing");
|
|
|
|
log_batch_size = get_option("Broker::log_batch_size")->AsCount();
|
|
default_log_topic_prefix = get_option("Broker::default_log_topic_prefix")->AsString()->CheckString();
|
|
log_topic_func = get_option("Broker::log_topic")->AsFunc();
|
|
log_id_type = id::find_type("Log::ID")->AsEnumType();
|
|
writer_id_type = id::find_type("Log::Writer")->AsEnumType();
|
|
zeek_table_manager = get_option("Broker::table_store_master")->AsBool();
|
|
zeek_table_db_directory = get_option("Broker::table_store_db_directory")->AsString()->CheckString();
|
|
|
|
// If Zeek's forwarding of network time to wallclock time was disabled,
|
|
// assume that also Broker does not use realtime and instead receives
|
|
// time via explicit AdvanceTime() calls.
|
|
if ( ! get_option("allow_network_time_forward")->AsBool() )
|
|
use_real_time = false;
|
|
|
|
detail::opaque_of_data_type = make_intrusive<OpaqueType>("Broker::Data");
|
|
detail::opaque_of_set_iterator = make_intrusive<OpaqueType>("Broker::SetIterator");
|
|
detail::opaque_of_table_iterator = make_intrusive<OpaqueType>("Broker::TableIterator");
|
|
detail::opaque_of_vector_iterator = make_intrusive<OpaqueType>("Broker::VectorIterator");
|
|
detail::opaque_of_record_iterator = make_intrusive<OpaqueType>("Broker::RecordIterator");
|
|
detail::opaque_of_store_handle = make_intrusive<OpaqueType>("Broker::Store");
|
|
vector_of_data_type = make_intrusive<VectorType>(id::find_type("Broker::Data"));
|
|
|
|
// Register as a "dont-count" source first, we may change that later.
|
|
iosource_mgr->Register(this, true);
|
|
|
|
broker::broker_options options;
|
|
options.disable_ssl = get_option("Broker::disable_ssl")->AsBool();
|
|
options.skip_ssl_init = true;
|
|
options.disable_forwarding = ! get_option("Broker::forward_messages")->AsBool();
|
|
options.use_real_time = use_real_time;
|
|
|
|
options.peer_buffer_size = get_option("Broker::peer_buffer_size")->AsCount();
|
|
auto peer_overflow_policy = get_option("Broker::peer_overflow_policy")->AsString()->CheckString();
|
|
if ( util::streq(peer_overflow_policy, "disconnect") ) {
|
|
options.peer_overflow_policy = broker::overflow_policy::disconnect;
|
|
}
|
|
else if ( util::streq(peer_overflow_policy, "drop_oldest") ) {
|
|
options.peer_overflow_policy = broker::overflow_policy::drop_oldest;
|
|
}
|
|
else if ( util::streq(peer_overflow_policy, "drop_newest") ) {
|
|
options.peer_overflow_policy = broker::overflow_policy::drop_newest;
|
|
}
|
|
else {
|
|
reporter->FatalError("Invalid Broker::peer_overflow_policy: %s", peer_overflow_policy);
|
|
}
|
|
|
|
options.web_socket_buffer_size = get_option("Broker::web_socket_buffer_size")->AsCount();
|
|
auto web_socket_overflow_policy = get_option("Broker::web_socket_overflow_policy")->AsString()->CheckString();
|
|
if ( util::streq(web_socket_overflow_policy, "disconnect") ) {
|
|
options.web_socket_overflow_policy = broker::overflow_policy::disconnect;
|
|
}
|
|
else if ( util::streq(web_socket_overflow_policy, "drop_oldest") ) {
|
|
options.web_socket_overflow_policy = broker::overflow_policy::drop_oldest;
|
|
}
|
|
else if ( util::streq(web_socket_overflow_policy, "drop_newest") ) {
|
|
options.web_socket_overflow_policy = broker::overflow_policy::drop_newest;
|
|
}
|
|
else {
|
|
reporter->FatalError("Invalid Broker::web_socket_overflow_policy: %s", web_socket_overflow_policy);
|
|
}
|
|
|
|
broker::configuration config{std::move(options)};
|
|
|
|
config.openssl_cafile(get_option("Broker::ssl_cafile")->AsString()->CheckString());
|
|
config.openssl_capath(get_option("Broker::ssl_capath")->AsString()->CheckString());
|
|
config.openssl_certificate(get_option("Broker::ssl_certificate")->AsString()->CheckString());
|
|
config.openssl_key(get_option("Broker::ssl_keyfile")->AsString()->CheckString());
|
|
config.openssl_passphrase(get_option("Broker::ssl_passphrase")->AsString()->CheckString());
|
|
|
|
auto scheduler_policy = get_option("Broker::scheduler_policy")->AsString()->CheckString();
|
|
|
|
if ( util::streq(scheduler_policy, "sharing") )
|
|
config.set("caf.scheduler.policy", "sharing");
|
|
else if ( util::streq(scheduler_policy, "stealing") )
|
|
config.set("caf.scheduler.policy", "stealing");
|
|
else
|
|
reporter->FatalError("Invalid Broker::scheduler_policy: %s", scheduler_policy);
|
|
|
|
auto max_threads_env = getenv("ZEEK_BROKER_MAX_THREADS");
|
|
|
|
if ( max_threads_env )
|
|
config.set("caf.scheduler.max-threads", atoi(max_threads_env));
|
|
else
|
|
config.set("caf.scheduler.max-threads", get_option("Broker::max_threads")->AsCount());
|
|
|
|
config.set("caf.work-stealing.moderate-sleep-duration",
|
|
broker::timespan(static_cast<unsigned>(get_option("Broker::moderate_sleep")->AsInterval() * 1e9)));
|
|
|
|
config.set("caf.work-stealing.relaxed-sleep-duration",
|
|
broker::timespan(static_cast<unsigned>(get_option("Broker::relaxed_sleep")->AsInterval() * 1e9)));
|
|
|
|
config.set("caf.work-stealing.aggressive-poll-attempts", get_option("Broker::aggressive_polls")->AsCount());
|
|
config.set("caf.work-stealing.moderate-poll-attempts", get_option("Broker::moderate_polls")->AsCount());
|
|
|
|
config.set("caf.work-stealing.aggressive-steal-interval", get_option("Broker::aggressive_interval")->AsCount());
|
|
config.set("caf.work-stealing.moderate-steal-interval", get_option("Broker::moderate_interval")->AsCount());
|
|
config.set("caf.work-stealing.relaxed-steal-interval", get_option("Broker::relaxed_interval")->AsCount());
|
|
|
|
// Hook up the logger.
|
|
auto checkLogSeverity = [](int level) {
|
|
if ( level < 0 || level > static_cast<int>(BrokerSeverityLevel::debug) ) {
|
|
reporter->FatalError("Invalid Broker::log_severity_level: %d", level);
|
|
}
|
|
};
|
|
auto logSeverityVal = static_cast<int>(get_option("Broker::log_severity_level")->AsEnum());
|
|
checkLogSeverity(logSeverityVal);
|
|
auto stderrSeverityVal = static_cast<int>(get_option("Broker::log_stderr_severity_level")->AsEnum());
|
|
checkLogSeverity(stderrSeverityVal);
|
|
auto adapterVerbosity = static_cast<BrokerSeverityLevel>(std::max(logSeverityVal, stderrSeverityVal));
|
|
auto queue = std::make_shared<LoggerQueue>();
|
|
auto pbstate = std::make_shared<PeerBufferState>(get_option("Broker::peer_buffer_size")->AsCount(),
|
|
get_option("Broker::buffer_stats_reset_interval")->AsDouble());
|
|
auto observer = std::make_shared<Observer>(adapterVerbosity, queue, pbstate);
|
|
broker::logger(observer); // *must* be called before creating the BrokerState
|
|
|
|
bstate = std::make_shared<BrokerState>(std::move(config), queue, pbstate);
|
|
bstate->logSeverity = static_cast<BrokerSeverityLevel>(logSeverityVal);
|
|
bstate->stderrSeverity = static_cast<BrokerSeverityLevel>(stderrSeverityVal);
|
|
|
|
if ( ! iosource_mgr->RegisterFd(bstate->subscriber.fd(), this) )
|
|
reporter->FatalError("Failed to register broker subscriber with iosource_mgr");
|
|
|
|
if ( ! iosource_mgr->RegisterFd(queue->FlareFd(), this) )
|
|
reporter->FatalError("Failed to register broker logger with iosource_mgr");
|
|
|
|
bstate->subscriber.add_topic(broker::topic::store_events(), true);
|
|
|
|
SetNodeId(broker::to_string(bstate->endpoint.node_id()));
|
|
|
|
InitializeBrokerStoreForwarding();
|
|
|
|
num_peers_metric =
|
|
telemetry_mgr->GaugeInstance("zeek", "broker_peers", {}, "Current number of peers connected via broker", "",
|
|
[]() { return static_cast<double>(broker_mgr->peer_count); });
|
|
|
|
num_stores_metric =
|
|
telemetry_mgr->GaugeInstance("zeek", "broker_stores", {}, "Current number of stores connected via broker", "",
|
|
[]() { return static_cast<double>(broker_mgr->data_stores.size()); });
|
|
|
|
num_pending_queries_metric =
|
|
telemetry_mgr->GaugeInstance("zeek", "broker_pending_queries", {}, "Current number of pending broker queries",
|
|
"", []() { return static_cast<double>(broker_mgr->pending_queries.size()); });
|
|
|
|
num_events_incoming_metric = telemetry_mgr->CounterInstance("zeek", "broker_incoming_events", {},
|
|
"Total number of incoming events via broker");
|
|
num_events_outgoing_metric = telemetry_mgr->CounterInstance("zeek", "broker_outgoing_events", {},
|
|
"Total number of outgoing events via broker");
|
|
num_logs_incoming_metric =
|
|
telemetry_mgr->CounterInstance("zeek", "broker_incoming_logs", {}, "Total number of incoming logs via broker");
|
|
num_logs_outgoing_metric =
|
|
telemetry_mgr->CounterInstance("zeek", "broker_outgoing_logs", {}, "Total number of outgoing logs via broker");
|
|
num_ids_incoming_metric =
|
|
telemetry_mgr->CounterInstance("zeek", "broker_incoming_ids", {}, "Total number of incoming ids via broker");
|
|
num_ids_outgoing_metric =
|
|
telemetry_mgr->CounterInstance("zeek", "broker_outgoing_ids", {}, "Total number of outgoing ids via broker");
|
|
}
|
|
|
|
void Manager::InitializeBrokerStoreForwarding() {
|
|
const auto& globals = zeek::detail::global_scope()->Vars();
|
|
|
|
for ( const auto& global : globals ) {
|
|
auto& id = global.second;
|
|
if ( id->HasVal() && id->GetAttr(zeek::detail::ATTR_BACKEND) ) {
|
|
const auto& attr = id->GetAttr(zeek::detail::ATTR_BACKEND);
|
|
auto e = static_cast<BifEnum::Broker::BackendType>(attr->GetExpr()->Eval(nullptr)->AsEnum());
|
|
auto storename = std::string("___sync_store_") + global.first;
|
|
id->GetVal()->AsTableVal()->SetBrokerStore(storename);
|
|
AddForwardedStore(storename, cast_intrusive<TableVal>(id->GetVal()));
|
|
|
|
// We only create masters here. For clones, we do all the work of setting up
|
|
// the forwarding - but we do not try to initialize the clone. We can only initialize
|
|
// the clone, once a node has a connection to a master. This is currently done in
|
|
// scriptland in scripts/base/frameworks/cluster/broker-stores.zeek. Once the ALM
|
|
// transport is ready we can change over to doing this here.
|
|
if ( ! zeek_table_manager )
|
|
continue;
|
|
|
|
auto backend = detail::to_backend_type(e);
|
|
auto suffix = ".store";
|
|
|
|
switch ( backend ) {
|
|
case broker::backend::sqlite: suffix = ".sqlite"; break;
|
|
default: break;
|
|
}
|
|
|
|
auto path = zeek_table_db_directory + "/" + storename + suffix;
|
|
|
|
MakeMaster(storename, backend, broker::backend_options{{"path", path}});
|
|
}
|
|
}
|
|
}
|
|
|
|
void Manager::DoTerminate() {
|
|
FlushLogBuffers();
|
|
|
|
iosource_mgr->UnregisterFd(bstate->subscriber.fd(), this);
|
|
|
|
iosource_mgr->UnregisterFd(bstate->loggerQueue->FlareFd(), this);
|
|
|
|
vector<string> stores_to_close;
|
|
|
|
for ( auto& x : data_stores )
|
|
stores_to_close.push_back(x.first);
|
|
|
|
for ( auto& x : stores_to_close )
|
|
// This doesn't loop directly over data_stores, because CloseStore
|
|
// modifies the map and invalidates iterators.
|
|
CloseStore(x);
|
|
|
|
ProcessLogEvents();
|
|
|
|
FlushLogBuffers();
|
|
}
|
|
|
|
bool Manager::Active() {
|
|
if ( bstate->endpoint.is_shutdown() )
|
|
return false;
|
|
|
|
if ( bound_port > 0 )
|
|
return true;
|
|
|
|
return peer_count > 0 || hub_count > 0;
|
|
}
|
|
|
|
void Manager::AdvanceTime(double seconds_since_unix_epoch) {
|
|
if ( bstate->endpoint.is_shutdown() )
|
|
return;
|
|
|
|
if ( bstate->endpoint.use_real_time() )
|
|
return;
|
|
|
|
auto secs = std::chrono::duration<double>(seconds_since_unix_epoch);
|
|
auto span = std::chrono::duration_cast<broker::timespan>(secs);
|
|
broker::timestamp next_time{span};
|
|
bstate->endpoint.advance_time(next_time);
|
|
}
|
|
|
|
void Manager::FlushPendingQueries() {
|
|
while ( ! pending_queries.empty() ) {
|
|
// possibly an infinite loop if a query can recursively
|
|
// generate more queries...
|
|
for ( auto& s : data_stores ) {
|
|
while ( ! s.second->proxy.mailbox().empty() ) {
|
|
auto response = s.second->proxy.receive();
|
|
ProcessStoreResponse(s.second, std::move(response));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
void Manager::ClearStores() {
|
|
FlushPendingQueries();
|
|
|
|
for ( const auto& [name, handle] : data_stores )
|
|
handle->store.clear();
|
|
}
|
|
|
|
uint16_t Manager::Listen(const string& addr, uint16_t port, BrokerProtocol type) {
|
|
if ( bstate->endpoint.is_shutdown() )
|
|
return 0;
|
|
|
|
switch ( type ) {
|
|
case BrokerProtocol::Native: bound_port = bstate->endpoint.listen(addr, port); break;
|
|
|
|
case BrokerProtocol::WebSocket: bound_port = bstate->endpoint.web_socket_listen(addr, port); break;
|
|
}
|
|
|
|
if ( bound_port == 0 )
|
|
Error("Failed to listen on %s:%" PRIu16, addr.empty() ? "INADDR_ANY" : addr.c_str(), port);
|
|
|
|
// Register as a "does-count" source now.
|
|
iosource_mgr->Register(this, false);
|
|
|
|
DBG_LOG(DBG_BROKER, "Listening on %s:%" PRIu16, addr.empty() ? "INADDR_ANY" : addr.c_str(), port);
|
|
|
|
return bound_port;
|
|
}
|
|
|
|
void Manager::Peer(const string& addr, uint16_t port, double retry) {
|
|
if ( bstate->endpoint.is_shutdown() )
|
|
return;
|
|
|
|
DBG_LOG(DBG_BROKER, "Starting to peer with %s:%" PRIu16 " (retry: %fs)", addr.c_str(), port, retry);
|
|
|
|
auto e = getenv("ZEEK_DEFAULT_CONNECT_RETRY");
|
|
|
|
if ( e )
|
|
retry = atoi(e);
|
|
|
|
if ( retry > 0.0 && retry < 1.0 )
|
|
// Ensure that it doesn't get turned into zero.
|
|
retry = 1.0;
|
|
|
|
auto secs = broker::timeout::seconds(static_cast<uint64_t>(retry));
|
|
bstate->endpoint.peer_nosync(addr, port, secs);
|
|
bstate->outbound_peerings.emplace(broker::network_info(addr, port));
|
|
|
|
auto counts_as_iosource = get_option("Broker::peer_counts_as_iosource")->AsBool();
|
|
|
|
if ( counts_as_iosource )
|
|
// Register as a "does-count" source now.
|
|
iosource_mgr->Register(this, false);
|
|
}
|
|
|
|
void Manager::PeerNoRetry(const string& addr, uint16_t port) {
|
|
if ( bstate->endpoint.is_shutdown() )
|
|
return;
|
|
|
|
DBG_LOG(DBG_BROKER, "Starting to peer with %s:%" PRIu16 " (no retry)", addr.c_str(), port);
|
|
|
|
bstate->endpoint.peer_nosync(addr, port, broker::timeout::seconds{0});
|
|
|
|
auto counts_as_iosource = get_option("Broker::peer_counts_as_iosource")->AsBool();
|
|
|
|
if ( counts_as_iosource )
|
|
// Register as a "does-count" source now.
|
|
iosource_mgr->Register(this, false);
|
|
}
|
|
|
|
void Manager::Unpeer(const string& addr, uint16_t port) {
|
|
if ( bstate->endpoint.is_shutdown() )
|
|
return;
|
|
|
|
DBG_LOG(DBG_BROKER, "Stopping to peer with %s:%" PRIu16, addr.c_str(), port);
|
|
|
|
FlushLogBuffers();
|
|
bstate->endpoint.unpeer_nosync(addr, port);
|
|
bstate->outbound_peerings.erase(broker::network_info(addr, port));
|
|
}
|
|
|
|
bool Manager::IsOutboundPeering(const string& addr, uint16_t port) const {
|
|
return bstate->outbound_peerings.find(broker::network_info(addr, port)) != bstate->outbound_peerings.end();
|
|
}
|
|
|
|
bool Manager::IsOutboundPeering(const broker::network_info& ni) const {
|
|
return bstate->outbound_peerings.find(ni) != bstate->outbound_peerings.end();
|
|
}
|
|
|
|
std::vector<broker::peer_info> Manager::Peers() const {
|
|
if ( bstate->endpoint.is_shutdown() )
|
|
return {};
|
|
|
|
return bstate->endpoint.peers();
|
|
}
|
|
|
|
std::string Manager::NodeID() const { return to_string(bstate->endpoint.node_id()); }
|
|
|
|
bool Manager::DoPublishEvent(const std::string& topic, cluster::detail::Event& event) {
|
|
bool do_publish = PLUGIN_HOOK_WITH_RESULT(HOOK_PUBLISH_EVENT, HookPublishEvent(*this, topic, event), true);
|
|
if ( ! do_publish )
|
|
return true;
|
|
|
|
auto maybe_ev = zeek::cluster::detail::to_broker_event(event);
|
|
if ( ! maybe_ev )
|
|
return false;
|
|
|
|
auto& ev = maybe_ev.value();
|
|
|
|
DBG_LOG(DBG_BROKER, "Publishing event: %s", RenderEvent(topic, std::string(ev.name()), ev.args()).c_str());
|
|
bstate->endpoint.publish(topic, ev.move_data());
|
|
num_events_outgoing_metric->Inc();
|
|
return true;
|
|
}
|
|
|
|
bool Manager::PublishEvent(string topic, std::string name, broker::vector args, double ts) {
|
|
if ( bstate->endpoint.is_shutdown() )
|
|
return true;
|
|
|
|
if ( peer_count == 0 && hub_count == 0 )
|
|
return true;
|
|
|
|
broker::vector meta;
|
|
if ( BifConst::EventMetadata::add_network_timestamp ) {
|
|
broker::vector entry{static_cast<broker::count>(zeek::detail::MetadataType::NetworkTimestamp),
|
|
broker::to_timestamp(ts)};
|
|
meta.emplace_back(std::move(entry));
|
|
}
|
|
|
|
broker::zeek::Event ev(name, args, meta);
|
|
DBG_LOG(DBG_BROKER, "Publishing event: %s", RenderEvent(topic, std::string(ev.name()), ev.args()).c_str());
|
|
bstate->endpoint.publish(std::move(topic), ev.move_data());
|
|
num_events_outgoing_metric->Inc();
|
|
return true;
|
|
}
|
|
|
|
bool Manager::PublishEvent(string topic, RecordVal* args) {
|
|
if ( bstate->endpoint.is_shutdown() )
|
|
return true;
|
|
|
|
if ( peer_count == 0 && hub_count == 0 )
|
|
return true;
|
|
|
|
if ( ! args->HasField(0) )
|
|
return false;
|
|
|
|
auto event_name = args->GetFieldAs<StringVal>(0)->CheckString();
|
|
auto vv = args->GetFieldAs<VectorVal>(1);
|
|
broker::vector xs;
|
|
xs.reserve(vv->Size());
|
|
|
|
for ( auto i = 0u; i < vv->Size(); ++i ) {
|
|
const auto& val = vv->RecordValAt(i)->GetField(0);
|
|
auto data_val = static_cast<detail::DataVal*>(val.get());
|
|
xs.emplace_back(data_val->data);
|
|
}
|
|
|
|
// At this point we come from script-land. This means that publishing of the event was
|
|
// explicitly triggered. Hence, the timestamp is set to the current event's time. This
|
|
// also means that timestamping cannot be manipulated from script-land for now.
|
|
auto ts = event_mgr.CurrentEventTime();
|
|
return PublishEvent(std::move(topic), event_name, std::move(xs), ts);
|
|
}
|
|
|
|
bool Manager::PublishIdentifier(std::string topic, std::string id) {
|
|
if ( bstate->endpoint.is_shutdown() )
|
|
return true;
|
|
|
|
if ( peer_count == 0 )
|
|
return true;
|
|
|
|
const auto& i = zeek::detail::global_scope()->Find(id);
|
|
|
|
if ( ! i )
|
|
return false;
|
|
|
|
const auto& val = i->GetVal();
|
|
|
|
if ( ! val )
|
|
// Probably could have a special case to also unset the value on the
|
|
// receiving side, but not sure what use that would be.
|
|
return false;
|
|
|
|
auto data = BrokerData{};
|
|
|
|
if ( ! data.Convert(val) ) {
|
|
Error("Failed to publish ID with unsupported type: %s (%s)", id.c_str(), type_name(val->GetType()->Tag()));
|
|
return false;
|
|
}
|
|
|
|
broker::zeek::IdentifierUpdate msg(std::move(id), std::move(data.value_));
|
|
DBG_LOG(DBG_BROKER, "Publishing id-update: %s", RenderMessage(topic, msg.as_data()).c_str());
|
|
bstate->endpoint.publish(std::move(topic), msg.move_data());
|
|
num_ids_outgoing_metric->Inc();
|
|
return true;
|
|
}
|
|
|
|
bool Manager::PublishLogCreate(EnumVal* stream, EnumVal* writer, const logging::WriterBackend::WriterInfo& info,
|
|
int num_fields, const threading::Field* const* fields,
|
|
const broker::endpoint_info& peer) {
|
|
if ( bstate->endpoint.is_shutdown() )
|
|
return true;
|
|
|
|
if ( peer_count == 0 )
|
|
return true;
|
|
|
|
auto stream_id = stream->GetType()->AsEnumType()->Lookup(stream->AsEnum());
|
|
|
|
if ( ! stream_id ) {
|
|
reporter->Error("Failed to remotely log: stream %" PRId64 " doesn't have name", stream->AsEnum());
|
|
return false;
|
|
}
|
|
|
|
auto writer_id = writer->GetType()->AsEnumType()->Lookup(writer->AsEnum());
|
|
|
|
if ( ! writer_id ) {
|
|
reporter->Error("Failed to remotely log: writer %" PRId64 " doesn't have name", writer->AsEnum());
|
|
return false;
|
|
}
|
|
|
|
auto writer_info = info.ToBroker();
|
|
|
|
broker::vector fields_data;
|
|
fields_data.reserve(num_fields);
|
|
|
|
for ( auto i = 0; i < num_fields; ++i ) {
|
|
auto field_data = detail::threading_field_to_data(fields[i]);
|
|
fields_data.push_back(std::move(field_data));
|
|
}
|
|
|
|
std::string topic = default_log_topic_prefix + stream_id;
|
|
auto bstream_id = broker::enum_value(std::move(stream_id));
|
|
auto bwriter_id = broker::enum_value(std::move(writer_id));
|
|
broker::zeek::LogCreate msg(std::move(bstream_id), std::move(bwriter_id), std::move(writer_info),
|
|
std::move(fields_data));
|
|
|
|
DBG_LOG(DBG_BROKER, "Publishing log creation: %s", RenderMessage(topic, msg.as_data()).c_str());
|
|
|
|
if ( peer.node != NoPeer.node )
|
|
// Direct message.
|
|
bstate->endpoint.publish(peer, std::move(topic), msg.move_data());
|
|
else
|
|
// Broadcast.
|
|
bstate->endpoint.publish(std::move(topic), msg.move_data());
|
|
|
|
return true;
|
|
}
|
|
|
|
bool Manager::PublishLogWrite(EnumVal* stream, EnumVal* writer, const string& path,
|
|
const logging::detail::LogRecord& rec) {
|
|
if ( bstate->endpoint.is_shutdown() )
|
|
return true;
|
|
|
|
if ( peer_count == 0 )
|
|
return true;
|
|
|
|
auto stream_id_num = stream->AsEnum();
|
|
auto stream_id = stream->GetType()->AsEnumType()->Lookup(stream_id_num);
|
|
|
|
if ( ! stream_id ) {
|
|
reporter->Error("Failed to remotely log: stream %" PRId64 " doesn't have name", stream->AsEnum());
|
|
return false;
|
|
}
|
|
|
|
auto writer_id = writer->GetType()->AsEnumType()->Lookup(writer->AsEnum());
|
|
|
|
if ( ! writer_id ) {
|
|
reporter->Error("Failed to remotely log: writer %" PRId64 " doesn't have name", writer->AsEnum());
|
|
return false;
|
|
}
|
|
|
|
zeek::detail::BinarySerializationFormat fmt;
|
|
char* data;
|
|
int len;
|
|
|
|
fmt.StartWrite();
|
|
|
|
// Cast to int for binary compatibility.
|
|
bool success = fmt.Write(static_cast<int>(rec.size()), "num_fields");
|
|
|
|
if ( ! success ) {
|
|
reporter->Error("Failed to remotely log stream %s: num_fields serialization failed", stream_id);
|
|
return false;
|
|
}
|
|
|
|
for ( size_t i = 0; i < rec.size(); ++i ) {
|
|
if ( ! rec[i].Write(&fmt) ) {
|
|
reporter->Error("Failed to remotely log stream %s: field %zu serialization failed", stream_id, i);
|
|
return false;
|
|
}
|
|
}
|
|
|
|
len = fmt.EndWrite(&data);
|
|
std::string serial_data(data, len);
|
|
free(data);
|
|
|
|
auto v = log_topic_func->Invoke(IntrusivePtr{NewRef{}, stream}, make_intrusive<StringVal>(path));
|
|
|
|
if ( ! v ) {
|
|
reporter->Error(
|
|
"Failed to remotely log: log_topic func did not return"
|
|
" a value for stream %s at path %s",
|
|
stream_id, std::string{path}.c_str());
|
|
return false;
|
|
}
|
|
|
|
std::string topic = v->AsString()->CheckString();
|
|
|
|
auto bstream_id = broker::enum_value(std::move(stream_id));
|
|
auto bwriter_id = broker::enum_value(std::move(writer_id));
|
|
broker::zeek::LogWrite msg(std::move(bstream_id), std::move(bwriter_id), std::move(path), std::move(serial_data));
|
|
|
|
DBG_LOG(DBG_BROKER, "Buffering log record: %s", RenderMessage(topic, msg.as_data()).c_str());
|
|
|
|
if ( log_buffers.size() <= (unsigned int)stream_id_num )
|
|
log_buffers.resize(stream_id_num + 1);
|
|
|
|
auto& lb = log_buffers[stream_id_num];
|
|
++lb.message_count;
|
|
lb.msgs[topic].add(std::move(msg));
|
|
|
|
if ( lb.message_count >= log_batch_size ) {
|
|
auto outgoing_logs = static_cast<double>(lb.Flush(bstate->endpoint, log_batch_size));
|
|
num_logs_outgoing_metric->Inc(outgoing_logs);
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
size_t Manager::LogBuffer::Flush(broker::endpoint& endpoint, size_t log_batch_size) {
|
|
if ( endpoint.is_shutdown() )
|
|
return 0;
|
|
|
|
if ( ! message_count )
|
|
// No logs buffered for this stream.
|
|
return 0;
|
|
|
|
for ( auto& [topic, pending_batch] : msgs ) {
|
|
if ( ! pending_batch.empty() )
|
|
endpoint.publish(topic, pending_batch.build());
|
|
}
|
|
|
|
auto rval = message_count;
|
|
message_count = 0;
|
|
return rval;
|
|
}
|
|
|
|
size_t Manager::FlushLogBuffers() {
|
|
DBG_LOG(DBG_BROKER, "Flushing all log buffers");
|
|
auto rval = 0u;
|
|
|
|
for ( auto& lb : log_buffers )
|
|
rval += lb.Flush(bstate->endpoint, log_batch_size);
|
|
|
|
num_logs_outgoing_metric->Inc(rval);
|
|
|
|
return rval;
|
|
}
|
|
|
|
void Manager::Error(const char* format, ...) {
|
|
va_list args;
|
|
va_start(args, format);
|
|
auto msg = util::vfmt(format, args);
|
|
va_end(args);
|
|
|
|
if ( script_scope )
|
|
emit_builtin_error(msg);
|
|
else
|
|
reporter->Error("%s", msg);
|
|
}
|
|
|
|
bool Manager::AutoPublishEvent(string topic, Val* event) {
|
|
if ( event->GetType()->Tag() != TYPE_FUNC ) {
|
|
Error("Broker::auto_publish must operate on an event");
|
|
return false;
|
|
}
|
|
|
|
auto event_val = event->AsFunc();
|
|
if ( event_val->Flavor() != FUNC_FLAVOR_EVENT ) {
|
|
Error("Broker::auto_publish must operate on an event");
|
|
return false;
|
|
}
|
|
|
|
auto handler = event_registry->Lookup(event_val->GetName());
|
|
if ( ! handler ) {
|
|
Error("Broker::auto_publish failed to lookup event '%s'", event_val->GetName().c_str());
|
|
return false;
|
|
}
|
|
|
|
DBG_LOG(DBG_BROKER, "Enabling auto-publishing of event %s to topic %s", handler->Name(), topic.c_str());
|
|
#pragma GCC diagnostic push
|
|
#pragma GCC diagnostic ignored "-Wdeprecated-declarations"
|
|
handler->AutoPublish(std::move(topic));
|
|
#pragma GCC diagnostic pop
|
|
|
|
return true;
|
|
}
|
|
|
|
bool Manager::AutoUnpublishEvent(const string& topic, Val* event) {
|
|
if ( event->GetType()->Tag() != TYPE_FUNC ) {
|
|
Error("Broker::auto_event_stop must operate on an event");
|
|
return false;
|
|
}
|
|
|
|
auto event_val = event->AsFunc();
|
|
|
|
if ( event_val->Flavor() != FUNC_FLAVOR_EVENT ) {
|
|
Error("Broker::auto_event_stop must operate on an event");
|
|
return false;
|
|
}
|
|
|
|
auto handler = event_registry->Lookup(event_val->GetName());
|
|
|
|
if ( ! handler ) {
|
|
Error("Broker::auto_event_stop failed to lookup event '%s'", event_val->GetName().c_str());
|
|
return false;
|
|
}
|
|
|
|
DBG_LOG(DBG_BROKER, "Disabling auto-publishing of event %s to topic %s", handler->Name(), topic.c_str());
|
|
#pragma GCC diagnostic push
|
|
#pragma GCC diagnostic ignored "-Wdeprecated-declarations"
|
|
handler->AutoUnpublish(topic);
|
|
#pragma GCC diagnostic pop
|
|
|
|
return true;
|
|
}
|
|
|
|
RecordVal* Manager::MakeEvent(ValPList* args, zeek::detail::Frame* frame) {
|
|
// Deprecated MakeEvent() version using ValPList - requires extra copy.
|
|
zeek::Args cargs;
|
|
cargs.reserve(args->size());
|
|
for ( auto* a : *args )
|
|
cargs.push_back({zeek::NewRef{}, a});
|
|
|
|
return MakeEvent(ArgsSpan{cargs}, frame)->Ref()->AsRecordVal();
|
|
}
|
|
|
|
zeek::RecordValPtr Manager::MakeEvent(ArgsSpan args, zeek::detail::Frame* frame) {
|
|
scoped_reporter_location srl{frame};
|
|
auto rval = zeek::make_intrusive<RecordVal>(BifType::Record::Broker::Event);
|
|
auto arg_vec = make_intrusive<VectorVal>(vector_of_data_type);
|
|
rval->Assign(1, arg_vec);
|
|
const Func* func = nullptr;
|
|
|
|
for ( size_t index = 0; index < args.size(); index++ ) {
|
|
const auto& arg_val = args[index];
|
|
if ( index == 0 ) {
|
|
// Event val must come first.
|
|
|
|
if ( arg_val->GetType()->Tag() != TYPE_FUNC ) {
|
|
Error("attempt to convert non-event into an event type (%s)",
|
|
zeek::obj_desc_short(arg_val.get()).c_str());
|
|
return rval;
|
|
}
|
|
|
|
func = arg_val->AsFunc();
|
|
|
|
if ( func->Flavor() != FUNC_FLAVOR_EVENT ) {
|
|
Error("attempt to convert non-event into an event type");
|
|
return rval;
|
|
}
|
|
|
|
auto num_args = static_cast<size_t>(func->GetType()->Params()->NumFields());
|
|
|
|
if ( num_args != args.size() - 1 ) {
|
|
Error("bad # of arguments: got %zu, expect %zu", args.size() - 1, num_args);
|
|
return rval;
|
|
}
|
|
|
|
rval->Assign(0, func->GetName());
|
|
continue;
|
|
}
|
|
|
|
auto got_type = arg_val->GetType();
|
|
const auto& expected_type = func->GetType()->ParamList()->GetTypes()[index - 1];
|
|
|
|
// If called with an unspecified table or set, adopt the expected type.
|
|
if ( got_type->Tag() == TYPE_TABLE && got_type->AsTableType()->IsUnspecifiedTable() )
|
|
if ( expected_type->Tag() == TYPE_TABLE && got_type->IsSet() == expected_type->IsSet() )
|
|
got_type = expected_type;
|
|
|
|
if ( ! same_type(got_type, expected_type) ) {
|
|
rval->Remove(0);
|
|
Error("event parameter #%zu type mismatch, got %s, expect %s", index,
|
|
obj_desc_short(got_type.get()).c_str(), obj_desc_short(expected_type.get()).c_str());
|
|
return rval;
|
|
}
|
|
|
|
RecordValPtr data_val;
|
|
|
|
if ( same_type(got_type, detail::DataVal::ScriptDataType()) )
|
|
data_val = {NewRef{}, arg_val->AsRecordVal()};
|
|
else
|
|
data_val = BrokerData::ToRecordVal(arg_val);
|
|
|
|
if ( ! data_val->HasField(0) ) {
|
|
rval->Remove(0);
|
|
Error("failed to convert param #%zu of type %s to broker data", index, type_name(got_type->Tag()));
|
|
return rval;
|
|
}
|
|
|
|
arg_vec->Assign(index - 1, std::move(data_val));
|
|
}
|
|
|
|
return rval;
|
|
}
|
|
|
|
bool Manager::DoSubscribe(const string& topic_prefix, SubscribeCallback cb) {
|
|
DBG_LOG(DBG_BROKER, "Subscribing to topic prefix %s", topic_prefix.c_str());
|
|
bstate->subscriber.add_topic(topic_prefix, ! run_state::detail::zeek_init_done);
|
|
|
|
if ( cb )
|
|
cb(topic_prefix, {CallbackStatus::NotImplemented});
|
|
|
|
return true;
|
|
}
|
|
|
|
bool Manager::Forward(string topic_prefix) {
|
|
for ( const auto& prefix : forwarded_prefixes )
|
|
if ( prefix == topic_prefix )
|
|
return false;
|
|
|
|
DBG_LOG(DBG_BROKER, "Forwarding topic prefix %s", topic_prefix.c_str());
|
|
Subscribe(topic_prefix);
|
|
forwarded_prefixes.emplace_back(std::move(topic_prefix));
|
|
return true;
|
|
}
|
|
|
|
bool Manager::DoUnsubscribe(const string& topic_prefix) {
|
|
for ( size_t i = 0; i < forwarded_prefixes.size(); ++i )
|
|
if ( forwarded_prefixes[i] == topic_prefix ) {
|
|
DBG_LOG(DBG_BROKER, "Unforwarding topic prefix %s", topic_prefix.c_str());
|
|
forwarded_prefixes.erase(forwarded_prefixes.begin() + i);
|
|
break;
|
|
}
|
|
|
|
DBG_LOG(DBG_BROKER, "Unsubscribing from topic prefix %s", topic_prefix.c_str());
|
|
bstate->subscriber.remove_topic(topic_prefix, ! run_state::detail::zeek_init_done);
|
|
return true;
|
|
}
|
|
|
|
void Manager::ProcessMessages() {
|
|
auto messages = bstate->subscriber.poll();
|
|
|
|
for ( auto& message : messages ) {
|
|
auto&& topic = broker::get_topic(message);
|
|
|
|
if ( broker::is_prefix(topic, broker::topic::statuses_str) ) {
|
|
if ( auto stat = broker::to<broker::status>(get_data(message)) ) {
|
|
ProcessStatus(*stat);
|
|
}
|
|
else {
|
|
auto str = to_string(message);
|
|
reporter->Warning("ignoring malformed Broker status event: %s", str.c_str());
|
|
}
|
|
continue;
|
|
}
|
|
|
|
if ( broker::is_prefix(topic, broker::topic::errors_str) ) {
|
|
if ( auto err = broker::to<broker::error>(get_data(message)) ) {
|
|
ProcessError(*err);
|
|
}
|
|
else {
|
|
auto str = to_string(message);
|
|
reporter->Warning("ignoring malformed Broker error event: %s", str.c_str());
|
|
}
|
|
continue;
|
|
}
|
|
|
|
if ( broker::is_prefix(topic, broker::topic::store_events_str) ) {
|
|
ProcessStoreEvent(broker::get_data(message).to_data());
|
|
continue;
|
|
}
|
|
|
|
try {
|
|
// Once we call a broker::move_* function, we force Broker to
|
|
// unshare the content of the message, i.e., copy the content to a
|
|
// different memory region if other threads keep references to the
|
|
// message. Since `topic` still points into the original memory
|
|
// region, we may no longer access it after this point.
|
|
auto topic_str = broker::get_topic_str(message);
|
|
broker::zeek::visit_as_message([this, topic_str](auto& msg) { ProcessMessage(topic_str, msg); },
|
|
std::move(message));
|
|
} catch ( std::runtime_error& e ) {
|
|
reporter->Warning("ignoring invalid Broker message: %s", +e.what());
|
|
continue;
|
|
}
|
|
}
|
|
}
|
|
|
|
namespace {
|
|
|
|
// Note: copied from Stmt.cc, might be worth to move to a common place.
|
|
EnumValPtr lookup_enum_val(const char* module_name, const char* name) {
|
|
const auto& id = zeek::detail::lookup_ID(name, module_name);
|
|
assert(id);
|
|
assert(id->IsEnumConst());
|
|
|
|
EnumType* et = id->GetType()->AsEnumType();
|
|
|
|
int index = et->Lookup(module_name, name);
|
|
assert(index >= 0);
|
|
|
|
return et->GetEnumVal(index);
|
|
}
|
|
|
|
} // namespace
|
|
|
|
void Manager::ProcessLogEvents() {
|
|
static auto ev_critical = lookup_enum_val("Broker", "LOG_CRITICAL");
|
|
static auto ev_error = lookup_enum_val("Broker", "LOG_ERROR");
|
|
static auto ev_warning = lookup_enum_val("Broker", "LOG_WARNING");
|
|
static auto ev_info = lookup_enum_val("Broker", "LOG_INFO");
|
|
static auto ev_verbose = lookup_enum_val("Broker", "LOG_VERBOSE");
|
|
static auto ev_debug = lookup_enum_val("Broker", "LOG_DEBUG");
|
|
|
|
auto evType = [](BrokerSeverityLevel lvl) {
|
|
switch ( lvl ) {
|
|
case BrokerSeverityLevel::critical: return ev_critical;
|
|
case BrokerSeverityLevel::error: return ev_error;
|
|
case BrokerSeverityLevel::warning: return ev_warning;
|
|
case BrokerSeverityLevel::info: return ev_info;
|
|
case BrokerSeverityLevel::verbose: return ev_verbose;
|
|
default: return ev_debug;
|
|
}
|
|
};
|
|
|
|
constexpr const char* severity_names_tbl[] = {"critical", "error", "warning", "info", "verbose", "debug"};
|
|
|
|
auto events = bstate->loggerQueue->Drain();
|
|
for ( auto& event : events ) {
|
|
auto severity = event->severity;
|
|
if ( bstate->logSeverity >= severity ) {
|
|
auto args = Args{};
|
|
args.reserve(3);
|
|
args.emplace_back(evType(severity));
|
|
args.emplace_back(make_intrusive<StringVal>(event->identifier));
|
|
args.emplace_back(make_intrusive<StringVal>(event->description));
|
|
event_mgr.Enqueue(::Broker::internal_log_event, std::move(args));
|
|
}
|
|
if ( bstate->stderrSeverity >= severity ) {
|
|
// Formatting the event->identifier string_view using "%.*s" - the explicit
|
|
// precision ".*" allows specifying the length of the following char* argument
|
|
// as string_views in general are not guaranteed to be null terminated.
|
|
fprintf(stderr, "[BROKER/%s] %.*s: %s\n", severity_names_tbl[static_cast<int>(severity)],
|
|
static_cast<int>(event->identifier.size()), event->identifier.data(), event->description.c_str());
|
|
}
|
|
}
|
|
}
|
|
|
|
void Manager::ProcessDataStore(detail::StoreHandleVal* store) {
|
|
auto num_available = store->proxy.mailbox().size();
|
|
|
|
if ( num_available > 0 ) {
|
|
auto responses = store->proxy.receive(num_available);
|
|
|
|
for ( auto& r : responses )
|
|
ProcessStoreResponse(store, std::move(r));
|
|
}
|
|
}
|
|
|
|
void Manager::ProcessDataStores() {
|
|
for ( auto& kvp : data_stores ) {
|
|
ProcessDataStore(kvp.second);
|
|
}
|
|
}
|
|
|
|
void Manager::ProcessFd(int fd, int flags) {
|
|
if ( fd == bstate->subscriber.fd() ) {
|
|
ProcessMessages();
|
|
}
|
|
else if ( fd == bstate->loggerQueue->FlareFd() ) {
|
|
ProcessLogEvents();
|
|
}
|
|
else {
|
|
for ( auto& kvp : data_stores ) {
|
|
if ( fd == kvp.second->proxy.mailbox().descriptor() ) {
|
|
ProcessDataStore(kvp.second);
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
void Manager::Process() {
|
|
ProcessMessages();
|
|
ProcessLogEvents();
|
|
ProcessDataStores();
|
|
}
|
|
|
|
void Manager::ProcessStoreEventInsertUpdate(const TableValPtr& table, const std::string& store_id,
|
|
const broker::data& key, const broker::data& data,
|
|
const broker::data& old_value, bool insert) {
|
|
auto type = "Insert";
|
|
if ( ! insert )
|
|
type = "Update";
|
|
|
|
if ( insert ) {
|
|
DBG_LOG(DBG_BROKER, "Store %s: Insert: %s:%s (%s:%s)", store_id.c_str(), to_string(key).c_str(),
|
|
to_string(data).c_str(), key.get_type_name(), data.get_type_name());
|
|
}
|
|
else {
|
|
DBG_LOG(DBG_BROKER, "Store %s: Update: %s->%s (%s)", store_id.c_str(), to_string(old_value).c_str(),
|
|
to_string(data).c_str(), data.get_type_name());
|
|
}
|
|
|
|
if ( table->GetType()->IsSet() && data.get_type() != broker::data::type::none ) {
|
|
reporter->Error("ProcessStoreEvent %s got %s when expecting set", type, data.get_type_name());
|
|
return;
|
|
}
|
|
|
|
const auto& its = table->GetType()->AsTableType()->GetIndexTypes();
|
|
ValPtr zeek_key;
|
|
auto key_copy = key;
|
|
if ( its.size() == 1 )
|
|
zeek_key = detail::data_to_val(key_copy, its[0].get());
|
|
else
|
|
zeek_key = detail::data_to_val(key_copy, table->GetType()->AsTableType()->GetIndices().get());
|
|
|
|
if ( ! zeek_key ) {
|
|
reporter->Error(
|
|
"ProcessStoreEvent %s: could not convert key \"%s\" for store \"%s\" while receiving "
|
|
"remote data. This probably means the tables have different types on different nodes.",
|
|
type, to_string(key).c_str(), store_id.c_str());
|
|
return;
|
|
}
|
|
|
|
if ( table->GetType()->IsSet() ) {
|
|
table->Assign(zeek_key, nullptr, false);
|
|
return;
|
|
}
|
|
|
|
// it is a table
|
|
auto data_copy = data;
|
|
auto zeek_value = detail::data_to_val(data_copy, table->GetType()->Yield().get());
|
|
if ( ! zeek_value ) {
|
|
reporter->Error(
|
|
"ProcessStoreEvent %s: could not convert value \"%s\" for key \"%s\" in "
|
|
"store \"%s\" while receiving remote data. This probably means the tables "
|
|
"have different types on different nodes.",
|
|
type, to_string(data).c_str(), to_string(key).c_str(), store_id.c_str());
|
|
return;
|
|
}
|
|
|
|
table->Assign(zeek_key, zeek_value, false);
|
|
}
|
|
|
|
void Manager::ProcessStoreEvent(broker::data msg) {
|
|
if ( auto insert = broker::store_event::insert::make(msg) ) {
|
|
auto storehandle = broker_mgr->LookupStore(insert.store_id());
|
|
if ( ! storehandle )
|
|
return;
|
|
|
|
const auto& table = storehandle->forward_to;
|
|
if ( ! table )
|
|
return;
|
|
|
|
// We sent this message. Ignore it.
|
|
if ( insert.publisher() == storehandle->store_pid )
|
|
return;
|
|
|
|
ProcessStoreEventInsertUpdate(table, insert.store_id(), insert.key(), insert.value(), {}, true);
|
|
}
|
|
else if ( auto update = broker::store_event::update::make(msg) ) {
|
|
auto storehandle = broker_mgr->LookupStore(update.store_id());
|
|
if ( ! storehandle )
|
|
return;
|
|
|
|
const auto& table = storehandle->forward_to;
|
|
if ( ! table )
|
|
return;
|
|
|
|
// We sent this message. Ignore it.
|
|
if ( update.publisher() == storehandle->store_pid )
|
|
return;
|
|
|
|
ProcessStoreEventInsertUpdate(table, update.store_id(), update.key(), update.new_value(), update.old_value(),
|
|
false);
|
|
}
|
|
else if ( auto erase = broker::store_event::erase::make(msg) ) {
|
|
auto storehandle = broker_mgr->LookupStore(erase.store_id());
|
|
if ( ! storehandle )
|
|
return;
|
|
|
|
auto table = storehandle->forward_to;
|
|
if ( ! table )
|
|
return;
|
|
|
|
// We sent this message. Ignore it.
|
|
if ( erase.publisher() == storehandle->store_pid )
|
|
return;
|
|
|
|
auto key = erase.key();
|
|
DBG_LOG(DBG_BROKER, "Store %s: Erase key %s", erase.store_id().c_str(), to_string(key).c_str());
|
|
|
|
const auto& its = table->GetType()->AsTableType()->GetIndexTypes();
|
|
ValPtr zeek_key;
|
|
if ( its.size() == 1 )
|
|
zeek_key = detail::data_to_val(key, its[0].get());
|
|
else
|
|
zeek_key = detail::data_to_val(key, table->GetType()->AsTableType()->GetIndices().get());
|
|
|
|
if ( ! zeek_key ) {
|
|
reporter->Error(
|
|
"ProcessStoreEvent: could not convert key \"%s\" for store \"%s\" "
|
|
"while receiving remote erase. This probably means the tables have "
|
|
"different types on different nodes.",
|
|
to_string(key).c_str(), insert.store_id().c_str());
|
|
return;
|
|
}
|
|
|
|
table->Remove(*zeek_key, false);
|
|
}
|
|
else if ( auto expire = broker::store_event::expire::make(msg) ) {
|
|
// We just ignore expiries - expiring information on the Zeek side is handled by Zeek
|
|
// itself.
|
|
#ifdef DEBUG
|
|
// let's only debug log for stores that we know.
|
|
auto storehandle = broker_mgr->LookupStore(expire.store_id());
|
|
if ( ! storehandle )
|
|
return;
|
|
|
|
auto table = storehandle->forward_to;
|
|
if ( ! table )
|
|
return;
|
|
|
|
DBG_LOG(DBG_BROKER, "Store %s: Store expired key %s", expire.store_id().c_str(),
|
|
to_string(expire.key()).c_str());
|
|
#endif /* DEBUG */
|
|
}
|
|
else {
|
|
reporter->Error("ProcessStoreEvent: Unhandled event type");
|
|
}
|
|
}
|
|
|
|
void Manager::ProcessMessage(std::string_view topic, broker::zeek::Invalid& ev) {
|
|
reporter->Warning("received invalid broker message: %s", broker::to_string(ev).c_str());
|
|
}
|
|
|
|
void Manager::ProcessMessage(std::string_view topic, broker::zeek::Batch& ev) {
|
|
ev.for_each([this, topic](auto& inner) { ProcessMessage(topic, inner); });
|
|
}
|
|
|
|
void Manager::ProcessMessage(std::string_view topic, broker::zeek::Event& ev) {
|
|
if ( ! ev.valid() ) {
|
|
reporter->Warning("received invalid broker Event: %s", broker::to_string(ev.as_data()).c_str());
|
|
return;
|
|
}
|
|
|
|
auto&& name = ev.name();
|
|
auto&& args = ev.args();
|
|
auto meta = cluster::detail::metadata_vector_from_broker_event(ev);
|
|
|
|
DBG_LOG(DBG_BROKER, "Process event: %s (with %zu metadata entries) %s", std::string{name}.c_str(),
|
|
meta ? meta->size() : 0, RenderMessage(args).c_str());
|
|
num_events_incoming_metric->Inc();
|
|
auto handler = event_registry->Lookup(name);
|
|
|
|
if ( ! handler )
|
|
return;
|
|
|
|
for ( const auto& p : forwarded_prefixes ) {
|
|
if ( p.size() > topic.size() )
|
|
continue;
|
|
|
|
if ( strncmp(p.data(), topic.data(), p.size()) != 0 )
|
|
continue;
|
|
|
|
DBG_LOG(DBG_BROKER, "Skip processing of forwarded event: %s %s", std::string{name}.c_str(),
|
|
RenderMessage(args).c_str());
|
|
return;
|
|
}
|
|
|
|
const auto& arg_types = handler->GetType(false)->ParamList()->GetTypes();
|
|
|
|
if ( arg_types.size() != args.size() ) {
|
|
reporter->Warning(
|
|
"got event message '%s' with invalid # of args,"
|
|
" got %zd, expected %zu",
|
|
std::string{name}.c_str(), args.size(), arg_types.size());
|
|
return;
|
|
}
|
|
|
|
Args vl;
|
|
vl.reserve(args.size());
|
|
|
|
for ( size_t i = 0; i < args.size(); ++i ) {
|
|
auto got_type = args[i].get_type_name();
|
|
const auto& expected_type = arg_types[i];
|
|
auto arg = args[i].to_data();
|
|
auto val = detail::data_to_val(arg, expected_type.get());
|
|
|
|
if ( val )
|
|
vl.emplace_back(std::move(val));
|
|
else {
|
|
auto expected_name = type_name(expected_type->Tag());
|
|
std::string msg_addl = util::fmt("got %s, expected %s", got_type, expected_name);
|
|
|
|
if ( strcmp(expected_name, "record") == 0 && strcmp("vector", got_type) == 0 ) {
|
|
// This means the vector elements didn't align with the record
|
|
// fields. Produce an error message that shows what we
|
|
// received.
|
|
std::string elements;
|
|
for ( auto&& e : broker_vector_from(args[i]) ) {
|
|
if ( ! elements.empty() )
|
|
elements += ", ";
|
|
|
|
elements += e.get_type_name();
|
|
}
|
|
|
|
msg_addl = util::fmt("got mismatching field types [%s] for record type '%s'", elements.c_str(),
|
|
expected_type->GetName().c_str());
|
|
}
|
|
|
|
reporter->Warning("failed to convert remote event '%s' arg #%zu, %s", std::string{name}.c_str(), i + 1,
|
|
msg_addl.c_str());
|
|
|
|
// If we got a vector and expected a function this is
|
|
// possibly because of a mismatch between
|
|
// anonymous-function bodies.
|
|
if ( strcmp(expected_name, "func") == 0 && strcmp("vector", got_type) == 0 )
|
|
reporter->Warning(
|
|
"when sending functions the receiver must have access to a"
|
|
" version of that function.\nFor anonymous functions, that "
|
|
"function must have the same body.");
|
|
|
|
break;
|
|
}
|
|
}
|
|
|
|
if ( vl.size() == args.size() )
|
|
event_mgr.Enqueue(std::move(meta), handler, std::move(vl), util::detail::SOURCE_BROKER);
|
|
}
|
|
|
|
bool Manager::ProcessMessage(std::string_view, broker::zeek::LogCreate& lc) {
|
|
DBG_LOG(DBG_BROKER, "Received log-create: %s", RenderMessage(lc.as_data()).c_str());
|
|
if ( ! lc.valid() ) {
|
|
reporter->Warning("received invalid broker LogCreate: %s", broker::to_string(lc.as_data()).c_str());
|
|
return false;
|
|
}
|
|
|
|
auto wrapped_stream_id = broker::data{lc.stream_id()};
|
|
auto stream_id = detail::data_to_val(wrapped_stream_id, log_id_type);
|
|
if ( ! stream_id ) {
|
|
reporter->Warning("failed to unpack remote log stream id");
|
|
return false;
|
|
}
|
|
|
|
auto wrapped_writer_id = broker::data{lc.writer_id()};
|
|
auto writer_id = detail::data_to_val(wrapped_writer_id, writer_id_type);
|
|
if ( ! writer_id ) {
|
|
reporter->Warning("failed to unpack remote log writer id");
|
|
return false;
|
|
}
|
|
|
|
auto writer_info = std::make_unique<logging::WriterBackend::WriterInfo>();
|
|
if ( ! writer_info->FromBroker(lc.writer_info().to_data()) ) {
|
|
reporter->Warning("failed to unpack remote log writer info");
|
|
return false;
|
|
}
|
|
|
|
// Get log fields.
|
|
if ( ! lc.fields_data().is_list() ) {
|
|
reporter->Warning("failed to unpack remote log fields");
|
|
return false;
|
|
}
|
|
auto&& fields_data = broker_vector_from(lc.fields_data());
|
|
|
|
auto num_fields = fields_data.size();
|
|
auto fields = new threading::Field*[num_fields];
|
|
|
|
for ( size_t i = 0; i < num_fields; ++i ) {
|
|
if ( auto field = detail::data_to_threading_field(fields_data[i]) )
|
|
fields[i] = field;
|
|
else {
|
|
reporter->Warning("failed to convert remote log field #%zu: %s", i,
|
|
broker::to_string(fields_data[i]).c_str());
|
|
delete[] fields;
|
|
return false;
|
|
}
|
|
}
|
|
|
|
if ( ! log_mgr->CreateWriterForRemoteLog(stream_id->AsEnumVal(), writer_id->AsEnumVal(), writer_info.release(),
|
|
num_fields, fields) ) {
|
|
ODesc d;
|
|
stream_id->Describe(&d);
|
|
reporter->Warning("failed to create remote log stream for %s locally", d.Description());
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
bool Manager::ProcessMessage(std::string_view, broker::zeek::LogWrite& lw) {
|
|
DBG_LOG(DBG_BROKER, "Received log-write: %s", RenderMessage(lw.as_data()).c_str());
|
|
|
|
if ( ! lw.valid() ) {
|
|
reporter->Warning("received invalid broker LogWrite: %s", broker::to_string(lw.as_data()).c_str());
|
|
return false;
|
|
}
|
|
|
|
num_logs_incoming_metric->Inc();
|
|
auto&& stream_id_name = lw.stream_id().name;
|
|
|
|
// Get stream ID.
|
|
auto wrapped_stream_id = broker::data{lw.stream_id()};
|
|
auto stream_id = detail::data_to_val(wrapped_stream_id, log_id_type);
|
|
|
|
if ( ! stream_id ) {
|
|
reporter->Warning("failed to unpack remote log stream id: %s", std::string{stream_id_name}.c_str());
|
|
return false;
|
|
}
|
|
|
|
// Get writer ID.
|
|
auto wrapped_writer_id = broker::data{lw.writer_id()};
|
|
auto writer_id = detail::data_to_val(wrapped_writer_id, writer_id_type);
|
|
if ( ! writer_id ) {
|
|
reporter->Warning("failed to unpack remote log writer id for stream: %s", std::string{stream_id_name}.c_str());
|
|
return false;
|
|
}
|
|
|
|
auto path = std::string{lw.path_str()};
|
|
|
|
auto serial_data = lw.serial_data_str();
|
|
|
|
zeek::detail::BinarySerializationFormat fmt;
|
|
fmt.StartRead(serial_data.data(), serial_data.size());
|
|
|
|
int num_fields;
|
|
bool success = fmt.Read(&num_fields, "num_fields");
|
|
|
|
if ( ! success ) {
|
|
reporter->Warning("failed to unserialize remote log num fields for stream: %s",
|
|
std::string{stream_id_name}.c_str());
|
|
return false;
|
|
}
|
|
|
|
logging::detail::LogRecord rec(num_fields);
|
|
|
|
for ( int i = 0; i < num_fields; ++i ) {
|
|
if ( ! rec[i].Read(&fmt) ) {
|
|
reporter->Warning("failed to unserialize remote log field %d for stream: %s", i,
|
|
std::string{stream_id_name}.c_str());
|
|
|
|
return false;
|
|
}
|
|
}
|
|
|
|
log_mgr->WriteFromRemote(stream_id->AsEnumVal(), writer_id->AsEnumVal(), path, std::move(rec));
|
|
fmt.EndRead();
|
|
return true;
|
|
}
|
|
|
|
bool Manager::ProcessMessage(std::string_view, broker::zeek::IdentifierUpdate& iu) {
|
|
DBG_LOG(DBG_BROKER, "Received id-update: %s", RenderMessage(iu.as_data()).c_str());
|
|
|
|
if ( ! iu.valid() ) {
|
|
reporter->Warning("received invalid broker IdentifierUpdate: %s", broker::to_string(iu.as_data()).c_str());
|
|
return false;
|
|
}
|
|
|
|
num_ids_incoming_metric->Inc();
|
|
auto id_name = std::string{iu.id_name()};
|
|
auto id_value = iu.id_value().to_data();
|
|
const auto& id = zeek::detail::global_scope()->Find(id_name);
|
|
|
|
if ( ! id ) {
|
|
reporter->Warning("Received id-update request for unknown id: %s", id_name.c_str());
|
|
return false;
|
|
}
|
|
|
|
auto val = detail::data_to_val(id_value, id->GetType().get());
|
|
|
|
if ( ! val ) {
|
|
reporter->Error("Failed to receive ID with unsupported type: %s (%s)", id_name.c_str(),
|
|
type_name(id->GetType()->Tag()));
|
|
return false;
|
|
}
|
|
|
|
id->SetVal(std::move(val));
|
|
return true;
|
|
}
|
|
|
|
void Manager::ProcessStatus(broker::status& stat) {
|
|
DBG_LOG(DBG_BROKER, "Received status message: %s", RenderMessage(stat).c_str());
|
|
|
|
auto ctx = stat.context();
|
|
|
|
EventHandlerPtr event;
|
|
switch ( stat.code() ) {
|
|
case broker::sc::unspecified: event = ::Broker::status; break;
|
|
|
|
case broker::sc::peer_added:
|
|
++peer_count;
|
|
assert(ctx);
|
|
log_mgr->SendAllWritersTo(*ctx);
|
|
event = ::Broker::peer_added;
|
|
break;
|
|
|
|
case broker::sc::peer_removed:
|
|
--peer_count;
|
|
event = ::Broker::peer_removed;
|
|
break;
|
|
|
|
case broker::sc::peer_lost:
|
|
--peer_count;
|
|
event = ::Broker::peer_lost;
|
|
break;
|
|
|
|
case broker::sc::endpoint_discovered: event = ::Broker::endpoint_discovered; break;
|
|
|
|
case broker::sc::endpoint_unreachable: event = ::Broker::endpoint_unreachable; break;
|
|
|
|
default: reporter->Warning("Unhandled Broker status: %s", to_string(stat).c_str()); break;
|
|
}
|
|
|
|
if ( ! event )
|
|
return;
|
|
|
|
static auto ei = id::find_type<RecordType>("Broker::EndpointInfo");
|
|
auto endpoint_info = make_intrusive<RecordVal>(ei);
|
|
|
|
if ( ctx ) {
|
|
endpoint_info->Assign(0, to_string(ctx->node));
|
|
static auto ni = id::find_type<RecordType>("Broker::NetworkInfo");
|
|
auto network_info = make_intrusive<RecordVal>(ni);
|
|
|
|
if ( ctx->network ) {
|
|
network_info->Assign(0, ctx->network->address.c_str());
|
|
network_info->Assign(1, val_mgr->Port(ctx->network->port, TRANSPORT_TCP));
|
|
}
|
|
else {
|
|
// TODO: are there any status messages where the ctx->network
|
|
// is not set and actually could be?
|
|
network_info->Assign(0, "<unknown>");
|
|
network_info->Assign(1, val_mgr->Port(0, TRANSPORT_TCP));
|
|
}
|
|
|
|
endpoint_info->Assign(1, std::move(network_info));
|
|
}
|
|
|
|
auto str = stat.message();
|
|
auto msg = make_intrusive<StringVal>(str ? *str : "");
|
|
|
|
event_mgr.Enqueue(event, std::move(endpoint_info), std::move(msg));
|
|
}
|
|
|
|
void Manager::ProcessError(broker::error& err) {
|
|
DBG_LOG(DBG_BROKER, "Received error message: %s", RenderMessage(err).c_str());
|
|
|
|
if ( ! ::Broker::error )
|
|
return;
|
|
|
|
auto int_code = static_cast<uint8_t>(err.code());
|
|
|
|
BifEnum::Broker::ErrorCode ec;
|
|
static auto enum_type = id::find_type<EnumType>("Broker::ErrorCode");
|
|
if ( enum_type->Lookup(int_code) )
|
|
ec = static_cast<BifEnum::Broker::ErrorCode>(int_code);
|
|
else {
|
|
reporter->Warning("Unknown Broker error code %u: mapped to unspecified enum value ",
|
|
static_cast<unsigned>(int_code));
|
|
ec = BifEnum::Broker::ErrorCode::UNSPECIFIED;
|
|
}
|
|
|
|
std::string msg;
|
|
// Note: we could also use to_string, but that would change the log output
|
|
// and we would have to update all baselines relying on this format.
|
|
if ( auto ctx = err.context() ) {
|
|
msg += '(';
|
|
msg += broker::to_string(ctx->node);
|
|
msg += ", ";
|
|
msg += broker::to_string(ctx->network);
|
|
msg += ", ";
|
|
if ( auto what = err.message() )
|
|
print_escaped(msg, *what);
|
|
else
|
|
msg += R"_("")_";
|
|
msg += ')';
|
|
}
|
|
else
|
|
msg = "(null)";
|
|
|
|
event_mgr.Enqueue(::Broker::error, BifType::Enum::Broker::ErrorCode->GetEnumVal(ec),
|
|
make_intrusive<StringVal>(msg));
|
|
}
|
|
|
|
void Manager::ProcessStoreResponse(detail::StoreHandleVal* s, broker::store::response response) {
|
|
DBG_LOG(DBG_BROKER, "Received store response: %s", RenderMessage(response).c_str());
|
|
|
|
auto request = pending_queries.find(std::make_pair(response.id, s));
|
|
|
|
if ( request == pending_queries.end() ) {
|
|
reporter->Warning("unmatched response to query %" PRIu64 " on store %s", response.id, s->store.name().c_str());
|
|
return;
|
|
}
|
|
|
|
if ( request->second->Disabled() ) {
|
|
// Trigger timer must have timed the query out already.
|
|
delete request->second;
|
|
pending_queries.erase(request);
|
|
return;
|
|
}
|
|
|
|
if ( response.answer ) {
|
|
BrokerData tmp{std::move(*response.answer)};
|
|
request->second->Result(detail::query_result(std::move(tmp).ToRecordVal()));
|
|
}
|
|
else if ( response.answer.error() == broker::ec::request_timeout ) {
|
|
// Fine, trigger's timeout takes care of things.
|
|
}
|
|
else if ( response.answer.error() == broker::ec::stale_data ) {
|
|
// It's sort of arbitrary whether to make this type of error successful
|
|
// query with a "fail" status versus going through the when stmt timeout
|
|
// code path. I think the timeout path is maybe more expected in order
|
|
// for failures like "no such key" to actually be distinguishable from
|
|
// this type of error (which is less easily handled programmatically).
|
|
}
|
|
else if ( response.answer.error() == broker::ec::no_such_key )
|
|
request->second->Result(detail::query_result());
|
|
else
|
|
reporter->InternalWarning("unknown store response status: %s", to_string(response.answer.error()).c_str());
|
|
|
|
delete request->second;
|
|
pending_queries.erase(request);
|
|
}
|
|
|
|
detail::StoreHandleVal* Manager::MakeMaster(const string& name, broker::backend type, broker::backend_options opts) {
|
|
if ( ! bstate ) {
|
|
if ( zeek::detail::current_scope() == zeek::detail::global_scope() )
|
|
reporter->Error("Broker stores cannot be created at the global scope");
|
|
|
|
return nullptr;
|
|
}
|
|
|
|
if ( bstate->endpoint.is_shutdown() )
|
|
return nullptr;
|
|
|
|
if ( LookupStore(name) )
|
|
return nullptr;
|
|
|
|
DBG_LOG(DBG_BROKER, "Creating master for data store %s", name.c_str());
|
|
|
|
auto it = opts.find("path");
|
|
|
|
if ( it == opts.end() )
|
|
it = opts.emplace("path", "").first;
|
|
|
|
if ( it->second == broker::data("") ) {
|
|
auto suffix = ".store";
|
|
|
|
switch ( type ) {
|
|
case broker::backend::sqlite: suffix = ".sqlite"; break;
|
|
default: break;
|
|
}
|
|
|
|
it->second = name + suffix;
|
|
}
|
|
|
|
auto result = bstate->endpoint.attach_master(name, type, std::move(opts));
|
|
if ( ! result ) {
|
|
Error("Failed to attach master store %s:", to_string(result.error()).c_str());
|
|
return nullptr;
|
|
}
|
|
|
|
auto handle = new detail::StoreHandleVal{*result};
|
|
Ref(handle);
|
|
|
|
data_stores.emplace(name, handle);
|
|
if ( ! iosource_mgr->RegisterFd(handle->proxy.mailbox().descriptor(), this) )
|
|
reporter->FatalError("Failed to register broker master mailbox descriptor with iosource_mgr");
|
|
|
|
PrepareForwarding(name);
|
|
|
|
if ( ! bstate->endpoint.use_real_time() )
|
|
// Wait for master to become available/responsive.
|
|
// Possibly avoids timeouts in scripts during unit tests.
|
|
handle->store.exists("");
|
|
|
|
BrokerStoreToZeekTable(name, handle);
|
|
|
|
return handle;
|
|
}
|
|
|
|
void Manager::BrokerStoreToZeekTable(const std::string& name, const detail::StoreHandleVal* handle) {
|
|
if ( ! handle->forward_to )
|
|
return;
|
|
|
|
auto keys = handle->store.keys();
|
|
if ( ! keys )
|
|
return;
|
|
|
|
auto set = get_if<broker::set>(&(keys->get_data()));
|
|
auto table = handle->forward_to;
|
|
const auto& its = table->GetType()->AsTableType()->GetIndexTypes();
|
|
bool is_set = table->GetType()->IsSet();
|
|
|
|
// disable &on_change notifications while filling the table.
|
|
table->DisableChangeNotifications();
|
|
|
|
for ( const auto& key : *set ) {
|
|
auto zeek_key = ValPtr{};
|
|
auto key_copy = key;
|
|
if ( its.size() == 1 )
|
|
zeek_key = detail::data_to_val(key_copy, its[0].get());
|
|
else
|
|
zeek_key = detail::data_to_val(key_copy, table->GetType()->AsTableType()->GetIndices().get());
|
|
|
|
if ( ! zeek_key ) {
|
|
reporter->Error(
|
|
"Failed to convert key \"%s\" while importing broker store to table "
|
|
"for store \"%s\". Aborting import.",
|
|
to_string(key).c_str(), name.c_str());
|
|
// just abort - this probably means the types are incompatible
|
|
table->EnableChangeNotifications();
|
|
return;
|
|
}
|
|
|
|
if ( is_set ) {
|
|
table->Assign(zeek_key, nullptr, false);
|
|
continue;
|
|
}
|
|
|
|
auto value = handle->store.get(key);
|
|
if ( ! value ) {
|
|
reporter->Error("Failed to load value for key %s while importing Broker store %s to table",
|
|
to_string(key).c_str(), name.c_str());
|
|
table->EnableChangeNotifications();
|
|
continue;
|
|
}
|
|
|
|
auto zeek_value = detail::data_to_val(*value, table->GetType()->Yield().get());
|
|
if ( ! zeek_value ) {
|
|
reporter->Error(
|
|
"Could not convert %s to table value while trying to import Broker "
|
|
"store %s. Aborting import.",
|
|
to_string(value).c_str(), name.c_str());
|
|
table->EnableChangeNotifications();
|
|
return;
|
|
}
|
|
|
|
table->Assign(zeek_key, zeek_value, false);
|
|
}
|
|
|
|
table->EnableChangeNotifications();
|
|
return;
|
|
}
|
|
|
|
detail::StoreHandleVal* Manager::MakeClone(const string& name, double resync_interval, double stale_interval,
|
|
double mutation_buffer_interval) {
|
|
if ( bstate->endpoint.is_shutdown() )
|
|
return nullptr;
|
|
|
|
if ( LookupStore(name) )
|
|
return nullptr;
|
|
|
|
DBG_LOG(DBG_BROKER, "Creating clone for data store %s", name.c_str());
|
|
|
|
auto result = bstate->endpoint.attach_clone(name, resync_interval, stale_interval, mutation_buffer_interval);
|
|
if ( ! result ) {
|
|
Error("Failed to attach clone store %s:", to_string(result.error()).c_str());
|
|
return nullptr;
|
|
}
|
|
|
|
auto handle = new detail::StoreHandleVal{*result};
|
|
Ref(handle);
|
|
|
|
if ( ! handle->proxy.valid() ) {
|
|
reporter->Error("Failed to create clone for data store %s", name.c_str());
|
|
return nullptr;
|
|
}
|
|
|
|
data_stores.emplace(name, handle);
|
|
if ( ! iosource_mgr->RegisterFd(handle->proxy.mailbox().descriptor(), this) )
|
|
reporter->FatalError("Failed to register broker clone mailbox descriptor with iosource_mgr");
|
|
PrepareForwarding(name);
|
|
return handle;
|
|
}
|
|
|
|
detail::StoreHandleVal* Manager::LookupStore(const string& name) {
|
|
auto i = data_stores.find(name);
|
|
return i == data_stores.end() ? nullptr : i->second;
|
|
}
|
|
|
|
bool Manager::CloseStore(const string& name) {
|
|
DBG_LOG(DBG_BROKER, "Closing data store %s", name.c_str());
|
|
|
|
auto s = data_stores.find(name);
|
|
if ( s == data_stores.end() )
|
|
return false;
|
|
|
|
iosource_mgr->UnregisterFd(s->second->proxy.mailbox().descriptor(), this);
|
|
|
|
for ( auto i = pending_queries.begin(); i != pending_queries.end(); )
|
|
if ( i->second->Store().name() == name ) {
|
|
i->second->Abort();
|
|
delete i->second;
|
|
i = pending_queries.erase(i);
|
|
}
|
|
else {
|
|
++i;
|
|
}
|
|
|
|
s->second->have_store = false;
|
|
s->second->store_pid = {};
|
|
s->second->proxy = {};
|
|
s->second->store = {};
|
|
Unref(s->second);
|
|
data_stores.erase(s);
|
|
return true;
|
|
}
|
|
|
|
bool Manager::TrackStoreQuery(detail::StoreHandleVal* handle, broker::request_id id, detail::StoreQueryCallback* cb) {
|
|
auto rval = pending_queries.emplace(std::make_pair(id, handle), cb).second;
|
|
|
|
if ( bstate->endpoint.use_real_time() )
|
|
return rval;
|
|
|
|
FlushPendingQueries();
|
|
return rval;
|
|
}
|
|
|
|
const Stats& Manager::GetStatistics() {
|
|
statistics.num_peers = peer_count;
|
|
statistics.num_stores = data_stores.size();
|
|
statistics.num_pending_queries = pending_queries.size();
|
|
|
|
statistics.num_events_incoming = static_cast<size_t>(num_events_incoming_metric->Value());
|
|
statistics.num_events_outgoing = static_cast<size_t>(num_events_outgoing_metric->Value());
|
|
statistics.num_logs_incoming = static_cast<size_t>(num_logs_incoming_metric->Value());
|
|
statistics.num_logs_outgoing = static_cast<size_t>(num_logs_outgoing_metric->Value());
|
|
statistics.num_ids_incoming = static_cast<size_t>(num_ids_incoming_metric->Value());
|
|
statistics.num_ids_outgoing = static_cast<size_t>(num_ids_outgoing_metric->Value());
|
|
|
|
return statistics;
|
|
}
|
|
|
|
TableValPtr Manager::GetPeeringStatsTable() { return bstate->peerBufferState->GetPeeringStatsTable(); }
|
|
|
|
bool Manager::AddForwardedStore(const std::string& name, TableValPtr table) {
|
|
if ( forwarded_stores.find(name) != forwarded_stores.end() ) {
|
|
reporter->Error("same &broker_store %s specified for two different variables", name.c_str());
|
|
return false;
|
|
}
|
|
|
|
DBG_LOG(DBG_BROKER, "Adding table forward for data store %s", name.c_str());
|
|
forwarded_stores.emplace(name, std::move(table));
|
|
|
|
PrepareForwarding(name);
|
|
return true;
|
|
}
|
|
|
|
void Manager::PrepareForwarding(const std::string& name) {
|
|
auto handle = LookupStore(name);
|
|
if ( ! handle )
|
|
return;
|
|
|
|
if ( forwarded_stores.find(name) == forwarded_stores.end() )
|
|
return;
|
|
|
|
handle->forward_to = forwarded_stores.at(name);
|
|
DBG_LOG(DBG_BROKER, "Resolved table forward for data store %s", name.c_str());
|
|
}
|
|
|
|
broker::hub Manager::MakeHub(broker::filter_type ft) {
|
|
++hub_count;
|
|
return bstate->endpoint.make_hub(std::move(ft));
|
|
}
|
|
|
|
void Manager::DestroyHub(broker::hub&& hub) { --hub_count; }
|
|
|
|
} // namespace zeek::Broker
|