Re: [pdm-devel] [PATCH proxmox-datacenter-manager 01/16] server: add locked sdn client helpers

[Stefan Hanreich <s.hanreich@proxmox.com>]

On 8/27/25 2:27 PM, Dominik Csapak wrote:
> some comments inline
> 
> On 8/27/25 1:35 PM, Stefan Hanreich wrote:
>> Add a new client that represents a remote with a locked SDN
>> configuration. It works by creating a new PveClient and then locking
>> the SDN configuration via the client. It ensures that, while the lock
>> is held, all methods are called with the proper lock secret. It also
>> provides helpers for applying / rollbacking the configuration and
>> releasing the lock.
>>
>> Additionally, a collection type is introduced, that can hold multiple
>> locked SDN clients. It provides a helper for executing SDN API calls
>> across multiple remotes and rolling back the SDN configuration of all
>> remotes if any API command on any remote fails. This client will be
>> used for making changes across all remotes in PDM worker tasks.
>>
>> After applying, we cannot rollback the configuration - so the
>> collection only logs any errors - waits for the execution to finish on
>> all remotes and then returns a Result indicating whether the operation
>> was successful on *all* remotes.
> 
> imho this is a bit confusingly worded. I think what you mean is
> that we can't rollback after *starting* to apply changes no?
> 
> so once we started applying on remotes, we cannot go back.
> which makes kinda sense, but the first sentence would read
> like 'after applying finished'

I will try to make this more clear in the next iteration!

>>
>> Signed-off-by: Stefan Hanreich <s.hanreich@proxmox.com>
>> ---
>>   server/src/lib.rs        |   1 +
>>   server/src/sdn_client.rs | 389 +++++++++++++++++++++++++++++++++++++++
>>   2 files changed, 390 insertions(+)
>>   create mode 100644 server/src/sdn_client.rs
>>
>> diff --git a/server/src/lib.rs b/server/src/lib.rs
>> index 485bc79..760de7d 100644
>> --- a/server/src/lib.rs
>> +++ b/server/src/lib.rs
>> @@ -13,6 +13,7 @@ pub mod task_utils;
>>     pub mod connection;
>>   pub mod pbs_client;
>> +pub mod sdn_client;
>>     #[cfg(any(remote_config = "faked", test))]
>>   pub mod test_support;
>> diff --git a/server/src/sdn_client.rs b/server/src/sdn_client.rs
>> new file mode 100644
>> index 0000000..0fde307
>> --- /dev/null
>> +++ b/server/src/sdn_client.rs
>> @@ -0,0 +1,389 @@
>> +use std::sync::Arc;
>> +use std::time::Duration;
>> +
>> +use anyhow::{self, bail, Context};
>> +
>> +use futures::{future::join_all, stream::FuturesUnordered, StreamExt,
>> TryFutureExt};
>> +use pdm_api_types::{remotes::Remote, RemoteUpid};
>> +use pve_api_types::{
>> +    client::PveClient, CreateSdnLock, CreateVnet, CreateZone,
>> PveUpid, ReleaseSdnLock, ReloadSdn,
>> +    RollbackSdn,
>> +};
>> +
>> +use crate::api::pve::{connect, get_remote};
>> +
>> +/// Wrapper for [`PveClient`] for representing a locked SDN
>> configuration.
>> +///
>> +/// It stores the client that has been locked, as well as the
>> lock_token that is required for
>> +/// making changes to the SDN configuration. It provides methods that
>> proxy the respective SDN
>> +/// endpoints, where it adds the lock_token when making the proxied
>> calls.
>> +pub struct LockedSdnClient {
>> +    secret: String,
>> +    client: Arc<dyn PveClient + Send + Sync>,
>> +}
>> +
>> +impl LockedSdnClient {
>> +    /// Creates a new PveClient for a given [`Remote`] and locks the
>> SDN configuration there.
>> +    ///
>> +    /// # Errors
>> +    ///
>> +    /// This function will return an error if locking the remote fails.
>> +    pub async fn new(
>> +        remote: &Remote,
>> +        allow_pending: impl Into<Option<bool>>,
> 
> i get why you'd maybe want to have this as an option, but
> the only call site simply passes in a boolean here,
> so we could simply take a boolean here too?

yeah, it's probably better to ask callers to explicitly allow / deny
pending.

>> +    ) -> Result<Self, anyhow::Error> {
>> +        let client = connect(remote)?;
>> +
>> +        let params = CreateSdnLock {
>> +            allow_pending: allow_pending.into(),
>> +        };
>> +
>> +        let secret = client.acquire_sdn_lock(params).await?;
>> +
>> +        Ok(Self { secret, client })
>> +    }
>> +
>> +    /// proxies [`PveClient::create_vnet`] and adds lock_token to the
>> passed parameters before
>> +    /// making the call.
>> +    pub async fn create_vnet(&self, mut params: CreateVnet) ->
>> Result<(), proxmox_client::Error> {
>> +        params.lock_token = Some(self.secret.clone());
>> +
>> +        self.client.create_vnet(params).await
>> +    }
>> +
>> +    /// proxies [`PveClient::create_zone`] and adds lock_token to the
>> passed parameters before
>> +    /// making the call.
>> +    pub async fn create_zone(&self, mut params: CreateZone) ->
>> Result<(), proxmox_client::Error> {
>> +        params.lock_token = Some(self.secret.clone());
>> +
>> +        self.client.create_zone(params).await
>> +    }
>> +
>> +    /// applies the changes made while the client was locked and
>> returns the original [`PveClient`] if the
>> +    /// changes have been applied successfully.
>> +    pub async fn apply_and_release(
>> +        self,
>> +    ) -> Result<(PveUpid, Arc<dyn PveClient + Send + Sync>),
>> proxmox_client::Error> {
>> +        let params = ReloadSdn {
>> +            lock_token: Some(self.secret.clone()),
>> +            release_lock: Some(true),
>> +        };
>> +
>> +        self.client
>> +            .sdn_apply(params)
>> +            .await
>> +            .map(move |upid| (upid, self.client))
>> +    }
>> +
>> +    /// releases the lock on the [`PveClient`] without applying
>> pending changes.
>> +    pub async fn release(
>> +        self,
>> +        force: impl Into<Option<bool>>,
>> +    ) -> Result<Arc<dyn PveClient + Send + Sync>,
>> proxmox_client::Error> {
>> +        let params = ReleaseSdnLock {
>> +            force: force.into(),
>> +            lock_token: Some(self.secret),
>> +        };
>> +
>> +        self.client.release_sdn_lock(params).await?;
>> +        Ok(self.client)
>> +    }
>> +
>> +    /// rolls back all pending changes and then releases the lock
>> +    pub async fn rollback_and_release(
>> +        self,
>> +    ) -> Result<Arc<dyn PveClient + Send + Sync>,
>> proxmox_client::Error> {
>> +        let params = RollbackSdn {
>> +            lock_token: Some(self.secret),
>> +            release_lock: Some(true),
>> +        };
>> +
>> +        self.client.rollback_sdn_changes(params).await?;
>> +        Ok(self.client)
>> +    }
>> +}
>> +
>> +/// Context for [`LockedSdnClient`] stored in [`LockedSdnClients`].
>> +pub struct LockedSdnClientContext<C> {
>> +    remote_id: String,
>> +    data: C,
>> +}
>> +
>> +impl<C> LockedSdnClientContext<C> {
>> +    fn new(remote_id: String, data: C) -> Self {
>> +        Self { remote_id, data }
>> +    }
>> +
>> +    pub fn remote_id(&self) -> &str {
>> +        &self.remote_id
>> +    }
>> +
>> +    pub fn data(&self) -> &C {
>> +        &self.data
>> +    }
>> +}
>> +
>> +/// A collection abstracting [`LockedSdnClient`] for multiple locked
>> remotes.
>> +///
>> +/// It can be used for running the same command across multiple
>> remotes, while automatically
>> +/// handling rollback and releasing locks in case of failures across
>> all remotes. If an API call
>> +/// made to one of the remotes fails, then this client will
>> automatically take care of rolling back
>> +/// all changes made during the transaction and then releasing the
>> locks.
>> +pub struct LockedSdnClients<C> {
>> +    clients: Vec<(LockedSdnClient, LockedSdnClientContext<C>)>,
>> +}
>> +
>> +impl<C> LockedSdnClients<C> {
>> +    /// A convenience function for creating locked clients for
>> multiple remotes.
>> +    ///
>> +    /// For each remote a Context can be specified, which will be
>> supplied to all callbacks that
>> +    /// are using this [`LockedSdnClients`] to make calls across all
>> remotes.
>> +    ///
>> +    /// # Errors
>> +    ///
>> +    /// This function will return an error if:
>> +    /// * the remote configuration cannot be read
>> +    /// * any of the supplied remotes is not contained in the
>> configuration
>> +    /// * locking the configuration on any remote fails
>> +    ///
>> +    /// If necessary, the configuration of all remotes will be
>> unlocked, if possible.
>> +    pub async fn from_remote_names<I: IntoIterator<Item = (String, C)>>(
>> +        remote_names: I,
>> +        allow_pending: bool,
>> +    ) -> Result<Self, anyhow::Error> {
>> +        let (remote_config, _) = pdm_config::remotes::config()?;
>> +
>> +        let mut clients = Vec::new();
>> +
>> +        for (remote_name, context) in remote_names {
>> +            let remote = get_remote(&remote_config, &remote_name)?;
>> +            proxmox_log::info!("acquiring lock for remote {}",
>> remote.id);
>> +
>> +            match LockedSdnClient::new(remote, allow_pending).await {
>> +                Ok(client) => {
>> +                    let context =
>> LockedSdnClientContext::new(remote_name, context);
>> +                    clients.push((client, context));
>> +                }
>> +                Err(error) => {
>> +                    proxmox_log::info!(
>> +                        "encountered an error when locking a remote,
>> releasing all locks"
>> +                    );
>> +
>> +                    for (client, _) in clients {
>> +                        proxmox_log::info!("releasing lock for remote
>> {}", remote.id);
>> +
>> +                        if let Err(error) =
>> client.release(false).await {
>> +                            proxmox_log::error!(
>> +                                "could not release lock for remote
>> {}: {error:#}",
>> +                                remote.id
>> +                            )
>> +                        }
>> +                    }
>> +
>> +                    return Err(error).with_context(|| {
>> +                        format!("could not lock sdn configuration for
>> remote {}", remote.id)
>> +                    });
>> +                }
>> +            };
>> +        }
>> +
>> +        Ok(Self { clients })
>> +    }
>> +
>> +    /// Executes the given callback for each [`LockedSdnClient`] in
>> this collection.
>> +    ///
>> +    /// On error, it tries to rollback the configuration of *all*
>> locked clients, releases the lock
>> +    /// and returns the error. If rollbacking fails, an error will be
>> logged and no further action
>> +    /// is taken.
>> +    pub async fn for_each<F>(self, callback: F) -> Result<Self,
>> anyhow::Error>
>> +    where
>> +        F: AsyncFn(
>> +            &LockedSdnClient,
>> +            &LockedSdnClientContext<C>,
>> +        ) -> Result<(), proxmox_client::Error>,
>> +    {
>> +        let futures = self.clients.iter().map(|(client, context)| {
>> +            callback(client, context)
>> +                .map_ok(|_| context.remote_id())
>> +                .map_err(|err| (err, context.remote_id()))
>> +        });
>> +
>> +        let mut errors = false;
>> +
>> +        for result in join_all(futures).await {
>> +            match result {
>> +                Ok(remote_id) => {
>> +                    proxmox_log::info!("succcessfully executed
>> transaction on remote {remote_id}");
>> +                }
>> +                Err((error, remote_id)) => {
>> +                    proxmox_log::error!(
>> +                        "failed to execute transaction on remote
>> {remote_id}: {error:#}",
>> +                    );
>> +                    errors = true;
>> +                }
>> +            }
>> +        }
>> +
>> +        if errors {
>> +            let mut rollback_futures = FuturesUnordered::new();
>> +
>> +            for (client, ctx) in self.clients {
>> +                let ctx = Arc::new(ctx);
>> +                let err_ctx = ctx.clone();
>> +
>> +                rollback_futures.push(
>> +                    client
>> +                        .rollback_and_release()
>> +                        .map_ok(|_| ctx)
>> +                        .map_err(|err| (err, err_ctx)),
>> +                );
>> +            }
>> +
>> +            while let Some(result) = rollback_futures.next().await {
>> +                match result {
>> +                    Ok(ctx) => {
>> +                        proxmox_log::info!(
>> +                            "successfully rolled back configuration
>> for remote {}",
>> +                            ctx.remote_id()
>> +                        )
>> +                    }
>> +                    Err((_, ctx)) => {
>> +                        proxmox_log::error!(
>> +                            "could not rollback configuration for
>> remote {}",
>> +                            ctx.remote_id()
>> +                        )
>> +                    }
>> +                }
>> +            }
> 
> 
> if we could not rollback_and_release, should we maybe call release for
> those ? The admin has to cleanup anyway, but is it possible for one to
> release such a lock (via the gui?). Not very sure about the lock
> semantics here.

I thought about it, but I think it is better to leave the remote locked
on failure since admins then get an error on the respective PVE remote
if they want to change some settings there.

They can then either explicitly rollback the changes manually or fix the
issues and apply the configuration anyway.

The most likely scenario for why this would fail is networking issues
imo. I think if the rollback_and_release call fails, there is a decent
chance that a subsequent release call will fail as well. This would
introduce a third level of requests + error handling, which I'm not sure
is worth the potential gain.

It is also analogous to e.g. failures when snapshotting a VM where the
lock needs to be forcibly released iirc.

I think the error message could be improved by indicating that the
configuration is left untouched and admins need to rollback / release
manually and potentially even a indicator / pointer on how to do it.


> also using FuturesUnordered here can make sense here, but it's not
> really parallel, since it only does work during await.
> 
> does it make a big difference (for many remotes) instead of just
> awaiting the futures directly? or joining them ?
> 
> i.e. having a large number of remotes unordered in the logs does not
> make it easier to read.
> 
> I'd either expect the remotes to be sorted, or the failed remotes to be
> separated from the successful ones (e.g. only logging the error ones
> separately and only a number for the succesful ones?)

I think I had some issues with just using a Vec of futures + join_all
(ownership + borrowing with clients / ctx) and using FuturesUnordered
resolved the issue. I'll check again more thoroughly on the root cause
and report back. I wanted to improve how the requests are made in the
client anyway, but didn't yet get around to it. We might also just want
to spawn blocking tasks in tokio instead.

Using FuturesOrdered would be fine if collecting into a Vec doesn't
work, but then we should maybe sort the clients in the constructor
(alphabetically by remote name?) to ensure deterministic/consistent
ordering any time any function in this collection is called.

Anyway, I'll look into improving this in the next version (most likely
by spawning blocking tasks in tokio itself)!


> same remark for 'apply_and_release' below
> 
>> +
>> +            bail!("running the transaction failed on at least one
>> remote!");
>> +        }
>> +
>> +        Ok(self)
>> +    }
>> +
>> +    // pve-http-server TCP connection timeout is 5 seconds, use a
>> lower amount with some margin for
>> +    // latency in order to avoid re-opening TCP connections for every
>> polling request.
>> +    const POLLING_INTERVAL: Duration = Duration::from_secs(3);
>> +
>> +    /// Convenience function for polling a running task on a PVE remote.
>> +    ///
>> +    /// It polls a given task on a given node, waiting for the task
>> to finish successfully.
>> +    ///
>> +    /// # Errors
>> +    ///
>> +    /// This function will return an error if:
>> +    /// * There was a problem querying the task status (this does not
>> necessarily mean the task failed).
>> +    /// * The task finished unsuccessfully.
>> +    async fn poll_task(
>> +        node: String,
>> +        upid: RemoteUpid,
>> +        client: Arc<dyn PveClient + Send + Sync>,
>> +    ) -> Result<RemoteUpid, anyhow::Error> {
>> +        loop {
>> +            tokio::time::sleep(Self::POLLING_INTERVAL).await;
>> +
>> +            let status = client.get_task_status(&node,
>> &upid.upid).await?;
>> +
>> +            if !status.is_running() {
>> +                if status.finished_successfully() == Some(true) {
>> +                    return Ok(upid);
>> +                } else {
>> +                    bail!(
>> +                        "task did not finish successfully on remote {}",
>> +                        upid.remote()
>> +                    );
>> +                }
>> +            }
>> +        }
>> +    }
>> +
>> +    /// Applies and Reloads the SDN configuration for all locked
>> clients.
>> +    ///
>> +    /// This function tries to apply the SDN configuration for all
>> supplied locked clients and, if
>> +    /// it was successful, to reload the SDN configuration of the
>> remote. It logs success and error
>> +    /// messages via proxmox_log. Rollbacking in cases of failure is
>> no longer possible, so this
>> +    /// function then returns an error if applying or reloading the
>> configuration was unsuccessful
>> +    /// on at least one remote.
>> +    ///
>> +    /// # Errors This function returns an error if applying or
>> reloading the configuration on one
>> +    /// of the remotes failed. It will always wait for all futures to
>> finish and only return an
>> +    /// error afterwards.
>> +    pub async fn apply_and_release(self) -> Result<(), anyhow::Error> {
>> +        let mut futures = FuturesUnordered::new();
>> +
>> +        for (client, context) in self.clients {
>> +            let ctx = Arc::new(context);
>> +            let err_ctx = ctx.clone();
>> +
>> +            futures.push(
>> +                client
>> +                    .apply_and_release()
>> +                    .map_ok(|(upid, client)| ((upid, client), ctx))
>> +                    .map_err(|err| (err, err_ctx)),
>> +            );
>> +        }
>> +
>> +        let mut reload_futures = FuturesUnordered::new();
>> +
>> +        while let Some(result) = futures.next().await {
>> +            match result {
>> +                Ok(((upid, client), ctx)) => {
>> +                    proxmox_log::info!(
>> +                        "successfully applied sdn config on remote {}",
>> +                        ctx.remote_id()
>> +                    );
>> +
>> +                    let Ok(remote_upid) =
>> +                        RemoteUpid::try_from((ctx.remote_id(),
>> upid.to_string().as_str()))
>> +                    else {
>> +                        proxmox_log::error!("invalid UPID received
>> from PVE: {upid}");
>> +                        continue;
>> +                    };
>> +
>> +                    reload_futures.push(
>> +                        Self::poll_task(upid.node.clone(),
>> remote_upid, client)
>> +                            .map_err(move |err| (err, ctx)),
>> +                    );
>> +                }
>> +                Err((error, ctx)) => {
>> +                    proxmox_log::error!(
>> +                        "failed to apply sdn configuration on remote
>> {}: {error:#}, not reloading",
>> +                        ctx.remote_id()
>> +                    );
>> +                }
>> +            }
>> +        }
>> +
>> +        proxmox_log::info!(
>> +            "Waiting for reload tasks to finish on all remotes, this
>> can take awhile"
>> +        );
>> +
>> +        let mut errors = false;
>> +
>> +        while let Some(result) = reload_futures.next().await {
>> +            match result {
>> +                Ok(upid) => {
>> +                    proxmox_log::info!(
>> +                        "successfully reloaded configuration on
>> remote {}",
>> +                        upid.remote()
>> +                    );
>> +                }
>> +                Err((error, ctx)) => {
>> +                    proxmox_log::error!(
>> +                        "could not reload configuration on remote {}:
>> {error:#}",
>> +                        ctx.remote_id()
>> +                    );
>> +
>> +                    errors = true;
>> +                }
>> +            }
>> +        }
>> +
>> +        if errors {
>> +            bail!("failed to apply configuration on at least one
>> remote");
>> +        }
>> +
>> +        Ok(())
>> +    }
>> +}
> 



_______________________________________________
pdm-devel mailing list
pdm-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pdm-devel