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;
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            },
41            Authenticated,
42        );
43        let session_client = client.startup(session).await?;
44        Ok(Self { session_client })
45    }
46
47    /// Push all current values from the given [SynchronizedParameters] that are
48    /// marked as modified to the [SystemParameterBackend] and reset their
49    /// modified status.
50    pub async fn push(&mut self, params: &mut SynchronizedParameters) {
51        for param in params.modified() {
52            let mut vars = BTreeMap::new();
53            info!(name = param.name, value = param.value, "updating parameter");
54            vars.insert(param.name.clone(), param.value.clone());
55            match self.session_client.set_system_vars(vars).await {
56                Ok(()) => {
57                    info!(name = param.name, value = param.value, "update success");
58                }
59                Err(error) => match error {
60                    AdapterError::ReadOnly => {
61                        info!(
62                            name = param.name,
63                            value = param.value,
64                            "cannot update system variable in read-only mode",
65                        );
66                    }
67                    error => {
68                        error!(
69                            name = param.name,
70                            value = param.value,
71                            "cannot update system variable: {}",
72                            error
73                        );
74                    }
75                },
76            }
77        }
78    }
79
80    /// Pull the current values for all [SynchronizedParameters] from the
81    /// [SystemParameterBackend].
82    pub async fn pull(&self, params: &mut SynchronizedParameters) {
83        let vars = self.session_client.get_system_vars().await;
84        for var in vars.iter() {
85            params.modify(var.name(), &var.value());
86        }
87    }
88}
mz_adapter/config/backend.rs

mz_adapter/config/
backend.rs
