redis_enterprise/crdb.rs
1//! Active-Active (CRDB) database management
2//!
3//! ## Overview
4//! - Create and manage Active-Active databases
5//! - Configure cross-region replication
6//! - Monitor CRDB status
7
8use crate::client::RestClient;
9use crate::error::Result;
10use serde::{Deserialize, Serialize};
11use serde_json::Value;
12use typed_builder::TypedBuilder;
13
14/// CRDB (Active-Active Database) information
15#[derive(Debug, Clone, Serialize, Deserialize)]
16pub struct Crdb {
17 /// The GUID of the Active-Active database
18 pub guid: String,
19 /// Name of Active-Active database
20 pub name: String,
21 /// Current status of the Active-Active database
22 pub status: String,
23 /// Database memory size limit, in bytes
24 pub memory_size: u64,
25 /// List of participating instances in the Active-Active setup
26 pub instances: Vec<CrdbInstance>,
27 /// Whether communication encryption is enabled
28 pub encryption: Option<bool>,
29 /// Database on-disk persistence policy
30 pub data_persistence: Option<String>,
31 /// Whether database replication is enabled
32 pub replication: Option<bool>,
33 /// Data eviction policy (e.g., 'allkeys-lru', 'volatile-lru')
34 pub eviction_policy: Option<String>,
35}
36
37/// CRDB instance information
38#[derive(Debug, Clone, Serialize, Deserialize)]
39pub struct CrdbInstance {
40 /// Unique instance ID
41 pub id: u32,
42 /// Cluster fully qualified name
43 pub cluster: String,
44 /// Human-readable cluster name
45 pub cluster_name: Option<String>,
46 /// Current status of this instance
47 pub status: String,
48 /// List of endpoint addresses for this instance
49 pub endpoints: Option<Vec<String>>,
50}
51
52/// Create CRDB request
53///
54/// # Examples
55///
56/// ```rust,no_run
57/// use redis_enterprise::{CreateCrdbRequest, CreateCrdbInstance};
58///
59/// let request = CreateCrdbRequest::builder()
60/// .name("global-cache")
61/// .memory_size(1024 * 1024 * 1024) // 1GB
62/// .instances(vec![
63/// CreateCrdbInstance::builder()
64/// .cluster("cluster1.example.com")
65/// .cluster_url("https://cluster1.example.com:9443")
66/// .username("admin")
67/// .password("password")
68/// .build(),
69/// CreateCrdbInstance::builder()
70/// .cluster("cluster2.example.com")
71/// .cluster_url("https://cluster2.example.com:9443")
72/// .username("admin")
73/// .password("password")
74/// .build()
75/// ])
76/// .encryption(true)
77/// .data_persistence("aof")
78/// .build();
79/// ```
80#[derive(Debug, Serialize, TypedBuilder)]
81pub struct CreateCrdbRequest {
82 /// Name of the Active-Active database
83 #[builder(setter(into))]
84 pub name: String,
85 /// Database memory size limit, in bytes
86 pub memory_size: u64,
87 /// List of participating cluster instances
88 pub instances: Vec<CreateCrdbInstance>,
89 /// Whether to encrypt communication between instances
90 #[serde(skip_serializing_if = "Option::is_none")]
91 #[builder(default, setter(strip_option))]
92 pub encryption: Option<bool>,
93 /// Database on-disk persistence policy ('disabled', 'aof', 'snapshot')
94 #[serde(skip_serializing_if = "Option::is_none")]
95 #[builder(default, setter(into, strip_option))]
96 pub data_persistence: Option<String>,
97 /// Data eviction policy when memory limit is reached
98 #[serde(skip_serializing_if = "Option::is_none")]
99 #[builder(default, setter(into, strip_option))]
100 pub eviction_policy: Option<String>,
101}
102
103/// Create CRDB instance
104#[derive(Debug, Serialize, TypedBuilder)]
105pub struct CreateCrdbInstance {
106 /// Cluster fully qualified name, used to uniquely identify the cluster
107 #[builder(setter(into))]
108 pub cluster: String,
109 /// Cluster access URL (for example `https://cluster1.example.com:9443`)
110 #[serde(skip_serializing_if = "Option::is_none")]
111 #[builder(default, setter(into, strip_option))]
112 pub cluster_url: Option<String>,
113 /// Username for cluster authentication
114 #[serde(skip_serializing_if = "Option::is_none")]
115 #[builder(default, setter(into, strip_option))]
116 pub username: Option<String>,
117 /// Password for cluster authentication
118 #[serde(skip_serializing_if = "Option::is_none")]
119 #[builder(default, setter(into, strip_option))]
120 pub password: Option<String>,
121}
122
123/// Module version change requested as part of a CRDB upgrade.
124///
125/// The module UIDs are deprecated by Redis Enterprise Software 7.8.2 and
126/// later, but remain part of the documented request contract for older
127/// supported cluster versions.
128#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, TypedBuilder)]
129pub struct CrdbModuleUpgrade {
130 /// UID of the currently installed module version.
131 #[serde(skip_serializing_if = "Option::is_none")]
132 #[builder(default, setter(into, strip_option))]
133 pub current_module: Option<String>,
134 /// UID of the module version to install.
135 #[serde(skip_serializing_if = "Option::is_none")]
136 #[builder(default, setter(into, strip_option))]
137 pub new_module: Option<String>,
138 /// Arguments to pass to the upgraded module.
139 #[serde(skip_serializing_if = "Option::is_none")]
140 #[builder(default, setter(into, strip_option))]
141 pub new_module_args: Option<String>,
142}
143
144/// Request for upgrading an Active-Active database.
145#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, TypedBuilder)]
146pub struct CrdbUpgradeRequest {
147 /// Allow the upgrade to discard data when required.
148 #[serde(skip_serializing_if = "Option::is_none")]
149 #[builder(default, setter(strip_option))]
150 pub force_discard: Option<bool>,
151 /// Restart the database even when an in-place upgrade is possible.
152 #[serde(skip_serializing_if = "Option::is_none")]
153 #[builder(default, setter(strip_option))]
154 pub force_restart: Option<bool>,
155 /// Retain the current CRDT protocol version after upgrading.
156 #[serde(skip_serializing_if = "Option::is_none")]
157 #[builder(default, setter(strip_option))]
158 pub keep_crdt_protocol_version: Option<bool>,
159 /// Confirm that data loss is acceptable when required by the upgrade.
160 #[serde(skip_serializing_if = "Option::is_none")]
161 #[builder(default, setter(strip_option))]
162 pub may_discard_data: Option<bool>,
163 /// Module changes to apply during the upgrade.
164 #[serde(skip_serializing_if = "Option::is_none")]
165 #[builder(default, setter(strip_option))]
166 pub modules: Option<Vec<CrdbModuleUpgrade>>,
167 /// Maximum number of shards to upgrade in parallel.
168 #[serde(skip_serializing_if = "Option::is_none")]
169 #[builder(default, setter(strip_option))]
170 pub parallel_shards_upgrade: Option<u64>,
171 /// Preserve existing database roles during the upgrade.
172 #[serde(skip_serializing_if = "Option::is_none")]
173 #[builder(default, setter(strip_option))]
174 pub preserve_roles: Option<bool>,
175 /// Redis version to install.
176 #[serde(skip_serializing_if = "Option::is_none")]
177 #[builder(default, setter(into, strip_option))]
178 pub redis_version: Option<String>,
179}
180
181/// Response wrapper for `GET /v1/crdbs`.
182///
183/// The API returns `{"crdbs": [...]}`. Kept private to the module since
184/// [`CrdbHandler::list`] unwraps it; only exposed in the public API via
185/// the flat `Vec<Crdb>` return.
186#[derive(Debug, Clone, Deserialize)]
187struct CrdbsListResponse {
188 #[serde(default)]
189 crdbs: Vec<Crdb>,
190}
191
192/// CRDB handler for managing Active-Active databases
193pub struct CrdbHandler {
194 client: RestClient,
195}
196
197impl CrdbHandler {
198 /// Create a new handler bound to the given REST client.
199 pub fn new(client: RestClient) -> Self {
200 CrdbHandler { client }
201 }
202
203 /// List all CRDBs.
204 ///
205 /// `GET /v1/crdbs`. The API wraps the array under a `crdbs` key
206 /// (`{"crdbs": [...]}`); this method unwraps it so callers get a
207 /// flat `Vec<Crdb>`.
208 pub async fn list(&self) -> Result<Vec<Crdb>> {
209 let resp: CrdbsListResponse = self.client.get("/v1/crdbs").await?;
210 Ok(resp.crdbs)
211 }
212
213 /// Get specific CRDB
214 pub async fn get(&self, guid: &str) -> Result<Crdb> {
215 self.client.get(&format!("/v1/crdbs/{}", guid)).await
216 }
217
218 /// Create new CRDB
219 pub async fn create(&self, request: CreateCrdbRequest) -> Result<Crdb> {
220 self.client.post("/v1/crdbs", &request).await
221 }
222
223 /// Update an existing CRDB.
224 ///
225 /// `PATCH /v1/crdbs/{crdb_guid}`. The Redis Enterprise REST API
226 /// documents the CRDB update verb as `PATCH`; the previous
227 /// implementation used `PUT` and returned 405 on recent cluster
228 /// versions.
229 pub async fn update(&self, guid: &str, updates: Value) -> Result<Crdb> {
230 let response = self
231 .client
232 .patch_raw(&format!("/v1/crdbs/{}", guid), updates)
233 .await?;
234 serde_json::from_value(response).map_err(Into::into)
235 }
236
237 /// Delete CRDB
238 pub async fn delete(&self, guid: &str) -> Result<()> {
239 self.client.delete(&format!("/v1/crdbs/{}", guid)).await
240 }
241
242 /// Get tasks for a CRDB.
243 ///
244 /// Redis Software exposes tasks as a global collection, not beneath an
245 /// individual CRDB. Preserve this convenience method by filtering the
246 /// canonical collection client-side.
247 pub async fn tasks(&self, guid: &str) -> Result<Value> {
248 let tasks: Vec<Value> = self.client.get("/v1/crdb_tasks").await?;
249 Ok(Value::Array(
250 tasks
251 .into_iter()
252 .filter(|task| task.get("crdb_guid").and_then(Value::as_str) == Some(guid))
253 .collect(),
254 ))
255 }
256
257 /// Flush all data from an Active-Active database.
258 ///
259 /// `PUT /v1/crdbs/{crdb_guid}/flush`. The request body is intentionally
260 /// `serde_json::Value` because the documented payload is a small set of
261 /// optional flags (e.g. `{}` for the default flush) whose accepted shape
262 /// is version-specific; pass `json!({})` for the common case.
263 pub async fn flush(&self, guid: &str, body: Value) -> Result<Value> {
264 self.client
265 .put_raw(&format!("/v1/crdbs/{}/flush", guid), body)
266 .await
267 }
268
269 /// Retrieve the health report for an Active-Active database.
270 ///
271 /// `GET /v1/crdbs/{crdb_guid}/health_report`. The response is returned
272 /// as `serde_json::Value` because the report is a richly structured
273 /// document whose shape evolves across cluster versions.
274 pub async fn health_report(&self, guid: &str) -> Result<Value> {
275 self.client
276 .get(&format!("/v1/crdbs/{}/health_report", guid))
277 .await
278 }
279
280 /// Purge data from an instance that was forcibly removed from an
281 /// Active-Active database.
282 ///
283 /// `PUT /v1/crdbs/{crdb_guid}/purge`. The body identifies the
284 /// instance(s) to purge; pass a `Value` matching the documented
285 /// shape for the cluster version under test.
286 pub async fn purge(&self, guid: &str, body: Value) -> Result<Value> {
287 self.client
288 .put_raw(&format!("/v1/crdbs/{}/purge", guid), body)
289 .await
290 }
291
292 /// Submit a configuration update against an Active-Active database.
293 ///
294 /// `POST /v1/crdbs/{crdb_guid}/updates`. The body shape is version-
295 /// specific; pass a `Value` with the desired field changes.
296 pub async fn updates(&self, guid: &str, body: Value) -> Result<Value> {
297 self.client
298 .post_raw(&format!("/v1/crdbs/{}/updates", guid), body)
299 .await
300 }
301
302 /// Upgrade an Active-Active database.
303 ///
304 /// Calls `POST /v1/crdbs/{crdb_guid}/upgrade`. The response is returned
305 /// as `serde_json::Value` because the currently documented CRDB task
306 /// object differs from the crate's legacy task model and varies across
307 /// supported cluster versions.
308 pub async fn upgrade(&self, guid: &str, request: CrdbUpgradeRequest) -> Result<Value> {
309 self.client
310 .post(&format!("/v1/crdbs/{}/upgrade", guid), &request)
311 .await
312 }
313}