1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
//! Shards management for Redis Enterprise
//!
//! ## Overview
//! - List and query resources
//! - Create and update configurations
//! - Monitor status and metrics
use crate::client::RestClient;
use crate::error::{RestError, Result};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::BTreeMap;
/// Response for a single metric query
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MetricResponse {
/// Interval label for the metric series.
pub interval: String,
/// List of Unix-epoch timestamps for the data points.
pub timestamps: Vec<i64>,
/// List of metric values, aligned to `timestamps`.
pub values: Vec<Value>,
}
/// Current shard dump-file loading state.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[non_exhaustive]
pub struct ShardLoading {
/// Percentage of bytes already loaded (0-100).
pub progress: Option<f64>,
/// Load status (`in_progress` or `idle`).
pub status: Option<String>,
/// Additive or version-specific loading fields.
#[serde(default, flatten, skip_serializing_if = "BTreeMap::is_empty")]
pub additional_fields: BTreeMap<String, Value>,
}
/// Shard information
#[derive(Debug, Clone, Serialize, Deserialize)]
#[non_exhaustive]
pub struct Shard {
/// Unique identifier (read-only).
pub uid: String,
/// Database (BDB) UID this entity belongs to.
pub bdb_uid: u32,
/// Node UID this entity belongs to.
pub node_uid: String,
/// Role.
pub role: String,
/// Current status.
pub status: String,
/// More detailed shard status.
pub detailed_status: Option<String>,
/// Current dump-file loading state.
pub loading: Option<ShardLoading>,
/// Time at which the shard information was collected.
pub report_timestamp: Option<String>,
/// Effective role reported by Redis Software 8.x.
pub actual_role: Option<String>,
/// Whether Redis Software has marked the shard for removal.
pub marked_for_removal: Option<bool>,
/// Hash slot range owned by this shard.
#[serde(skip_serializing_if = "Option::is_none")]
pub slots: Option<String>,
/// Used memory in bytes.
#[serde(skip_serializing_if = "Option::is_none")]
pub used_memory: Option<u64>,
/// Percent progress (0-100) of in-flight backup.
#[serde(skip_serializing_if = "Option::is_none")]
pub backup_progress: Option<f64>,
/// Percent progress (0-100) of in-flight import.
#[serde(skip_serializing_if = "Option::is_none")]
pub import_progress: Option<f64>,
/// All nodes that this shard is associated with
pub all_nodes: Option<Vec<u32>>,
/// Assigned slots for this shard
pub assigned_slots: Option<String>,
/// Client certificate subject validation type
pub client_cert_subject_validation_type: Option<String>,
/// Redis info for this shard
pub redis_info: Option<Value>,
/// Roles assigned to this shard
pub roles: Option<Vec<String>>,
/// Additive or version-specific shard fields.
#[serde(default, flatten, skip_serializing_if = "BTreeMap::is_empty")]
pub additional_fields: BTreeMap<String, Value>,
}
/// Shard stats information
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ShardStats {
/// Unique identifier (read-only).
pub uid: String,
/// Per-interval metric series for the resource.
pub intervals: Vec<StatsInterval>,
}
/// One interval of statistics, with aligned timestamps and values.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StatsInterval {
/// Interval label for the metric series.
pub interval: String,
/// List of Unix-epoch timestamps for the data points.
pub timestamps: Vec<i64>,
/// List of metric values, aligned to `timestamps`.
pub values: Vec<Value>,
}
/// Shard handler for managing shards
pub struct ShardHandler {
client: RestClient,
}
impl ShardHandler {
/// Create a new handler bound to the given REST client.
pub fn new(client: RestClient) -> Self {
ShardHandler { client }
}
/// List all shards
pub async fn list(&self) -> Result<Vec<Shard>> {
self.client.get("/v1/shards").await
}
/// Get specific shard information
pub async fn get(&self, uid: &str) -> Result<Shard> {
self.client.get(&format!("/v1/shards/{}", uid)).await
}
/// Get shard statistics
pub async fn stats(&self, uid: &str) -> Result<ShardStats> {
self.client.get(&format!("/v1/shards/stats/{}", uid)).await
}
/// Get the first interval of one metric from canonical shard statistics.
pub async fn stats_metric(&self, uid: &str, metric: &str) -> Result<MetricResponse> {
let stats = self.stats(uid).await?;
let interval = stats.intervals.into_iter().next().ok_or_else(|| {
RestError::ParseError(format!("shard {uid} returned no statistics intervals"))
})?;
let values = interval
.values
.into_iter()
.map(|sample| {
sample.get(metric).cloned().ok_or_else(|| {
RestError::ParseError(format!(
"metric {metric:?} is absent from shard {uid} statistics"
))
})
})
.collect::<Result<Vec<_>>>()?;
Ok(MetricResponse {
interval: interval.interval,
timestamps: interval.timestamps,
values,
})
}
// raw variant removed: use stats_metric()
/// Get shards for a specific database
pub async fn list_by_database(&self, bdb_uid: u32) -> Result<Vec<Shard>> {
self.client
.get(&format!("/v1/bdbs/{}/shards", bdb_uid))
.await
}
/// Get shards for a specific node.
///
/// Redis Software does not expose a node-scoped shard route. Fetch the
/// documented global collection and filter it client-side instead.
pub async fn list_by_node(&self, node_uid: u32) -> Result<Vec<Shard>> {
let node_uid = node_uid.to_string();
Ok(self
.list()
.await?
.into_iter()
.filter(|shard| shard.node_uid == node_uid)
.collect())
}
// Aggregate raw helpers removed; use StatsHandler for aggregates
/// Global failover - POST /v1/shards/actions/failover
pub async fn failover_all(&self, body: ShardActionRequest) -> Result<Action> {
self.client.post("/v1/shards/actions/failover", &body).await
}
/// Global migrate - POST /v1/shards/actions/migrate
pub async fn migrate_all(&self, body: ShardActionRequest) -> Result<Action> {
self.client.post("/v1/shards/actions/migrate", &body).await
}
/// Per-shard failover - POST /v1/shards/{uid}/actions/failover
pub async fn failover(&self, uid: &str, body: ShardActionRequest) -> Result<Action> {
self.client
.post(&format!("/v1/shards/{}/actions/failover", uid), &body)
.await
}
/// Per-shard migrate - POST /v1/shards/{uid}/actions/migrate
pub async fn migrate(&self, uid: &str, body: ShardActionRequest) -> Result<Action> {
self.client
.post(&format!("/v1/shards/{}/actions/migrate", uid), &body)
.await
}
}
/// Request body for shard action endpoints (failover, migrate).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ShardActionRequest {
/// List of shard UIDs targeted by this action.
#[serde(skip_serializing_if = "Option::is_none")]
pub shard_uids: Option<Vec<String>>,
}
/// Response from a shard action endpoint with the tracking UID.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Action {
/// Action UID for tracking async operations (read-only).
pub action_uid: String,
/// Current status.
#[serde(skip_serializing_if = "Option::is_none")]
pub status: Option<String>,
}