Line data Source code
1 : //! Utils for dumping full state of the safekeeper.
2 :
3 : use std::fs;
4 : use std::fs::DirEntry;
5 : use std::io::BufReader;
6 : use std::io::Read;
7 : use std::path::PathBuf;
8 : use std::sync::Arc;
9 :
10 : use anyhow::bail;
11 : use anyhow::Result;
12 : use camino::Utf8Path;
13 : use chrono::{DateTime, Utc};
14 : use postgres_ffi::XLogSegNo;
15 : use postgres_ffi::MAX_SEND_SIZE;
16 : use serde::Deserialize;
17 : use serde::Serialize;
18 :
19 : use sha2::{Digest, Sha256};
20 : use utils::id::NodeId;
21 : use utils::id::TenantTimelineId;
22 : use utils::id::{TenantId, TimelineId};
23 : use utils::lsn::Lsn;
24 :
25 : use crate::safekeeper::TermHistory;
26 : use crate::send_wal::WalSenderState;
27 : use crate::state::TimelineMemState;
28 : use crate::state::TimelinePersistentState;
29 : use crate::wal_storage::WalReader;
30 : use crate::GlobalTimelines;
31 : use crate::SafeKeeperConf;
32 :
33 : /// Various filters that influence the resulting JSON output.
34 0 : #[derive(Debug, Serialize, Deserialize, Clone)]
35 : pub struct Args {
36 : /// Dump all available safekeeper state. False by default.
37 : pub dump_all: bool,
38 :
39 : /// Dump control_file content. Uses value of `dump_all` by default.
40 : pub dump_control_file: bool,
41 :
42 : /// Dump in-memory state. Uses value of `dump_all` by default.
43 : pub dump_memory: bool,
44 :
45 : /// Dump all disk files in a timeline directory. Uses value of `dump_all` by default.
46 : pub dump_disk_content: bool,
47 :
48 : /// Dump full term history. True by default.
49 : pub dump_term_history: bool,
50 :
51 : /// Filter timelines by tenant_id.
52 : pub tenant_id: Option<TenantId>,
53 :
54 : /// Filter timelines by timeline_id.
55 : pub timeline_id: Option<TimelineId>,
56 : }
57 :
58 : /// Response for debug dump request.
59 : #[derive(Debug, Serialize)]
60 : pub struct Response {
61 : pub start_time: DateTime<Utc>,
62 : pub finish_time: DateTime<Utc>,
63 : pub timelines: Vec<TimelineDumpSer>,
64 : pub timelines_count: usize,
65 : pub config: Config,
66 : }
67 :
68 : pub struct TimelineDumpSer {
69 : pub tli: Arc<crate::timeline::Timeline>,
70 : pub args: Args,
71 : pub runtime: Arc<tokio::runtime::Runtime>,
72 : }
73 :
74 : impl std::fmt::Debug for TimelineDumpSer {
75 0 : fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
76 0 : f.debug_struct("TimelineDumpSer")
77 0 : .field("tli", &self.tli.ttid)
78 0 : .field("args", &self.args)
79 0 : .finish()
80 0 : }
81 : }
82 :
83 : impl Serialize for TimelineDumpSer {
84 0 : fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
85 0 : where
86 0 : S: serde::Serializer,
87 0 : {
88 0 : let dump = self
89 0 : .runtime
90 0 : .block_on(build_from_tli_dump(self.tli.clone(), self.args.clone()));
91 0 : dump.serialize(serializer)
92 0 : }
93 : }
94 :
95 0 : async fn build_from_tli_dump(timeline: Arc<crate::timeline::Timeline>, args: Args) -> Timeline {
96 0 : let control_file = if args.dump_control_file {
97 0 : let mut state = timeline.get_state().await.1;
98 0 : if !args.dump_term_history {
99 0 : state.acceptor_state.term_history = TermHistory(vec![]);
100 0 : }
101 0 : Some(state)
102 : } else {
103 0 : None
104 : };
105 :
106 0 : let memory = if args.dump_memory {
107 0 : Some(timeline.memory_dump().await)
108 : } else {
109 0 : None
110 : };
111 :
112 0 : let disk_content = if args.dump_disk_content {
113 : // build_disk_content can fail, but we don't want to fail the whole
114 : // request because of that.
115 0 : build_disk_content(&timeline.timeline_dir).ok()
116 : } else {
117 0 : None
118 : };
119 :
120 0 : Timeline {
121 0 : tenant_id: timeline.ttid.tenant_id,
122 0 : timeline_id: timeline.ttid.timeline_id,
123 0 : control_file,
124 0 : memory,
125 0 : disk_content,
126 0 : }
127 0 : }
128 :
129 : /// Safekeeper configuration.
130 0 : #[derive(Debug, Serialize, Deserialize)]
131 : pub struct Config {
132 : pub id: NodeId,
133 : pub workdir: PathBuf,
134 : pub listen_pg_addr: String,
135 : pub listen_http_addr: String,
136 : pub no_sync: bool,
137 : pub max_offloader_lag_bytes: u64,
138 : pub wal_backup_enabled: bool,
139 : }
140 :
141 0 : #[derive(Debug, Serialize, Deserialize)]
142 : pub struct Timeline {
143 : pub tenant_id: TenantId,
144 : pub timeline_id: TimelineId,
145 : pub control_file: Option<TimelinePersistentState>,
146 : pub memory: Option<Memory>,
147 : pub disk_content: Option<DiskContent>,
148 : }
149 :
150 0 : #[derive(Debug, Serialize, Deserialize)]
151 : pub struct Memory {
152 : pub is_cancelled: bool,
153 : pub peers_info_len: usize,
154 : pub walsenders: Vec<WalSenderState>,
155 : pub wal_backup_active: bool,
156 : pub active: bool,
157 : pub num_computes: u32,
158 : pub last_removed_segno: XLogSegNo,
159 : pub epoch_start_lsn: Lsn,
160 : pub mem_state: TimelineMemState,
161 :
162 : // PhysicalStorage state.
163 : pub write_lsn: Lsn,
164 : pub write_record_lsn: Lsn,
165 : pub flush_lsn: Lsn,
166 : pub file_open: bool,
167 : }
168 :
169 0 : #[derive(Debug, Serialize, Deserialize)]
170 : pub struct DiskContent {
171 : pub files: Vec<FileInfo>,
172 : }
173 :
174 0 : #[derive(Debug, Serialize, Deserialize)]
175 : pub struct FileInfo {
176 : pub name: String,
177 : pub size: u64,
178 : pub created: DateTime<Utc>,
179 : pub modified: DateTime<Utc>,
180 : pub start_zeroes: u64,
181 : pub end_zeroes: u64,
182 : // TODO: add sha256 checksum
183 : }
184 :
185 : /// Build debug dump response, using the provided [`Args`] filters.
186 0 : pub async fn build(args: Args) -> Result<Response> {
187 0 : let start_time = Utc::now();
188 0 : let timelines_count = GlobalTimelines::timelines_count();
189 :
190 0 : let ptrs_snapshot = if args.tenant_id.is_some() && args.timeline_id.is_some() {
191 : // If both tenant_id and timeline_id are specified, we can just get the
192 : // timeline directly, without taking a snapshot of the whole list.
193 0 : let ttid = TenantTimelineId::new(args.tenant_id.unwrap(), args.timeline_id.unwrap());
194 0 : if let Ok(tli) = GlobalTimelines::get(ttid) {
195 0 : vec![tli]
196 : } else {
197 0 : vec![]
198 : }
199 : } else {
200 : // Otherwise, take a snapshot of the whole list.
201 0 : GlobalTimelines::get_all()
202 : };
203 :
204 0 : let mut timelines = Vec::new();
205 0 : let runtime = Arc::new(
206 0 : tokio::runtime::Builder::new_current_thread()
207 0 : .build()
208 0 : .unwrap(),
209 0 : );
210 0 : for tli in ptrs_snapshot {
211 0 : let ttid = tli.ttid;
212 0 : if let Some(tenant_id) = args.tenant_id {
213 0 : if tenant_id != ttid.tenant_id {
214 0 : continue;
215 0 : }
216 0 : }
217 0 : if let Some(timeline_id) = args.timeline_id {
218 0 : if timeline_id != ttid.timeline_id {
219 0 : continue;
220 0 : }
221 0 : }
222 :
223 0 : timelines.push(TimelineDumpSer {
224 0 : tli,
225 0 : args: args.clone(),
226 0 : runtime: runtime.clone(),
227 0 : });
228 : }
229 :
230 0 : let config = GlobalTimelines::get_global_config();
231 0 :
232 0 : Ok(Response {
233 0 : start_time,
234 0 : finish_time: Utc::now(),
235 0 : timelines,
236 0 : timelines_count,
237 0 : config: build_config(config),
238 0 : })
239 0 : }
240 :
241 : /// Builds DiskContent from a directory path. It can fail if the directory
242 : /// is deleted between the time we get the path and the time we try to open it.
243 0 : fn build_disk_content(path: &Utf8Path) -> Result<DiskContent> {
244 0 : let mut files = Vec::new();
245 0 : for entry in fs::read_dir(path)? {
246 0 : if entry.is_err() {
247 0 : continue;
248 0 : }
249 0 : let file = build_file_info(entry?);
250 0 : if file.is_err() {
251 0 : continue;
252 0 : }
253 0 : files.push(file?);
254 : }
255 :
256 0 : Ok(DiskContent { files })
257 0 : }
258 :
259 : /// Builds FileInfo from DirEntry. Sometimes it can return an error
260 : /// if the file is deleted between the time we get the DirEntry
261 : /// and the time we try to open it.
262 0 : fn build_file_info(entry: DirEntry) -> Result<FileInfo> {
263 0 : let metadata = entry.metadata()?;
264 0 : let path = entry.path();
265 0 : let name = path
266 0 : .file_name()
267 0 : .and_then(|x| x.to_str())
268 0 : .unwrap_or("")
269 0 : .to_owned();
270 0 : let mut file = fs::File::open(path)?;
271 0 : let mut reader = BufReader::new(&mut file).bytes().filter_map(|x| x.ok());
272 0 :
273 0 : let start_zeroes = reader.by_ref().take_while(|&x| x == 0).count() as u64;
274 0 : let mut end_zeroes = 0;
275 0 : for b in reader {
276 0 : if b == 0 {
277 0 : end_zeroes += 1;
278 0 : } else {
279 0 : end_zeroes = 0;
280 0 : }
281 : }
282 :
283 : Ok(FileInfo {
284 0 : name,
285 0 : size: metadata.len(),
286 0 : created: DateTime::from(metadata.created()?),
287 0 : modified: DateTime::from(metadata.modified()?),
288 0 : start_zeroes,
289 0 : end_zeroes,
290 : })
291 0 : }
292 :
293 : /// Converts SafeKeeperConf to Config, filtering out the fields that are not
294 : /// supposed to be exposed.
295 0 : fn build_config(config: SafeKeeperConf) -> Config {
296 0 : Config {
297 0 : id: config.my_id,
298 0 : workdir: config.workdir.into(),
299 0 : listen_pg_addr: config.listen_pg_addr,
300 0 : listen_http_addr: config.listen_http_addr,
301 0 : no_sync: config.no_sync,
302 0 : max_offloader_lag_bytes: config.max_offloader_lag_bytes,
303 0 : wal_backup_enabled: config.wal_backup_enabled,
304 0 : }
305 0 : }
306 :
307 0 : #[derive(Debug, Clone, Deserialize, Serialize)]
308 : pub struct TimelineDigestRequest {
309 : pub from_lsn: Lsn,
310 : pub until_lsn: Lsn,
311 : }
312 :
313 0 : #[derive(Debug, Serialize, Deserialize)]
314 : pub struct TimelineDigest {
315 : pub sha256: String,
316 : }
317 :
318 0 : pub async fn calculate_digest(
319 0 : tli: &Arc<crate::timeline::Timeline>,
320 0 : request: TimelineDigestRequest,
321 0 : ) -> Result<TimelineDigest> {
322 0 : if request.from_lsn > request.until_lsn {
323 0 : bail!("from_lsn is greater than until_lsn");
324 0 : }
325 0 :
326 0 : let conf = GlobalTimelines::get_global_config();
327 0 : let (_, persisted_state) = tli.get_state().await;
328 :
329 0 : if persisted_state.timeline_start_lsn > request.from_lsn {
330 0 : bail!("requested LSN is before the start of the timeline");
331 0 : }
332 :
333 0 : let mut wal_reader = WalReader::new(
334 0 : conf.workdir.clone(),
335 0 : tli.timeline_dir.clone(),
336 0 : &persisted_state,
337 0 : request.from_lsn,
338 0 : true,
339 0 : )?;
340 :
341 0 : let mut hasher = Sha256::new();
342 0 : let mut buf = [0u8; MAX_SEND_SIZE];
343 0 :
344 0 : let mut bytes_left = (request.until_lsn.0 - request.from_lsn.0) as usize;
345 0 : while bytes_left > 0 {
346 0 : let bytes_to_read = std::cmp::min(buf.len(), bytes_left);
347 0 : let bytes_read = wal_reader.read(&mut buf[..bytes_to_read]).await?;
348 0 : if bytes_read == 0 {
349 0 : bail!("wal_reader.read returned 0 bytes");
350 0 : }
351 0 : hasher.update(&buf[..bytes_read]);
352 0 : bytes_left -= bytes_read;
353 : }
354 :
355 0 : let digest = hasher.finalize();
356 0 : let digest = hex::encode(digest);
357 0 : Ok(TimelineDigest { sha256: digest })
358 0 : }
|
