Commit 206ba3f2 authored by Dominik Charousset's avatar Dominik Charousset

Add metric families for streams

parent b3478a09
......@@ -204,6 +204,31 @@ public:
/// Counts how many messages are currently waiting in the mailbox.
telemetry::int_gauge_family* mailbox_size_family = nullptr;
struct {
// -- inbound ------------------------------------------------------------
/// Counts the total number of processed stream elements from upstream.
telemetry::int_counter_family* processed_elements_family = nullptr;
/// Tracks how many stream elements from upstream are currently buffered.
telemetry::int_gauge_family* input_buffer_size_family = nullptr;
// -- outbound -----------------------------------------------------------
/// Counts the total number of elements that have been pushed downstream.
telemetry::int_counter_family* pushed_elements_family = nullptr;
/// Counts the total number of batches that have been pushed downstream.
telemetry::int_counter_family* pushed_batches_family = nullptr;
/// Tracks how many stream elements are currently waiting in the output
/// buffer due to insufficient credit.
telemetry::int_gauge_family* output_buffer_size_family = nullptr;
}
/// Wraps streaming-related actor metric families.
stream;
};
/// @warning The system stores a reference to `cfg`, which means the
......
......@@ -77,6 +77,30 @@ public:
telemetry::int_gauge* mailbox_size = nullptr;
};
/// Optional metrics for inbound stream traffic collected by individual actors
/// when configured to do so.
struct inbound_stream_metrics_t {
/// Counts the total number of processed stream elements from upstream.
telemetry::int_counter* processed_elements = nullptr;
/// Tracks how many stream elements from upstream are currently buffered.
telemetry::int_gauge* input_buffer_size = nullptr;
};
/// Optional metrics for outbound stream traffic collected by individual
/// actors when configured to do so.
struct outbound_stream_metrics_t {
/// Counts the total number of elements that have been pushed downstream.
telemetry::int_counter* pushed_elements = nullptr;
/// Counts the total number of batches that have been pushed downstream.
telemetry::int_counter* pushed_batches = nullptr;
/// Tracks how many stream elements are currently waiting in the output
/// buffer due to insufficient credit.
telemetry::int_gauge* output_buffer_size = nullptr;
};
// -- constructors, destructors, and assignment operators --------------------
local_actor(actor_config& cfg);
......@@ -382,6 +406,11 @@ public:
return metrics_;
}
bool has_metrics_enabled() const noexcept {
// Either all fields are null or none is.
return metrics_.processing_time != nullptr;
}
template <class ActorHandle>
ActorHandle eval_opts(spawn_options opts, ActorHandle res) {
if (has_monitor_flag(opts))
......
......@@ -254,6 +254,22 @@ auto make_actor_metric_families(telemetry::metric_registry& reg) {
"Time a message waits in the mailbox before processing.", "seconds"),
reg.gauge_family("caf.actor", "mailbox-size", {"name"},
"Number of messages in the mailbox."),
{
reg.counter_family("caf.actor.stream", "processed-elements",
{"name", "slot"},
"Number of processed stream elements from upstream."),
reg.gauge_family("caf.actor.stream", "input_buffer_size",
{"name", "slot"},
"Number of buffered stream elements from upstream."),
reg.counter_family(
"caf.actor.stream", "pushed-elements", {"name", "slot"},
"Number of elements that have been pushed downstream."),
reg.counter_family("caf.actor.stream", "pushed-batches", {"name", "slot"},
"Number of batches that have been pushed downstream."),
reg.gauge_family("caf.actor.stream", "output-buffer-size",
{"name", "slot"},
"Number of buffered output stream elements."),
},
};
}
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment