mz/command/
region.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License in the LICENSE file at the
6// root of this repository, or online at
7//
8//     http://www.apache.org/licenses/LICENSE-2.0
9//
10// Unless required by applicable law or agreed to in writing, software
11// distributed under the License is distributed on an "AS IS" BASIS,
12// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13// See the License for the specific language governing permissions and
14// limitations under the License.
15
16//! Implementation of the `mz region` command.
17//!
18//! Consult the user-facing documentation for details.
19//!
20use std::time::Duration;
21
22use crate::{context::RegionContext, error::Error};
23
24use mz_cloud_api::client::{cloud_provider::CloudProvider, region::RegionState};
25use mz_ore::retry::Retry;
26use reqwest::StatusCode;
27use serde::{Deserialize, Serialize};
28use tabled::Tabled;
29
30/// Enable a region in the profile organization.
31///
32/// In cases where the organization has already enabled the region
33/// the command will try to run a version update. Resulting
34/// in a downtime for a short period.
35pub async fn enable(
36    cx: RegionContext,
37    version: Option<String>,
38    environmentd_extra_arg: Option<Vec<String>>,
39    environmentd_cpu_allocation: Option<String>,
40    environmentd_memory_allocation: Option<String>,
41) -> Result<(), Error> {
42    let loading_spinner = cx
43        .output_formatter()
44        .loading_spinner("Retrieving information...");
45    let cloud_provider = cx.get_cloud_provider().await?;
46
47    loading_spinner.set_message("Enabling the region...");
48
49    let environmentd_extra_arg: Vec<String> = environmentd_extra_arg.unwrap_or_else(Vec::new);
50
51    // Loop region creation.
52    // After 6 minutes it will timeout.
53    Retry::default()
54        .max_duration(Duration::from_secs(720))
55        .clamp_backoff(Duration::from_secs(1))
56        .retry_async(|_| async {
57            let _ = cx
58                .cloud_client()
59                .create_region(
60                    version.clone(),
61                    environmentd_extra_arg.clone(),
62                    environmentd_cpu_allocation.clone(),
63                    environmentd_memory_allocation.clone(),
64                    cloud_provider.clone(),
65                )
66                .await?;
67            Ok(())
68        })
69        .await
70        .map_err(|e| Error::TimeoutError(Box::new(e)))?;
71
72    loading_spinner.set_message("Waiting for the region to be online...");
73
74    // Loop retrieving the region and checking the SQL connection for 6 minutes.
75    // After 6 minutes it will timeout.
76    Retry::default()
77        .max_duration(Duration::from_secs(720))
78        .clamp_backoff(Duration::from_secs(1))
79        .retry_async(|_| async {
80            let region = cx.get_region().await?;
81
82            match region.region_state {
83                RegionState::EnablementPending => {
84                    loading_spinner.set_message("Waiting for the region to be ready...");
85                    Err(Error::NotReadyRegion)
86                }
87                RegionState::DeletionPending => Err(Error::CommandExecutionError(
88                    "This region is pending deletion!".to_string(),
89                )),
90                RegionState::SoftDeleted => Err(Error::CommandExecutionError(
91                    "This region has been marked soft-deleted!".to_string(),
92                )),
93                RegionState::Enabled => match region.region_info {
94                    Some(region_info) => {
95                        loading_spinner.set_message("Waiting for the region to be resolvable...");
96                        if region_info.resolvable {
97                            let claims = cx.admin_client().claims().await?;
98                            let user = claims.user()?;
99                            if cx.sql_client().is_ready(&region_info, user)? {
100                                return Ok(());
101                            }
102                            Err(Error::NotPgReadyError)
103                        } else {
104                            Err(Error::NotResolvableRegion)
105                        }
106                    }
107                    None => Err(Error::NotReadyRegion),
108                },
109            }
110        })
111        .await
112        .map_err(|e| Error::TimeoutError(Box::new(e)))?;
113
114    loading_spinner.finish_with_message(format!("Region in {} is now online", cloud_provider.id));
115
116    Ok(())
117}
118
119/// Disable a region in the profile organization.
120///
121/// This command can take several minutes to complete.
122pub async fn disable(cx: RegionContext, hard: bool) -> Result<(), Error> {
123    let loading_spinner = cx
124        .output_formatter()
125        .loading_spinner("Retrieving information...");
126
127    let cloud_provider = cx.get_cloud_provider().await?;
128
129    // The `delete_region` method retries disabling a region,
130    // has an inner timeout, and manages a `504` response.
131    // For any other type of error response, we handle it here
132    // with a retry loop.
133    Retry::default()
134        .max_duration(Duration::from_secs(720))
135        .clamp_backoff(Duration::from_secs(1))
136        .retry_async(|_| async {
137            loading_spinner.set_message("Disabling region...");
138            match cx
139                .cloud_client()
140                .delete_region(cloud_provider.clone(), hard)
141                .await
142            {
143                Ok(()) => {
144                    loading_spinner.finish_with_message("Region disabled.");
145                    Ok(())
146                }
147                // A 404 means no region exists, so the desired state already
148                // holds. Retrying cannot change the response, so report
149                // success instead of burning the whole retry budget.
150                Err(mz_cloud_api::error::Error::Api(err))
151                    if err.status_code == StatusCode::NOT_FOUND =>
152                {
153                    loading_spinner.finish_with_message("Region already disabled.");
154                    Ok(())
155                }
156                Err(e) => Err(e.into()),
157            }
158        })
159        .await
160}
161
162/// Lists all the available regions and their status.
163pub async fn list(cx: RegionContext) -> Result<(), Error> {
164    let output_formatter = cx.output_formatter();
165    let loading_spinner = output_formatter.loading_spinner("Retrieving regions...");
166
167    #[derive(Deserialize, Serialize, Tabled)]
168    pub struct Region<'a> {
169        #[tabled(rename = "Region")]
170        region: String,
171        #[tabled(rename = "Status")]
172        status: &'a str,
173    }
174
175    let cloud_providers: Vec<CloudProvider> = cx.cloud_client().list_cloud_regions().await?;
176    let mut regions: Vec<Region> = vec![];
177
178    for cloud_provider in cloud_providers {
179        match cx.cloud_client().get_region(cloud_provider.clone()).await {
180            Ok(_) => regions.push(Region {
181                region: cloud_provider.id,
182                status: "enabled",
183            }),
184            Err(mz_cloud_api::error::Error::EmptyRegion) => regions.push(Region {
185                region: cloud_provider.id,
186                status: "disabled",
187            }),
188            Err(err) => {
189                println!("Error: {:?}", err)
190            }
191        }
192    }
193
194    loading_spinner.finish_and_clear();
195    output_formatter.output_table(regions)?;
196    Ok(())
197}
198
199/// Shows the health of the profile region followed by the HTTP and SQL endpoints.
200pub async fn show(cx: RegionContext) -> Result<(), Error> {
201    // Sharing the reference of the context in multiple places makes
202    // it necesarry to wrap in an `alloc::rc`.
203
204    let output_formatter = cx.output_formatter();
205    let loading_spinner = output_formatter.loading_spinner("Retrieving region...");
206
207    let region_info = cx.get_region_info().await?;
208
209    loading_spinner.set_message("Checking environment health...");
210    let claims = cx.admin_client().claims().await?;
211    let sql_client = cx.sql_client();
212    let environment_health = match sql_client.is_ready(&region_info, claims.user()?) {
213        Ok(healthy) => match healthy {
214            true => "yes",
215            _ => "no",
216        },
217        Err(_) => "no",
218    };
219
220    loading_spinner.finish_and_clear();
221    output_formatter.output_scalar(Some(&format!("Healthy: \t{}", environment_health)))?;
222    output_formatter.output_scalar(Some(&format!("SQL address: \t{}", region_info.sql_address)))?;
223    output_formatter.output_scalar(Some(&format!("HTTP URL: \t{}", region_info.http_address)))?;
224
225    Ok(())
226}
mz/command/region.rs

mz/command/
region.rs
