Skip to content

node: pause subgraphs before reassigning in graphman #6160

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

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

node: pause subgraphs before reassigning in graphman #6160

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
8 changes: 4 additions & 4 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

15 changes: 14 additions & 1 deletion node/src/bin/manager.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -169,6 +169,14 @@ pub enum Command {
deployment: DeploymentSearch,
/// The name of the node that should index the deployment
node: String,
/// Sleep for this many seconds between pausing and reassigning subgraphs
#[clap(
long,
short,
default_value = "20",
value_parser = parse_duration_in_secs
)]
sleep: Duration,
},
/// Unassign a deployment
Unassign {
Expand Down Expand Up @@ -1230,7 +1238,11 @@ async fn main() -> anyhow::Result<()> {
let deployment = make_deployment_selector(deployment);
commands::deployment::unassign::run(primary_pool, notifications_sender, deployment)
}
Reassign { deployment, node } => {
Reassign {
deployment,
node,
sleep,
} => {
let notifications_sender = ctx.notification_sender();
let primary_pool = ctx.primary_pool();
let deployment = make_deployment_selector(deployment);
Expand All @@ -1240,6 +1252,7 @@ async fn main() -> anyhow::Result<()> {
notifications_sender,
deployment,
&node,
sleep,
)
}
Pause { deployment } => {
Expand Down
27 changes: 23 additions & 4 deletions node/src/manager/commands/deployment/reassign.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
use std::sync::Arc;
use std::time::Duration;

use anyhow::Result;
use graph::prelude::NodeId;
Expand All @@ -12,10 +13,22 @@ use graphman::deployment::DeploymentSelector;
pub fn run(
primary_pool: ConnectionPool,
notification_sender: Arc<NotificationSender>,
deployment: DeploymentSelector,
deployment_selector: DeploymentSelector,
node: &NodeId,
delay: Duration,
) -> Result<()> {
let deployment = load_deployment(primary_pool.clone(), &deployment)?;
super::pause::run(
primary_pool.clone(),
notification_sender.clone(),
deployment_selector.clone(),
)?;

println!(
"Waiting {}s to make sure pausing was processed ...",
delay.as_secs()
);
Comment on lines +26 to +29
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The delay is not actually used, i think what you meant to do is to call sleep() here too with the delay


let deployment = load_deployment(primary_pool.clone(), &deployment_selector)?;
let curr_node = deployment.assigned_node(primary_pool.clone())?;
let reassign_msg = match &curr_node {
Some(curr_node) => format!(
Expand All @@ -28,8 +41,8 @@ pub fn run(
println!("{}", reassign_msg);

let reassign_result = reassign_deployment(
primary_pool,
notification_sender,
primary_pool.clone(),
notification_sender.clone(),
&deployment,
node,
curr_node,
Comment on lines +44 to 48
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

minor nitpick: if the pause::run succeeds and actual reassign fails the resume will never run and the subgraph would be paused unintentionally.

Expand All @@ -50,5 +63,11 @@ pub fn run(
}
}

super::resume::run(
primary_pool,
notification_sender,
deployment_selector.clone(),
)?;

Ok(())
}
Loading