Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(memory): Implement shared memory state across Relay #3821

Merged
merged 40 commits into from
Jul 22, 2024
Merged

feat(memory): Implement shared memory state across Relay #3821

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
40 commits
Select commit Hold shift + click to select a range
90fb599
feat(memory): Implement shared memory state across Relay
iambriccardo Jul 15, 2024
b0ca49d
feat(memory): Add a way to track memory usage
iambriccardo Jul 16, 2024
7c8d357
Fix
iambriccardo Jul 16, 2024
cc4b577
Fix
iambriccardo Jul 16, 2024
c449dfd
Fix
iambriccardo Jul 16, 2024
2dc4e13
Fix
iambriccardo Jul 16, 2024
6bfd213
Fix
iambriccardo Jul 16, 2024
584a868
Fix
iambriccardo Jul 16, 2024
e7608a8
Remove usage of floats
iambriccardo Jul 16, 2024
75a3ffc
Improve
iambriccardo Jul 16, 2024
1069847
Improve
iambriccardo Jul 16, 2024
719bea7
Improve
iambriccardo Jul 16, 2024
45b35ee
Improve
iambriccardo Jul 16, 2024
6a8be96
Improve
iambriccardo Jul 17, 2024
f4a7d39
fi
iambriccardo Jul 17, 2024
8ceac2e
Fix
iambriccardo Jul 17, 2024
cc0596d
Fix
iambriccardo Jul 18, 2024
1d3b3e8
Fix
iambriccardo Jul 18, 2024
fa67244
Fix
iambriccardo Jul 18, 2024
08a6e85
Fix
iambriccardo Jul 18, 2024
bd9e159
Fix
iambriccardo Jul 18, 2024
450311d
Fix
iambriccardo Jul 18, 2024
707cb07
Fix
iambriccardo Jul 18, 2024
8670f0d
Fix
iambriccardo Jul 18, 2024
dfc5a9b
Fix
iambriccardo Jul 18, 2024
464ca43
Fix
iambriccardo Jul 18, 2024
e2ca3d4
Merge
iambriccardo Jul 18, 2024
05b1317
Fix
iambriccardo Jul 18, 2024
6823d7d
Update relay-server/src/endpoints/common.rs
iambriccardo Jul 19, 2024
78abada
Update relay-server/src/utils/memory.rs
iambriccardo Jul 19, 2024
8755d87
Update relay-server/src/utils/memory.rs
iambriccardo Jul 19, 2024
f0f2c9f
Update relay-server/src/utils/memory.rs
iambriccardo Jul 19, 2024
aafba46
Update relay-server/src/utils/memory.rs
iambriccardo Jul 19, 2024
998fdf2
Update relay-server/src/utils/memory.rs
iambriccardo Jul 19, 2024
aa1f39e
Update relay-server/src/utils/memory.rs
iambriccardo Jul 19, 2024
735d52c
Improve
iambriccardo Jul 19, 2024
f434135
Merge
iambriccardo Jul 19, 2024
d22dfa6
Merge
iambriccardo Jul 19, 2024
9204a7f
Merge
iambriccardo Jul 19, 2024
8a70bf2
Changelog
iambriccardo Jul 19, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Fix
  • Loading branch information
@iambriccardo
iambriccardo committed Jul 18, 2024
commit 1d3b3e88115444cb8b2463590cdd43daa4a0e297
4 changes: 2 additions & 2 deletions relay-server/src/endpoints/common.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -296,7 +296,7 @@ fn queue_envelope(
// Split off the envelopes by item type.
let envelopes = ProcessingGroup::split_envelope(*managed_envelope.take_envelope());
for (group, envelope) in envelopes {
let envelope = ManagedEnvelope::standalone(
let envelope = ManagedEnvelope::new(
envelope,
state.outcome_aggregator().clone(),
state.test_store().clone(),
Expand Down Expand Up @@ -335,7 +335,7 @@ pub async fn handle_envelope(
return Err(BadStoreRequest::QueueFailed);
iambriccardo marked this conversation as resolved.
Show resolved Hide resolved
};

let mut managed_envelope = ManagedEnvelope::standalone(
let mut managed_envelope = ManagedEnvelope::new(
envelope,
state.outcome_aggregator().clone(),
state.test_store().clone(),
Expand Down
2 changes: 1 addition & 1 deletion relay-server/src/service.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ use crate::services::relays::{RelayCache, RelayCacheService};
use crate::services::store::StoreService;
use crate::services::test_store::{TestStore, TestStoreService};
use crate::services::upstream::{UpstreamRelay, UpstreamRelayService};
use crate::utils::{BufferGuard, MemoryStat, MemoryStatConfig};
use crate::utils::{MemoryStat, MemoryStatConfig};

/// Indicates the type of failure of the server.
#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash, thiserror::Error)]
Expand Down
2 changes: 1 addition & 1 deletion relay-server/src/services/health_check.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ use crate::services::metrics::{AcceptsMetrics, Aggregator};
use crate::services::project_cache::{ProjectCache, SpoolHealth};
use crate::services::upstream::{IsAuthenticated, UpstreamRelay};
use crate::statsd::RelayTimers;
use crate::utils::{MemoryStat, MemoryStatConfig};
use crate::utils::MemoryStatConfig;

/// Checks whether Relay is alive and healthy based on its variant.
#[derive(Clone, Copy, Debug, serde::Deserialize)]
Expand Down
10 changes: 5 additions & 5 deletions relay-server/src/services/processor.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2295,7 +2295,7 @@ impl EnvelopeProcessorService {
envelope.add_item(item);
}

let envelope = ManagedEnvelope::standalone(
let envelope = ManagedEnvelope::new(
envelope,
self.inner.addrs.outcome_aggregator.clone(),
self.inner.addrs.test_store.clone(),
Expand Down Expand Up @@ -2589,7 +2589,7 @@ impl EnvelopeProcessorService {
item.set_payload(ContentType::Json, serde_json::to_vec(&buckets).unwrap());
envelope.add_item(item);

let mut envelope = ManagedEnvelope::standalone(
let mut envelope = ManagedEnvelope::new(
envelope,
self.inner.addrs.outcome_aggregator.clone(),
self.inner.addrs.test_store.clone(),
Expand Down Expand Up @@ -2725,7 +2725,7 @@ impl EnvelopeProcessorService {
let mut envelope = Envelope::from_request(None, RequestMeta::outbound(dsn));
envelope.add_item(item);

let envelope = ManagedEnvelope::standalone(
let envelope = ManagedEnvelope::new(
envelope,
self.inner.addrs.outcome_aggregator.clone(),
self.inner.addrs.test_store.clone(),
Expand Down Expand Up @@ -3397,7 +3397,7 @@ mod tests {
assert_eq!(envelopes.len(), 1);

let (group, envelope) = envelopes.pop().unwrap();
let envelope = ManagedEnvelope::standalone(envelope, outcome_aggregator, test_store, group);
let envelope = ManagedEnvelope::new(envelope, outcome_aggregator, test_store, group);

let message = ProcessEnvelope {
envelope,
Expand Down Expand Up @@ -3454,7 +3454,7 @@ mod tests {
envelope.add_item(item);

let (outcome_aggregator, test_store) = testutils::processor_services();
let managed_envelope = ManagedEnvelope::standalone(
let managed_envelope = ManagedEnvelope::new(
envelope,
outcome_aggregator,
test_store,
Expand Down
7 changes: 3 additions & 4 deletions relay-server/src/services/processor/dynamic_sampling.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -283,7 +283,7 @@ mod tests {
use crate::testutils::{
self, create_test_processor, new_envelope, state_with_rule_and_condition,
};
use crate::utils::{ManagedEnvelope, Semaphore as TestSemaphore};
use crate::utils::ManagedEnvelope;

use super::*;

Expand Down Expand Up @@ -323,7 +323,7 @@ mod tests {
let (group, envelope) = envelopes.pop().unwrap();

let message = ProcessEnvelope {
envelope: ManagedEnvelope::standalone(envelope, outcome_aggregator, test_store, group),
envelope: ManagedEnvelope::new(envelope, outcome_aggregator, test_store, group),
project_state: Arc::new(ProjectState::allowed()),
sampling_project_state,
reservoir_counters: ReservoirCounters::default(),
Expand Down Expand Up @@ -471,7 +471,6 @@ mod tests {
project_id: ProjectId::new(42),
managed_envelope: ManagedEnvelope::new(
envelope,
TestSemaphore::new(42).try_acquire().unwrap(),
outcome_aggregator.clone(),
test_store.clone(),
ProcessingGroup::Transaction,
Expand Down Expand Up @@ -774,7 +773,7 @@ mod tests {
Some(Arc::new(state))
},
project_id: ProjectId::new(1),
managed_envelope: ManagedEnvelope::standalone(
managed_envelope: ManagedEnvelope::new(
envelope,
Addr::dummy(),
Addr::dummy(),
Expand Down
9 changes: 4 additions & 5 deletions relay-server/src/services/processor/profile.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -259,7 +259,7 @@ mod tests {
assert_eq!(envelopes.len(), 1);

let (group, envelope) = envelopes.pop().unwrap();
let envelope = ManagedEnvelope::standalone(envelope, Addr::dummy(), Addr::dummy(), group);
let envelope = ManagedEnvelope::new(envelope, Addr::dummy(), Addr::dummy(), group);

let message = ProcessEnvelope {
envelope,
Expand Down Expand Up @@ -389,7 +389,7 @@ mod tests {
assert_eq!(envelopes.len(), 1);

let (group, envelope) = envelopes.pop().unwrap();
let envelope = ManagedEnvelope::standalone(envelope, Addr::dummy(), Addr::dummy(), group);
let envelope = ManagedEnvelope::new(envelope, Addr::dummy(), Addr::dummy(), group);

let message = ProcessEnvelope {
envelope,
Expand Down Expand Up @@ -458,8 +458,7 @@ mod tests {
assert_eq!(envelopes.len(), 1);

let (group, envelope) = envelopes.pop().unwrap();
let envelope =
ManagedEnvelope::standalone(envelope.clone(), Addr::dummy(), Addr::dummy(), group);
let envelope = ManagedEnvelope::new(envelope.clone(), Addr::dummy(), Addr::dummy(), group);

let message = ProcessEnvelope {
envelope,
Expand Down Expand Up @@ -530,7 +529,7 @@ mod tests {
assert_eq!(envelopes.len(), 1);

let (group, envelope) = envelopes.pop().unwrap();
let envelope = ManagedEnvelope::standalone(envelope, Addr::dummy(), Addr::dummy(), group);
let envelope = ManagedEnvelope::new(envelope, Addr::dummy(), Addr::dummy(), group);

let message = ProcessEnvelope {
envelope,
Expand Down
10 changes: 5 additions & 5 deletions relay-server/src/services/processor/report.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -315,7 +315,7 @@ mod tests {
assert_eq!(envelopes.len(), 1);
let (group, envelope) = envelopes.pop().unwrap();

let envelope = ManagedEnvelope::standalone(envelope, outcome_aggregator, test_store, group);
let envelope = ManagedEnvelope::new(envelope, outcome_aggregator, test_store, group);
let message = ProcessEnvelope {
envelope,
project_state: Arc::new(ProjectState::allowed()),
Expand Down Expand Up @@ -368,7 +368,7 @@ mod tests {
let mut envelopes = ProcessingGroup::split_envelope(*envelope);
assert_eq!(envelopes.len(), 1);
let (group, envelope) = envelopes.pop().unwrap();
let envelope = ManagedEnvelope::standalone(envelope, outcome_aggregator, test_store, group);
let envelope = ManagedEnvelope::new(envelope, outcome_aggregator, test_store, group);

let message = ProcessEnvelope {
envelope,
Expand Down Expand Up @@ -431,7 +431,7 @@ mod tests {
assert_eq!(envelopes.len(), 1);

let (group, envelope) = envelopes.pop().unwrap();
let envelope = ManagedEnvelope::standalone(envelope, outcome_aggregator, test_store, group);
let envelope = ManagedEnvelope::new(envelope, outcome_aggregator, test_store, group);
let message = ProcessEnvelope {
envelope,
project_state: Arc::new(ProjectState::allowed()),
Expand Down Expand Up @@ -471,7 +471,7 @@ mod tests {

let (group, envelope) = envelopes.pop().unwrap();

let envelope = ManagedEnvelope::standalone(envelope, outcome_aggregator, test_store, group);
let envelope = ManagedEnvelope::new(envelope, outcome_aggregator, test_store, group);
let message = ProcessEnvelope {
envelope,
project_state: Arc::new(ProjectState::allowed()),
Expand Down Expand Up @@ -518,7 +518,7 @@ mod tests {
let mut envelopes = ProcessingGroup::split_envelope(*envelope);
assert_eq!(envelopes.len(), 1);
let (group, envelope) = envelopes.pop().unwrap();
let envelope = ManagedEnvelope::standalone(envelope, outcome_aggregator, test_store, group);
let envelope = ManagedEnvelope::new(envelope, outcome_aggregator, test_store, group);

let message = ProcessEnvelope {
envelope,
Expand Down
2 changes: 1 addition & 1 deletion relay-server/src/services/processor/span/processing.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -674,7 +674,7 @@ mod tests {
..Default::default()
};

let managed_envelope = ManagedEnvelope::standalone(
let managed_envelope = ManagedEnvelope::new(
dummy_envelope,
Addr::dummy(),
Addr::dummy(),
Expand Down
2 changes: 1 addition & 1 deletion relay-server/src/services/project.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1452,7 +1452,7 @@ mod tests {
let (outcome_aggregator, mut outcome_aggregator_rx) = Addr::custom();
let (test_store, _) = Addr::custom();

let managed_envelope = ManagedEnvelope::standalone(
let managed_envelope = ManagedEnvelope::new(
envelope,
outcome_aggregator.clone(),
test_store,
Expand Down
Loading
Loading