Skip to content

Commit 7b9c81e

Browse files
authored
Selene manager: store work proportional to changes, not record count (#139)
* Selene manager: build the metrics text when it is scraped The owner thread rebuilt the whole metrics text after every command batch, walking every record twice per namespace. A scrape now asks the owner for the text, so an idle store does no per-record work and a busy one does it once per scrape. Design-Signed-Off-By: yatesdr; design-review=1c01 * Selene manager: refresh durability only for records that can change Every publish, disk-write completion and compaction completion recomputed durability for every record in the namespace, sorted by depth. The store now keeps the set of records that are not durable yet and a refresh visits the touched record, and that set only when the record has children and has just become durable; replacing a durable record still rescans, since its descendants may stop being durable. Recovery still rebuilds the set from the whole tree. Design-Signed-Off-By: yatesdr; design-review=1c01 * Selene manager: schedule disk writes from the not-yet-durable set schedule() walked every record in every namespace on each pump to find the few that still need a disk write. It now walks the set of records that are not durable yet, which the durability refresh already maintains. Design-Signed-Off-By: yatesdr; design-review=1c01 * Selene manager: build the catalog snapshot without sorting The snapshot sorted every key before building the entries, but its only consumer upserts them into a hash index, so order is unspecified and the sort is gone. Design-Signed-Off-By: yatesdr; design-review=1c01 * Selene manager: iterate a bucket in place instead of copying and sorting it in_bucket cloned and sorted the whole bucket on every boundary of every lookup. The bucket is now an ordered set, so the lookup iterates it without a copy, in value order, which is the order the old sort produced; the candidate order and the scan budget cutoff are unchanged. Design-Signed-Off-By: yatesdr; design-review=1c01 * Selene manager: scale test for a publish into a large store An ignored, release-only test that holds 100k records and a 10k backlog of records not yet on disk, asserts the work after a publish (durability refresh and scheduling) takes under a millisecond, and checks that the incremental refreshes agree with a full rescan. Design-Signed-Off-By: yatesdr; design-review=1c01
1 parent 9920d90 commit 7b9c81e

4 files changed

Lines changed: 271 additions & 50 deletions

File tree

‎selene-manager/docs/operating.md‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -235,7 +235,8 @@ the total; a nonzero value that does not fall means an instance is hung, not a l
235235
## Metrics
236236

237237
`GET /metrics` on `--metrics HOST:PORT` returns Prometheus text (`selene_` prefix, content type
238-
`text/plain; version=0.0.4`); only that path is served.
238+
`text/plain; version=0.0.4`); only that path is served. Store series are built when scraped, on the store thread; if it
239+
does not answer within 1 s the previous store text is served.
239240

240241
| Series | Meaning | Alert on |
241242
|---|---|---|

‎selene-manager/src/control.rs‎

Lines changed: 14 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -180,7 +180,6 @@ pub fn serve(config: Config, stop: Arc<AtomicBool>) -> Result<()> {
180180
}
181181
// Bind every listener before any thread starts, so a startup failure leaves no threads
182182
// and no locks behind; the guard below also removes the socket it creates.
183-
let metrics_text = Arc::new(Mutex::new(String::new()));
184183
let metrics_address = config.metrics.map(|address| address.to_string());
185184
let metrics_listener = bind_metrics(metrics_address.as_deref())?;
186185
let network_listener = match config.listen {
@@ -211,12 +210,11 @@ pub fn serve(config: Config, stop: Arc<AtomicBool>) -> Result<()> {
211210
if let Some(listener) = metrics_listener {
212211
cleanup.metrics = Some(serve_metrics(
213212
listener,
214-
metrics_text.clone(),
213+
commands.clone(),
215214
pool_handle.shared.clone(),
216215
stop.clone(),
217216
));
218217
}
219-
let owner_metrics = metrics_text.clone();
220218
let region_error = Arc::new(Mutex::new(None));
221219
let owner_error = region_error.clone();
222220
let owner_stop = stop.clone();
@@ -253,7 +251,6 @@ pub fn serve(config: Config, stop: Arc<AtomicBool>) -> Result<()> {
253251
}
254252
}
255253
store.pump();
256-
*owner_metrics.lock().unwrap() = store.metrics_text();
257254
}
258255
}));
259256
let stop_notice = Arc::new(Notify::new());
@@ -454,11 +451,12 @@ fn bind_metrics(address: Option<&str>) -> Result<Option<TcpListener>> {
454451

455452
fn serve_metrics(
456453
listener: TcpListener,
457-
text: Arc<Mutex<String>>,
454+
commands: tokio::sync::mpsc::Sender<StoreCommand>,
458455
pool: Arc<crate::pool::Shared>,
459456
stop: Arc<AtomicBool>,
460457
) -> thread::JoinHandle<()> {
461458
thread::spawn(move || {
459+
let mut store_text = String::new();
462460
while !stop.load(Ordering::Acquire) {
463461
match listener.accept() {
464462
Ok((mut stream, _)) => {
@@ -468,7 +466,17 @@ fn serve_metrics(
468466
let count = stream.read(&mut request).unwrap_or(0);
469467
let (status, body) = if request[..count].starts_with(b"GET /metrics ") {
470468
("200 OK", {
471-
let mut body = text.lock().unwrap().clone();
469+
// The owner builds the store's text at scrape time; if it does
470+
// not answer within 1 s, the last text is served again.
471+
let (reply, text) = std::sync::mpsc::channel();
472+
if let Some(fresh) = commands
473+
.try_send(StoreCommand::Metrics { reply })
474+
.ok()
475+
.and_then(|()| text.recv_timeout(Duration::from_secs(1)).ok())
476+
{
477+
store_text = fresh;
478+
}
479+
let mut body = store_text.clone();
472480
body.push_str(&pool.metrics.lock().unwrap());
473481
body
474482
})

‎selene-manager/src/pool/index.rs‎

Lines changed: 6 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@
1212
1313
use crate::Key;
1414
use flare_selene::shared::Tier;
15-
use std::collections::{HashMap, HashSet};
15+
use std::collections::{BTreeSet, HashMap};
1616

1717
pub type InstanceId = [u8; 16];
1818
pub type Bucket = [u8; 16];
@@ -42,7 +42,7 @@ pub type EntryRef = (SourceId, Key, Tier);
4242
#[derive(Default)]
4343
pub struct ScopeIndex {
4444
sources: HashMap<SourceId, HashMap<(Key, Tier), Entry>>,
45-
buckets: HashMap<Bucket, HashSet<EntryRef>>,
45+
buckets: HashMap<Bucket, BTreeSet<EntryRef>>,
4646
entries: usize,
4747
}
4848

@@ -119,14 +119,8 @@ impl ScopeIndex {
119119
}
120120

121121
/// The entries whose bucket is `bucket`, in a stable order.
122-
pub fn in_bucket(&self, bucket: &Bucket) -> Vec<EntryRef> {
123-
let mut refs: Vec<EntryRef> = self
124-
.buckets
125-
.get(bucket)
126-
.map(|set| set.iter().copied().collect())
127-
.unwrap_or_default();
128-
refs.sort_unstable();
129-
refs
122+
pub fn in_bucket(&self, bucket: &Bucket) -> impl Iterator<Item = EntryRef> + '_ {
123+
self.buckets.get(bucket).into_iter().flatten().copied()
130124
}
131125

132126
fn drop_bucket(&mut self, source: SourceId, entry: &Entry) {
@@ -257,7 +251,7 @@ mod tests {
257251
assert_eq!(index.get(SOURCE, entry.key, Tier::Dram).unwrap().bytes, 77);
258252
assert!(index.remove(SOURCE, entry.key, Tier::Dram));
259253
assert!(!index.remove(SOURCE, entry.key, Tier::Dram));
260-
assert!(index.in_bucket(&[9; 16]).is_empty());
254+
assert!(index.in_bucket(&[9; 16]).next().is_none());
261255
}
262256

263257
#[test]
@@ -279,7 +273,7 @@ mod tests {
279273
index.upsert(SOURCE, entry(256, 2, Some((128, 1)), 2, Tier::Dram));
280274
index.replace(SOURCE, [entry(64, 7, None, 1, Tier::Dram)]);
281275
assert_eq!(index.len(), 1);
282-
assert_eq!(index.in_bucket(&[9; 16]).len(), 1);
276+
assert_eq!(index.in_bucket(&[9; 16]).count(), 1);
283277
assert_eq!(index.clear(SOURCE), 1);
284278
assert_eq!(index.len(), 0);
285279
assert!(index.sources.is_empty());

0 commit comments

Comments
 (0)