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