mz_compute_client/
logging.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//! Compute layer logging configuration.
11
12use std::collections::BTreeMap;
13use std::time::Duration;
14
15use mz_repr::{GlobalId, RelationDesc, SqlScalarType};
16use serde::{Deserialize, Serialize};
17
18/// Logging configuration.
19///
20/// Setting `enable_logging` to `false` specifies that logging is disabled.
21//
22// Ideally we'd want to instead signal disabled logging by leaving `index_logs`
23// empty. Unfortunately, we have to always provide `index_logs`, because we must
24// install the logging dataflows even on replicas that have logging disabled. See database-issues#4545.
25#[derive(Debug, Default, Clone, PartialEq, Serialize, Deserialize)]
26pub struct LoggingConfig {
27    /// The logging interval
28    pub interval: Duration,
29    /// Whether logging is enabled
30    pub enable_logging: bool,
31    /// Whether we should report logs for the log-processing dataflows
32    pub log_logging: bool,
33    /// Logs to keep in an arrangement
34    pub index_logs: BTreeMap<LogVariant, GlobalId>,
35}
36
37/// TODO(database-issues#7533): Add documentation.
38#[derive(
39    Hash,
40    Eq,
41    PartialEq,
42    Ord,
43    PartialOrd,
44    Debug,
45    Clone,
46    Copy,
47    Serialize,
48    Deserialize
49)]
50pub enum LogVariant {
51    /// TODO(database-issues#7533): Add documentation.
52    Timely(TimelyLog),
53    /// TODO(database-issues#7533): Add documentation.
54    Differential(DifferentialLog),
55    /// TODO(database-issues#7533): Add documentation.
56    Compute(ComputeLog),
57}
58
59impl From<TimelyLog> for LogVariant {
60    fn from(value: TimelyLog) -> Self {
61        Self::Timely(value)
62    }
63}
64
65impl From<DifferentialLog> for LogVariant {
66    fn from(value: DifferentialLog) -> Self {
67        Self::Differential(value)
68    }
69}
70
71impl From<ComputeLog> for LogVariant {
72    fn from(value: ComputeLog) -> Self {
73        Self::Compute(value)
74    }
75}
76
77/// TODO(database-issues#7533): Add documentation.
78#[derive(
79    Hash,
80    Eq,
81    Ord,
82    PartialEq,
83    PartialOrd,
84    Debug,
85    Clone,
86    Copy,
87    Serialize,
88    Deserialize
89)]
90pub enum TimelyLog {
91    /// TODO(database-issues#7533): Add documentation.
92    Operates,
93    /// TODO(database-issues#7533): Add documentation.
94    Channels,
95    /// TODO(database-issues#7533): Add documentation.
96    Elapsed,
97    /// TODO(database-issues#7533): Add documentation.
98    Histogram,
99    /// TODO(database-issues#7533): Add documentation.
100    Addresses,
101    /// TODO(database-issues#7533): Add documentation.
102    Parks,
103    /// TODO(database-issues#7533): Add documentation.
104    MessagesSent,
105    /// TODO(database-issues#7533): Add documentation.
106    MessagesReceived,
107    /// TODO(database-issues#7533): Add documentation.
108    Reachability,
109    /// TODO(database-issues#7533): Add documentation.
110    BatchesSent,
111    /// TODO(database-issues#7533): Add documentation.
112    BatchesReceived,
113}
114
115/// TODO(database-issues#7533): Add documentation.
116#[derive(
117    Hash,
118    Eq,
119    Ord,
120    PartialEq,
121    PartialOrd,
122    Debug,
123    Clone,
124    Copy,
125    Serialize,
126    Deserialize
127)]
128pub enum DifferentialLog {
129    /// TODO(database-issues#7533): Add documentation.
130    ArrangementBatches,
131    /// TODO(database-issues#7533): Add documentation.
132    ArrangementRecords,
133    /// TODO(database-issues#7533): Add documentation.
134    Sharing,
135    /// TODO(database-issues#7533): Add documentation.
136    BatcherRecords,
137    /// TODO(database-issues#7533): Add documentation.
138    BatcherSize,
139    /// TODO(database-issues#7533): Add documentation.
140    BatcherCapacity,
141    /// TODO(database-issues#7533): Add documentation.
142    BatcherAllocations,
143}
144
145/// Variants of compute introspection sources.
146#[derive(
147    Hash,
148    Eq,
149    PartialEq,
150    Ord,
151    PartialOrd,
152    Debug,
153    Clone,
154    Copy,
155    Serialize,
156    Deserialize
157)]
158pub enum ComputeLog {
159    /// Installed dataflow exports.
160    DataflowCurrent,
161    /// Dataflow write frontiers.
162    FrontierCurrent,
163    /// Pending peeks.
164    PeekCurrent,
165    /// A histogram over peek durations.
166    PeekDuration,
167    /// Dataflow import frontiers.
168    ImportFrontierCurrent,
169    /// Arrangement heap sizes.
170    ArrangementHeapSize,
171    /// Arrangement heap capacities.
172    ArrangementHeapCapacity,
173    /// Arrangement heap allocations.
174    ArrangementHeapAllocations,
175    /// Counts of errors in exported collections.
176    ErrorCount,
177    /// Hydration times of exported collections.
178    HydrationTime,
179    /// Hydration status of dataflow operators.
180    OperatorHydrationStatus,
181    /// Mappings from `GlobalId`/`LirId`` pairs to dataflow addresses.
182    LirMapping,
183    /// Mappings from dataflows to `GlobalId`s.
184    DataflowGlobal,
185    /// Prometheus metrics gathered from the metrics registry.
186    PrometheusMetrics,
187    /// Resource usage observations of each replica process.
188    ResourceUsage,
189}
190
191impl LogVariant {
192    /// By which columns should the logs be indexed.
193    ///
194    /// This is distinct from the `keys` property of the type, which indicates uniqueness.
195    /// When keys exist these are good choices for indexing, but when they do not we still
196    /// require index guidance.
197    pub fn index_by(&self) -> Vec<usize> {
198        let desc = self.desc();
199        let arity = desc.arity();
200        desc.typ()
201            .keys
202            .get(0)
203            .cloned()
204            .unwrap_or_else(|| (0..arity).collect())
205    }
206
207    /// Relation schemas for the logs.
208    ///
209    /// This types need to agree with the values that are produced
210    /// in `logging::compute::construct` and with the description in
211    /// `catalog/src/builtin/mz_introspection.rs`.
212    pub fn desc(&self) -> RelationDesc {
213        match self {
214            LogVariant::Timely(TimelyLog::Operates) => RelationDesc::builder()
215                .with_column("id", SqlScalarType::UInt64.nullable(false))
216                .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
217                .with_column("name", SqlScalarType::String.nullable(false))
218                .with_key(vec![0, 1])
219                .finish(),
220
221            LogVariant::Timely(TimelyLog::Channels) => RelationDesc::builder()
222                .with_column("id", SqlScalarType::UInt64.nullable(false))
223                .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
224                .with_column("from_index", SqlScalarType::UInt64.nullable(false))
225                .with_column("from_port", SqlScalarType::UInt64.nullable(false))
226                .with_column("to_index", SqlScalarType::UInt64.nullable(false))
227                .with_column("to_port", SqlScalarType::UInt64.nullable(false))
228                .with_column("type", SqlScalarType::String.nullable(false))
229                .with_key(vec![0, 1])
230                .finish(),
231
232            LogVariant::Timely(TimelyLog::Elapsed) => RelationDesc::builder()
233                .with_column("id", SqlScalarType::UInt64.nullable(false))
234                .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
235                .finish(),
236
237            LogVariant::Timely(TimelyLog::Histogram) => RelationDesc::builder()
238                .with_column("id", SqlScalarType::UInt64.nullable(false))
239                .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
240                .with_column("duration_ns", SqlScalarType::UInt64.nullable(false))
241                .finish(),
242
243            LogVariant::Timely(TimelyLog::Addresses) => RelationDesc::builder()
244                .with_column("id", SqlScalarType::UInt64.nullable(false))
245                .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
246                .with_column(
247                    "address",
248                    SqlScalarType::List {
249                        element_type: Box::new(SqlScalarType::UInt64),
250                        custom_id: None,
251                    }
252                    .nullable(false),
253                )
254                .with_key(vec![0, 1])
255                .finish(),
256
257            LogVariant::Timely(TimelyLog::Parks) => RelationDesc::builder()
258                .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
259                .with_column("slept_for_ns", SqlScalarType::UInt64.nullable(false))
260                .with_column("requested_ns", SqlScalarType::UInt64.nullable(false))
261                .finish(),
262
263            LogVariant::Timely(TimelyLog::BatchesReceived) => RelationDesc::builder()
264                .with_column("channel_id", SqlScalarType::UInt64.nullable(false))
265                .with_column("from_worker_id", SqlScalarType::UInt64.nullable(false))
266                .with_column("to_worker_id", SqlScalarType::UInt64.nullable(false))
267                .finish(),
268
269            LogVariant::Timely(TimelyLog::BatchesSent) => RelationDesc::builder()
270                .with_column("channel_id", SqlScalarType::UInt64.nullable(false))
271                .with_column("from_worker_id", SqlScalarType::UInt64.nullable(false))
272                .with_column("to_worker_id", SqlScalarType::UInt64.nullable(false))
273                .finish(),
274
275            LogVariant::Timely(TimelyLog::MessagesReceived) => RelationDesc::builder()
276                .with_column("channel_id", SqlScalarType::UInt64.nullable(false))
277                .with_column("from_worker_id", SqlScalarType::UInt64.nullable(false))
278                .with_column("to_worker_id", SqlScalarType::UInt64.nullable(false))
279                .finish(),
280
281            LogVariant::Timely(TimelyLog::MessagesSent) => RelationDesc::builder()
282                .with_column("channel_id", SqlScalarType::UInt64.nullable(false))
283                .with_column("from_worker_id", SqlScalarType::UInt64.nullable(false))
284                .with_column("to_worker_id", SqlScalarType::UInt64.nullable(false))
285                .finish(),
286
287            LogVariant::Timely(TimelyLog::Reachability) => RelationDesc::builder()
288                .with_column("id", SqlScalarType::UInt64.nullable(false))
289                .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
290                .with_column("source", SqlScalarType::UInt64.nullable(false))
291                .with_column("port", SqlScalarType::UInt64.nullable(false))
292                .with_column("update_type", SqlScalarType::String.nullable(false))
293                .with_column("time", SqlScalarType::MzTimestamp.nullable(true))
294                .finish(),
295
296            LogVariant::Differential(DifferentialLog::ArrangementBatches)
297            | LogVariant::Differential(DifferentialLog::ArrangementRecords)
298            | LogVariant::Differential(DifferentialLog::Sharing)
299            | LogVariant::Differential(DifferentialLog::BatcherRecords)
300            | LogVariant::Differential(DifferentialLog::BatcherSize)
301            | LogVariant::Differential(DifferentialLog::BatcherCapacity)
302            | LogVariant::Differential(DifferentialLog::BatcherAllocations)
303            | LogVariant::Compute(ComputeLog::ArrangementHeapSize)
304            | LogVariant::Compute(ComputeLog::ArrangementHeapCapacity)
305            | LogVariant::Compute(ComputeLog::ArrangementHeapAllocations) => {
306                RelationDesc::builder()
307                    .with_column("operator_id", SqlScalarType::UInt64.nullable(false))
308                    .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
309                    .finish()
310            }
311
312            LogVariant::Compute(ComputeLog::DataflowCurrent) => RelationDesc::builder()
313                .with_column("export_id", SqlScalarType::String.nullable(false))
314                .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
315                .with_column("dataflow_id", SqlScalarType::UInt64.nullable(false))
316                .with_key(vec![0, 1])
317                .finish(),
318
319            LogVariant::Compute(ComputeLog::FrontierCurrent) => RelationDesc::builder()
320                .with_column("export_id", SqlScalarType::String.nullable(false))
321                .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
322                .with_column("time", SqlScalarType::MzTimestamp.nullable(false))
323                .with_key(vec![0, 1])
324                .finish(),
325
326            LogVariant::Compute(ComputeLog::ImportFrontierCurrent) => RelationDesc::builder()
327                .with_column("export_id", SqlScalarType::String.nullable(false))
328                .with_column("import_id", SqlScalarType::String.nullable(false))
329                .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
330                .with_column("time", SqlScalarType::MzTimestamp.nullable(false))
331                .with_key(vec![0, 1, 2])
332                .finish(),
333
334            LogVariant::Compute(ComputeLog::PeekCurrent) => RelationDesc::builder()
335                .with_column("id", SqlScalarType::Uuid.nullable(false))
336                .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
337                .with_column("object_id", SqlScalarType::String.nullable(false))
338                .with_column("type", SqlScalarType::String.nullable(false))
339                .with_column("time", SqlScalarType::MzTimestamp.nullable(false))
340                .with_key(vec![0, 1])
341                .finish(),
342
343            LogVariant::Compute(ComputeLog::PeekDuration) => RelationDesc::builder()
344                .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
345                .with_column("type", SqlScalarType::String.nullable(false))
346                .with_column("duration_ns", SqlScalarType::UInt64.nullable(false))
347                .finish(),
348
349            LogVariant::Compute(ComputeLog::ErrorCount) => RelationDesc::builder()
350                .with_column("export_id", SqlScalarType::String.nullable(false))
351                .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
352                .with_column("count", SqlScalarType::Int64.nullable(false))
353                .with_key(vec![0, 1])
354                .finish(),
355
356            LogVariant::Compute(ComputeLog::HydrationTime) => RelationDesc::builder()
357                .with_column("export_id", SqlScalarType::String.nullable(false))
358                .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
359                .with_column("time_ns", SqlScalarType::UInt64.nullable(true))
360                .with_column(
361                    "installed_at",
362                    SqlScalarType::TimestampTz { precision: None }.nullable(false),
363                )
364                .with_column(
365                    "started_at",
366                    SqlScalarType::TimestampTz { precision: None }.nullable(true),
367                )
368                .with_column(
369                    "hydrated_at",
370                    SqlScalarType::TimestampTz { precision: None }.nullable(true),
371                )
372                .with_key(vec![0, 1])
373                .finish(),
374
375            LogVariant::Compute(ComputeLog::OperatorHydrationStatus) => RelationDesc::builder()
376                .with_column("export_id", SqlScalarType::String.nullable(false))
377                .with_column("lir_id", SqlScalarType::UInt64.nullable(false))
378                .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
379                .with_column("hydrated", SqlScalarType::Bool.nullable(false))
380                .with_key(vec![0, 1, 2])
381                .finish(),
382
383            LogVariant::Compute(ComputeLog::LirMapping) => RelationDesc::builder()
384                .with_column("global_id", SqlScalarType::String.nullable(false))
385                .with_column("lir_id", SqlScalarType::UInt64.nullable(false))
386                .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
387                .with_column("operator", SqlScalarType::String.nullable(false))
388                .with_column("parent_lir_id", SqlScalarType::UInt64.nullable(true))
389                .with_column("nesting", SqlScalarType::UInt16.nullable(false))
390                .with_column("operator_id_start", SqlScalarType::UInt64.nullable(false))
391                .with_column("operator_id_end", SqlScalarType::UInt64.nullable(false))
392                .with_key(vec![0, 1, 2])
393                .finish(),
394
395            LogVariant::Compute(ComputeLog::DataflowGlobal) => RelationDesc::builder()
396                .with_column("id", SqlScalarType::UInt64.nullable(false))
397                .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
398                .with_column("global_id", SqlScalarType::String.nullable(false))
399                .with_key(vec![0, 1, 2])
400                .finish(),
401
402            LogVariant::Compute(ComputeLog::PrometheusMetrics) => RelationDesc::builder()
403                .with_column("process_id", SqlScalarType::UInt64.nullable(false))
404                .with_column("metric_name", SqlScalarType::String.nullable(false))
405                .with_column("metric_type", SqlScalarType::String.nullable(false))
406                .with_column(
407                    "labels",
408                    SqlScalarType::Map {
409                        value_type: Box::new(SqlScalarType::String),
410                        custom_id: None,
411                    }
412                    .nullable(false),
413                )
414                .with_column("value", SqlScalarType::Float64.nullable(false))
415                .with_column("help", SqlScalarType::String.nullable(false))
416                .with_key(vec![0, 1, 3])
417                .finish(),
418
419            LogVariant::Compute(ComputeLog::ResourceUsage) => RelationDesc::builder()
420                .with_column("process_id", SqlScalarType::UInt64.nullable(false))
421                .with_column("source", SqlScalarType::String.nullable(false))
422                .with_column("metric", SqlScalarType::String.nullable(false))
423                .with_column("value", SqlScalarType::UInt64.nullable(false))
424                .with_key(vec![0, 1, 2])
425                .finish(),
426        }
427    }
428}
mz_compute_client/logging.rs

mz_compute_client/
logging.rs
