Yehowshua/ZIm
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fix collab (#8298)

Browse source 
Co-Authored-By: Marshall <marshall@zed.dev>

We broke it by deploying two servers simultaneously.

Release Notes:

- N/A

Co-authored-by: Marshall <marshall@zed.dev>
This commit is contained in:
Conrad Irwin 2024-02-23 12:23:15 -07:00 • committed by GitHub
parent a11ebe01ff
commit 7c514d044f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 39 additions and 20 deletions
Download patch file Download diff file Expand all files Collapse all files

2
crates/collab/k8s/collab.template.yml
View file

@ -58,7 +58,7 @@ spec:
- name: ${ZED_SERVICE_NAME} - name: ${ZED_SERVICE_NAME}
image: "${ZED_IMAGE_ID}" image: "${ZED_IMAGE_ID}"
args: args:
- serve - serve ${ZED_SERVICE_NAME}
ports: ports:
- containerPort: 8080 - containerPort: 8080
protocol: TCP protocol: TCP

8
crates/collab/src/api.rs
View file

@ -25,7 +25,7 @@ use tracing::instrument;
pub use extensions::fetch_extensions_from_blob_store_periodically; pub use extensions::fetch_extensions_from_blob_store_periodically;
pub fn routes(rpc_server: Arc<rpc::Server>, state: Arc<AppState>) -> Router<Body> { pub fn routes(rpc_server: Option<Arc<rpc::Server>>, state: Arc<AppState>) -> Router<Body> {
Router::new() Router::new()
.route("/user", get(get_authenticated_user)) .route("/user", get(get_authenticated_user))
.route("/users/:id/access_tokens", post(create_access_token)) .route("/users/:id/access_tokens", post(create_access_token))
@ -136,8 +136,12 @@ async fn trace_panic(panic: Json<Panic>) -> Result<()> {
} }
async fn get_rpc_server_snapshot( async fn get_rpc_server_snapshot(
Extension(rpc_server): Extension<Arc<rpc::Server>>, Extension(rpc_server): Extension<Option<Arc<rpc::Server>>>,
) -> Result<ErasedJson> { ) -> Result<ErasedJson> {
let Some(rpc_server) = rpc_server else {
return Err(Error::Internal(anyhow!("rpc server is not available")));
};
Ok(ErasedJson::pretty(rpc_server.snapshot().await)) Ok(ErasedJson::pretty(rpc_server.snapshot().await))
} }

25
crates/collab/src/main.rs
View file

@ -28,7 +28,8 @@ async fn main() -> Result<()> {
); );
} }
match args().skip(1).next().as_deref() { let mut args = args().skip(1);
match args.next().as_deref() {
Some("version") => { Some("version") => {
println!("collab v{} ({})", VERSION, REVISION.unwrap_or("unknown")); println!("collab v{} ({})", VERSION, REVISION.unwrap_or("unknown"));
} }
@ -36,6 +37,8 @@ async fn main() -> Result<()> {
run_migrations().await?; run_migrations().await?;
} }
Some("serve") => { Some("serve") => {
let is_api_only = args.next().is_some_and(|arg| arg == "api");
let config = envy::from_env::<Config>().expect("error loading config"); let config = envy::from_env::<Config>().expect("error loading config");
init_tracing(&config); init_tracing(&config);
@ -46,18 +49,27 @@ async fn main() -> Result<()> {
let listener = TcpListener::bind(&format!("0.0.0.0:{}", state.config.http_port)) let listener = TcpListener::bind(&format!("0.0.0.0:{}", state.config.http_port))
.expect("failed to bind TCP listener"); .expect("failed to bind TCP listener");
let rpc_server = if !is_api_only {
let epoch = state let epoch = state
.db .db
.create_server(&state.config.zed_environment) .create_server(&state.config.zed_environment)
.await?; .await?;
let rpc_server = collab::rpc::Server::new(epoch, state.clone(), Executor::Production); let rpc_server =
collab::rpc::Server::new(epoch, state.clone(), Executor::Production);
rpc_server.start().await?; rpc_server.start().await?;
Some(rpc_server)
} else {
None
};
fetch_extensions_from_blob_store_periodically(state.clone(), Executor::Production); fetch_extensions_from_blob_store_periodically(state.clone(), Executor::Production);
let app = collab::api::routes(rpc_server.clone(), state.clone()) let mut app = collab::api::routes(rpc_server.clone(), state.clone());
.merge(collab::rpc::routes(rpc_server.clone())) if let Some(rpc_server) = rpc_server.clone() {
.merge( app = app.merge(collab::rpc::routes(rpc_server))
}
app = app.merge(
Router::new() Router::new()
.route("/", get(handle_root)) .route("/", get(handle_root))
.route("/healthz", get(handle_liveness_probe)) .route("/healthz", get(handle_liveness_probe))
@ -77,7 +89,10 @@ async fn main() -> Result<()> {
futures::pin_mut!(sigterm, sigint); futures::pin_mut!(sigterm, sigint);
futures::future::select(sigterm, sigint).await; futures::future::select(sigterm, sigint).await;
tracing::info!("Received interrupt signal"); tracing::info!("Received interrupt signal");
if let Some(rpc_server) = rpc_server {
rpc_server.teardown(); rpc_server.teardown();
}
}) })
.await?; .await?;
} }