mtop 0.16.2

mtop: top for Memcached
Documentation 
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
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245

use mtop_client::{ServerID, SlabItems, Slabs, Stats};
use std::collections::{HashMap, VecDeque};
use std::fmt;
use std::sync::Arc;
use tokio::runtime::Handle;
use tokio::sync::Mutex;

#[derive(Debug, Clone, Hash, Eq, PartialEq)]
#[repr(transparent)]
pub struct Host(String);

impl Host {
    pub fn from(id: &ServerID) -> Self {
        Self(id.to_string())
    }
}

impl fmt::Display for Host {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.0)
    }
}

#[derive(Debug, Clone)]
pub struct StatsDelta {
    pub previous: ServerStats,
    pub current: ServerStats,
    pub seconds: u64,
}

#[derive(Debug, Clone)]
pub struct ServerStats {
    pub stats: Stats,
    pub slabs: Slabs,
    pub items: SlabItems,
}

#[derive(Debug)]
pub struct StatsQueue {
    queues: Mutex<HashMap<Host, VecDeque<ServerStats>>>,
    max_size: usize,
}

impl StatsQueue {
    pub fn new(max_size: usize) -> Self {
        Self {
            max_size,
            queues: Mutex::new(HashMap::new()),
        }
    }

    pub async fn insert(&self, host: Host, stats: Stats, slabs: Slabs, items: SlabItems) {
        let mut map = self.queues.lock().await;
        let q = map.entry(host).or_default();

        if let Some(prev) = q.back() {
            if stats.uptime == prev.stats.uptime {
                // The most recent entry in the queue has the same uptime as the measurement we're
                // inserting now. This can happen when the server is under heavy load and it isn't
                // able to update the variable it uses to keep track of uptime. Instead of clearing
                // the entire measurement queue, skip this insert instead (less disruptive to users
                // to miss a measurement than cause the UI to reset).
                tracing::debug!(
                    message = "server uptime did not advance, dropping measurement",
                    old_uptime = prev.stats.uptime,
                    current_uptime = stats.uptime,
                );
                return;
            } else if stats.uptime < prev.stats.uptime {
                // The most recent entry in the queue has a higher uptime than the measurement
                // we're inserting now. This means there was a restart and counters are all reset
                // to 0. Clear the existing queue to avoid underflow (since we assume counters can
                // only increase).
                tracing::debug!(
                    message = "server uptime reset detected, clearing measurement queue",
                    old_uptime = prev.stats.uptime,
                    current_uptime = stats.uptime
                );
                q.clear();
            } else if stats.server_time <= prev.stats.server_time {
                // Make sure that any calculations that depend on the number of seconds elapsed
                // since the last measurement don't divide by zero. Realistically this shouldn't
                // happen unless we get really unlucky with the frequency of updates or the
                // memcached server is being purposefully malicious but it doesn't hurt to be
                // defensive.
                tracing::debug!(
                    message = "server timestamp did not advance, dropping measurement",
                    old_timestamp = prev.stats.server_time,
                    current_timestamp = stats.server_time
                );
                return;
            }
        }

        q.push_back(ServerStats { stats, slabs, items });
        if q.len() > self.max_size {
            q.pop_front();
        }
    }

    pub async fn read_delta(&self, host: &Host) -> Option<StatsDelta> {
        let map = self.queues.lock().await;
        map.get(host).and_then(|q| match (q.front(), q.back()) {
            // The delta is only valid if there are more than two entries in the queue. This
            // avoids division by zero errors (since the time for the entries would be the same).
            (Some(previous), Some(current)) if q.len() >= 2 => {
                let seconds = (current.stats.server_time - previous.stats.server_time) as u64;
                Some(StatsDelta {
                    previous: previous.clone(),
                    current: current.clone(),
                    seconds,
                })
            }
            _ => None,
        })
    }

    pub async fn read_last(&self, host: &Host) -> Option<ServerStats> {
        let map = self.queues.lock().await;
        map.get(host).and_then(|q| q.back().cloned())
    }
}

#[derive(Debug)]
pub struct BlockingStatsQueue {
    queue: Arc<StatsQueue>,
    handle: Handle,
}

impl BlockingStatsQueue {
    pub fn new(queue: Arc<StatsQueue>, handle: Handle) -> Self {
        Self { queue, handle }
    }

    pub fn insert(&self, host: Host, stats: Stats, slabs: Slabs, items: SlabItems) {
        self.handle.block_on(self.queue.insert(host, stats, slabs, items))
    }

    pub fn read_delta(&self, host: &Host) -> Option<StatsDelta> {
        self.handle.block_on(self.queue.read_delta(host))
    }

    pub fn read_last(&self, host: &Host) -> Option<ServerStats> {
        self.handle.block_on(self.queue.read_last(host))
    }
}

#[cfg(test)]
mod test {
    use super::{Host, StatsQueue};
    use mtop_client::{ServerID, SlabItems, Slabs, Stats};

    fn fixture(uptime: u64, server_time: i64) -> (Host, Stats, Slabs, SlabItems) {
        let host = Host::from(&ServerID::from(("localhost", 11211)));
        let stats = Stats {
            uptime,
            server_time,
            ..Default::default()
        };
        let slabs = Slabs::default();
        let items = SlabItems::default();

        (host, stats, slabs, items)
    }

    #[tokio::test]
    async fn test_stats_queue_insert_same_uptime() {
        let queue = StatsQueue::new(10);

        let (host, stats, slabs, items) = fixture(1, 1);
        queue.insert(host, stats, slabs, items).await;

        let (host, stats, slabs, items) = fixture(1, 2);
        queue.insert(host.clone(), stats, slabs, items).await;

        // Uptime didn't advance so the second entry should be discarded.
        let last = queue.read_last(&host).await.unwrap();
        assert_eq!(1, last.stats.uptime);
        assert_eq!(1, last.stats.server_time);
    }

    #[tokio::test]
    async fn test_stats_queue_insert_lower_uptime() {
        let queue = StatsQueue::new(10);

        let (host, stats, slabs, items) = fixture(10, 1);
        queue.insert(host, stats, slabs, items).await;

        let (host, stats, slabs, items) = fixture(1, 2);
        queue.insert(host.clone(), stats, slabs, items).await;

        // Uptime reset so the queue should be cleared before the second entry is inserted.
        let last = queue.read_last(&host).await.unwrap();
        assert_eq!(1, last.stats.uptime);
        assert_eq!(2, last.stats.server_time);
    }

    #[tokio::test]
    async fn test_stats_queue_insert_same_server_time() {
        let queue = StatsQueue::new(10);

        let (host, stats, slabs, items) = fixture(1, 1);
        queue.insert(host, stats, slabs, items).await;

        let (host, stats, slabs, items) = fixture(2, 1);
        queue.insert(host.clone(), stats, slabs, items).await;

        // Server time didn't advance so the second entry should be discarded.
        let last = queue.read_last(&host).await.unwrap();
        assert_eq!(1, last.stats.uptime);
        assert_eq!(1, last.stats.server_time);
    }

    #[tokio::test]
    async fn test_stats_queue_insert_lower_server_time() {
        let queue = StatsQueue::new(10);

        let (host, stats, slabs, items) = fixture(1, 10);
        queue.insert(host, stats, slabs, items).await;

        let (host, stats, slabs, items) = fixture(2, 1);
        queue.insert(host.clone(), stats, slabs, items).await;

        // Server time went backwards so the second entry should be discarded.
        let last = queue.read_last(&host).await.unwrap();
        assert_eq!(1, last.stats.uptime);
        assert_eq!(10, last.stats.server_time);
    }

    #[tokio::test]
    async fn test_stats_queue_insert_success() {
        let queue = StatsQueue::new(10);

        let (host, stats, slabs, items) = fixture(1, 1);
        queue.insert(host, stats, slabs, items).await;

        let (host, stats, slabs, items) = fixture(2, 2);
        queue.insert(host.clone(), stats, slabs, items).await;

        // Uptime and server time advanced so the second entry should be stored.
        let last = queue.read_last(&host).await.unwrap();
        assert_eq!(2, last.stats.uptime);
        assert_eq!(2, last.stats.server_time);
    }
}