1use mz_catalog::memory::objects::{Cluster, ClusterReplica};
11use mz_ore::sql;
12use mz_ore::sql::Sql;
13
14use super::sql::SqlRequest;
15
16#[derive(Debug)]
17pub(crate) struct PrometheusSqlQuery<'a> {
18 pub(crate) metric_name: &'a str,
19 pub(crate) help: &'a str,
20 pub(crate) query: &'static str,
21 pub(crate) value_column_name: &'a str,
22 pub(crate) per_replica: bool,
23}
24
25impl<'a> PrometheusSqlQuery<'a> {
26 pub(crate) fn to_sql_request(
27 &self,
28 cluster: Option<(&Cluster, &ClusterReplica)>,
29 ) -> SqlRequest {
30 let query = Sql::new(self.query);
31 let query = if let Some((cluster, replica)) = cluster {
32 sql!(
33 "SET auto_route_catalog_queries = false; SET CLUSTER = {}; SET CLUSTER_REPLICA = {}; {}",
34 Sql::literal(&cluster.name),
35 Sql::literal(&replica.name),
36 query
37 )
38 } else {
39 sql!(
40 "SET auto_route_catalog_queries = true; RESET CLUSTER; RESET CLUSTER_REPLICA; {}",
41 query
42 )
43 };
44 SqlRequest::Simple { query }
45 }
46}
47
48pub(crate) static FRONTIER_METRIC_QUERIES: &[PrometheusSqlQuery] = &[
49 PrometheusSqlQuery {
50 metric_name: "mz_write_frontier",
51 help: "The global write frontiers of compute and storage collections.",
52 query: "SELECT
53 object_id AS collection_id,
54 coalesce(write_frontier::text::uint8, 18446744073709551615::uint8) AS write_frontier
55 FROM mz_internal.mz_frontiers
56 WHERE object_id NOT LIKE 't%';",
57 value_column_name: "write_frontier",
58 per_replica: false,
59 },
60 PrometheusSqlQuery {
61 metric_name: "mz_read_frontier",
62 help: "The global read frontiers of compute and storage collections.",
63 query: "SELECT
64 object_id AS collection_id,
65 coalesce(read_frontier::text::uint8, 18446744073709551615::uint8) AS read_frontier
66 FROM mz_internal.mz_frontiers
67 WHERE object_id NOT LIKE 't%';",
68 value_column_name: "read_frontier",
69 per_replica: false,
70 },
71 PrometheusSqlQuery {
72 metric_name: "mz_replica_write_frontiers",
73 help: "The per-replica write frontiers of compute and storage collections.",
74 query: "SELECT
75 object_id AS collection_id,
76 coalesce(write_frontier::text::uint8, 18446744073709551615::uint8) AS write_frontier,
77 cluster_id AS instance_id,
78 replica_id AS replica_id
79 FROM mz_catalog.mz_cluster_replica_frontiers
80 JOIN mz_cluster_replicas ON (id = replica_id)
81 WHERE object_id NOT LIKE 't%';",
82 value_column_name: "write_frontier",
83 per_replica: false,
84 },
85 PrometheusSqlQuery {
86 metric_name: "mz_replica_write_frontiers",
87 help: "The per-replica write frontiers of compute and storage collections.",
88 query: "SELECT
89 object_id AS collection_id,
90 coalesce(write_frontier::text::uint8, 18446744073709551615::uint8) AS write_frontier,
91 cluster_id AS instance_id,
92 replica_id AS replica_id
93 FROM mz_catalog.mz_cluster_replica_frontiers
94 JOIN mz_cluster_replicas ON (id = replica_id)
95 WHERE object_id NOT LIKE 't%';",
96 value_column_name: "write_frontier",
97 per_replica: false,
98 },
99];
100
101pub(crate) static USAGE_METRIC_QUERIES: &[PrometheusSqlQuery] = &[
102 PrometheusSqlQuery {
103 metric_name: "mz_compute_cluster_status",
104 help: "Reports the name, ID, size, and availability zone of each cluster replica. Value is always 1.",
105 query: "SELECT
106 1 AS status,
107 cr.cluster_id AS compute_cluster_id,
108 cr.id AS compute_replica_id,
109 c.name AS compute_cluster_name,
110 cr.name AS compute_replica_name,
111 COALESCE(cr.size, '') AS size,
112 COALESCE(cr.availability_zone, '') AS availability_zone,
113 mz_version() AS mz_version
114 FROM
115 mz_cluster_replicas AS cr
116 JOIN mz_clusters AS c
117 ON cr.cluster_id = c.id",
118 value_column_name: "status",
119 per_replica: false,
120 },
121 PrometheusSqlQuery {
122 metric_name: "mz_workload_clusters",
123 help: "Reports the workload_class of each user cluster. Value is always 1.",
124 query: "select
125 c.id as cluster_id,
126 c.name as cluster_name,
127 coalesce(wc.workload_class,'false') as workload_class,
128 1 as value
129 from mz_clusters c
130 join mz_internal.mz_cluster_workload_classes wc
131 on c.id = wc.id",
132 value_column_name: "value",
133 per_replica: false,
134 },
135 PrometheusSqlQuery {
136 metric_name: "mz_indexes_count",
137 help: "Number of active indexes in the instance, by the type of relation on which the index is built.",
138 query: "select
139 o.type as relation_type
140 , count(i.id) as indexes
141 from mz_catalog.mz_indexes i
142 join mz_catalog.mz_objects o
143 on i.on_id = o.id
144 where i.id like 'u%'
145 group by relation_type",
146 value_column_name: "indexes",
147 per_replica: false,
148 },
149 PrometheusSqlQuery {
150 metric_name: "mz_sources_count",
151 help: "Number of active sources in the instance, by type, envelope type, and size of source.",
152 query: "SELECT
153 type,
154 COALESCE(envelope_type, '<none>') AS envelope_type,
155 COALESCE(mz_cluster_replicas.size, '') AS size,
156 count(mz_sources.id) AS sources
157 FROM
158 mz_sources, mz_cluster_replicas
159 WHERE mz_sources.id LIKE 'u%'
160 AND mz_sources.type != 'subsource'
161 AND mz_sources.cluster_id = mz_cluster_replicas.cluster_id
162 GROUP BY
163 type, envelope_type, mz_cluster_replicas.size",
164 value_column_name: "sources",
165 per_replica: false,
166 },
167 PrometheusSqlQuery {
168 metric_name: "mz_sinks_count",
169 help: "Number of active sinks in the instance, by type, envelope type, and size.",
170 query: "SELECT
171 type,
172 COALESCE(envelope_type, '<none>') AS envelope_type,
173 COALESCE(mz_cluster_replicas.size, '') AS size,
174 count(mz_sinks.id) AS sinks
175 FROM
176 mz_sinks, mz_cluster_replicas
177 WHERE mz_sinks.id LIKE 'u%'
178 AND mz_sinks.cluster_id = mz_cluster_replicas.cluster_id
179 GROUP BY type, envelope_type, mz_cluster_replicas.size",
180 value_column_name: "sinks",
181 per_replica: false,
182 },
183 PrometheusSqlQuery {
184 metric_name: "mz_connections_count",
185 help: "Number of active connections in the instance, by type.",
186 query: "select
187 type
188 , count(id) as connections
189 from mz_connections
190 where id like 'u%'
191 group by type",
192 value_column_name: "connections",
193 per_replica: false,
194 },
195];
196
197pub(crate) static COMPUTE_METRIC_QUERIES: &[PrometheusSqlQuery] = &[
198 PrometheusSqlQuery {
199 metric_name: "mz_arrangement_count",
200 help: "The number of arrangements in a dataflow.",
201 query: "WITH
202 arrangements AS (
203 SELECT DISTINCT operator_id AS id
204 FROM mz_internal.mz_arrangement_records_raw
205 ),
206 collections AS (
207 SELECT
208 id,
209 CASE
210 WHEN starts_with(export_id, 't') THEN 'transient'
211 ELSE export_id
212 END AS export_id
213 FROM
214 mz_internal.mz_dataflow_addresses,
215 mz_internal.mz_compute_exports
216 WHERE address[1] = dataflow_id
217 )
218 SELECT
219 COALESCE(export_id, 'none') AS collection_id,
220 count(*) as count
221 FROM arrangements
222 LEFT JOIN collections USING (id)
223 GROUP BY export_id",
224 value_column_name: "count",
225 per_replica: true,
226 },
227 PrometheusSqlQuery {
228 metric_name: "mz_arrangement_record_count",
229 help: "The number of records in all arrangements in a dataflow.",
230 query: "WITH
231 collections AS (
232 SELECT
233 id,
234 CASE
235 WHEN starts_with(export_id, 't') THEN 'transient'
236 ELSE export_id
237 END AS export_id
238 FROM
239 mz_internal.mz_dataflow_addresses,
240 mz_internal.mz_compute_exports
241 WHERE address[1] = dataflow_id
242 )
243 SELECT
244 worker_id,
245 COALESCE(export_id, 'none') AS collection_id,
246 count(*) as count
247 FROM mz_internal.mz_arrangement_records_raw
248 LEFT JOIN collections ON (operator_id = id)
249 GROUP BY worker_id, export_id",
250 value_column_name: "count",
251 per_replica: true,
252 },
253 PrometheusSqlQuery {
254 metric_name: "mz_arrangement_batch_count",
255 help: "The number of batches in all arrangements in a dataflow.",
256 query: "WITH
257 collections AS (
258 SELECT
259 id,
260 CASE
261 WHEN starts_with(export_id, 't') THEN 'transient'
262 ELSE export_id
263 END AS export_id
264 FROM
265 mz_internal.mz_dataflow_addresses,
266 mz_internal.mz_compute_exports
267 WHERE address[1] = dataflow_id
268 )
269 SELECT
270 worker_id,
271 COALESCE(export_id, 'none') AS collection_id,
272 count(*) as count
273 FROM mz_internal.mz_arrangement_batches_raw
274 LEFT JOIN collections ON (operator_id = id)
275 GROUP BY worker_id, export_id",
276 value_column_name: "count",
277 per_replica: true,
278 },
279 PrometheusSqlQuery {
280 metric_name: "mz_arrangement_size_bytes",
281 help: "The size of all arrangements in a dataflow.",
282 query: "WITH
283 collections AS (
284 SELECT
285 id,
286 CASE
287 WHEN starts_with(export_id, 't') THEN 'transient'
288 ELSE export_id
289 END AS export_id
290 FROM
291 mz_internal.mz_dataflow_addresses,
292 mz_internal.mz_compute_exports
293 WHERE address[1] = dataflow_id
294 )
295 SELECT
296 worker_id,
297 COALESCE(export_id, 'none') AS collection_id,
298 count(*) as count
299 FROM mz_internal.mz_arrangement_heap_size_raw
300 LEFT JOIN collections ON (operator_id = id)
301 GROUP BY worker_id, export_id",
302 value_column_name: "count",
303 per_replica: true,
304 },
305 PrometheusSqlQuery {
306 metric_name: "mz_arrangement_capacity_bytes",
307 help: "The capacity of all arrangements in all dataflows.",
308 query: "SELECT
309 worker_id,
310 count(*) as count
311 FROM mz_internal.mz_arrangement_heap_capacity_raw
312 GROUP BY worker_id",
313 value_column_name: "count",
314 per_replica: true,
315 },
316 PrometheusSqlQuery {
317 metric_name: "mz_arrangement_allocation_count",
318 help: "The number of allocations in all arrangements in all dataflows.",
319 query: "SELECT
320 worker_id,
321 count(*) as count
322 FROM mz_internal.mz_arrangement_heap_allocations_raw
323 GROUP BY worker_id",
324 value_column_name: "count",
325 per_replica: true,
326 },
327 PrometheusSqlQuery {
328 metric_name: "mz_compute_replica_park_duration_seconds_total",
329 help: "The total time workers were parked since restart.",
330 query: "SELECT
331 worker_id,
332 sum(slept_for_ns * count)::float8 / 1000000000 AS duration_s
333 FROM mz_internal.mz_scheduling_parks_histogram_per_worker
334 GROUP BY worker_id",
335 value_column_name: "duration_s",
336 per_replica: true,
337 },
338 PrometheusSqlQuery {
339 metric_name: "mz_compute_replica_peek_count",
340 help: "The number of pending peeks.",
341 query: "SELECT worker_id, count(*) as count
342 FROM mz_internal.mz_active_peeks_per_worker
343 GROUP BY worker_id",
344 value_column_name: "count",
345 per_replica: true,
346 },
347 PrometheusSqlQuery {
348 metric_name: "mz_dataflow_elapsed_seconds_total",
349 help: "The total time spent computing a dataflow.",
350 query: "SELECT
351 worker_id,
352 CASE
353 WHEN starts_with(export_id, 't') THEN 'transient'
354 ELSE export_id
355 END AS collection_id,
356 sum(elapsed_ns)::float8 / 1000000000 AS elapsed_s
357 FROM
358 mz_internal.mz_scheduling_elapsed_per_worker AS s,
359 mz_internal.mz_dataflow_operators AS o,
360 mz_internal.mz_dataflow_addresses AS a,
361 mz_internal.mz_compute_exports AS e
362 WHERE
363 o.id = s.id AND
364 o.id = a.id AND
365 list_length(a.address) = 1 AND
366 e.dataflow_id = a.address[1]
367 GROUP BY worker_id, collection_id",
368 value_column_name: "elapsed_s",
369 per_replica: true,
370 },
371 PrometheusSqlQuery {
372 metric_name: "mz_dataflow_error_count",
373 help: "The number of errors in a dataflow",
374 query: "SELECT
375 CASE
376 WHEN starts_with(export_id, 't') THEN 'transient'
377 ELSE export_id
378 END AS collection_id,
379 count::uint8 as count
380 FROM mz_internal.mz_compute_error_counts",
381 value_column_name: "count",
382 per_replica: true,
383 },
384];
385
386pub(crate) static STORAGE_METRIC_QUERIES: &[PrometheusSqlQuery] = &[
387 PrometheusSqlQuery {
388 metric_name: "mz_storage_objects",
389 help: "Nicely labeled information about existing sources and sinks.",
390 value_column_name: "value",
391 per_replica: false,
392 query: "
393 WITH
394 -- All user, non- progress or subsource sources.
395 top_level_sources AS (
396 SELECT id, connection_id, type, envelope_type, cluster_id
397 FROM mz_sources
398 WHERE id LIKE 'u%' AND type NOT IN ('progress', 'subsource')
399 ),
400 -- Sources enriched with the type of the core connection.
401 source_and_conns AS (
402 SELECT top_level_sources.*, mc.type AS connection_type
403 FROM top_level_sources
404 LEFT OUTER JOIN mz_connections mc ON mc.id = top_level_sources.connection_id
405 ),
406
407 -- All user sinks.
408 top_level_sinks AS (
409 SELECT id, connection_id, type, envelope_type, cluster_id
410 FROM mz_sinks
411 WHERE id LIKE 'u%'
412 ),
413 -- Sinks enriched with the type of the core connection.
414 sink_and_conns AS (
415 SELECT top_level_sinks.*, mc.type AS connection_type
416 FROM top_level_sinks
417 LEFT OUTER JOIN mz_connections mc ON mc.id = top_level_sinks.connection_id
418 ),
419
420 -- All objects we care about
421 object_and_conns AS (
422 SELECT * FROM source_and_conns
423 UNION ALL
424 SELECT * FROM sink_and_conns
425 ),
426
427 -- The networking the object connection uses, if any.
428 networking AS (
429 SELECT object_and_conns.id, mc.id AS networking_id, mc.type AS networking_type
430 FROM object_and_conns
431 JOIN mz_internal.mz_object_dependencies mod ON object_and_conns.connection_id = mod.object_id
432 JOIN mz_connections mc ON mc.id = mod.referenced_object_id
433 -- Not required but made explicit
434 WHERE object_and_conns.connection_id IS NOT NULL
435 ),
436 -- The connection the format of the object uses, if any. This uses `mz_object_dependencies`
437 -- and a filter to find non-core objects the object depends on.
438 format_conns AS (
439 SELECT object_and_conns.id, mc.id AS format_connection_id, mc.type AS format_connection
440 FROM mz_internal.mz_object_dependencies mod
441 JOIN object_and_conns ON mod.object_id = object_and_conns.id
442 JOIN mz_connections mc ON mc.id = mod.referenced_object_id
443 WHERE mc.id NOT IN (SELECT connection_id FROM top_level_sources UNION ALL SELECT connection_id FROM top_level_sinks)
444 -- Not required but made explicit
445 AND object_and_conns.connection_id IS NOT NULL
446 ),
447 -- The networking used by `format_conns`, if any.
448 format_conn_deps AS (
449 SELECT format_conns.id, mc.id AS format_connection_networking_id, mc.type AS format_connection_networking
450 FROM format_conns
451 JOIN mz_internal.mz_object_dependencies mod ON mod.object_id = format_conns.format_connection_id
452 JOIN mz_connections mc
453 ON mc.id = mod.referenced_object_id
454 ),
455
456 -- When aggregating values that are known to be the same, we just use `MAX` for simplicity.
457
458 -- source_and_conns LEFT JOINed with the networking and format connection information.
459 -- Missing networking/format connections are coalesced to `none`, and aggregated with a comma.
460 -- This is because sources can have multiple type of networking (i.e. different kafka brokers with
461 -- different configuration), and multiple connections for formats (for key and value formats).
462 --
463 -- The actual format type is not yet included, as it depends on https://github.com/MaterializeInc/materialize/pull/23880
464 sources AS (
465 SELECT
466 -- Whether its a source or sink
467 'source' AS type,
468 1 AS value,
469 source_and_conns.id AS id,
470 -- What type of source/sink it is
471 MAX(type) AS object_type,
472 COALESCE(MAX(connection_type), 'none') AS connection_type,
473 COALESCE(MAX(envelope_type), 'none') AS envelope_type,
474 STRING_AGG(
475 DISTINCT COALESCE(networking_type, 'none'),
476 ','
477 ORDER BY COALESCE(networking_type, 'none') ASC
478 ) AS networking_type,
479 STRING_AGG(
480 DISTINCT COALESCE(format_connection, 'none'),
481 ','
482 ORDER BY COALESCE(format_connection, 'none') ASC
483 ) AS format_connection,
484 STRING_AGG(
485 DISTINCT COALESCE(format_connection_networking, 'none'),
486 ','
487 ORDER BY COALESCE(format_connection_networking, 'none') ASC
488 ) AS format_connection_networking,
489 MAX(cluster_id) AS cluster_id
490 FROM source_and_conns
491 LEFT OUTER JOIN networking ON networking.id = source_and_conns.id
492 LEFT OUTER JOIN format_conns ON format_conns.id = source_and_conns.id
493 LEFT OUTER JOIN format_conn_deps ON format_conn_deps.id = source_and_conns.id
494 GROUP BY source_and_conns.id
495 ),
496
497 sinks AS (
498 SELECT
499 'sink' AS type,
500 1 AS value,
501 sink_and_conns.id AS id,
502 MAX(type) AS object_type,
503 COALESCE(MAX(connection_type), 'none') AS connection_type,
504 COALESCE(MAX(envelope_type), 'none') AS envelope_type,
505 STRING_AGG(
506 DISTINCT COALESCE(networking_type, 'none'),
507 ','
508 ORDER BY COALESCE(networking_type, 'none') ASC
509 ) AS networking_type,
510 -- Sinks can only have 1 format connection but we aggregate
511 -- for consistency.
512 STRING_AGG(
513 DISTINCT COALESCE(format_connection, 'none'),
514 ','
515 ORDER BY COALESCE(format_connection, 'none') ASC
516 ) AS format_connection,
517 STRING_AGG(
518 DISTINCT COALESCE(format_connection_networking, 'none'),
519 ','
520 ORDER BY COALESCE(format_connection_networking, 'none') ASC
521 ) AS format_connection_networking,
522 MAX(cluster_id) AS cluster_id
523 FROM sink_and_conns
524 LEFT OUTER JOIN networking ON networking.id = sink_and_conns.id
525 LEFT OUTER JOIN format_conns ON format_conns.id = sink_and_conns.id
526 LEFT OUTER JOIN format_conn_deps ON format_conn_deps.id = sink_and_conns.id
527 GROUP BY sink_and_conns.id
528 ),
529
530 -- everything without replicas
531 together AS (
532 SELECT * FROM sources
533 UNION ALL
534 SELECT * from sinks
535 ),
536
537 with_cluster_replicas AS (
538 SELECT
539 -- `together.*` doesn't work because we need to aggregate the columns :(
540 MAX(together.id) AS id,
541 MAX(together.type) AS type,
542 MAX(together.object_type) AS object_type,
543 -- We just report 1 to the gauge. The `replica_id` labels aggregates the 0-many replicas associated
544 -- with this object.
545 MAX(together.value) AS value,
546 MAX(together.connection_type) AS connection_type,
547 MAX(together.envelope_type) AS envelope_type,
548 MAX(together.networking_type) AS networking_type,
549 MAX(together.format_connection) AS format_connection,
550 MAX(together.format_connection_networking) AS format_connection_networking,
551 MAX(together.cluster_id) AS cluster_id,
552 mcr.id AS replica_id,
553 -- Coalesce to 0 when there is no replica. This ensures `with_pod` below will generate
554 -- 1 row for the object. Also, -1 because `generate_series` is inclusive.
555 COALESCE((mcrs.processes - 1)::int, 0) AS processes
556 FROM together
557 LEFT OUTER JOIN mz_cluster_replicas mcr ON together.cluster_id = mcr.cluster_id
558 LEFT OUTER JOIN mz_catalog.mz_cluster_replica_sizes mcrs ON mcr.size = mcrs.size
559 GROUP BY together.id, mcr.id, mcrs.processes
560 ),
561
562 with_pod AS (
563 SELECT
564 id,
565 type,
566 object_type,
567 value,
568 connection_type,
569 envelope_type,
570 networking_type,
571 format_connection,
572 format_connection_networking,
573 cluster_id,
574 COALESCE(replica_id, 'none') AS replica_id,
575 -- When `replica_id` is NULL`, this coalesces to `none`.
576 COALESCE('cluster-' || cluster_id || '-replica-' || replica_id || '-' || generated, 'none') AS synthesized_pod
577 FROM with_cluster_replicas, generate_series(0, processes) generated
578 )
579
580 SELECT * FROM with_pod;
581 ",
582 },
583];