rust/neon
1
0
Fork 0
mirror of https://github.com/neondatabase/neon.git synced 2026-01-04 20:12:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

proxy: Revert ControlPlaneEvent back to struct (#9649)

Browse Source 
Due to neondatabase/cloud#19815 we need to be more tolerant when reading
events.
This commit is contained in:
Folke Behrens
2024-11-05 22:32:33 +01:00
committed by GitHub
parent fcde40d600
commit 754d2950a3
1 changed files with 47 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

77
proxy/src/cache/endpoints.rs vendored
View File

@@ -19,26 +19,28 @@ use crate::rate_limiter::GlobalRateLimiter;
use crate::redis::connection_with_credentials_provider::ConnectionWithCredentialsProvider;
use crate::types::EndpointId;
#[allow(clippy::enum_variant_names)]
#[derive(Deserialize, Debug, Clone)]
#[serde(tag = "type", rename_all(deserialize = "snake_case"))]
enum ControlPlaneEvent {
EndpointCreated { endpoint_created: EndpointCreated },
BranchCreated { branch_created: BranchCreated },
ProjectCreated { project_created: ProjectCreated },
// TODO: this could be an enum, but events in Redis need to be fixed first.
// ProjectCreated was sent with type:branch_created. So we ignore type.
#[derive(Deserialize, Debug, Clone, PartialEq)]
struct ControlPlaneEvent {
endpoint_created: Option<EndpointCreated>,
branch_created: Option<BranchCreated>,
project_created: Option<ProjectCreated>,
#[serde(rename = "type")]
_type: Option<String>,
}
#[derive(Deserialize, Debug, Clone)]
#[derive(Deserialize, Debug, Clone, PartialEq)]
struct EndpointCreated {
endpoint_id: String,
}
#[derive(Deserialize, Debug, Clone)]
#[derive(Deserialize, Debug, Clone, PartialEq)]
struct BranchCreated {
branch_id: String,
}
#[derive(Deserialize, Debug, Clone)]
#[derive(Deserialize, Debug, Clone, PartialEq)]
struct ProjectCreated {
project_id: String,
}
@@ -104,24 +106,28 @@ impl EndpointsCache {
}
fn insert_event(&self, event: ControlPlaneEvent) {
let counter = match event {
ControlPlaneEvent::EndpointCreated { endpoint_created } => {
self.endpoints
.insert(EndpointIdInt::from(&endpoint_created.endpoint_id.into()));
RedisEventsCount::EndpointCreated
}
ControlPlaneEvent::BranchCreated { branch_created } => {
self.branches
.insert(BranchIdInt::from(&branch_created.branch_id.into()));
RedisEventsCount::BranchCreated
}
ControlPlaneEvent::ProjectCreated { project_created } => {
self.projects
.insert(ProjectIdInt::from(&project_created.project_id.into()));
RedisEventsCount::ProjectCreated
}
};
Metrics::get().proxy.redis_events_count.inc(counter);
if let Some(endpoint_created) = event.endpoint_created {
self.endpoints
.insert(EndpointIdInt::from(&endpoint_created.endpoint_id.into()));
Metrics::get()
.proxy
.redis_events_count
.inc(RedisEventsCount::EndpointCreated);
} else if let Some(branch_created) = event.branch_created {
self.branches
.insert(BranchIdInt::from(&branch_created.branch_id.into()));
Metrics::get()
.proxy
.redis_events_count
.inc(RedisEventsCount::BranchCreated);
} else if let Some(project_created) = event.project_created {
self.projects
.insert(ProjectIdInt::from(&project_created.project_id.into()));
Metrics::get()
.proxy
.redis_events_count
.inc(RedisEventsCount::ProjectCreated);
}
}
pub async fn do_read(
@@ -235,11 +241,22 @@ impl EndpointsCache {
#[cfg(test)]
mod tests {
use super::ControlPlaneEvent;
use super::*;
#[test]
fn test_parse_control_plane_event() {
let s = r#"{"branch_created":null,"endpoint_created":{"endpoint_id":"ep-rapid-thunder-w0qqw2q9"},"project_created":null,"type":"endpoint_created"}"#;
serde_json::from_str::<ControlPlaneEvent>(s).unwrap();
assert_eq!(
serde_json::from_str::<ControlPlaneEvent>(s).unwrap(),
ControlPlaneEvent {
endpoint_created: Some(EndpointCreated {
endpoint_id: "ep-rapid-thunder-w0qqw2q9".into()
}),
branch_created: None,
project_created: None,
_type: Some("endpoint_created".into()),
}
);
}
}