mz_compute/logging/
resource_usage.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10//! Logging dataflow for the resource usage of the replica's processes.
11//!
12//! The observations are read by `mz_metrics::usage`, on a task that keeps sampling while the
13//! timely workers are busy. This dataflow only reports them, so a worker that stalls delays the
14//! report without losing an observation: the values it reads are either kernel-maintained
15//! high-water marks or the sampler's own folded peaks, neither of which a late read can miss.
16//!
17//! One row per `(process_id, source, metric)`, so a metric that moves every sample does not drag
18//! the stable ones through a retraction with it.
19
20use std::collections::BTreeMap;
21use std::rc::Rc;
22use std::time::{Duration, Instant};
23
24use mz_metrics::usage::{MetricKey, observations};
25use mz_ore::cast::CastFrom;
26use mz_ore::collections::CollectionExt;
27use mz_repr::{Datum, Timestamp};
28use mz_row_spine::RowRowBuilder;
29use mz_timely_util::columnar::builder::ColumnBuilder;
30use mz_timely_util::columnar::{Col2ValBatcher, batcher, columnar_exchange};
31use timely::dataflow::Scope;
32use timely::dataflow::channels::pact::ExchangeCore;
33use timely::dataflow::operators::generic::OutputBuilder;
34use timely::dataflow::operators::generic::builder_rc::OperatorBuilder;
35
36use crate::extensions::arrange::MzArrangeCore;
37use crate::logging::{
38    ComputeLog, LogCollection, LogVariant, PermutedRowPacker, downgrade_to_interval_boundary,
39    emit_snapshot_diff,
40};
41use crate::typedefs::RowRowSpine;
42
43/// The return type of [`construct`].
44pub(super) struct Return {
45    /// Collections to export.
46    pub collections: BTreeMap<LogVariant, LogCollection>,
47}
48
49/// Constructs the logging dataflow fragment for process resource usage.
50pub(super) fn construct(
51    scope: Scope<'_, Timestamp>,
52    config: &mz_compute_client::logging::LoggingConfig,
53    now: Instant,
54    start_offset: Duration,
55    workers_per_process: usize,
56) -> Return {
57    let variant = LogVariant::Compute(ComputeLog::ResourceUsage);
58    let mut collections = BTreeMap::new();
59    let interval_ms = std::cmp::max(1, config.interval.as_millis());
60
61    if !config.index_logs.contains_key(&variant) {
62        return Return { collections };
63    }
64
65    let process_id = scope.index() / workers_per_process;
66    let enable = scope.index() % workers_per_process == 0;
67
68    let mut builder = OperatorBuilder::new("ResourceUsage".to_string(), scope.clone());
69    let (output, stream) = builder.new_output();
70    let mut output = OutputBuilder::<_, ColumnBuilder<_>>::from(output);
71
72    let operator_info = builder.operator_info();
73    builder.build(move |capabilities| {
74        // Usage is per-process, so only one worker per process reports it. Drop the capability for
75        // disabled workers so the frontier can advance without this operator holding it back.
76        let mut cap = enable.then_some(capabilities.into_element());
77        let activator = scope.activator_for(operator_info.address);
78
79        let mut prev: BTreeMap<MetricKey, u64> = BTreeMap::new();
80        let mut packer = PermutedRowPacker::new(ComputeLog::ResourceUsage);
81
82        move |_frontiers| {
83            let Some(cap) = &mut cap else { return };
84
85            // The capability is downgraded on this operator's own timer rather than on the
86            // sampler's, so a sampler that stops ticking cannot freeze this collection's frontier.
87            let ts =
88                downgrade_to_interval_boundary(cap, &activator, now, start_offset, interval_ms);
89
90            let current = observations().unwrap_or_default();
91            if prev == current {
92                return;
93            }
94
95            let mut output = output.activate();
96            let mut session = output.session_with_builder(&cap);
97            emit_snapshot_diff(
98                &mut session,
99                &mut packer,
100                &prev,
101                &current,
102                ts,
103                |packer, key, value| pack_row(packer, process_id, *key, *value),
104            );
105
106            prev = current;
107        }
108    });
109
110    let exchange = ExchangeCore::<ColumnBuilder<_>, _>::new_core(
111        columnar_exchange::<mz_repr::Row, mz_repr::Row, Timestamp, mz_repr::Diff>,
112    );
113    let trace = stream
114        .mz_arrange_core::<
115            _,
116            batcher::Chunker<_>,
117            Col2ValBatcher<_, _, _, _>,
118            RowRowBuilder<_, _>,
119            RowRowSpine<_, _>,
120        >(exchange, "Arrange ResourceUsage")
121        .trace;
122    let token: Rc<dyn std::any::Any> = Rc::new(());
123    let collection = LogCollection { trace, token };
124    collections.insert(variant, collection);
125
126    Return { collections }
127}
128
129/// Pack one observation into key/value row pairs.
130fn pack_row(
131    packer: &mut PermutedRowPacker,
132    process_id: usize,
133    (source, metric): MetricKey,
134    value: u64,
135) -> (&mz_repr::RowRef, &mz_repr::RowRef) {
136    packer.pack_by_index(|row_packer, index| match index {
137        0 => row_packer.push(Datum::UInt64(u64::cast_from(process_id))),
138        1 => row_packer.push(Datum::String(source)),
139        2 => row_packer.push(Datum::String(metric)),
140        3 => row_packer.push(Datum::UInt64(value)),
141        _ => unreachable!("unexpected column index {index}"),
142    })
143}
mz_compute/logging/resource_usage.rs

mz_compute/logging/
resource_usage.rs
