mz_adapter/config/
backend.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10use std::collections::BTreeMap;
11
12use mz_auth::{Authenticated, AuthenticatorKind};
13use mz_sql::session::user::SYSTEM_USER;
14use tracing::{error, info};
15use uuid::Uuid;
16
17use crate::config::SynchronizedParameters;
18use crate::session::SessionConfig;
19use crate::{AdapterError, Client, SessionClient};
20
21/// A backend client for pushing and pulling [SynchronizedParameters].
22///
23/// Pulling is required in order to catch concurrent changes before pushing
24/// modified values in the [crate::config::system_parameter_sync].
25pub struct SystemParameterBackend {
26    session_client: SessionClient,
27}
28
29impl SystemParameterBackend {
30    pub async fn new(client: Client) -> Result<Self, AdapterError> {
31        let conn_id = client.new_conn_id()?;
32        let session = client.new_session(
33            SessionConfig {
34                conn_id,
35                uuid: Uuid::new_v4(),
36                user: SYSTEM_USER.name.clone(),
37                client_ip: None,
38                external_metadata_rx: None,
39                helm_chart_version: None,
40                authenticator_kind: AuthenticatorKind::None,
41            },
42            Authenticated,
43        );
44        let session_client = client.startup(session).await?;
45        Ok(Self { session_client })
46    }
47
48    /// Push all current values from the given [SynchronizedParameters] that are
49    /// marked as modified to the [SystemParameterBackend] and reset their
50    /// modified status.
51    pub async fn push(&mut self, params: &mut SynchronizedParameters) {
52        for param in params.modified() {
53            let mut vars = BTreeMap::new();
54            info!(name = param.name, value = param.value, "updating parameter");
55            vars.insert(param.name.clone(), param.value.clone());
56            match self.session_client.set_system_vars(vars).await {
57                Ok(()) => {
58                    info!(name = param.name, value = param.value, "update success");
59                }
60                Err(error) => match error {
61                    AdapterError::ReadOnly => {
62                        info!(
63                            name = param.name,
64                            value = param.value,
65                            "cannot update system variable in read-only mode",
66                        );
67                    }
68                    error => {
69                        error!(
70                            name = param.name,
71                            value = param.value,
72                            "cannot update system variable: {}",
73                            error
74                        );
75                    }
76                },
77            }
78        }
79    }
80
81    /// Pull the current values for all [SynchronizedParameters] from the
82    /// [SystemParameterBackend].
83    pub async fn pull(&self, params: &mut SynchronizedParameters) {
84        let vars = self.session_client.get_system_vars().await;
85        for var in vars.iter() {
86            params.modify(var.name(), &var.value());
87        }
88    }
89}
mz_adapter/config/backend.rs

mz_adapter/config/
backend.rs
