Line data Source code
1 : use crate::tenant::mgr::TenantManager;
2 : use crate::{context::RequestContext, tenant::timeline::logical_size::CurrentLogicalSize};
3 : use chrono::{DateTime, Utc};
4 : use consumption_metrics::EventType;
5 : use futures::stream::StreamExt;
6 : use std::{sync::Arc, time::SystemTime};
7 : use utils::{
8 : id::{TenantId, TimelineId},
9 : lsn::Lsn,
10 : };
11 :
12 : use super::{Cache, RawMetric};
13 :
14 : /// Name of the metric, used by `MetricsKey` factory methods and `deserialize_cached_events`
15 : /// instead of static str.
16 : // Do not rename any of these without first consulting with data team and partner
17 : // management.
18 432 : #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
19 : pub(super) enum Name {
20 : /// Timeline last_record_lsn, absolute
21 : #[serde(rename = "written_size")]
22 : WrittenSize,
23 : /// Timeline last_record_lsn, incremental
24 : #[serde(rename = "written_data_bytes_delta")]
25 : WrittenSizeDelta,
26 : /// Timeline logical size
27 : #[serde(rename = "timeline_logical_size")]
28 : LogicalSize,
29 : /// Tenant remote size
30 : #[serde(rename = "remote_storage_size")]
31 : RemoteSize,
32 : /// Tenant resident size
33 : #[serde(rename = "resident_size")]
34 : ResidentSize,
35 : /// Tenant synthetic size
36 : #[serde(rename = "synthetic_storage_size")]
37 : SyntheticSize,
38 : }
39 :
40 : /// Key that uniquely identifies the object this metric describes.
41 : ///
42 : /// This is a denormalization done at the MetricsKey const methods; these should not be constructed
43 : /// elsewhere.
44 0 : #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
45 : pub(crate) struct MetricsKey {
46 : pub(super) tenant_id: TenantId,
47 :
48 : #[serde(skip_serializing_if = "Option::is_none")]
49 : pub(super) timeline_id: Option<TimelineId>,
50 :
51 : pub(super) metric: Name,
52 : }
53 :
54 : impl MetricsKey {
55 282 : const fn absolute_values(self) -> AbsoluteValueFactory {
56 282 : AbsoluteValueFactory(self)
57 282 : }
58 90 : const fn incremental_values(self) -> IncrementalValueFactory {
59 90 : IncrementalValueFactory(self)
60 90 : }
61 : }
62 :
63 : /// Helper type which each individual metric kind can return to produce only absolute values.
64 : struct AbsoluteValueFactory(MetricsKey);
65 :
66 : impl AbsoluteValueFactory {
67 264 : const fn at(self, time: DateTime<Utc>, val: u64) -> RawMetric {
68 264 : let key = self.0;
69 264 : (key, (EventType::Absolute { time }, val))
70 264 : }
71 :
72 30 : fn key(&self) -> &MetricsKey {
73 30 : &self.0
74 30 : }
75 : }
76 :
77 : /// Helper type which each individual metric kind can return to produce only incremental values.
78 : struct IncrementalValueFactory(MetricsKey);
79 :
80 : impl IncrementalValueFactory {
81 : #[allow(clippy::wrong_self_convention)]
82 90 : const fn from_until(
83 90 : self,
84 90 : prev_end: DateTime<Utc>,
85 90 : up_to: DateTime<Utc>,
86 90 : val: u64,
87 90 : ) -> RawMetric {
88 90 : let key = self.0;
89 90 : // cannot assert prev_end < up_to because these are realtime clock based
90 90 : let when = EventType::Incremental {
91 90 : start_time: prev_end,
92 90 : stop_time: up_to,
93 90 : };
94 90 : (key, (when, val))
95 90 : }
96 :
97 36 : fn key(&self) -> &MetricsKey {
98 36 : &self.0
99 36 : }
100 : }
101 :
102 : // the static part of a MetricsKey
103 : impl MetricsKey {
104 : /// Absolute value of [`Timeline::get_last_record_lsn`].
105 : ///
106 : /// [`Timeline::get_last_record_lsn`]: crate::tenant::Timeline::get_last_record_lsn
107 96 : const fn written_size(tenant_id: TenantId, timeline_id: TimelineId) -> AbsoluteValueFactory {
108 96 : MetricsKey {
109 96 : tenant_id,
110 96 : timeline_id: Some(timeline_id),
111 96 : metric: Name::WrittenSize,
112 96 : }
113 96 : .absolute_values()
114 96 : }
115 :
116 : /// Values will be the difference of the latest [`MetricsKey::written_size`] to what we
117 : /// previously sent, starting from the previously sent incremental time range ending at the
118 : /// latest absolute measurement.
119 90 : const fn written_size_delta(
120 90 : tenant_id: TenantId,
121 90 : timeline_id: TimelineId,
122 90 : ) -> IncrementalValueFactory {
123 90 : MetricsKey {
124 90 : tenant_id,
125 90 : timeline_id: Some(timeline_id),
126 90 : metric: Name::WrittenSizeDelta,
127 90 : }
128 90 : .incremental_values()
129 90 : }
130 :
131 : /// Exact [`Timeline::get_current_logical_size`].
132 : ///
133 : /// [`Timeline::get_current_logical_size`]: crate::tenant::Timeline::get_current_logical_size
134 78 : const fn timeline_logical_size(
135 78 : tenant_id: TenantId,
136 78 : timeline_id: TimelineId,
137 78 : ) -> AbsoluteValueFactory {
138 78 : MetricsKey {
139 78 : tenant_id,
140 78 : timeline_id: Some(timeline_id),
141 78 : metric: Name::LogicalSize,
142 78 : }
143 78 : .absolute_values()
144 78 : }
145 :
146 : /// [`Tenant::remote_size`]
147 : ///
148 : /// [`Tenant::remote_size`]: crate::tenant::Tenant::remote_size
149 36 : const fn remote_storage_size(tenant_id: TenantId) -> AbsoluteValueFactory {
150 36 : MetricsKey {
151 36 : tenant_id,
152 36 : timeline_id: None,
153 36 : metric: Name::RemoteSize,
154 36 : }
155 36 : .absolute_values()
156 36 : }
157 :
158 : /// Sum of [`Timeline::resident_physical_size`] for each `Tenant`.
159 : ///
160 : /// [`Timeline::resident_physical_size`]: crate::tenant::Timeline::resident_physical_size
161 36 : const fn resident_size(tenant_id: TenantId) -> AbsoluteValueFactory {
162 36 : MetricsKey {
163 36 : tenant_id,
164 36 : timeline_id: None,
165 36 : metric: Name::ResidentSize,
166 36 : }
167 36 : .absolute_values()
168 36 : }
169 :
170 : /// [`Tenant::cached_synthetic_size`] as refreshed by [`calculate_synthetic_size_worker`].
171 : ///
172 : /// [`Tenant::cached_synthetic_size`]: crate::tenant::Tenant::cached_synthetic_size
173 : /// [`calculate_synthetic_size_worker`]: super::calculate_synthetic_size_worker
174 36 : const fn synthetic_size(tenant_id: TenantId) -> AbsoluteValueFactory {
175 36 : MetricsKey {
176 36 : tenant_id,
177 36 : timeline_id: None,
178 36 : metric: Name::SyntheticSize,
179 36 : }
180 36 : .absolute_values()
181 36 : }
182 : }
183 :
184 0 : pub(super) async fn collect_all_metrics(
185 0 : tenant_manager: &Arc<TenantManager>,
186 0 : cached_metrics: &Cache,
187 0 : ctx: &RequestContext,
188 0 : ) -> Vec<RawMetric> {
189 : use pageserver_api::models::TenantState;
190 :
191 0 : let started_at = std::time::Instant::now();
192 :
193 0 : let tenants = match tenant_manager.list_tenants() {
194 0 : Ok(tenants) => tenants,
195 0 : Err(err) => {
196 0 : tracing::error!("failed to list tenants: {:?}", err);
197 0 : return vec![];
198 : }
199 : };
200 :
201 0 : let tenants = futures::stream::iter(tenants).filter_map(|(id, state, _)| async move {
202 0 : if state != TenantState::Active || !id.is_shard_zero() {
203 0 : None
204 : } else {
205 0 : tenant_manager
206 0 : .get_attached_tenant_shard(id)
207 0 : .ok()
208 0 : .map(|tenant| (id.tenant_id, tenant))
209 : }
210 0 : });
211 :
212 0 : let res = collect(tenants, cached_metrics, ctx).await;
213 :
214 0 : tracing::info!(
215 0 : elapsed_ms = started_at.elapsed().as_millis(),
216 0 : total = res.len(),
217 0 : "collected metrics"
218 : );
219 :
220 0 : res
221 0 : }
222 :
223 0 : async fn collect<S>(tenants: S, cache: &Cache, ctx: &RequestContext) -> Vec<RawMetric>
224 0 : where
225 0 : S: futures::stream::Stream<Item = (TenantId, Arc<crate::tenant::Tenant>)>,
226 0 : {
227 0 : let mut current_metrics: Vec<RawMetric> = Vec::new();
228 0 :
229 0 : let mut tenants = std::pin::pin!(tenants);
230 :
231 0 : while let Some((tenant_id, tenant)) = tenants.next().await {
232 0 : let mut tenant_resident_size = 0;
233 :
234 0 : for timeline in tenant.list_timelines() {
235 0 : let timeline_id = timeline.timeline_id;
236 0 :
237 0 : match TimelineSnapshot::collect(&timeline, ctx) {
238 0 : Ok(Some(snap)) => {
239 0 : snap.to_metrics(
240 0 : tenant_id,
241 0 : timeline_id,
242 0 : Utc::now(),
243 0 : &mut current_metrics,
244 0 : cache,
245 0 : );
246 0 : }
247 0 : Ok(None) => {}
248 0 : Err(e) => {
249 0 : tracing::error!(
250 0 : "failed to get metrics values for tenant {tenant_id} timeline {}: {e:#?}",
251 0 : timeline.timeline_id
252 : );
253 0 : continue;
254 : }
255 : }
256 :
257 0 : tenant_resident_size += timeline.resident_physical_size();
258 : }
259 :
260 0 : let snap = TenantSnapshot::collect(&tenant, tenant_resident_size);
261 0 : snap.to_metrics(tenant_id, Utc::now(), cache, &mut current_metrics);
262 : }
263 :
264 0 : current_metrics
265 0 : }
266 :
267 : /// In-between abstraction to allow testing metrics without actual Tenants.
268 : struct TenantSnapshot {
269 : resident_size: u64,
270 : remote_size: u64,
271 : synthetic_size: u64,
272 : }
273 :
274 : impl TenantSnapshot {
275 : /// Collect tenant status to have metrics created out of it.
276 : ///
277 : /// `resident_size` is calculated of the timelines we had access to for other metrics, so we
278 : /// cannot just list timelines here.
279 0 : fn collect(t: &Arc<crate::tenant::Tenant>, resident_size: u64) -> Self {
280 0 : TenantSnapshot {
281 0 : resident_size,
282 0 : remote_size: t.remote_size(),
283 0 : // Note that this metric is calculated in a separate bgworker
284 0 : // Here we only use cached value, which may lag behind the real latest one
285 0 : synthetic_size: t.cached_synthetic_size(),
286 0 : }
287 0 : }
288 :
289 12 : fn to_metrics(
290 12 : &self,
291 12 : tenant_id: TenantId,
292 12 : now: DateTime<Utc>,
293 12 : cached: &Cache,
294 12 : metrics: &mut Vec<RawMetric>,
295 12 : ) {
296 12 : let remote_size = MetricsKey::remote_storage_size(tenant_id).at(now, self.remote_size);
297 12 :
298 12 : let resident_size = MetricsKey::resident_size(tenant_id).at(now, self.resident_size);
299 :
300 12 : let synthetic_size = {
301 12 : let factory = MetricsKey::synthetic_size(tenant_id);
302 12 : let mut synthetic_size = self.synthetic_size;
303 12 :
304 12 : if synthetic_size == 0 {
305 12 : if let Some((_, value)) = cached.get(factory.key()) {
306 6 : // use the latest value from previous session
307 6 : synthetic_size = *value;
308 6 : }
309 0 : }
310 :
311 12 : if synthetic_size != 0 {
312 : // only send non-zeroes because otherwise these show up as errors in logs
313 6 : Some(factory.at(now, synthetic_size))
314 : } else {
315 6 : None
316 : }
317 : };
318 :
319 12 : metrics.extend(
320 12 : [Some(remote_size), Some(resident_size), synthetic_size]
321 12 : .into_iter()
322 12 : .flatten(),
323 12 : );
324 12 : }
325 : }
326 :
327 : /// Internal type to make timeline metric production testable.
328 : ///
329 : /// As this value type contains all of the information needed from a timeline to produce the
330 : /// metrics, it can easily be created with different values in test.
331 : struct TimelineSnapshot {
332 : loaded_at: (Lsn, SystemTime),
333 : last_record_lsn: Lsn,
334 : current_exact_logical_size: Option<u64>,
335 : }
336 :
337 : impl TimelineSnapshot {
338 : /// Collect the metrics from an actual timeline.
339 : ///
340 : /// Fails currently only when [`Timeline::get_current_logical_size`] fails.
341 : ///
342 : /// [`Timeline::get_current_logical_size`]: crate::tenant::Timeline::get_current_logical_size
343 0 : fn collect(
344 0 : t: &Arc<crate::tenant::Timeline>,
345 0 : ctx: &RequestContext,
346 0 : ) -> anyhow::Result<Option<Self>> {
347 0 : if !t.is_active() {
348 : // no collection for broken or stopping needed, we will still keep the cached values
349 : // though at the caller.
350 0 : Ok(None)
351 : } else {
352 0 : let loaded_at = t.loaded_at;
353 0 : let last_record_lsn = t.get_last_record_lsn();
354 :
355 0 : let current_exact_logical_size = {
356 0 : let span = tracing::info_span!("collect_metrics_iteration", tenant_id = %t.tenant_shard_id.tenant_id, timeline_id = %t.timeline_id);
357 0 : let size = span.in_scope(|| {
358 0 : t.get_current_logical_size(
359 0 : crate::tenant::timeline::GetLogicalSizePriority::Background,
360 0 : ctx,
361 0 : )
362 0 : });
363 0 : match size {
364 : // Only send timeline logical size when it is fully calculated.
365 0 : CurrentLogicalSize::Exact(ref size) => Some(size.into()),
366 0 : CurrentLogicalSize::Approximate(_) => None,
367 : }
368 : };
369 :
370 0 : Ok(Some(TimelineSnapshot {
371 0 : loaded_at,
372 0 : last_record_lsn,
373 0 : current_exact_logical_size,
374 0 : }))
375 : }
376 0 : }
377 :
378 : /// Produce the timeline consumption metrics into the `metrics` argument.
379 36 : fn to_metrics(
380 36 : &self,
381 36 : tenant_id: TenantId,
382 36 : timeline_id: TimelineId,
383 36 : now: DateTime<Utc>,
384 36 : metrics: &mut Vec<RawMetric>,
385 36 : cache: &Cache,
386 36 : ) {
387 36 : let timeline_written_size = u64::from(self.last_record_lsn);
388 36 :
389 36 : let written_size_delta_key = MetricsKey::written_size_delta(tenant_id, timeline_id);
390 36 :
391 36 : let last_stop_time = cache
392 36 : .get(written_size_delta_key.key())
393 36 : .map(|(until, _val)| {
394 18 : until
395 18 : .incremental_timerange()
396 18 : .expect("never create EventType::Absolute for written_size_delta")
397 18 : .end
398 36 : });
399 36 :
400 36 : let (key, written_size_now) =
401 36 : MetricsKey::written_size(tenant_id, timeline_id).at(now, timeline_written_size);
402 36 :
403 36 : // by default, use the last sent written_size as the basis for
404 36 : // calculating the delta. if we don't yet have one, use the load time value.
405 36 : let prev = cache
406 36 : .get(&key)
407 36 : .map(|(prev_at, prev)| {
408 24 : // use the prev time from our last incremental update, or default to latest
409 24 : // absolute update on the first round.
410 24 : let prev_at = prev_at
411 24 : .absolute_time()
412 24 : .expect("never create EventType::Incremental for written_size");
413 24 : let prev_at = last_stop_time.unwrap_or(prev_at);
414 24 : (*prev_at, *prev)
415 36 : })
416 36 : .unwrap_or_else(|| {
417 12 : // if we don't have a previous point of comparison, compare to the load time
418 12 : // lsn.
419 12 : let (disk_consistent_lsn, loaded_at) = &self.loaded_at;
420 12 : (DateTime::from(*loaded_at), disk_consistent_lsn.0)
421 36 : });
422 36 :
423 36 : let up_to = now;
424 :
425 36 : if let Some(delta) = written_size_now.1.checked_sub(prev.1) {
426 24 : let key_value = written_size_delta_key.from_until(prev.0, up_to, delta);
427 24 : // written_size_delta
428 24 : metrics.push(key_value);
429 24 : // written_size
430 24 : metrics.push((key, written_size_now));
431 24 : } else {
432 12 : // the cached value was ahead of us, report zero until we've caught up
433 12 : metrics.push(written_size_delta_key.from_until(prev.0, up_to, 0));
434 12 : // the cached value was ahead of us, report the same until we've caught up
435 12 : metrics.push((key, (written_size_now.0, prev.1)));
436 12 : }
437 :
438 : {
439 36 : let factory = MetricsKey::timeline_logical_size(tenant_id, timeline_id);
440 36 : let current_or_previous = self
441 36 : .current_exact_logical_size
442 36 : .or_else(|| cache.get(factory.key()).map(|(_, val)| *val));
443 :
444 36 : if let Some(size) = current_or_previous {
445 24 : metrics.push(factory.at(now, size));
446 24 : }
447 : }
448 36 : }
449 : }
450 :
451 : #[cfg(test)]
452 : mod tests;
453 :
454 : #[cfg(test)]
455 : pub(crate) use tests::metric_examples;
|
