Struct relay_config::aggregator::AggregatorServiceConfig
source · pub struct AggregatorServiceConfig {
- pub aggregator: AggregatorConfig,
+ pub aggregator: AggregatorConfig,
pub max_total_bucket_bytes: Option<usize>,
pub max_flush_bytes: usize,
pub flush_interval_ms: u64,
}Expand description
Parameters used for metric aggregation.
-Fields§
§aggregator: AggregatorConfigThe config used by the internal aggregator.
+Fields§
§aggregator: AggregatorConfigThe config used by the internal aggregator.
max_total_bucket_bytes: Option<usize>Maximum amount of bytes used for metrics aggregation.
When aggregating metrics, Relay keeps track of how many bytes a metric takes in memory. This is only an approximation and does not take into account things such as pre-allocation diff --git a/relay_config/struct.Config.html b/relay_config/struct.Config.html index 4f26d51f86..06508d9114 100644 --- a/relay_config/struct.Config.html +++ b/relay_config/struct.Config.html @@ -165,7 +165,7 @@
sourcepub fn cogs_granularity(&self) -> Duration
pub fn cogs_granularity(&self) -> Duration
Granularity for COGS measurements.
sourcepub fn cogs_max_queue_size(&self) -> u64
pub fn cogs_max_queue_size(&self) -> u64
Maximum amount of COGS measurements buffered in memory.
sourcepub fn cogs_relay_resource_id(&self) -> &str
pub fn cogs_relay_resource_id(&self) -> &str
Resource ID to use for Relay COGS measurements.
-sourcepub fn permissive_aggregator_config(&self) -> AggregatorConfig
pub fn permissive_aggregator_config(&self) -> AggregatorConfig
Creates an AggregatorConfig that is compatible with every other aggregator.
sourcepub fn permissive_aggregator_config(&self) -> AggregatorConfig
pub fn permissive_aggregator_config(&self) -> AggregatorConfig
Creates an [AggregatorConfig] that is compatible with every other aggregator.
A lossless aggregator can be put in front of any of the configured aggregators without losing data that the configured aggregator would keep. This is useful for pre-aggregating metrics together in a single aggregator instance.
sourcepub fn default_aggregator_config(&self) -> &AggregatorServiceConfig
pub fn default_aggregator_config(&self) -> &AggregatorServiceConfig
Returns configuration for the default metrics aggregator.
diff --git a/relay_event_schema/processor/derive.ProcessValue.html b/relay_event_schema/processor/derive.ProcessValue.html index a3e79c530c..777a2e8655 100644 --- a/relay_event_schema/processor/derive.ProcessValue.html +++ b/relay_event_schema/processor/derive.ProcessValue.html @@ -1,4 +1,4 @@ -Derive Macro relay_event_schema::processor::ProcessValue
source · #[derive(ProcessValue)]
+ProcessValue in relay_event_schema::processor - Rust Derive Macro relay_event_schema::processor::ProcessValue
#[derive(ProcessValue)]
{
// Attributes available to this derive:
#[metastructure]
diff --git a/relay_filter/fn.should_filter.html b/relay_filter/fn.should_filter.html
index 575bb253df..0b838c214b 100644
--- a/relay_filter/fn.should_filter.html
+++ b/relay_filter/fn.should_filter.html
@@ -1,4 +1,4 @@
-should_filter in relay_filter - Rust Function relay_filter::should_filter
source · pub fn should_filter<F: Filterable + Getter>(
+should_filter in relay_filter - Rust Function relay_filter::should_filter
source · pub fn should_filter<F: Filterable + Getter>(
item: &F,
client_ip: Option<IpAddr>,
config: &ProjectFiltersConfig,
diff --git a/relay_filter/generic/index.html b/relay_filter/generic/index.html
index 8b55209a02..efac86fa60 100644
--- a/relay_filter/generic/index.html
+++ b/relay_filter/generic/index.html
@@ -1,4 +1,4 @@
-relay_filter::generic - Rust Module relay_filter::generic
source · Expand description
Implements generic filtering based on the [RuleCondition] DSL.
+relay_filter::generic - Rust Module relay_filter::generic
source · Expand description
Implements generic filtering based on the RuleCondition DSL.
Multiple generic filters can be defined and they are going to be checked in FIFO order. The
first one that matches, will result in the event being discarded with a FilterStatKey
identifying the matching filter.
diff --git a/relay_filter/index.html b/relay_filter/index.html
index dd9e40d2ba..d7bb89f244 100644
--- a/relay_filter/index.html
+++ b/relay_filter/index.html
@@ -6,4 +6,4 @@
web crawlers (filter events sent by user agents known to be web crawlers)
legacy browsers (filter events originating from legacy browsers, can be configured)
-Re-exports§
pub use crate::csp::matches_any_origin;pub use crate::generic::are_generic_filters_supported;
Modules§
- Implements filtering for events caused by problematic browsers extensions.
- Implements event filtering based on the client ip address.
- Implements event filtering for events originating from CSP endpoints
- Implements event filtering based on the error message
- Implements generic filtering based on the [
RuleCondition] DSL. - Implements filtering for events originating from legacy browsers.
- Implements filtering for events originating from the localhost
- Implements event filtering based on whether the endpoint called is a healthcheck endpoint.
- Filters events coming from user agents known to be web crawlers.
Structs§
- Configuration for the client ips filter.
- Configuration for the CSP filter.
- Configuration for the error messages filter.
- Common configuration for event filters.
- Configuration for a generic filter.
- Configuration for generic filters.
- Map of generic filters, mapping from the id to the filter itself.
- Configuration for transaction name filter.
- Configuration for the legacy browsers filter.
- Configuration for all event filters from project configs.
- Configuration for the releases filter.
Enums§
- Identifies which filter dropped an event for which reason.
- A browser class to be filtered by the legacy browser filter.
Traits§
- A data item to which filters can be applied.
Functions§
- Checks whether an event should be filtered for a particular configuration.
Re-exports§
pub use crate::csp::matches_any_origin;pub use crate::generic::are_generic_filters_supported;
Modules§
- Implements filtering for events caused by problematic browsers extensions.
- Implements event filtering based on the client ip address.
- Implements event filtering for events originating from CSP endpoints
- Implements event filtering based on the error message
- Implements generic filtering based on the
RuleCondition DSL. - Implements filtering for events originating from legacy browsers.
- Implements filtering for events originating from the localhost
- Implements event filtering based on whether the endpoint called is a healthcheck endpoint.
- Filters events coming from user agents known to be web crawlers.
Structs§
- Configuration for the client ips filter.
- Configuration for the CSP filter.
- Configuration for the error messages filter.
- Common configuration for event filters.
- Configuration for a generic filter.
- Configuration for generic filters.
- Map of generic filters, mapping from the id to the filter itself.
- Configuration for transaction name filter.
- Configuration for the legacy browsers filter.
- Configuration for all event filters from project configs.
- Configuration for the releases filter.
Enums§
- Identifies which filter dropped an event for which reason.
- A browser class to be filtered by the legacy browser filter.
Traits§
- A data item to which filters can be applied.
Functions§
- Checks whether an event should be filtered for a particular configuration.
GenericFilterConfig in relay_filter - Rust Struct relay_filter::GenericFilterConfig
source · pub struct GenericFilterConfig {
pub id: String,
pub is_enabled: bool,
- pub condition: Option<RuleCondition>,
+ pub condition: Option<RuleCondition>,
}
Expand description
Configuration for a generic filter.
Fields§
§id: StringUnique identifier of the generic filter.
§is_enabled: boolSpecifies whether this filter is enabled.
-§condition: Option<RuleCondition>The condition for the filter.
+§condition: Option<RuleCondition>The condition for the filter.
Implementations§
Trait Implementations§
source§impl Clone for GenericFilterConfig
source§fn clone(&self) -> GenericFilterConfig
Returns a copy of the value. Read more1.0.0 · source§fn clone_from(&mut self, source: &Self)
Performs copy-assignment from source. Read moresource§impl Debug for GenericFilterConfig
source§impl Default for GenericFilterConfig
source§fn default() -> GenericFilterConfig
Returns the “default value” for a type. Read moresource§impl<'de> Deserialize<'de> for GenericFilterConfig
source§fn deserialize<__D>(__deserializer: __D) -> Result<Self, __D::Error>where
__D: Deserializer<'de>,
Deserialize this value from the given Serde deserializer. Read moresource§impl PartialEq for GenericFilterConfig
source§fn eq(&self, other: &GenericFilterConfig) -> bool
This method tests for self and other values to be equal, and is used
diff --git a/relay_filter/trait.Filterable.html b/relay_filter/trait.Filterable.html
index a932c6881a..007a330ee3 100644
--- a/relay_filter/trait.Filterable.html
+++ b/relay_filter/trait.Filterable.html
@@ -1,20 +1,20 @@
Filterable in relay_filter - Rust Trait relay_filter::Filterable
source · pub trait Filterable {
// Required methods
- fn csp(&self) -> Option<&Csp>;
- fn exceptions(&self) -> Option<&Values<Exception>>;
+ fn csp(&self) -> Option<&Csp>;
+ fn exceptions(&self) -> Option<&Values<Exception>>;
fn ip_addr(&self) -> Option<&str>;
- fn logentry(&self) -> Option<&LogEntry>;
+ fn logentry(&self) -> Option<&LogEntry>;
fn release(&self) -> Option<&str>;
fn transaction(&self) -> Option<&str>;
fn url(&self) -> Option<Url>;
fn user_agent(&self) -> Option<&str>;
}
Expand description
A data item to which filters can be applied.
-Required Methods§
sourcefn exceptions(&self) -> Option<&Values<Exception>>
The exception values of the item. Only for error events.
+Required Methods§
sourcefn exceptions(&self) -> Option<&Values<Exception>>
The exception values of the item. Only for error events.
sourcefn transaction(&self) -> Option<&str>
The transaction name. Only for transaction events.
sourcefn url(&self) -> Option<Url>
The URL from which the request originates. Used for localhost filtering.
sourcefn user_agent(&self) -> Option<&str>
The user agent of the client that sent the data.
-Implementations on Foreign Types§
source§impl Filterable for Event
source§fn csp(&self) -> Option<&Csp>
source§fn exceptions(&self) -> Option<&Values<Exception>>
source§fn ip_addr(&self) -> Option<&str>
source§fn logentry(&self) -> Option<&LogEntry>
source§fn release(&self) -> Option<&str>
source§fn transaction(&self) -> Option<&str>
source§fn url(&self) -> Option<Url>
source§fn user_agent(&self) -> Option<&str>
source§impl Filterable for Replay
source§fn csp(&self) -> Option<&Csp>
source§fn exceptions(&self) -> Option<&Values<Exception>>
source§fn ip_addr(&self) -> Option<&str>
source§fn logentry(&self) -> Option<&LogEntry>
source§fn release(&self) -> Option<&str>
source§fn transaction(&self) -> Option<&str>
source§fn url(&self) -> Option<Url>
source§fn user_agent(&self) -> Option<&str>
source§impl Filterable for Span
source§fn csp(&self) -> Option<&Csp>
source§fn exceptions(&self) -> Option<&Values<Exception>>
source§fn ip_addr(&self) -> Option<&str>
source§fn logentry(&self) -> Option<&LogEntry>
source§fn release(&self) -> Option<&str>
source§fn transaction(&self) -> Option<&str>
source§fn url(&self) -> Option<Url>
source§fn user_agent(&self) -> Option<&str>
Implementors§
\ No newline at end of file
+Implementations on Foreign Types§
source§impl Filterable for Event
source§fn csp(&self) -> Option<&Csp>
source§fn exceptions(&self) -> Option<&Values<Exception>>
source§fn ip_addr(&self) -> Option<&str>
source§fn logentry(&self) -> Option<&LogEntry>
source§fn release(&self) -> Option<&str>
source§fn transaction(&self) -> Option<&str>
source§fn url(&self) -> Option<Url>
source§fn user_agent(&self) -> Option<&str>
source§impl Filterable for Replay
source§fn csp(&self) -> Option<&Csp>
source§fn exceptions(&self) -> Option<&Values<Exception>>
source§fn ip_addr(&self) -> Option<&str>
source§fn logentry(&self) -> Option<&LogEntry>
source§fn release(&self) -> Option<&str>
source§fn transaction(&self) -> Option<&str>
source§fn url(&self) -> Option<Url>
source§fn user_agent(&self) -> Option<&str>
source§impl Filterable for Span
source§fn csp(&self) -> Option<&Csp>
source§fn exceptions(&self) -> Option<&Values<Exception>>
source§fn ip_addr(&self) -> Option<&str>
source§fn logentry(&self) -> Option<&LogEntry>
source§fn release(&self) -> Option<&str>
source§fn transaction(&self) -> Option<&str>
source§fn url(&self) -> Option<Url>
source§fn user_agent(&self) -> Option<&str>
Implementors§
\ No newline at end of file
diff --git a/relay_metrics/aggregator/struct.AggregateMetricsError.html b/relay_metrics/aggregator/enum.AggregateMetricsError.html
similarity index 52%
rename from relay_metrics/aggregator/struct.AggregateMetricsError.html
rename to relay_metrics/aggregator/enum.AggregateMetricsError.html
index caf979b770..b4d3557a42 100644
--- a/relay_metrics/aggregator/struct.AggregateMetricsError.html
+++ b/relay_metrics/aggregator/enum.AggregateMetricsError.html
@@ -1,7 +1,22 @@
-AggregateMetricsError in relay_metrics::aggregator - Rust Struct relay_metrics::aggregator::AggregateMetricsError
source · pub struct AggregateMetricsError { /* private fields */ }
Expand description
Any error that may occur during aggregation.
-Trait Implementations§
source§impl Debug for AggregateMetricsError
source§impl Display for AggregateMetricsError
source§impl Error for AggregateMetricsError
1.30.0 · source§fn source(&self) -> Option<&(dyn Error + 'static)>
The lower-level source of this error, if any. Read more1.0.0 · source§fn description(&self) -> &str
👎Deprecated since 1.42.0: use the Display impl or to_string()source§impl PartialEq for AggregateMetricsError
source§fn eq(&self, other: &AggregateMetricsError) -> bool
This method tests for self and other values to be equal, and is used
+AggregateMetricsError in relay_metrics::aggregator - Rust pub enum AggregateMetricsError {
+ InvalidCharacters(MetricName),
+ UnsupportedNamespace(MetricNamespace),
+ InvalidTimestamp(UnixTimestamp),
+ InvalidTypes,
+ InvalidStringLength(MetricName),
+ TotalLimitExceeded,
+ ProjectLimitExceeded,
+}
Expand description
Any error that may occur during aggregation.
+Variants§
§InvalidCharacters(MetricName)
A metric bucket had invalid characters in the metric name.
+§UnsupportedNamespace(MetricNamespace)
A metric bucket had an unknown namespace in the metric name.
+§InvalidTimestamp(UnixTimestamp)
A metric bucket’s timestamp was out of the configured acceptable range.
+§InvalidTypes
Internal error: Attempted to merge two metric buckets of different types.
+§InvalidStringLength(MetricName)
A metric bucket had a too long string (metric name or a tag key/value).
+§TotalLimitExceeded
A metric bucket is too large for the global bytes limit.
+§ProjectLimitExceeded
A metric bucket is too large for the per-project bytes limit.
+Trait Implementations§
source§impl Debug for AggregateMetricsError
source§impl Display for AggregateMetricsError
source§impl Error for AggregateMetricsError
1.30.0 · source§fn source(&self) -> Option<&(dyn Error + 'static)>
The lower-level source of this error, if any. Read more1.0.0 · source§fn description(&self) -> &str
👎Deprecated since 1.42.0: use the Display impl or to_string()source§impl PartialEq for AggregateMetricsError
source§fn eq(&self, other: &AggregateMetricsError) -> bool
This method tests for self and other values to be equal, and is used
by ==.source§impl StructuralPartialEq for AggregateMetricsError
Auto Trait Implementations§
§impl Freeze for AggregateMetricsError
§impl RefUnwindSafe for AggregateMetricsError
§impl Send for AggregateMetricsError
§impl Sync for AggregateMetricsError
§impl Unpin for AggregateMetricsError
§impl UnwindSafe for AggregateMetricsError
Blanket Implementations§
source§impl StructuralPartialEq for AggregateMetricsError
Auto Trait Implementations§
§impl Freeze for AggregateMetricsError
§impl RefUnwindSafe for AggregateMetricsError
§impl Send for AggregateMetricsError
§impl Sync for AggregateMetricsError
§impl Unpin for AggregateMetricsError
§impl UnwindSafe for AggregateMetricsError
Blanket Implementations§
source§impl<T> BorrowMut<T> for Twhere
T: ?Sized,
source§fn borrow_mut(&mut self) -> &mut T
Mutably borrows from an owned value. Read moresource§impl<T> From<T> for T
source§fn from(t: T) -> T
Returns the argument unchanged.
diff --git a/relay_metrics/aggregator/enum.FlushBatching.html b/relay_metrics/aggregator/enum.FlushBatching.html
index be6b1e9a86..b4a9c35a7c 100644
--- a/relay_metrics/aggregator/enum.FlushBatching.html
+++ b/relay_metrics/aggregator/enum.FlushBatching.html
@@ -1,4 +1,4 @@
-FlushBatching in relay_metrics::aggregator - Rust Enum relay_metrics::aggregator::FlushBatching
source · pub enum FlushBatching {
+FlushBatching in relay_metrics::aggregator - Rust Enum relay_metrics::aggregator::FlushBatching
source · pub enum FlushBatching {
Project,
Bucket,
Partition,
@@ -17,9 +17,9 @@
It should only be used in Relays with the http.global_metrics option set since when
encoding metrics via the global endpoint we can leverage partitioning.
§None
Do not apply shift.
-
Trait Implementations§
source§impl Clone for FlushBatching
source§fn clone(&self) -> FlushBatching
Returns a copy of the value. Read more1.0.0 · source§fn clone_from(&mut self, source: &Self)
Performs copy-assignment from source. Read moresource§impl Debug for FlushBatching
source§impl Default for FlushBatching
source§fn default() -> FlushBatching
Returns the “default value” for a type. Read moresource§impl<'de> Deserialize<'de> for FlushBatching
source§fn deserialize<__D>(__deserializer: __D) -> Result<Self, __D::Error>where
- __D: Deserializer<'de>,
Deserialize this value from the given Serde deserializer. Read moresource§impl Serialize for FlushBatching
source§impl Copy for FlushBatching
Auto Trait Implementations§
§impl Freeze for FlushBatching
§impl RefUnwindSafe for FlushBatching
§impl Send for FlushBatching
§impl Sync for FlushBatching
§impl Unpin for FlushBatching
§impl UnwindSafe for FlushBatching
Blanket Implementations§
Trait Implementations§
source§impl Clone for FlushBatching
source§fn clone(&self) -> FlushBatching
Returns a copy of the value. Read more1.0.0 · source§fn clone_from(&mut self, source: &Self)
Performs copy-assignment from source. Read moresource§impl Debug for FlushBatching
source§impl Default for FlushBatching
source§fn default() -> FlushBatching
Returns the “default value” for a type. Read moresource§impl<'de> Deserialize<'de> for FlushBatching
source§fn deserialize<__D>(__deserializer: __D) -> Result<Self, __D::Error>where
+ __D: Deserializer<'de>,
Deserialize this value from the given Serde deserializer. Read moresource§impl Serialize for FlushBatching
source§impl Copy for FlushBatching
Auto Trait Implementations§
§impl Freeze for FlushBatching
§impl RefUnwindSafe for FlushBatching
§impl Send for FlushBatching
§impl Sync for FlushBatching
§impl Unpin for FlushBatching
§impl UnwindSafe for FlushBatching
Blanket Implementations§
source§impl<T> BorrowMut<T> for Twhere
T: ?Sized,
source§fn borrow_mut(&mut self) -> &mut T
Mutably borrows from an owned value. Read moresource§impl<T> DynClone for Twhere
diff --git a/relay_metrics/aggregator/fn.tags_cost.html b/relay_metrics/aggregator/fn.tags_cost.html
index d40006733b..286ccd6cb6 100644
--- a/relay_metrics/aggregator/fn.tags_cost.html
+++ b/relay_metrics/aggregator/fn.tags_cost.html
@@ -1,4 +1,4 @@
-tags_cost in relay_metrics::aggregator - Rust Function relay_metrics::aggregator::tags_cost
source · pub fn tags_cost(tags: &BTreeMap<String, String>) -> usize
Expand description
Estimates the number of bytes needed to encode the tags.
+tags_cost in relay_metrics::aggregator - Rust Function relay_metrics::aggregator::tags_cost
source · pub fn tags_cost(tags: &BTreeMap<String, String>) -> usize
Expand description
Estimates the number of bytes needed to encode the tags.
Note that this does not necessarily match the exact memory footprint of the tags,
because data structures or their serialization have overheads.
relay_metrics::aggregator - Rust Module relay_metrics::aggregator
source · Expand description
Core functionality of metrics aggregation.
-Structs§
- Any error that may occur during aggregation.
- A collector of
Bucket submissions. - Parameters used by the
Aggregator.
Enums§
- Configuration value for
AggregatorConfig::flush_batching.
Functions§
- Estimates the number of bytes needed to encode the tags.
relay_metrics::aggregator - Rust Module relay_metrics::aggregator
source · Expand description
Core functionality of metrics aggregation.
+Structs§
- A collector of
Bucket submissions. - Parameters used by the
Aggregator.
Enums§
- Any error that may occur during aggregation.
- Configuration value for
AggregatorConfig::flush_batching.
Functions§
- Estimates the number of bytes needed to encode the tags.
Aggregator in relay_metrics::aggregator - Rust Struct relay_metrics::aggregator::Aggregator
source · pub struct Aggregator { /* private fields */ }
Expand description
A collector of Bucket submissions.
+Aggregator in relay_metrics::aggregator - Rust Struct relay_metrics::aggregator::Aggregator
source · pub struct Aggregator { /* private fields */ }
Expand description
A collector of Bucket submissions.
§Aggregation
Each metric is dispatched into the a Bucket depending on its project key (DSN), name, type,
unit, tags and timestamp. The bucket timestamp is rounded to the precision declared by the
@@ -14,34 +14,27 @@
§Conflicts
Metrics are uniquely identified by the combination of their name, type and unit. It is allowed
to send metrics of different types and units under the same name. For example, sending a metric
once as set and once as distribution will result in two actual metrics being recorded.
-Implementations§
source§impl Aggregator
sourcepub fn new(config: AggregatorConfig) -> Self
Create a new aggregator.
-sourcepub fn named(name: String, config: AggregatorConfig) -> Self
Like Self::new, but with a provided name.
-sourcepub fn bucket_count(&self) -> usize
Returns the number of buckets in the aggregator.
-sourcepub fn totals_cost_exceeded(&self, max_total_cost: Option<usize>) -> bool
Returns true if the cost trackers value is larger than the given max cost.
-sourcepub fn into_buckets(self) -> Vec<Bucket>
Converts this aggregator into a vector of Bucket.
-sourcepub fn pop_flush_buckets(
+
Implementations§
source§impl Aggregator
sourcepub fn new(config: AggregatorConfig) -> Self
Create a new aggregator.
+sourcepub fn named(name: String, config: AggregatorConfig) -> Self
Like Self::new, but with a provided name.
+sourcepub fn bucket_count(&self) -> usize
Returns the number of buckets in the aggregator.
+sourcepub fn totals_cost_exceeded(&self, max_total_cost: Option<usize>) -> bool
Returns true if the cost trackers value is larger than the given max cost.
+sourcepub fn into_buckets(self) -> Vec<Bucket>
Converts this aggregator into a vector of Bucket.
+sourcepub fn pop_flush_buckets(
&mut self,
force: bool,
) -> HashMap<Option<u64>, HashMap<ProjectKey, Vec<Bucket>>>
Pop and return the partitions with buckets that are eligible for flushing out according to
bucket interval.
If no partitioning is enabled, the function will return a single None partition.
Note that this function is primarily intended for tests.
-sourcepub fn merge(
&mut self,
project_key: ProjectKey,
bucket: Bucket,
max_total_bucket_bytes: Option<usize>,
-) -> Result<(), AggregateMetricsError>
Merge a preaggregated bucket into this aggregator.
+) -> Result<(), AggregateMetricsError>Merge a preaggregated bucket into this aggregator.
If no bucket exists for the given bucket key, a new bucket will be created.
-sourcepub fn merge_all(
- &mut self,
- project_key: ProjectKey,
- buckets: impl IntoIterator<Item = Bucket>,
- max_total_bucket_bytes: Option<usize>,
-)
Merges all given buckets into this aggregator.
-Buckets that do not exist yet will be created.
-Trait Implementations§
Auto Trait Implementations§
§impl Freeze for Aggregator
§impl RefUnwindSafe for Aggregator
§impl Send for Aggregator
§impl Sync for Aggregator
§impl Unpin for Aggregator
§impl UnwindSafe for Aggregator
Blanket Implementations§
Trait Implementations§
Auto Trait Implementations§
§impl Freeze for Aggregator
§impl RefUnwindSafe for Aggregator
§impl Send for Aggregator
§impl Sync for Aggregator
§impl Unpin for Aggregator
§impl UnwindSafe for Aggregator
Blanket Implementations§
source§impl<T> BorrowMut<T> for Twhere
T: ?Sized,
source§fn borrow_mut(&mut self) -> &mut T
Mutably borrows from an owned value. Read moresource§impl<T> From<T> for T
source§fn from(t: T) -> T
Returns the argument unchanged.
diff --git a/relay_metrics/aggregator/struct.AggregatorConfig.html b/relay_metrics/aggregator/struct.AggregatorConfig.html
index 96c6ed773c..b1a62e3527 100644
--- a/relay_metrics/aggregator/struct.AggregatorConfig.html
+++ b/relay_metrics/aggregator/struct.AggregatorConfig.html
@@ -1,4 +1,4 @@
-AggregatorConfig in relay_metrics::aggregator - Rust Struct relay_metrics::aggregator::AggregatorConfig
source · pub struct AggregatorConfig {
+AggregatorConfig in relay_metrics::aggregator - Rust Struct relay_metrics::aggregator::AggregatorConfig
source · pub struct AggregatorConfig {
pub bucket_interval: u64,
pub initial_delay: u64,
pub max_secs_in_past: u64,
@@ -41,11 +41,11 @@
bucket interval at the same time.
For example, the aggregator can choose to shift by the same value all buckets within a given
partition, effectively allowing all the elements of that partition to be flushed together.
-
Implementations§
source§impl AggregatorConfig
sourcepub fn timestamp_range(&self) -> Range<UnixTimestamp>
Returns the valid range for metrics timestamps.
+Implementations§
source§impl AggregatorConfig
sourcepub fn timestamp_range(&self) -> Range<UnixTimestamp>
Returns the valid range for metrics timestamps.
Metrics or buckets outside of this range should be discarded.
-Trait Implementations§
source§impl Clone for AggregatorConfig
source§fn clone(&self) -> AggregatorConfig
Returns a copy of the value. Read more1.0.0 · source§fn clone_from(&mut self, source: &Self)
Performs copy-assignment from source. Read moresource§impl Debug for AggregatorConfig
source§impl Default for AggregatorConfig
source§impl<'de> Deserialize<'de> for AggregatorConfigwhere
- AggregatorConfig: Default,
source§fn deserialize<__D>(__deserializer: __D) -> Result<Self, __D::Error>where
- __D: Deserializer<'de>,
Deserialize this value from the given Serde deserializer. Read moreTrait Implementations§
source§impl Clone for AggregatorConfig
source§fn clone(&self) -> AggregatorConfig
Returns a copy of the value. Read more1.0.0 · source§fn clone_from(&mut self, source: &Self)
Performs copy-assignment from source. Read moresource§impl Debug for AggregatorConfig
source§impl Default for AggregatorConfig
source§impl<'de> Deserialize<'de> for AggregatorConfigwhere
+ AggregatorConfig: Default,
source§fn deserialize<__D>(__deserializer: __D) -> Result<Self, __D::Error>where
+ __D: Deserializer<'de>,
Deserialize this value from the given Serde deserializer. Read moreAuto Trait Implementations§
§impl Freeze for AggregatorConfig
§impl RefUnwindSafe for AggregatorConfig
§impl Send for AggregatorConfig
§impl Sync for AggregatorConfig
§impl Unpin for AggregatorConfig
§impl UnwindSafe for AggregatorConfig
Blanket Implementations§
source§impl<T> BorrowMut<T> for Twhere
diff --git a/relay_metrics/all.html b/relay_metrics/all.html
index dadd98818f..8ec9b45f98 100644
--- a/relay_metrics/all.html
+++ b/relay_metrics/all.html
@@ -1 +1 @@
-List of all items in this crate List of all items
Structs
- Bucket
- BucketMetadata
- BucketView
- BucketsView
- BucketsViewBySizeIter
- BucketsViewIter
- CustomUnit
- FiniteF64
- GaugeValue
- MetricName
- MetricResourceIdentifier
- ParseBuckets
- ParseFiniteFloatError
- ParseMetricError
- ParseMetricUnitError
- SetView
- TryFromFloatError
- UnixTimestamp
- aggregator::AggregateMetricsError
- aggregator::Aggregator
- aggregator::AggregatorConfig
- cogs::ByCount
- cogs::BySize
- meta::Location
- meta::MetaAggregator
- meta::MetricMeta
- meta::RedisMetricMetaStore
- meta::StartOfDayUnixTimestamp
Enums
- BucketValue
- BucketViewValue
- DurationUnit
- FractionUnit
- InformationUnit
- MetricNamespace
- MetricType
- MetricUnit
- UnescapeError
- aggregator::FlushBatching
- meta::Item
Macros
Functions
Type Aliases
List of all items in this crate List of all items
Structs
- Bucket
- BucketMetadata
- BucketView
- BucketsView
- BucketsViewBySizeIter
- BucketsViewIter
- CustomUnit
- FiniteF64
- GaugeValue
- MetricName
- MetricResourceIdentifier
- ParseBuckets
- ParseFiniteFloatError
- ParseMetricError
- ParseMetricUnitError
- SetView
- TryFromFloatError
- UnixTimestamp
- aggregator::Aggregator
- aggregator::AggregatorConfig
- cogs::ByCount
- cogs::BySize
- meta::Location
- meta::MetaAggregator
- meta::MetricMeta
- meta::RedisMetricMetaStore
- meta::StartOfDayUnixTimestamp
Enums
- BucketValue
- BucketViewValue
- DurationUnit
- FractionUnit
- InformationUnit
- MetricNamespace
- MetricType
- MetricUnit
- UnescapeError
- aggregator::AggregateMetricsError
- aggregator::FlushBatching
- meta::Item
Macros
Functions
Type Aliases
process_check_in in relay_monitors - Rust Function relay_monitors::process_check_in
source · pub fn process_check_in(
payload: &[u8],
- project_id: ProjectId,
+ project_id: ProjectId,
) -> Result<ProcessedCheckInResult, ProcessCheckInError>
Expand description
Normalizes a monitor check-in payload.
otel_to_sentry_span in relay_spans - Rust Function relay_spans::otel_to_sentry_span
source · pub fn otel_to_sentry_span(otel_span: Span) -> Span
Expand description
Transform an OtelSpan to a Sentry span.
+otel_to_sentry_span in relay_spans - Rust Function relay_spans::otel_to_sentry_span
source · pub fn otel_to_sentry_span(otel_span: Span) -> Span
Expand description
Transform an OtelSpan to a Sentry span.
"bit"), corresponding to 1/8 of a byte.\nAn aggregation of metric values.\nRelay internal metadata for a metric bucket.\nThe aggregated value of a metric bucket.\nA view into a metrics bucket. Sometimes also called a …\nA view into the datapoints of a BucketValue.\nA view into a slice of metric buckets.\nIterator slicing a BucketsView into smaller views …\nIterator yielding all items contained in a BucketsView.\nByte ("byte").\nCounts instances of an event.\nCounts instances of an event (MetricType::Counter).\nA counter metric.\nType used for Counter metric\nUser-defined metrics directly sent by SDKs and …\nUser-defined units without built-in conversion or default.\nCustom user-defined units without builtin conversion.\nDay ("day"), 86,400 seconds.\nBuilds a statistical distribution over values reported.\nBuilds a statistical distribution over values reported (…\nA distribution metric.\nType of distribution entries\nA distribution of values within a Bucket.\nA time duration, defaulting to "millisecond".\nTime duration units used in MetricUnit::Duration.\nSmallest positive normal value.\nExabyte ("exabyte"), 10^18 bytes.\nExbibyte ("exbibyte"), 2^60 bytes.\nA finite 64-bit floating point type.\nFractions such as percentages, defaulting to "ratio".\nUnits of fraction used in MetricUnit::Fraction.\nStores absolute snapshots of values.\nStores absolute snapshots of values.\nA gauage metric.\nType used for Gauge entries\nA snapshot of values within a Bucket.\nGibibyte ("gibibyte"), 2^30 bytes.\nGigabyte ("gigabyte"), 10^9 bytes.\nHour ("hour"), 3600 seconds.\nSize of information derived from bytes, defaulting to …\nSize of information derived from bytes, used in …\nKibibyte ("kibibyte"), 2^10 bytes.\nKilobyte ("kilobyte"), 10^3 bytes.\nLargest finite value.\nSmallest finite value.\nMebibyte ("mebibyte"), 2^20 bytes.\nMegabyte ("megabyte"), 10^6 bytes.\nOptimized string represenation of a metric name.\nThe namespace of a metric.\nA unique identifier for metrics including typing and …\nThe type of a MetricResourceIdentifier, determining its …\nThe unit of measurement of a metric value.\nMicrosecond ("microsecond"), 10^-6 seconds.\nMillisecond ("millisecond"), 10^-3 seconds.\nMinute ("minute"), 60 seconds.\nNanosecond ("nanosecond"), 10^-9 seconds.\nUntyped value without a unit ("").\nIterator over parsed metrics returned from …\nError type returned when parsing FiniteF64 fails.\nAn error returned when metrics or MRIs cannot be parsed.\nAn error parsing a MetricUnit or one of its variants.\nPebibyte ("pebibyte"), 2^50 bytes.\nRatio expressed as a fraction of 100. 100% equals a ratio …\nPetabyte ("petabyte"), 10^15 bytes.\nMetrics extracted from profile functions.\nFloating point fraction of 1.\nFull second ("second").\nMetrics extracted from sessions.\nCounts the number of unique reported values.\nCounts the number of unique reported values.\nA set metric.\nType used for set elements in Set metric\nA set of unique values.\nA view into the datapoints of a set metric.\nMetrics extracted from spans.\nMetric stats.\nTebibyte ("tebibyte"), 2^40 bytes.\nTerabyte ("terabyte"), 10^12 bytes.\nMetrics extracted from transaction events.\nError type returned when conversion to FiniteF64 fails.\nUnescaper’s Error.\nA unix timestamp (full seconds elapsed since 1970-01-01 …\nAn unknown and unsupported metric.\nWeek ("week"), 604,800 seconds.\nComputes the absolute value of self.\nCore functionality of metrics aggregation.\nReturns all namespaces/variants of this enum.\nReturns the timestamp as chrono datetime.\nReturns the number of seconds since the UNIX epoch start.\nReturns the same bucket view as a bucket view over a slice.\nReturn the shortcode for this metric type.\nReturns the string representation for this metric type.\nReturns the string representation for this metric unit.\nReturns the string representation for this duration unit.\nReturns the string representation for this information …\nReturns the string representation for this fraction unit.\nReturns the string representation of this unit.\nReturns the average of all values reported in this bucket.\nIterator which slices the source view into segments with …\nCOGS related metric utilities.\nEstimates the number of bytes needed to encode the bucket …\nThe number of times this bucket was updated with a new …\nReturns a bucket value representing a counter with the …\nCreates a DistributionValue containing the given arguments.\nCreates a DistributionValue containing the given arguments.\nReturns a bucket value representing a distribution with a …\nEstimates the number of bytes needed to serialize the …\nIs true if this metric was extracted from a …\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nCreates a unix timestamp from the given chrono DateTime.\nConverts the given Instant into a UNIX timestamp.\nCreates a unix timestamp from the given number of seconds.\nCreates a unix timestamp from the given system time.\nReturns a bucket value representing a gauge with a single …\nReturns true if metric stats are enabled for this …\nInserts a new value into the gauge.\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nConverts the MRI into an owned version with a static …\nWhether the metadata does not contain more information …\nReturns true if this bucket contains no values.\nReturns whether the view contains any buckets.\nReturns true if this bucket view contains no values.\nReturns true if this set contains no values.\nReturns true if the metric_unit is None.\nIterator over all buckets in the view.\nIterator over all datapoints contained in this set metric.\nThe last value reported in the bucket.\nReturns the number of raw data points in this value.\nReturns the amount of partial or full buckets in the view.\nNumber of raw datapoints in this view.\nAmount of datapoints contained within the set view.\nReturns the maximum of two numbers.\nThe maximum value reported in the bucket.\nMerges two gauge snapshots.\nMerges the given bucket_value into self.\nMerges another metadata object into the current one.\nHow many times the bucket was merged.\nFunctionality for aggregating and storing of metrics …\nReturns the metadata for this bucket.\nRelay internal metadata for a metric bucket.\nReturns the minimum of two numbers.\nThe minimum value reported in the bucket.\nName of the bucket.\nThe display name of the metric in the allowed character …\nThe name of the metric in MRI (metric resource identifier) …\nExtracts the namespace from a well formed MRI.\nThe namespace for this metric.\nCreates a fresh metadata instance.\nCreates a finite float if the value is finite.\nCreates a new buckets view containing all data from the …\nCreates a new bucket view of a bucket.\nCreates a finite float without checking whether the value …\nReturns the current timestamp.\nParses and validates an MRI.\nParses a CustomUnit from a string.\nParses a single metric aggregate from the raw protocol.\nParses a set of metric aggregates from the raw protocol.\nParses an MRI from a string and a separate type.\nReceived timestamp of the first metric in this bucket.\nRemoves the value of the specified tag.\nAdds two numbers, saturating at the maximum and minimum …\nDivides two numbers, saturating at the maximum and minimum …\nMultiplies two numbers, saturating at the maximum and …\nAdds two numbers, saturating at the maximum and minimum …\nSelects a sub-view of the current view.\nReturns a bucket value representing a set with a single …\nReturns a bucket value representing a set with a single …\nReturns a bucket value representing a set with a single …\nCreates a gauge snapshot from a single value.\nCalculates a split for this bucket if its estimated …\nThe sum of all values reported in the bucket.\nReturns the value of the specified tag if it exists.\nReturns the value of the specified tag if it exists.\nName of the bucket.\nA list of tags adding dimensions to the metric for …\nTimestamp of the bucket.\nThe start time of the bucket’s time window.\nReturns the plain f64.\nExtracts the namespace from a well formed MRI.\nExtracts the type from a well formed MRI.\nReturns the type of this value.\nType of the value of the bucket view.\nThe type of a metric, determining its aggregation and …\nThe verbatim unit name of the metric value.\nValue of the bucket view.\nThe type and aggregated values of this bucket.\nWidth of the bucket.\nThe length of the time window in seconds.\nAny error that may occur during aggregation.\nA collector of Bucket submissions.\nParameters used by the Aggregator.\nShifts the flush time by an offset based on the bucket key …\nConfiguration value for AggregatorConfig::flush_batching.\nDo not apply shift.\nShifts the flush time by an offset based on the partition …\nShifts the flush time by an offset based on the ProjectKey.\nReturns the number of buckets in the aggregator.\nDetermines the wall clock time interval for buckets in …\nThe batching mode for the flushing of the aggregator.\nThe number of logical partitions that can receive flushed …\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nThe initial delay in seconds to wait before flushing a …\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nConverts this aggregator into a vector of Bucket.\nThe length the name of a metric is allowed to be.\nMaximum amount of bytes used for metrics aggregation per …\nThe time in seconds that a timestamp may be in the future.\nThe age in seconds of the oldest allowed bucket timestamp.\nThe length the tag key is allowed to be.\nThe length the tag value is allowed to be.\nMerge a preaggregated bucket into this aggregator.\nMerges all given buckets into this aggregator.\nReturns the name of the aggregator.\nLike Self::new, but with a provided name.\nCreate a new aggregator.\nPop and return the partitions with buckets that are …\nEstimates the number of bytes needed to encode the tags.\nReturns the valid range for metrics timestamps.\nReturns true if the cost trackers value is larger than the …\nCOGS estimator based on the bucket count.\nCOGS estimator based on the estimated size of each bucket …\nReturns the argument unchanged.\nReturns the argument unchanged.\nCalls U::from(self).\nCalls U::from(self).\nA metadata item.\nA code location.\nA location metadata pointing to the code location where …\nA metrics meta aggregator.\nA metric metadata item.\nRedis metric meta\nA Unix timestamp that is truncated to the start of the day.\nUnknown item.\nThe absolute file path.\nAdds a new meta item to the aggregator.\nReturns the underlying unix timestamp, truncated to the …\nRemove all contained state related to a project.\nSource code of the current line (lineno).\nThe relative file path.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nThe containing function name.\nRetrieves all currently relevant metric meta for a project.\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nThe line number.\nThe contained metadata mapped by MRI.\nThe containing module name or path.\nCreates a new Redis metrics meta store.\nCreates a new metrics meta aggregator.\nCreates a new StartOfDayUnixTimestamp from a timestamp by …\nSource code of the lines after lineno.\nSource code leading up to lineno.\nStores metric metadata in Redis.\nTimestamp scope for the contained metadata.")
\ No newline at end of file
+searchState.loadedDescShard("relay_metrics", 0, "Metric protocol, aggregation and processing for Sentry.\nBit ("bit"), corresponding to 1/8 of a byte.\nAn aggregation of metric values.\nRelay internal metadata for a metric bucket.\nThe aggregated value of a metric bucket.\nA view into a metrics bucket. Sometimes also called a …\nA view into the datapoints of a BucketValue.\nA view into a slice of metric buckets.\nIterator slicing a BucketsView into smaller views …\nIterator yielding all items contained in a BucketsView.\nByte ("byte").\nCounts instances of an event.\nCounts instances of an event (MetricType::Counter).\nA counter metric.\nType used for Counter metric\nUser-defined metrics directly sent by SDKs and …\nUser-defined units without built-in conversion or default.\nCustom user-defined units without builtin conversion.\nDay ("day"), 86,400 seconds.\nBuilds a statistical distribution over values reported.\nBuilds a statistical distribution over values reported (…\nA distribution metric.\nType of distribution entries\nA distribution of values within a Bucket.\nA time duration, defaulting to "millisecond".\nTime duration units used in MetricUnit::Duration.\nSmallest positive normal value.\nExabyte ("exabyte"), 10^18 bytes.\nExbibyte ("exbibyte"), 2^60 bytes.\nA finite 64-bit floating point type.\nFractions such as percentages, defaulting to "ratio".\nUnits of fraction used in MetricUnit::Fraction.\nStores absolute snapshots of values.\nStores absolute snapshots of values.\nA gauage metric.\nType used for Gauge entries\nA snapshot of values within a Bucket.\nGibibyte ("gibibyte"), 2^30 bytes.\nGigabyte ("gigabyte"), 10^9 bytes.\nHour ("hour"), 3600 seconds.\nSize of information derived from bytes, defaulting to …\nSize of information derived from bytes, used in …\nKibibyte ("kibibyte"), 2^10 bytes.\nKilobyte ("kilobyte"), 10^3 bytes.\nLargest finite value.\nSmallest finite value.\nMebibyte ("mebibyte"), 2^20 bytes.\nMegabyte ("megabyte"), 10^6 bytes.\nOptimized string represenation of a metric name.\nThe namespace of a metric.\nA unique identifier for metrics including typing and …\nThe type of a MetricResourceIdentifier, determining its …\nThe unit of measurement of a metric value.\nMicrosecond ("microsecond"), 10^-6 seconds.\nMillisecond ("millisecond"), 10^-3 seconds.\nMinute ("minute"), 60 seconds.\nNanosecond ("nanosecond"), 10^-9 seconds.\nUntyped value without a unit ("").\nIterator over parsed metrics returned from …\nError type returned when parsing FiniteF64 fails.\nAn error returned when metrics or MRIs cannot be parsed.\nAn error parsing a MetricUnit or one of its variants.\nPebibyte ("pebibyte"), 2^50 bytes.\nRatio expressed as a fraction of 100. 100% equals a ratio …\nPetabyte ("petabyte"), 10^15 bytes.\nMetrics extracted from profile functions.\nFloating point fraction of 1.\nFull second ("second").\nMetrics extracted from sessions.\nCounts the number of unique reported values.\nCounts the number of unique reported values.\nA set metric.\nType used for set elements in Set metric\nA set of unique values.\nA view into the datapoints of a set metric.\nMetrics extracted from spans.\nMetric stats.\nTebibyte ("tebibyte"), 2^40 bytes.\nTerabyte ("terabyte"), 10^12 bytes.\nMetrics extracted from transaction events.\nError type returned when conversion to FiniteF64 fails.\nUnescaper’s Error.\nA unix timestamp (full seconds elapsed since 1970-01-01 …\nAn unknown and unsupported metric.\nWeek ("week"), 604,800 seconds.\nComputes the absolute value of self.\nCore functionality of metrics aggregation.\nReturns all namespaces/variants of this enum.\nReturns the timestamp as chrono datetime.\nReturns the number of seconds since the UNIX epoch start.\nReturns the same bucket view as a bucket view over a slice.\nReturn the shortcode for this metric type.\nReturns the string representation for this metric type.\nReturns the string representation for this metric unit.\nReturns the string representation for this duration unit.\nReturns the string representation for this information …\nReturns the string representation for this fraction unit.\nReturns the string representation of this unit.\nReturns the average of all values reported in this bucket.\nIterator which slices the source view into segments with …\nCOGS related metric utilities.\nEstimates the number of bytes needed to encode the bucket …\nThe number of times this bucket was updated with a new …\nReturns a bucket value representing a counter with the …\nCreates a DistributionValue containing the given arguments.\nCreates a DistributionValue containing the given arguments.\nReturns a bucket value representing a distribution with a …\nEstimates the number of bytes needed to serialize the …\nIs true if this metric was extracted from a …\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nCreates a unix timestamp from the given chrono DateTime.\nConverts the given Instant into a UNIX timestamp.\nCreates a unix timestamp from the given number of seconds.\nCreates a unix timestamp from the given system time.\nReturns a bucket value representing a gauge with a single …\nReturns true if metric stats are enabled for this …\nInserts a new value into the gauge.\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nConverts the MRI into an owned version with a static …\nWhether the metadata does not contain more information …\nReturns true if this bucket contains no values.\nReturns whether the view contains any buckets.\nReturns true if this bucket view contains no values.\nReturns true if this set contains no values.\nReturns true if the metric_unit is None.\nIterator over all buckets in the view.\nIterator over all datapoints contained in this set metric.\nThe last value reported in the bucket.\nReturns the number of raw data points in this value.\nReturns the amount of partial or full buckets in the view.\nNumber of raw datapoints in this view.\nAmount of datapoints contained within the set view.\nReturns the maximum of two numbers.\nThe maximum value reported in the bucket.\nMerges two gauge snapshots.\nMerges the given bucket_value into self.\nMerges another metadata object into the current one.\nHow many times the bucket was merged.\nFunctionality for aggregating and storing of metrics …\nReturns the metadata for this bucket.\nRelay internal metadata for a metric bucket.\nReturns the minimum of two numbers.\nThe minimum value reported in the bucket.\nName of the bucket.\nThe display name of the metric in the allowed character …\nThe name of the metric in MRI (metric resource identifier) …\nExtracts the namespace from a well formed MRI.\nThe namespace for this metric.\nCreates a fresh metadata instance.\nCreates a finite float if the value is finite.\nCreates a new buckets view containing all data from the …\nCreates a new bucket view of a bucket.\nCreates a finite float without checking whether the value …\nReturns the current timestamp.\nParses and validates an MRI.\nParses a CustomUnit from a string.\nParses a single metric aggregate from the raw protocol.\nParses a set of metric aggregates from the raw protocol.\nParses an MRI from a string and a separate type.\nReceived timestamp of the first metric in this bucket.\nRemoves the value of the specified tag.\nAdds two numbers, saturating at the maximum and minimum …\nDivides two numbers, saturating at the maximum and minimum …\nMultiplies two numbers, saturating at the maximum and …\nAdds two numbers, saturating at the maximum and minimum …\nSelects a sub-view of the current view.\nReturns a bucket value representing a set with a single …\nReturns a bucket value representing a set with a single …\nReturns a bucket value representing a set with a single …\nCreates a gauge snapshot from a single value.\nCalculates a split for this bucket if its estimated …\nThe sum of all values reported in the bucket.\nReturns the value of the specified tag if it exists.\nReturns the value of the specified tag if it exists.\nName of the bucket.\nA list of tags adding dimensions to the metric for …\nTimestamp of the bucket.\nThe start time of the bucket’s time window.\nReturns the plain f64.\nExtracts the namespace from a well formed MRI.\nExtracts the type from a well formed MRI.\nReturns the type of this value.\nType of the value of the bucket view.\nThe type of a metric, determining its aggregation and …\nThe verbatim unit name of the metric value.\nValue of the bucket view.\nThe type and aggregated values of this bucket.\nWidth of the bucket.\nThe length of the time window in seconds.\nAny error that may occur during aggregation.\nA collector of Bucket submissions.\nParameters used by the Aggregator.\nShifts the flush time by an offset based on the bucket key …\nConfiguration value for AggregatorConfig::flush_batching.\nA metric bucket had invalid characters in the metric name.\nA metric bucket had a too long string (metric name or a …\nA metric bucket’s timestamp was out of the configured …\nInternal error: Attempted to merge two metric buckets of …\nDo not apply shift.\nShifts the flush time by an offset based on the partition …\nShifts the flush time by an offset based on the ProjectKey.\nA metric bucket is too large for the per-project bytes …\nA metric bucket is too large for the global bytes limit.\nA metric bucket had an unknown namespace in the metric …\nReturns the number of buckets in the aggregator.\nDetermines the wall clock time interval for buckets in …\nThe batching mode for the flushing of the aggregator.\nThe number of logical partitions that can receive flushed …\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nThe initial delay in seconds to wait before flushing a …\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nConverts this aggregator into a vector of Bucket.\nThe length the name of a metric is allowed to be.\nMaximum amount of bytes used for metrics aggregation per …\nThe time in seconds that a timestamp may be in the future.\nThe age in seconds of the oldest allowed bucket timestamp.\nThe length the tag key is allowed to be.\nThe length the tag value is allowed to be.\nMerge a preaggregated bucket into this aggregator.\nReturns the name of the aggregator.\nLike Self::new, but with a provided name.\nCreate a new aggregator.\nPop and return the partitions with buckets that are …\nEstimates the number of bytes needed to encode the tags.\nReturns the valid range for metrics timestamps.\nReturns true if the cost trackers value is larger than the …\nCOGS estimator based on the bucket count.\nCOGS estimator based on the estimated size of each bucket …\nReturns the argument unchanged.\nReturns the argument unchanged.\nCalls U::from(self).\nCalls U::from(self).\nA metadata item.\nA code location.\nA location metadata pointing to the code location where …\nA metrics meta aggregator.\nA metric metadata item.\nRedis metric meta\nA Unix timestamp that is truncated to the start of the day.\nUnknown item.\nThe absolute file path.\nAdds a new meta item to the aggregator.\nReturns the underlying unix timestamp, truncated to the …\nRemove all contained state related to a project.\nSource code of the current line (lineno).\nThe relative file path.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nReturns the argument unchanged.\nThe containing function name.\nRetrieves all currently relevant metric meta for a project.\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nCalls U::from(self).\nThe line number.\nThe contained metadata mapped by MRI.\nThe containing module name or path.\nCreates a new Redis metrics meta store.\nCreates a new metrics meta aggregator.\nCreates a new StartOfDayUnixTimestamp from a timestamp by …\nSource code of the lines after lineno.\nSource code leading up to lineno.\nStores metric metadata in Redis.\nTimestamp scope for the contained metadata.")
\ No newline at end of file
diff --git a/src/relay_metrics/aggregator.rs.html b/src/relay_metrics/aggregator.rs.html
index bb2c1f7343..c49b415348 100644
--- a/src/relay_metrics/aggregator.rs.html
+++ b/src/relay_metrics/aggregator.rs.html
@@ -1681,58 +1681,9 @@
1681
1682
1683
-1684
-1685
-1686
-1687
-1688
-1689
-1690
-1691
-1692
-1693
-1694
-1695
-1696
-1697
-1698
-1699
-1700
-1701
-1702
-1703
-1704
-1705
-1706
-1707
-1708
-1709
-1710
-1711
-1712
-1713
-1714
-1715
-1716
-1717
-1718
-1719
-1720
-1721
-1722
-1723
-1724
-1725
-1726
-1727
-1728
-1729
-1730
-1731
//! Core functionality of metrics aggregation.
use std::collections::BTreeMap;
-use std::error::Error;
use std::hash::{Hash, Hasher};
use std::time::Duration;
use std::{fmt, mem};
@@ -1754,20 +1705,7 @@
/// Any error that may occur during aggregation.
#[derive(Debug, Error, PartialEq)]
-#[error("failed to aggregate metrics: {kind}")]
-pub struct AggregateMetricsError {
- kind: AggregateMetricsErrorKind,
-}
-
-impl From<AggregateMetricsErrorKind> for AggregateMetricsError {
- fn from(kind: AggregateMetricsErrorKind) -> Self {
- AggregateMetricsError { kind }
- }
-}
-
-#[derive(Debug, Error, PartialEq)]
-#[allow(clippy::enum_variant_names)]
-enum AggregateMetricsErrorKind {
+pub enum AggregateMetricsError {
/// A metric bucket had invalid characters in the metric name.
#[error("found invalid characters: {0}")]
InvalidCharacters(MetricName),
@@ -2067,12 +2005,12 @@
&mut self,
value: BucketValue,
metadata: BucketMetadata,
- ) -> Result<usize, AggregateMetricsErrorKind> {
+ ) -> Result<usize, AggregateMetricsError> {
let cost_before = self.value.cost();
self.value
.merge(value)
- .map_err(|_| AggregateMetricsErrorKind::InvalidTypes)?;
+ .map_err(|_| AggregateMetricsError::InvalidTypes)?;
self.metadata.merge(metadata);
Ok(self.value.cost().saturating_sub(cost_before))
@@ -2129,7 +2067,7 @@
relay_log::configure_scope(|scope| {
scope.set_extra("bucket.project_key", project_key.as_str().to_owned().into());
});
- return Err(AggregateMetricsErrorKind::TotalLimitExceeded.into());
+ return Err(AggregateMetricsError::TotalLimitExceeded);
}
if let Some(max_project_cost) = max_project_cost {
@@ -2142,7 +2080,7 @@
relay_log::configure_scope(|scope| {
scope.set_extra("bucket.project_key", project_key.as_str().to_owned().into());
});
- return Err(AggregateMetricsErrorKind::ProjectLimitExceeded.into());
+ return Err(AggregateMetricsError::ProjectLimitExceeded);
}
}
@@ -2435,7 +2373,7 @@
aggregator = &self.name,
);
- return Err(AggregateMetricsErrorKind::InvalidTimestamp(timestamp).into());
+ return Err(AggregateMetricsError::InvalidTimestamp(timestamp));
}
Ok(bucket_ts)
@@ -2514,7 +2452,7 @@
});
if let Some(error) = error {
- return Err(error.into());
+ return Err(error);
}
if !updated {
@@ -2542,31 +2480,6 @@
Ok(())
}
-
- /// Merges all given `buckets` into this aggregator.
- ///
- /// Buckets that do not exist yet will be created.
- pub fn merge_all(
- &mut self,
- project_key: ProjectKey,
- buckets: impl IntoIterator<Item = Bucket>,
- max_total_bucket_bytes: Option<usize>,
- ) {
- for bucket in buckets.into_iter() {
- if let Err(error) = self.merge(project_key, bucket, max_total_bucket_bytes) {
- match &error.kind {
- // Ignore invalid timestamp errors.
- AggregateMetricsErrorKind::InvalidTimestamp(_) => {}
- _other => {
- relay_log::error!(
- tags.aggregator = self.name,
- bucket.error = &error as &dyn Error
- );
- }
- }
- }
- }
- }
}
impl fmt::Debug for Aggregator {
@@ -2605,7 +2518,7 @@
aggregator_config.max_name_length,
key.metric_name
);
- return Err(AggregateMetricsErrorKind::InvalidStringLength(key.metric_name).into());
+ return Err(AggregateMetricsError::InvalidStringLength(key.metric_name));
}
normalize_metric_name(&mut key)?;
@@ -2618,16 +2531,16 @@
Ok(mri) => {
if matches!(mri.namespace, MetricNamespace::Unsupported) {
relay_log::debug!("invalid metric namespace {:?}", &key.metric_name);
- return Err(AggregateMetricsErrorKind::UnsupportedNamespace(mri.namespace).into());
+ return Err(AggregateMetricsError::UnsupportedNamespace(mri.namespace));
}
mri.to_string().into()
}
Err(_) => {
relay_log::debug!("invalid metric name {:?}", &key.metric_name);
- return Err(
- AggregateMetricsErrorKind::InvalidCharacters(key.metric_name.clone()).into(),
- );
+ return Err(AggregateMetricsError::InvalidCharacters(
+ key.metric_name.clone(),
+ ));
}
};
@@ -3137,9 +3050,8 @@
assert!(matches!(
aggregator
.get_bucket_timestamp(UnixTimestamp::from_secs(u64::MAX), 2)
- .unwrap_err()
- .kind,
- AggregateMetricsErrorKind::InvalidTimestamp(_)
+ .unwrap_err(),
+ AggregateMetricsError::InvalidTimestamp(_)
));
}
@@ -3265,9 +3177,7 @@
assert!(matches!(
validation.unwrap_err(),
- AggregateMetricsError {
- kind: AggregateMetricsErrorKind::InvalidStringLength(_)
- }
+ AggregateMetricsError::InvalidStringLength(_),
));
let short_metric_long_tag_key = BucketKey {
@@ -3329,11 +3239,8 @@
.unwrap();
assert_eq!(
- aggregator
- .merge(project_key, bucket, Some(1))
- .unwrap_err()
- .kind,
- AggregateMetricsErrorKind::TotalLimitExceeded
+ aggregator.merge(project_key, bucket, Some(1)).unwrap_err(),
+ AggregateMetricsError::TotalLimitExceeded
);
}
@@ -3358,11 +3265,8 @@
aggregator.merge(project_key, bucket.clone(), None).unwrap();
assert_eq!(
- aggregator
- .merge(project_key, bucket, None)
- .unwrap_err()
- .kind,
- AggregateMetricsErrorKind::ProjectLimitExceeded
+ aggregator.merge(project_key, bucket, None).unwrap_err(),
+ AggregateMetricsError::ProjectLimitExceeded
);
}
diff --git a/src/relay_server/service.rs.html b/src/relay_server/service.rs.html
index 439681cfa8..250f2df974 100644
--- a/src/relay_server/service.rs.html
+++ b/src/relay_server/service.rs.html
@@ -431,6 +431,7 @@
431
432
433
+434
use std::convert::Infallible;
use std::fmt;
use std::sync::Arc;
@@ -623,8 +624,9 @@
config.default_aggregator_config().clone(),
config.secondary_aggregator_configs().clone(),
Some(project_cache.clone().recipient()),
- )
- .start();
+ );
+ let aggregator_handle = aggregator.handle();
+ let aggregator = aggregator.start();
let metric_stats = MetricStats::new(
config.clone(),
@@ -701,7 +703,7 @@
let health_check = HealthCheckService::new(
config.clone(),
MemoryChecker::new(memory_stat.clone(), config.clone()),
- aggregator.clone(),
+ aggregator_handle,
upstream_relay.clone(),
project_cache.clone(),
)
diff --git a/src/relay_server/services/health_check.rs.html b/src/relay_server/services/health_check.rs.html
index b1be357656..dd8dfa92c4 100644
--- a/src/relay_server/services/health_check.rs.html
+++ b/src/relay_server/services/health_check.rs.html
@@ -263,9 +263,6 @@
263
264
265
-266
-267
-268
use std::sync::Arc;
use relay_config::Config;
@@ -274,7 +271,7 @@
use tokio::sync::watch;
use tokio::time::{timeout, Instant};
-use crate::services::metrics::{AcceptsMetrics, Aggregator};
+use crate::services::metrics::RouterHandle;
use crate::services::project_cache::{ProjectCache, SpoolHealth};
use crate::services::upstream::{IsAuthenticated, UpstreamRelay};
use crate::statsd::RelayTimers;
@@ -352,7 +349,7 @@
pub struct HealthCheckService {
config: Arc<Config>,
memory_checker: MemoryChecker,
- aggregator: Addr<Aggregator>,
+ aggregator: RouterHandle,
upstream_relay: Addr<UpstreamRelay>,
project_cache: Addr<ProjectCache>,
}
@@ -364,7 +361,7 @@
pub fn new(
config: Arc<Config>,
memory_checker: MemoryChecker,
- aggregator: Addr<Aggregator>,
+ aggregator: RouterHandle,
upstream_relay: Addr<UpstreamRelay>,
project_cache: Addr<ProjectCache>,
) -> Self {
@@ -415,10 +412,7 @@
}
async fn aggregator_probe(&self) -> Status {
- self.aggregator
- .send(AcceptsMetrics)
- .await
- .map_or(Status::Unhealthy, Status::from)
+ Status::from(self.aggregator.can_accept_metrics())
}
async fn spool_health_probe(&self) -> Status {
diff --git a/src/relay_server/services/metrics/aggregator.rs.html b/src/relay_server/services/metrics/aggregator.rs.html
index da02722cd6..9c93873f3b 100644
--- a/src/relay_server/services/metrics/aggregator.rs.html
+++ b/src/relay_server/services/metrics/aggregator.rs.html
@@ -393,16 +393,57 @@
393
394
395
-
use std::time::Duration;
+396
+397
+398
+399
+400
+401
+402
+403
+404
+405
+406
+407
+408
+409
+410
+411
+412
+413
+414
+415
+416
+417
+418
+419
+420
+421
+422
+423
+424
+425
+426
+427
+428
+429
+430
+431
+432
+433
+434
+435
+436
+
use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+use std::time::Duration;
use hashbrown::HashMap;
use relay_base_schema::project::ProjectKey;
use relay_config::AggregatorServiceConfig;
+use relay_metrics::aggregator::AggregateMetricsError;
use relay_metrics::{aggregator, Bucket};
-use relay_system::{
- AsyncResponse, Controller, FromMessage, Interface, NoResponse, Recipient, Sender, Service,
- Shutdown,
-};
+use relay_system::{Controller, FromMessage, Interface, NoResponse, Recipient, Service, Shutdown};
use crate::statsd::{RelayCounters, RelayHistograms, RelayTimers};
@@ -418,21 +459,18 @@
/// Receivers must implement a handler for the [`FlushBuckets`] message.
#[derive(Debug)]
pub enum Aggregator {
- /// The health check message which makes sure that the service can accept the requests now.
- AcceptsMetrics(AcceptsMetrics, Sender<bool>),
/// Merge the buckets.
MergeBuckets(MergeBuckets),
/// Message is used only for tests to get the current number of buckets in `AggregatorService`.
#[cfg(test)]
- BucketCountInquiry(BucketCountInquiry, Sender<usize>),
+ BucketCountInquiry(BucketCountInquiry, relay_system::Sender<usize>),
}
impl Aggregator {
/// Returns the name of the message variant.
pub fn variant(&self) -> &'static str {
match self {
- Aggregator::AcceptsMetrics(_, _) => "AcceptsMetrics",
Aggregator::MergeBuckets(_) => "MergeBuckets",
#[cfg(test)]
Aggregator::BucketCountInquiry(_, _) => "BucketCountInquiry",
@@ -442,13 +480,6 @@
impl Interface for Aggregator {}
-impl FromMessage<AcceptsMetrics> for Aggregator {
- type Response = AsyncResponse<bool>;
- fn from_message(message: AcceptsMetrics, sender: Sender<bool>) -> Self {
- Self::AcceptsMetrics(message, sender)
- }
-}
-
impl FromMessage<MergeBuckets> for Aggregator {
type Response = NoResponse;
fn from_message(message: MergeBuckets, _: ()) -> Self {
@@ -458,16 +489,12 @@
#[cfg(test)]
impl FromMessage<BucketCountInquiry> for Aggregator {
- type Response = AsyncResponse<usize>;
- fn from_message(message: BucketCountInquiry, sender: Sender<usize>) -> Self {
+ type Response = relay_system::AsyncResponse<usize>;
+ fn from_message(message: BucketCountInquiry, sender: relay_system::Sender<usize>) -> Self {
Self::BucketCountInquiry(message, sender)
}
}
-/// Check whether the aggregator has not (yet) exceeded its total limits. Used for health checks.
-#[derive(Debug)]
-pub struct AcceptsMetrics;
-
/// Used only for testing the `AggregatorService`.
#[cfg(test)]
#[derive(Debug)]
@@ -501,6 +528,7 @@
receiver: Option<Recipient<FlushBuckets, NoResponse>>,
max_total_bucket_bytes: Option<usize>,
flush_interval_ms: u64,
+ can_accept_metrics: Arc<AtomicBool>,
}
impl AggregatorService {
@@ -521,21 +549,23 @@
config: AggregatorServiceConfig,
receiver: Option<Recipient<FlushBuckets, NoResponse>>,
) -> Self {
+ let aggregator = aggregator::Aggregator::named(name, config.aggregator);
Self {
receiver,
state: AggregatorState::Running,
max_total_bucket_bytes: config.max_total_bucket_bytes,
- aggregator: aggregator::Aggregator::named(name, config.aggregator),
flush_interval_ms: config.flush_interval_ms,
+ can_accept_metrics: Arc::new(AtomicBool::new(
+ !aggregator.totals_cost_exceeded(config.max_total_bucket_bytes),
+ )),
+ aggregator,
}
}
- fn handle_accepts_metrics(&self, sender: Sender<bool>) {
- let result = !self
- .aggregator
- .totals_cost_exceeded(self.max_total_bucket_bytes);
-
- sender.send(result);
+ pub fn handle(&self) -> AggregatorHandle {
+ AggregatorHandle {
+ can_accept_metrics: Arc::clone(&self.can_accept_metrics),
+ }
}
/// Sends the [`FlushBuckets`] message to the receiver in the fire and forget fashion. It is up
@@ -547,6 +577,13 @@
let force_flush = matches!(&self.state, AggregatorState::ShuttingDown);
let partitions = self.aggregator.pop_flush_buckets(force_flush);
+ self.can_accept_metrics.store(
+ !self
+ .aggregator
+ .totals_cost_exceeded(self.max_total_bucket_bytes),
+ Ordering::Relaxed,
+ );
+
if partitions.is_empty() {
return;
}
@@ -591,8 +628,41 @@
project_key,
buckets,
} = msg;
- self.aggregator
- .merge_all(project_key, buckets, self.max_total_bucket_bytes);
+
+ for bucket in buckets.into_iter() {
+ match self
+ .aggregator
+ .merge(project_key, bucket, self.max_total_bucket_bytes)
+ {
+ // Ignore invalid timestamp errors.
+ Err(AggregateMetricsError::InvalidTimestamp(_)) => {}
+ Err(AggregateMetricsError::TotalLimitExceeded) => {
+ relay_log::error!(
+ tags.aggregator = self.aggregator.name(),
+ "aggregator limit exceeded"
+ );
+ self.can_accept_metrics.store(false, Ordering::Relaxed);
+ break;
+ }
+ Err(AggregateMetricsError::ProjectLimitExceeded) => {
+ relay_log::error!(
+ tags.aggregator = self.aggregator.name(),
+ tags.project_key = project_key.as_str(),
+ "project metrics limit exceeded for project {project_key}"
+ );
+ break;
+ }
+ Err(error) => {
+ relay_log::error!(
+ tags.aggregator = self.aggregator.name(),
+ tags.project_key = project_key.as_str(),
+ bucket.error = &error as &dyn std::error::Error,
+ "failed to aggregate metric bucket"
+ );
+ }
+ Ok(()) => {}
+ };
+ }
}
fn handle_message(&mut self, message: Aggregator) {
@@ -602,7 +672,6 @@
message = ty,
{
match message {
- Aggregator::AcceptsMetrics(_, sender) => self.handle_accepts_metrics(sender),
Aggregator::MergeBuckets(msg) => self.handle_merge_buckets(msg),
#[cfg(test)]
Aggregator::BucketCountInquiry(_, sender) => {
@@ -678,6 +747,19 @@
}
}
+/// Provides sync access to the state of the [`AggregatorService`].
+#[derive(Debug, Clone)]
+pub struct AggregatorHandle {
+ can_accept_metrics: Arc<AtomicBool>,
+}
+
+impl AggregatorHandle {
+ /// Returns `true` if the aggregator can still accept metrics.
+ pub fn can_accept_metrics(&self) -> bool {
+ self.can_accept_metrics.load(Ordering::Relaxed)
+ }
+}
+
#[cfg(test)]
mod tests {
use std::collections::BTreeMap;
diff --git a/src/relay_server/services/metrics/router.rs.html b/src/relay_server/services/metrics/router.rs.html
index e739dd8f96..5ff0014ba2 100644
--- a/src/relay_server/services/metrics/router.rs.html
+++ b/src/relay_server/services/metrics/router.rs.html
@@ -156,22 +156,16 @@
156
157
158
-159
-160
-161
-162
//! Routing logic for metrics. Metrics from different namespaces may be routed to different aggregators,
//! with their own limits, bucket intervals, etc.
-use futures::stream::FuturesUnordered;
-use futures::StreamExt;
-
+use relay_config::aggregator::Condition;
use relay_config::{AggregatorServiceConfig, ScopedAggregatorConfig};
use relay_metrics::MetricNamespace;
use relay_system::{Addr, NoResponse, Recipient, Service};
use crate::services::metrics::{
- AcceptsMetrics, Aggregator, AggregatorService, FlushBuckets, MergeBuckets,
+ Aggregator, AggregatorHandle, AggregatorService, FlushBuckets, MergeBuckets,
};
use crate::statsd::RelayTimers;
use crate::utils;
@@ -179,13 +173,11 @@
/// Service that routes metrics & metric buckets to the appropriate aggregator.
///
/// Each aggregator gets its own configuration.
-/// Metrics are routed to the first aggregator which matches the configuration's
-/// [`Condition`](relay_config::aggregator::Condition).
+/// Metrics are routed to the first aggregator which matches the configuration's [`Condition`].
/// If no condition matches, the metric/bucket is routed to the `default_aggregator`.
pub struct RouterService {
- default_config: AggregatorServiceConfig,
- secondary_configs: Vec<ScopedAggregatorConfig>,
- receiver: Option<Recipient<FlushBuckets, NoResponse>>,
+ default: AggregatorService,
+ secondary: Vec<(AggregatorService, Condition)>,
}
impl RouterService {
@@ -195,11 +187,24 @@
secondary_configs: Vec<ScopedAggregatorConfig>,
receiver: Option<Recipient<FlushBuckets, NoResponse>>,
) -> Self {
- Self {
- default_config,
- secondary_configs,
- receiver,
+ let mut secondary = Vec::new();
+
+ for c in secondary_configs {
+ let service = AggregatorService::named(c.name, c.config, receiver.clone());
+ secondary.push((service, c.condition));
+ }
+
+ let default = AggregatorService::new(default_config, receiver);
+ Self { default, secondary }
+ }
+
+ pub fn handle(&self) -> RouterHandle {
+ let mut handles = vec![self.default.handle()];
+ for (aggregator, _) in &self.secondary {
+ handles.push(aggregator.handle());
}
+
+ RouterHandle(handles)
}
}
@@ -237,30 +242,22 @@
impl StartedRouter {
fn start(router: RouterService) -> Self {
- let RouterService {
- default_config,
- secondary_configs,
- receiver,
- } = router;
+ let RouterService { default, secondary } = router;
- let secondary = secondary_configs
+ let secondary = secondary
.into_iter()
- .map(|c| {
- let addr = AggregatorService::named(c.name, c.config, receiver.clone()).start();
- (c.condition, addr)
- })
- .map(|(cond, agg)| {
+ .map(|(aggregator, condition)| {
let namespaces: Vec<_> = MetricNamespace::all()
.into_iter()
- .filter(|&namespace| cond.matches(Some(namespace)))
+ .filter(|&namespace| condition.matches(Some(namespace)))
.collect();
- (agg, namespaces)
+ (aggregator.start(), namespaces)
})
.collect();
Self {
- default: AggregatorService::new(default_config, receiver).start(),
+ default: default.start(),
secondary,
}
}
@@ -272,22 +269,6 @@
message = ty,
{
match message {
- Aggregator::AcceptsMetrics(_, sender) => {
- let mut requests = self
- .secondary
- .iter()
- .map(|(agg, _)| agg.send(AcceptsMetrics))
- .chain(Some(self.default.send(AcceptsMetrics)))
- .collect::<FuturesUnordered<_>>();
-
- tokio::spawn(async move {
- let mut accepts = true;
- while let Some(req) = requests.next().await {
- accepts &= req.unwrap_or_default();
- }
- sender.send(accepts);
- });
- }
Aggregator::MergeBuckets(msg) => self.handle_merge_buckets(msg),
#[cfg(test)]
Aggregator::BucketCountInquiry(_, _sender) => (), // not supported
@@ -322,4 +303,15 @@
}
}
}
+
+/// Provides sync access to the state of the [`RouterService`].
+#[derive(Clone, Debug)]
+pub struct RouterHandle(Vec<AggregatorHandle>);
+
+impl RouterHandle {
+ /// Returns `true` if all the aggregators can still accept metrics.
+ pub fn can_accept_metrics(&self) -> bool {
+ self.0.iter().all(|ah| ah.can_accept_metrics())
+ }
+}