netget/server/redis/
cluster.rs

1//! Cluster mode: this server presents itself as one master of a Redis Cluster whose shape the
2//! operator declares in the `cluster_nodes` startup parameter, and which the model can change
3//! at runtime with `redis_cluster_set_slot`.
4//!
5//! There is no cluster bus and no gossip. Several NetGet Redis servers form a cluster only in
6//! the sense a client can observe: each is told the same topology, each answers the
7//! `CLUSTER` introspection commands from it, and each redirects a command for a key whose hash
8//! slot it does not own with `-MOVED <slot> <host>:<port>` — which is everything a cluster
9//! client (`redis-cli -c`, redis-rs's `ClusterClient`) navigates by. The slot arithmetic is
10//! `utils::redis_keys`, shared with NetGet's own cluster client.
11//!
12//! Slot migration follows `CLUSTER SETSLOT`'s states. A slot this node owns and has marked
13//! *migrating* to another node answers every key in it with `-ASK <slot> <target>` — Redis
14//! does so only for keys it no longer holds, and this server holds no keys, so every key has
15//! "already moved". A slot this node is *importing* is served only to a connection that sent
16//! `ASKING` immediately before; any other command for it gets `-MOVED` to the owner.
17use super::resp_value::Resp;
18use crate::utils::redis_keys::CLUSTER_SLOTS;
19use anyhow::{bail, ensure, Context, Result};
20use serde_json::Value;
21use std::collections::BTreeMap;
22use std::net::SocketAddr;
23
24/// Most nodes a topology may name. A real cluster tops out around 1000; a hand-written
25/// startup parameter with more than this is a mistake, and every `CLUSTER NODES` reply is
26/// proportional to it.
27pub const MAX_CLUSTER_NODES: usize = 256;
28
29/// One master in the declared topology.
30#[derive(Debug, Clone, PartialEq)]
31pub struct ClusterNode {
32    pub id: String,
33    pub host: String,
34    pub port: u16,
35    pub epoch: u64,
36}
37
38impl ClusterNode {
39    pub fn addr(&self) -> String {
40        format!("{}:{}", self.host, self.port)
41    }
42}
43
44/// Where a command goes.
45#[derive(Debug, Clone, PartialEq)]
46pub enum Route {
47    /// Served here.
48    Local,
49    /// `-MOVED <slot> <addr>`.
50    Moved(u16, String),
51    /// `-ASK <slot> <addr>`.
52    Ask(u16, String),
53    /// Keys in more than one slot.
54    CrossSlot,
55    /// The slot is assigned to no node.
56    Down(u16),
57}
58
59impl Route {
60    /// The RESP error this route answers with, or `None` for [`Route::Local`].
61    pub fn error(&self) -> Option<String> {
62        match self {
63            Route::Local => None,
64            Route::Moved(slot, addr) => Some(format!("MOVED {slot} {addr}")),
65            Route::Ask(slot, addr) => Some(format!("ASK {slot} {addr}")),
66            Route::CrossSlot => {
67                Some("CROSSSLOT Keys in request don't hash to the same slot".to_string())
68            }
69            Route::Down(_) => Some("CLUSTERDOWN Hash slot not served".to_string()),
70        }
71    }
72}
73
74/// The cluster as this node sees it.
75#[derive(Debug, Clone)]
76pub struct ClusterState {
77    pub nodes: Vec<ClusterNode>,
78    pub myself: usize,
79    /// Owner of each slot, as an index into `nodes`.
80    owner: Vec<Option<usize>>,
81    migrating: BTreeMap<u16, usize>,
82    importing: BTreeMap<u16, usize>,
83    current_epoch: u64,
84}
85
86/// A node id: 40 lowercase hex characters. Derived from `host:port` when the topology does not
87/// give one, so every server handed the same topology names every node the same way.
88fn derive_id(addr: &str) -> String {
89    // FNV-1a, three lanes with different offsets, truncated to 160 bits: stable across
90    // processes and Rust versions, which `DefaultHasher` does not promise.
91    let mut out = String::with_capacity(48);
92    for seed in [
93        0xcbf2_9ce4_8422_2325u64,
94        0x8422_2325_cbf2_9ce4,
95        0x9ce4_cbf2_2325_8422,
96    ] {
97        let mut h = seed;
98        for b in addr.bytes() {
99            h ^= u64::from(b);
100            h = h.wrapping_mul(0x0000_0100_0000_01b3);
101        }
102        out.push_str(&format!("{h:016x}"));
103    }
104    out.truncate(40);
105    out
106}
107
108fn valid_host(host: &str) -> bool {
109    !host.is_empty()
110        && host.len() <= 255
111        && host
112            .bytes()
113            .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'.' | b'-' | b'_' | b':'))
114}
115
116/// Parse a slot spec: an integer, `"N"` or `"A-B"`.
117pub fn parse_slots(v: &Value) -> Result<std::ops::RangeInclusive<u16>> {
118    let (a, b) = match v {
119        Value::Number(n) => {
120            let n = n.as_u64().context("a slot is a non-negative integer")?;
121            (n, n)
122        }
123        Value::String(s) => match s.split_once('-') {
124            Some((a, b)) => (
125                a.trim().parse::<u64>().context("slot range start")?,
126                b.trim().parse::<u64>().context("slot range end")?,
127            ),
128            None => {
129                let n = s.trim().parse::<u64>().context("a slot is an integer")?;
130                (n, n)
131            }
132        },
133        _ => bail!("a slot is an integer or a \"start-end\" string"),
134    };
135    ensure!(
136        a <= b && b < u64::from(CLUSTER_SLOTS),
137        "slots run from 0 to {}, start first",
138        CLUSTER_SLOTS - 1
139    );
140    Ok(a as u16..=b as u16)
141}
142
143impl ClusterState {
144    /// Build the topology from the `cluster_nodes` startup parameter.
145    ///
146    /// Each entry is `{"host", "port", "slots": ["0-5460", …], "id"?, "myself"?}`. Exactly one
147    /// entry is `myself`; its `host` and `port` default to the address this server bound, so a
148    /// server started on port 0 can still describe itself. Two nodes may not claim one slot.
149    pub fn from_param(value: &Value, bound: SocketAddr) -> Result<Self> {
150        let entries = value
151            .as_array()
152            .context("cluster_nodes is an array of nodes")?;
153        ensure!(!entries.is_empty(), "cluster_nodes names no node");
154        ensure!(
155            entries.len() <= MAX_CLUSTER_NODES,
156            "cluster_nodes names more than {MAX_CLUSTER_NODES} nodes"
157        );
158        let mut nodes = Vec::new();
159        let mut myself = None;
160        let mut owner = vec![None; usize::from(CLUSTER_SLOTS)];
161        for (i, entry) in entries.iter().enumerate() {
162            let is_me = entry
163                .get("myself")
164                .and_then(Value::as_bool)
165                .unwrap_or(false);
166            if is_me {
167                ensure!(
168                    myself.is_none(),
169                    "more than one cluster node is marked myself"
170                );
171                myself = Some(i);
172            }
173            let host = match entry.get("host").and_then(Value::as_str) {
174                Some(h) => h.to_string(),
175                None if is_me => {
176                    if bound.ip().is_unspecified() {
177                        "127.0.0.1".to_string()
178                    } else {
179                        bound.ip().to_string()
180                    }
181                }
182                None => bail!("cluster node {i} has no host"),
183            };
184            ensure!(
185                valid_host(&host),
186                "cluster node {i}'s host is not a hostname or IP address"
187            );
188            let port = match entry.get("port").and_then(Value::as_u64) {
189                Some(p) if (1..=65535).contains(&p) => p as u16,
190                Some(_) => bail!("cluster node {i}'s port is out of range"),
191                None if is_me => bound.port(),
192                None => bail!("cluster node {i} has no port"),
193            };
194            let addr = format!("{host}:{port}");
195            let id = match entry.get("id").and_then(Value::as_str) {
196                Some(id) => {
197                    ensure!(
198                        id.len() == 40 && id.bytes().all(|b| b.is_ascii_hexdigit()),
199                        "cluster node {i}'s id is not 40 hex characters"
200                    );
201                    id.to_ascii_lowercase()
202                }
203                None => derive_id(&addr),
204            };
205            if let Some(slots) = entry.get("slots") {
206                let specs = slots
207                    .as_array()
208                    .context("a node's slots is an array of \"start-end\" ranges")?;
209                for spec in specs {
210                    for slot in parse_slots(spec)? {
211                        let cell = &mut owner[usize::from(slot)];
212                        ensure!(cell.is_none(), "slot {slot} is claimed by two nodes");
213                        *cell = Some(i);
214                    }
215                }
216            }
217            nodes.push(ClusterNode {
218                id,
219                host,
220                port,
221                epoch: i as u64 + 1,
222            });
223        }
224        let myself = myself.context("no cluster node is marked myself: true")?;
225        let mut seen = std::collections::HashSet::new();
226        for n in &nodes {
227            ensure!(
228                seen.insert(n.id.clone()),
229                "two cluster nodes share id {}",
230                n.id
231            );
232        }
233        let current_epoch = nodes.len() as u64;
234        Ok(ClusterState {
235            nodes,
236            myself,
237            owner,
238            migrating: BTreeMap::new(),
239            importing: BTreeMap::new(),
240            current_epoch,
241        })
242    }
243
244    pub fn me(&self) -> &ClusterNode {
245        &self.nodes[self.myself]
246    }
247
248    pub fn owner_of(&self, slot: u16) -> Option<&ClusterNode> {
249        self.owner[usize::from(slot)].map(|i| &self.nodes[i])
250    }
251
252    /// Where a command whose keys hash to `slots` goes. `asking` is the connection's one-shot
253    /// `ASKING` flag.
254    pub fn route(&self, slots: &[u16], asking: bool) -> Route {
255        let Some(&slot) = slots.first() else {
256            return Route::Local;
257        };
258        if slots.iter().any(|s| *s != slot) {
259            return Route::CrossSlot;
260        }
261        match self.owner[usize::from(slot)] {
262            Some(i) if i == self.myself => match self.migrating.get(&slot) {
263                Some(&target) => Route::Ask(slot, self.nodes[target].addr()),
264                None => Route::Local,
265            },
266            _ if asking && self.importing.contains_key(&slot) => Route::Local,
267            Some(i) => Route::Moved(slot, self.nodes[i].addr()),
268            None => Route::Down(slot),
269        }
270    }
271
272    fn node_index(&self, addr_or_id: &str) -> Result<usize> {
273        self.nodes
274            .iter()
275            .position(|n| n.addr() == addr_or_id || n.id == addr_or_id)
276            .with_context(|| format!("no cluster node {addr_or_id}"))
277    }
278
279    /// `CLUSTER SETSLOT`, as an action: `state` is `node` (assign to `node`), `migrating` (this
280    /// node's slot, moving to `node`), `importing` (from `node`), or `stable` (clear both).
281    pub fn set_slot(
282        &mut self,
283        slots: std::ops::RangeInclusive<u16>,
284        state: &str,
285        node: Option<&str>,
286    ) -> Result<String> {
287        let target = node.map(|n| self.node_index(n)).transpose()?;
288        let count = slots.clone().count();
289        match state {
290            "node" => {
291                let t = target.context("state node needs the node that now owns the slots")?;
292                for s in slots {
293                    self.owner[usize::from(s)] = Some(t);
294                    self.migrating.remove(&s);
295                    self.importing.remove(&s);
296                }
297                self.current_epoch += 1;
298                self.nodes[t].epoch = self.current_epoch;
299                Ok(format!(
300                    "{count} slot(s) now owned by {}",
301                    self.nodes[t].addr()
302                ))
303            }
304            "migrating" => {
305                let t = target.context("state migrating needs the node the slots move to")?;
306                for s in slots.clone() {
307                    ensure!(
308                        self.owner[usize::from(s)] == Some(self.myself),
309                        "slot {s} is not owned by this node, so it cannot migrate from it"
310                    );
311                }
312                for s in slots {
313                    self.migrating.insert(s, t);
314                }
315                Ok(format!(
316                    "{count} slot(s) migrating to {}",
317                    self.nodes[t].addr()
318                ))
319            }
320            "importing" => {
321                let t = target.context("state importing needs the node the slots come from")?;
322                for s in slots {
323                    self.importing.insert(s, t);
324                }
325                Ok(format!(
326                    "{count} slot(s) importing from {}",
327                    self.nodes[t].addr()
328                ))
329            }
330            "stable" => {
331                for s in slots {
332                    self.migrating.remove(&s);
333                    self.importing.remove(&s);
334                }
335                Ok(format!("{count} slot(s) stable"))
336            }
337            other => bail!("state is node, migrating, importing or stable, not {other}"),
338        }
339    }
340
341    /// Contiguous runs of slots with one owner: `(start, end, node index)`.
342    fn ranges(&self) -> Vec<(u16, u16, usize)> {
343        let mut out: Vec<(u16, u16, usize)> = Vec::new();
344        for slot in 0..CLUSTER_SLOTS {
345            let Some(owner) = self.owner[usize::from(slot)] else {
346                continue;
347            };
348            match out.last_mut() {
349                Some((_, end, o)) if *o == owner && *end + 1 == slot => *end = slot,
350                _ => out.push((slot, slot, owner)),
351            }
352        }
353        out
354    }
355
356    fn assigned(&self) -> usize {
357        self.owner.iter().filter(|o| o.is_some()).count()
358    }
359
360    fn node_entry(&self, i: usize) -> Resp {
361        let n = &self.nodes[i];
362        // Redis 7's fourth element is a map of extra networking metadata; empty here.
363        Resp::Array(vec![
364            Resp::bulk(&n.host),
365            Resp::Int(i64::from(n.port)),
366            Resp::bulk(&n.id),
367            Resp::Map(vec![]),
368        ])
369    }
370
371    /// `CLUSTER SLOTS`.
372    pub fn slots_reply(&self) -> Resp {
373        Resp::Array(
374            self.ranges()
375                .into_iter()
376                .map(|(start, end, owner)| {
377                    Resp::Array(vec![
378                        Resp::Int(i64::from(start)),
379                        Resp::Int(i64::from(end)),
380                        self.node_entry(owner),
381                    ])
382                })
383                .collect(),
384        )
385    }
386
387    /// `CLUSTER SHARDS`: one shard per master, each with its slot ranges and its one node.
388    pub fn shards_reply(&self) -> Resp {
389        let ranges = self.ranges();
390        Resp::Array(
391            self.nodes
392                .iter()
393                .enumerate()
394                .map(|(i, n)| {
395                    let slots = ranges
396                        .iter()
397                        .filter(|(_, _, o)| *o == i)
398                        .flat_map(|(s, e, _)| [Resp::Int(i64::from(*s)), Resp::Int(i64::from(*e))])
399                        .collect();
400                    Resp::Map(vec![
401                        (Resp::bulk("slots"), Resp::Array(slots)),
402                        (
403                            Resp::bulk("nodes"),
404                            Resp::Array(vec![Resp::Map(vec![
405                                (Resp::bulk("id"), Resp::bulk(&n.id)),
406                                (Resp::bulk("port"), Resp::Int(i64::from(n.port))),
407                                (Resp::bulk("ip"), Resp::bulk(&n.host)),
408                                (Resp::bulk("endpoint"), Resp::bulk(&n.host)),
409                                (Resp::bulk("role"), Resp::bulk("master")),
410                                (Resp::bulk("replication-offset"), Resp::Int(0)),
411                                (Resp::bulk("health"), Resp::bulk("online")),
412                            ])]),
413                        ),
414                    ])
415                })
416                .collect(),
417        )
418    }
419
420    /// `CLUSTER NODES`: one line per node in Redis's own format.
421    pub fn nodes_reply(&self) -> Resp {
422        let ranges = self.ranges();
423        let mut text = String::new();
424        for (i, n) in self.nodes.iter().enumerate() {
425            let flags = if i == self.myself {
426                "myself,master"
427            } else {
428                "master"
429            };
430            text.push_str(&format!(
431                "{} {}:{}@{} {} - 0 0 {} connected",
432                n.id,
433                n.host,
434                n.port,
435                u32::from(n.port) + 10000,
436                flags,
437                n.epoch
438            ));
439            for (s, e, _) in ranges.iter().filter(|(_, _, o)| *o == i) {
440                if s == e {
441                    text.push_str(&format!(" {s}"));
442                } else {
443                    text.push_str(&format!(" {s}-{e}"));
444                }
445            }
446            if i == self.myself {
447                for (slot, t) in &self.migrating {
448                    text.push_str(&format!(" [{slot}->-{}]", self.nodes[*t].id));
449                }
450                for (slot, t) in &self.importing {
451                    text.push_str(&format!(" [{slot}-<-{}]", self.nodes[*t].id));
452                }
453            }
454            text.push('\n');
455        }
456        Resp::Verbatim(text)
457    }
458
459    /// `CLUSTER INFO`.
460    pub fn info_reply(&self) -> Resp {
461        let assigned = self.assigned();
462        let state = if assigned == usize::from(CLUSTER_SLOTS) {
463            "ok"
464        } else {
465            "fail"
466        };
467        let size = self
468            .nodes
469            .iter()
470            .enumerate()
471            .filter(|(i, _)| self.owner.contains(&Some(*i)))
472            .count();
473        let lines = [
474            format!("cluster_state:{state}"),
475            format!("cluster_slots_assigned:{assigned}"),
476            format!("cluster_slots_ok:{assigned}"),
477            "cluster_slots_pfail:0".to_string(),
478            "cluster_slots_fail:0".to_string(),
479            format!("cluster_known_nodes:{}", self.nodes.len()),
480            format!("cluster_size:{size}"),
481            format!("cluster_current_epoch:{}", self.current_epoch),
482            format!("cluster_my_epoch:{}", self.me().epoch),
483            "cluster_stats_messages_sent:0".to_string(),
484            "cluster_stats_messages_received:0".to_string(),
485            "total_cluster_links_buffer_limit_exceeded:0".to_string(),
486        ];
487        Resp::Verbatim(lines.iter().map(|l| format!("{l}\r\n")).collect())
488    }
489
490    /// The topology as JSON, for the model and the dashboard.
491    pub fn snapshot(&self) -> Value {
492        let ranges = self.ranges();
493        serde_json::json!({
494            "myself": self.me().addr(),
495            "nodes": self.nodes.iter().enumerate().map(|(i, n)| serde_json::json!({
496                "id": n.id,
497                "addr": n.addr(),
498                "slots": ranges.iter().filter(|(_, _, o)| *o == i)
499                    .map(|(s, e, _)| format!("{s}-{e}")).collect::<Vec<_>>(),
500            })).collect::<Vec<_>>(),
501            "migrating": self.migrating.iter().map(|(s, t)| (s.to_string(), self.nodes[*t].addr())).collect::<BTreeMap<_, _>>(),
502            "importing": self.importing.iter().map(|(s, t)| (s.to_string(), self.nodes[*t].addr())).collect::<BTreeMap<_, _>>(),
503            "slots_assigned": self.assigned(),
504        })
505    }
506}
netget/server/redis/cluster.rs

netget/server/redis/
cluster.rs
