Line data Source code
1 : //! A tool for working with read traces generated by the pageserver.
2 : use std::collections::HashMap;
3 : use std::path::PathBuf;
4 : use std::str::FromStr;
5 : use std::{
6 : fs::{read_dir, File},
7 : io::BufReader,
8 : };
9 :
10 : use pageserver_api::models::{
11 : PagestreamFeMessage, PagestreamGetPageRequest, PagestreamProtocolVersion,
12 : };
13 : use utils::id::{ConnectionId, TenantId, TimelineId};
14 :
15 : use clap::{Parser, Subcommand};
16 :
17 : /// Utils for working with pageserver read traces. For generating
18 : /// traces, see the `trace_read_requests` tenant config option.
19 0 : #[derive(Parser, Debug)]
20 : #[command(author, version, about, long_about = None)]
21 : struct Args {
22 : /// Path of trace directory
23 : #[arg(short, long)]
24 0 : path: PathBuf,
25 :
26 : #[command(subcommand)]
27 : command: Command,
28 : }
29 :
30 : /// What to do with the read trace
31 0 : #[derive(Subcommand, Debug)]
32 : enum Command {
33 : /// List traces in the directory
34 : List,
35 :
36 : /// Print the traces in text format
37 : Dump,
38 :
39 : /// Print stats and anomalies about the traces
40 : Analyze,
41 : }
42 :
43 : // HACK This function will change and improve as we see what kind of analysis is useful.
44 : // Currently it collects the difference in blkno of consecutive GetPage requests,
45 : // and counts the frequency of each value. This information is useful in order to:
46 : // - see how sequential a workload is by seeing how often the delta is 1
47 : // - detect any prefetching anomalies by looking for negative deltas during seqscan
48 0 : fn analyze_trace<R: std::io::Read>(mut reader: R) {
49 0 : let mut total = 0; // Total requests traced
50 0 : let mut cross_rel = 0; // Requests that ask for different rel than previous request
51 0 : let mut deltas = HashMap::<i32, u32>::new(); // Consecutive blkno differences
52 0 : let mut prev: Option<PagestreamGetPageRequest> = None;
53 :
54 : // Compute stats
55 0 : while let Ok(msg) = PagestreamFeMessage::parse(&mut reader, PagestreamProtocolVersion::V2) {
56 0 : match msg {
57 0 : PagestreamFeMessage::Exists(_) => {}
58 0 : PagestreamFeMessage::Nblocks(_) => {}
59 0 : PagestreamFeMessage::GetSlruSegment(_) => {}
60 0 : PagestreamFeMessage::GetPage(req) => {
61 0 : total += 1;
62 :
63 0 : if let Some(prev) = prev {
64 0 : if prev.rel == req.rel {
65 0 : let delta = (req.blkno as i32) - (prev.blkno as i32);
66 0 : deltas.entry(delta).and_modify(|c| *c += 1).or_insert(1);
67 0 : } else {
68 0 : cross_rel += 1;
69 0 : }
70 0 : }
71 0 : prev = Some(req);
72 : }
73 0 : PagestreamFeMessage::DbSize(_) => {}
74 : };
75 : }
76 :
77 : // Print stats.
78 0 : let mut other = deltas.len();
79 0 : deltas.retain(|_, count| *count > 300);
80 0 : other -= deltas.len();
81 0 : dbg!(total);
82 0 : dbg!(cross_rel);
83 0 : dbg!(other);
84 0 : dbg!(deltas);
85 0 : }
86 :
87 0 : fn dump_trace<R: std::io::Read>(mut reader: R) {
88 0 : while let Ok(msg) = PagestreamFeMessage::parse(&mut reader, PagestreamProtocolVersion::V2) {
89 0 : println!("{msg:?}");
90 0 : }
91 0 : }
92 :
93 : #[derive(Debug)]
94 : struct TraceFile {
95 : #[allow(dead_code)]
96 : pub tenant_id: TenantId,
97 :
98 : #[allow(dead_code)]
99 : pub timeline_id: TimelineId,
100 :
101 : #[allow(dead_code)]
102 : pub connection_id: ConnectionId,
103 :
104 : pub path: PathBuf,
105 : }
106 :
107 0 : fn get_trace_files(traces_dir: &PathBuf) -> anyhow::Result<Vec<TraceFile>> {
108 0 : let mut trace_files = Vec::<TraceFile>::new();
109 :
110 : // Trace files are organized as {tenant_id}/{timeline_id}/{connection_id}
111 0 : for tenant_dir in read_dir(traces_dir)? {
112 0 : let entry = tenant_dir?;
113 0 : let path = entry.path();
114 0 : let tenant_id = TenantId::from_str(path.file_name().unwrap().to_str().unwrap())?;
115 :
116 0 : for timeline_dir in read_dir(path)? {
117 0 : let entry = timeline_dir?;
118 0 : let path = entry.path();
119 0 : let timeline_id = TimelineId::from_str(path.file_name().unwrap().to_str().unwrap())?;
120 :
121 0 : for trace_dir in read_dir(path)? {
122 0 : let entry = trace_dir?;
123 0 : let path = entry.path();
124 0 : let connection_id =
125 0 : ConnectionId::from_str(path.file_name().unwrap().to_str().unwrap())?;
126 :
127 0 : trace_files.push(TraceFile {
128 0 : tenant_id,
129 0 : timeline_id,
130 0 : connection_id,
131 0 : path,
132 0 : });
133 : }
134 : }
135 : }
136 :
137 0 : Ok(trace_files)
138 0 : }
139 :
140 0 : fn main() -> anyhow::Result<()> {
141 0 : let args = Args::parse();
142 0 :
143 0 : match args.command {
144 : Command::List => {
145 0 : for trace_file in get_trace_files(&args.path)? {
146 0 : println!("{trace_file:?}");
147 0 : }
148 : }
149 : Command::Dump => {
150 0 : for trace_file in get_trace_files(&args.path)? {
151 0 : let file = File::open(trace_file.path.clone())?;
152 0 : let reader = BufReader::new(file);
153 0 : dump_trace(reader);
154 : }
155 : }
156 : Command::Analyze => {
157 0 : for trace_file in get_trace_files(&args.path)? {
158 0 : println!("analyzing {trace_file:?}");
159 0 : let file = File::open(trace_file.path.clone())?;
160 0 : let reader = BufReader::new(file);
161 0 : analyze_trace(reader);
162 : }
163 : }
164 : }
165 :
166 0 : Ok(())
167 0 : }
|
