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::GetSlruSegment(_) => {}
64 0 : PagestreamFeMessage::GetPage(req) => {
65 0 : total += 1;
66 :
67 0 : if let Some(prev) = prev {
68 0 : if prev.rel == req.rel {
69 0 : let delta = (req.blkno as i32) - (prev.blkno as i32);
70 0 : deltas.entry(delta).and_modify(|c| *c += 1).or_insert(1);
71 0 : } else {
72 0 : cross_rel += 1;
73 0 : }
74 0 : }
75 0 : prev = Some(req);
76 : }
77 0 : PagestreamFeMessage::DbSize(_) => {}
78 : };
79 : }
80 :
81 : // Print stats.
82 0 : let mut other = deltas.len();
83 0 : deltas.retain(|_, count| *count > 300);
84 0 : other -= deltas.len();
85 0 : dbg!(total);
86 0 : dbg!(cross_rel);
87 0 : dbg!(other);
88 0 : dbg!(deltas);
89 0 : }
90 :
91 0 : fn dump_trace<R: std::io::Read>(mut reader: R) {
92 0 : while let Ok(msg) = PagestreamFeMessage::parse(&mut reader) {
93 0 : println!("{msg:?}");
94 0 : }
95 0 : }
96 :
97 0 : #[derive(Debug)]
98 : struct TraceFile {
99 : #[allow(dead_code)]
100 : pub tenant_id: TenantId,
101 :
102 : #[allow(dead_code)]
103 : pub timeline_id: TimelineId,
104 :
105 : #[allow(dead_code)]
106 : pub connection_id: ConnectionId,
107 :
108 : pub path: PathBuf,
109 : }
110 :
111 0 : fn get_trace_files(traces_dir: &PathBuf) -> anyhow::Result<Vec<TraceFile>> {
112 0 : let mut trace_files = Vec::<TraceFile>::new();
113 :
114 : // Trace files are organized as {tenant_id}/{timeline_id}/{connection_id}
115 0 : for tenant_dir in read_dir(traces_dir)? {
116 0 : let entry = tenant_dir?;
117 0 : let path = entry.path();
118 0 : let tenant_id = TenantId::from_str(path.file_name().unwrap().to_str().unwrap())?;
119 :
120 0 : for timeline_dir in read_dir(path)? {
121 0 : let entry = timeline_dir?;
122 0 : let path = entry.path();
123 0 : let timeline_id = TimelineId::from_str(path.file_name().unwrap().to_str().unwrap())?;
124 :
125 0 : for trace_dir in read_dir(path)? {
126 0 : let entry = trace_dir?;
127 0 : let path = entry.path();
128 0 : let connection_id =
129 0 : ConnectionId::from_str(path.file_name().unwrap().to_str().unwrap())?;
130 :
131 0 : trace_files.push(TraceFile {
132 0 : tenant_id,
133 0 : timeline_id,
134 0 : connection_id,
135 0 : path,
136 0 : });
137 : }
138 : }
139 : }
140 :
141 0 : Ok(trace_files)
142 0 : }
143 :
144 0 : fn main() -> anyhow::Result<()> {
145 0 : let args = Args::parse();
146 0 :
147 0 : match args.command {
148 : Command::List => {
149 0 : for trace_file in get_trace_files(&args.path)? {
150 0 : println!("{trace_file:?}");
151 0 : }
152 : }
153 : Command::Dump => {
154 0 : for trace_file in get_trace_files(&args.path)? {
155 0 : let file = File::open(trace_file.path.clone())?;
156 0 : let reader = BufReader::new(file);
157 0 : dump_trace(reader);
158 : }
159 : }
160 : Command::Analyze => {
161 0 : for trace_file in get_trace_files(&args.path)? {
162 0 : println!("analyzing {trace_file:?}");
163 0 : let file = File::open(trace_file.path.clone())?;
164 0 : let reader = BufReader::new(file);
165 0 : analyze_trace(reader);
166 : }
167 : }
168 0 : Command::Draw => todo!(),
169 0 : Command::Replay => todo!(),
170 : }
171 :
172 0 : Ok(())
173 0 : }
|
