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