mz_audit_log/
lib.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//! Audit log data structures.
11//!
12//! The audit log is logging that is produced by user actions and consumed
13//! by users in the form of the `mz_catalog.mz_audit_events` SQL table and
14//! by the cloud management layer for billing and introspection. This crate
15//! is designed to make the production and consumption of the logs type
16//! safe. Events and their metadata are versioned and the data structures
17//! replicated here so that if the data change in some other crate, a
18//! new version here can be made. This avoids needing to poke at the data
19//! when reading it to determine what it means and should have full backward
20//! compatibility. This is its own crate so that production and consumption can
21//! be in different processes and production is not allowed to specify private
22//! data structures unknown to the reader.
23//!
24//! `EventDetails::as_json` produces the JSON that
25//! `mz_catalog.mz_audit_events.details` exposes. The durable catalog stores
26//! the proto twin from `mz_catalog_protos::objects::audit_log_event_v1`,
27//! which `parse_catalog_audit_log_details` (in
28//! `src/expr/src/scalar/func/impls/jsonb.rs`) reshapes back into
29//! `as_json`'s output. Changes here that shift that output (new variants,
30//! renamed fields, added `skip_serializing_if`, field-name diffs against
31//! the proto) need matching updates there. The round-trip is covered by
32//! the property test in `src/catalog/tests/audit_log_details.rs`.
33
34use std::time::Duration;
35
36use mz_ore::now::EpochMillis;
37use proptest_derive::Arbitrary;
38use serde::{Deserialize, Serialize};
39
40/// New version variants should be added if fields need to be added, changed, or removed.
41#[derive(
42    Clone,
43    Debug,
44    Serialize,
45    Deserialize,
46    PartialOrd,
47    PartialEq,
48    Eq,
49    Ord,
50    Hash,
51    Arbitrary
52)]
53pub enum VersionedEvent {
54    V1(EventV1),
55}
56
57impl VersionedEvent {
58    /// Create a new event. This function must always require and produce the most
59    /// recent variant of VersionedEvent. `id` must be a globally increasing,
60    /// ordered number such that sorting by it on all events yields the order
61    /// of events by users. It is insufficient to use `occurred_at` (even at
62    /// nanosecond precision) due to clock unpredictability.
63    pub fn new(
64        id: u64,
65        event_type: EventType,
66        object_type: ObjectType,
67        details: EventDetails,
68        user: Option<String>,
69        occurred_at: EpochMillis,
70    ) -> Self {
71        Self::V1(EventV1::new(
72            id,
73            event_type,
74            object_type,
75            details,
76            user,
77            occurred_at,
78        ))
79    }
80
81    // Implement deserialize and serialize so writers and readers don't have to
82    // coordinate about which Serializer to use.
83    pub fn deserialize(data: &[u8]) -> Result<Self, anyhow::Error> {
84        Ok(serde_json::from_slice(data)?)
85    }
86
87    pub fn serialize(&self) -> Vec<u8> {
88        serde_json::to_vec(self).expect("must serialize")
89    }
90
91    /// Returns a globally sortable event order. All event versions must have this
92    /// field.
93    pub fn sortable_id(&self) -> u64 {
94        match self {
95            VersionedEvent::V1(ev) => ev.id,
96        }
97    }
98}
99
100#[derive(
101    Clone,
102    Debug,
103    Serialize,
104    Deserialize,
105    PartialOrd,
106    PartialEq,
107    Eq,
108    Ord,
109    Hash,
110    Arbitrary
111)]
112#[serde(rename_all = "kebab-case")]
113pub enum EventType {
114    Create,
115    Drop,
116    Alter,
117    Grant,
118    Revoke,
119    Comment,
120}
121
122impl EventType {
123    pub fn as_title_case(&self) -> &'static str {
124        match self {
125            EventType::Create => "Created",
126            EventType::Drop => "Dropped",
127            EventType::Alter => "Altered",
128            EventType::Grant => "Granted",
129            EventType::Revoke => "Revoked",
130            EventType::Comment => "Comment",
131        }
132    }
133}
134
135serde_plain::derive_display_from_serialize!(EventType);
136
137#[derive(
138    Clone,
139    Copy,
140    Debug,
141    Serialize,
142    Deserialize,
143    PartialOrd,
144    PartialEq,
145    Eq,
146    Ord,
147    Hash,
148    Arbitrary
149)]
150#[serde(rename_all = "kebab-case")]
151pub enum ObjectType {
152    Cluster,
153    ClusterReplica,
154    Connection,
155    ContinualTask,
156    Database,
157    Func,
158    Index,
159    MaterializedView,
160    MetricSink,
161    NetworkPolicy,
162    Role,
163    Secret,
164    Schema,
165    Sink,
166    Source,
167    System,
168    Table,
169    Type,
170    View,
171}
172
173impl ObjectType {
174    pub fn as_title_case(&self) -> &'static str {
175        match self {
176            ObjectType::Cluster => "Cluster",
177            ObjectType::ClusterReplica => "Cluster Replica",
178            ObjectType::Connection => "Connection",
179            ObjectType::ContinualTask => "Continual Task",
180            ObjectType::Database => "Database",
181            ObjectType::Func => "Function",
182            ObjectType::Index => "Index",
183            ObjectType::MaterializedView => "Materialized View",
184            ObjectType::MetricSink => "Metric Sink",
185            ObjectType::NetworkPolicy => "Network Policy",
186            ObjectType::Role => "Role",
187            ObjectType::Schema => "Schema",
188            ObjectType::Secret => "Secret",
189            ObjectType::Sink => "Sink",
190            ObjectType::Source => "Source",
191            ObjectType::System => "System",
192            ObjectType::Table => "Table",
193            ObjectType::Type => "Type",
194            ObjectType::View => "View",
195        }
196    }
197}
198
199serde_plain::derive_display_from_serialize!(ObjectType);
200
201#[derive(
202    Clone,
203    Debug,
204    Serialize,
205    Deserialize,
206    PartialOrd,
207    PartialEq,
208    Eq,
209    Ord,
210    Hash,
211    Arbitrary
212)]
213pub enum EventDetails {
214    #[serde(rename = "CreateComputeReplicaV1")] // historical name
215    CreateClusterReplicaV1(CreateClusterReplicaV1),
216    CreateClusterReplicaV2(CreateClusterReplicaV2),
217    CreateClusterReplicaV3(CreateClusterReplicaV3),
218    CreateClusterReplicaV4(CreateClusterReplicaV4),
219    #[serde(rename = "DropComputeReplicaV1")] // historical name
220    DropClusterReplicaV1(DropClusterReplicaV1),
221    DropClusterReplicaV2(DropClusterReplicaV2),
222    DropClusterReplicaV3(DropClusterReplicaV3),
223    CreateSourceSinkV1(CreateSourceSinkV1),
224    CreateSourceSinkV2(CreateSourceSinkV2),
225    CreateSourceSinkV3(CreateSourceSinkV3),
226    CreateSourceSinkV4(CreateSourceSinkV4),
227    CreateIndexV1(CreateIndexV1),
228    CreateMaterializedViewV1(CreateMaterializedViewV1),
229    AlterApplyReplacementV1(AlterApplyReplacementV1),
230    AlterSetClusterV1(AlterSetClusterV1),
231    AlterSourceSinkV1(AlterSourceSinkV1),
232    GrantRoleV1(GrantRoleV1),
233    GrantRoleV2(GrantRoleV2),
234    RevokeRoleV1(RevokeRoleV1),
235    RevokeRoleV2(RevokeRoleV2),
236    UpdatePrivilegeV1(UpdatePrivilegeV1),
237    AlterDefaultPrivilegeV1(AlterDefaultPrivilegeV1),
238    UpdateOwnerV1(UpdateOwnerV1),
239    IdFullNameV1(IdFullNameV1),
240    RenameClusterV1(RenameClusterV1),
241    RenameClusterReplicaV1(RenameClusterReplicaV1),
242    AlterClusterReconfigurationV1(AlterClusterReconfigurationV1),
243    ClusterHydrationBurstV1(ClusterHydrationBurstV1),
244    RenameItemV1(RenameItemV1),
245    IdNameV1(IdNameV1),
246    SchemaV1(SchemaV1),
247    SchemaV2(SchemaV2),
248    UpdateItemV1(UpdateItemV1),
249    RenameSchemaV1(RenameSchemaV1),
250    AlterRetainHistoryV1(AlterRetainHistoryV1),
251    AlterAddColumnV1(AlterAddColumnV1),
252    AlterSourceTimestampIntervalV1(AlterSourceTimestampIntervalV1),
253    ToNewIdV1(ToNewIdV1),
254    FromPreviousIdV1(FromPreviousIdV1),
255    SetV1(SetV1),
256    ResetAllV1,
257    RotateKeysV1(RotateKeysV1),
258    CreateRoleV1(CreateRoleV1),
259}
260
261#[derive(
262    Clone,
263    Debug,
264    Serialize,
265    Deserialize,
266    PartialOrd,
267    PartialEq,
268    Eq,
269    Ord,
270    Hash,
271    Arbitrary
272)]
273pub struct SetV1 {
274    pub name: String,
275    pub value: Option<String>,
276}
277
278#[derive(
279    Clone,
280    Debug,
281    Serialize,
282    Deserialize,
283    PartialOrd,
284    PartialEq,
285    Eq,
286    Ord,
287    Hash,
288    Arbitrary
289)]
290pub struct RotateKeysV1 {
291    pub id: String,
292    pub name: String,
293}
294
295#[derive(
296    Clone,
297    Debug,
298    Serialize,
299    Deserialize,
300    PartialOrd,
301    PartialEq,
302    Eq,
303    Ord,
304    Hash,
305    Arbitrary
306)]
307pub struct IdFullNameV1 {
308    pub id: String,
309    #[serde(flatten)]
310    pub name: FullNameV1,
311}
312
313#[derive(
314    Clone,
315    Debug,
316    Serialize,
317    Deserialize,
318    PartialOrd,
319    PartialEq,
320    Eq,
321    Ord,
322    Hash,
323    Arbitrary
324)]
325pub struct FullNameV1 {
326    pub database: String,
327    pub schema: String,
328    pub item: String,
329}
330
331#[derive(
332    Clone,
333    Debug,
334    Serialize,
335    Deserialize,
336    PartialOrd,
337    PartialEq,
338    Eq,
339    Ord,
340    Hash,
341    Arbitrary
342)]
343pub struct IdNameV1 {
344    pub id: String,
345    pub name: String,
346}
347
348/// A transition in the lifecycle of a background cluster reconfiguration (a
349/// `reconfiguration` record on a managed cluster), recorded so an operator can
350/// trace a background `ALTER CLUSTER` from start to its resolution.
351///
352/// The replica creates and drops the reconfiguration induces are recorded
353/// separately, carrying [`CreateOrDropClusterReplicaReasonV1::Reconfiguration`];
354/// this event family records the cluster-level transitions those replica
355/// lifecycle events hang off of.
356#[derive(
357    Clone,
358    Debug,
359    Serialize,
360    Deserialize,
361    PartialOrd,
362    PartialEq,
363    Eq,
364    Ord,
365    Hash,
366    Arbitrary
367)]
368#[serde(rename_all = "kebab-case")]
369pub enum ReconfigurationLifecycleV1 {
370    /// A reconfiguration record was written or re-targeted: the cluster is now
371    /// converging onto a new target shape.
372    Started,
373    /// The realized config cut over to the target and the record was marked
374    /// finalized: a hydrated success under either `ON TIMEOUT` action
375    /// (including one that hydrated after the deadline, since success takes
376    /// precedence) or a forced `ON TIMEOUT COMMIT` cut-over of a
377    /// not-yet-hydrated target past the deadline. Which of the two it was is
378    /// recorded in [`AlterClusterReconfigurationV1::forced`].
379    Finalized,
380    /// The deadline fired with the target not hydrated under `ON TIMEOUT
381    /// ROLLBACK`: the record was marked timed out with the realized config
382    /// untouched and the target replicas dropped, reverting to the
383    /// pre-reconfiguration set. Emitted exactly once per timeout. The status
384    /// transition is durable, so it cannot re-fire. This event is the timeout's
385    /// papertrail, alongside the retained record.
386    TimedOut,
387    /// An in-flight reconfiguration was cancelled by re-targeting the record
388    /// back to the cluster's still-realized shape (the ALTER-back cancel path);
389    /// the controller drops the in-flight target replicas.
390    Cancelled,
391    /// The controller could not create the target replicas within the resource
392    /// budget and aborted the reconfiguration: the record was marked resource
393    /// exhausted with the realized config untouched, reverting to the
394    /// pre-reconfiguration set (like a rollback, but triggered by the budget
395    /// rather than the deadline). The status transition is durable, so it
396    /// cannot re-fire.
397    ResourceExhausted,
398}
399
400/// A cluster-level transition in a background reconfiguration's lifecycle.
401///
402/// `deadline` is the reconfiguration's active deadline as a millisecond
403/// `mz_timestamp`, recorded on every transition so an operator can correlate
404/// the transition with the originating `ALTER`.
405#[derive(
406    Clone,
407    Debug,
408    Serialize,
409    Deserialize,
410    PartialOrd,
411    PartialEq,
412    Eq,
413    Ord,
414    Hash,
415    Arbitrary
416)]
417pub struct AlterClusterReconfigurationV1 {
418    pub cluster_id: String,
419    pub cluster_name: String,
420    pub transition: ReconfigurationLifecycleV1,
421    /// On a `finalized` transition: whether the cut-over was forced by `ON
422    /// TIMEOUT COMMIT` at the deadline rather than reached by hydration.
423    /// `None` on every other transition.
424    #[serde(default)]
425    pub forced: Option<bool>,
426    pub target_size: String,
427    pub target_replication_factor: u32,
428    pub target_availability_zones: Vec<String>,
429    pub target_logging: ClusterReplicaLoggingV1,
430    pub deadline: Option<u64>,
431}
432
433/// A managed cluster's introspection-logging config, recorded on a
434/// reconfiguration event so the papertrail captures an introspection-only
435/// `ALTER` (which otherwise leaves `target_size` and
436/// `target_replication_factor` unchanged from the realized shape). Mirrors the
437/// durable `ReplicaLogging`: `log_logging` is `INTROSPECTION DEBUGGING`,
438/// `interval` is `INTROSPECTION INTERVAL` (`None` disables introspection).
439#[derive(
440    Clone,
441    Debug,
442    Serialize,
443    Deserialize,
444    PartialOrd,
445    PartialEq,
446    Eq,
447    Ord,
448    Hash,
449    Arbitrary
450)]
451pub struct ClusterReplicaLoggingV1 {
452    pub log_logging: bool,
453    pub interval: Option<Duration>,
454}
455
456#[derive(
457    Clone,
458    Debug,
459    Serialize,
460    Deserialize,
461    PartialOrd,
462    PartialEq,
463    Eq,
464    Ord,
465    Hash,
466    Arbitrary
467)]
468pub struct CreateRoleV1 {
469    pub id: String,
470    pub name: String,
471    pub auto_provision_source: Option<String>,
472}
473
474/// A transition in the lifecycle of a hydration burst (a `burst` record on a
475/// managed cluster), recorded so an operator can trace a controller-initiated
476/// burst from start to teardown.
477///
478/// The burst replica's create and drop are recorded separately, carrying
479/// [`CreateOrDropClusterReplicaReasonV1::HydrationBurst`]; this event family
480/// records the cluster-level transitions those replica lifecycle events hang off.
481#[derive(
482    Clone,
483    Debug,
484    Serialize,
485    Deserialize,
486    PartialOrd,
487    PartialEq,
488    Eq,
489    Ord,
490    Hash,
491    Arbitrary
492)]
493#[serde(rename_all = "kebab-case")]
494pub enum HydrationBurstLifecycleV1 {
495    /// A `burst` record was written: the controller is now running a burst
496    /// replica to accelerate hydration.
497    Started,
498    /// The `burst` record was cleared: the burst replica is torn down. Why is
499    /// recorded in [`ClusterHydrationBurstV1::finish_cause`].
500    Finished,
501}
502
503/// Why a hydration burst finished, recorded on a `finished` transition.
504#[derive(
505    Clone,
506    Copy,
507    Debug,
508    Serialize,
509    Deserialize,
510    PartialOrd,
511    PartialEq,
512    Eq,
513    Ord,
514    Hash,
515    Arbitrary
516)]
517#[serde(rename_all = "kebab-case")]
518pub enum BurstFinishCauseV1 {
519    /// The steady replica set hydrated and the linger duration elapsed.
520    LingerElapsed,
521    /// The burst is no longer warranted by current config: the auto-scaling
522    /// policy was removed or its hydration size changed, the cluster was
523    /// turned off, or burst was disabled environment-wide.
524    NoLongerWarranted,
525}
526
527/// A cluster-level transition in a hydration burst's lifecycle.
528#[derive(
529    Clone,
530    Debug,
531    Serialize,
532    Deserialize,
533    PartialOrd,
534    PartialEq,
535    Eq,
536    Ord,
537    Hash,
538    Arbitrary
539)]
540pub struct ClusterHydrationBurstV1 {
541    pub cluster_id: String,
542    pub cluster_name: String,
543    pub transition: HydrationBurstLifecycleV1,
544    /// On a `finished` transition: why the burst tore down. `None` on `started`.
545    #[serde(default)]
546    pub finish_cause: Option<BurstFinishCauseV1>,
547    /// The size of the burst replica the record runs.
548    pub burst_size: String,
549}
550
551#[derive(
552    Clone,
553    Debug,
554    Serialize,
555    Deserialize,
556    PartialOrd,
557    PartialEq,
558    Eq,
559    Ord,
560    Hash,
561    Arbitrary
562)]
563pub struct RenameItemV1 {
564    pub id: String,
565    pub old_name: FullNameV1,
566    pub new_name: FullNameV1,
567}
568
569#[derive(
570    Clone,
571    Debug,
572    Serialize,
573    Deserialize,
574    PartialOrd,
575    PartialEq,
576    Eq,
577    Ord,
578    Hash,
579    Arbitrary
580)]
581pub struct RenameClusterV1 {
582    pub id: String,
583    pub old_name: String,
584    pub new_name: String,
585}
586
587#[derive(
588    Clone,
589    Debug,
590    Serialize,
591    Deserialize,
592    PartialOrd,
593    PartialEq,
594    Eq,
595    Ord,
596    Hash,
597    Arbitrary
598)]
599pub struct RenameClusterReplicaV1 {
600    pub cluster_id: String,
601    pub replica_id: String,
602    pub old_name: String,
603    pub new_name: String,
604}
605
606#[derive(
607    Clone,
608    Debug,
609    Serialize,
610    Deserialize,
611    PartialOrd,
612    PartialEq,
613    Eq,
614    Ord,
615    Hash,
616    Arbitrary
617)]
618pub struct DropClusterReplicaV1 {
619    pub cluster_id: String,
620    pub cluster_name: String,
621    // Events that predate v0.32.0 will not have this field set.
622    #[serde(skip_serializing_if = "Option::is_none")]
623    pub replica_id: Option<String>,
624    pub replica_name: String,
625}
626
627#[derive(
628    Clone,
629    Debug,
630    Serialize,
631    Deserialize,
632    PartialOrd,
633    PartialEq,
634    Eq,
635    Ord,
636    Hash,
637    Arbitrary
638)]
639pub struct DropClusterReplicaV2 {
640    pub cluster_id: String,
641    pub cluster_name: String,
642    pub replica_id: Option<String>,
643    pub replica_name: String,
644    pub reason: CreateOrDropClusterReplicaReasonV1,
645    #[serde(skip_serializing_if = "Option::is_none")]
646    pub scheduling_policies: Option<SchedulingDecisionsWithReasonsV1>,
647}
648
649#[derive(
650    Clone,
651    Debug,
652    Serialize,
653    Deserialize,
654    PartialOrd,
655    PartialEq,
656    Eq,
657    Ord,
658    Hash,
659    Arbitrary
660)]
661pub struct DropClusterReplicaV3 {
662    pub cluster_id: String,
663    pub cluster_name: String,
664    pub replica_id: Option<String>,
665    pub replica_name: String,
666    pub reason: CreateOrDropClusterReplicaReasonV1,
667    #[serde(skip_serializing_if = "Option::is_none")]
668    pub scheduling_policies: Option<SchedulingDecisionsWithReasonsV2>,
669}
670
671#[derive(
672    Clone,
673    Debug,
674    Serialize,
675    Deserialize,
676    PartialOrd,
677    PartialEq,
678    Eq,
679    Ord,
680    Hash,
681    Arbitrary
682)]
683pub struct CreateClusterReplicaV1 {
684    pub cluster_id: String,
685    pub cluster_name: String,
686    // Events that predate v0.32.0 will not have this field set.
687    #[serde(skip_serializing_if = "Option::is_none")]
688    pub replica_id: Option<String>,
689    pub replica_name: String,
690    pub logical_size: String,
691    pub disk: bool,
692    pub billed_as: Option<String>,
693    pub internal: bool,
694}
695
696#[derive(
697    Clone,
698    Debug,
699    Serialize,
700    Deserialize,
701    PartialOrd,
702    PartialEq,
703    Eq,
704    Ord,
705    Hash,
706    Arbitrary
707)]
708pub struct CreateClusterReplicaV2 {
709    pub cluster_id: String,
710    pub cluster_name: String,
711    pub replica_id: Option<String>,
712    pub replica_name: String,
713    pub logical_size: String,
714    pub disk: bool,
715    pub billed_as: Option<String>,
716    pub internal: bool,
717    pub reason: CreateOrDropClusterReplicaReasonV1,
718    #[serde(skip_serializing_if = "Option::is_none")]
719    pub scheduling_policies: Option<SchedulingDecisionsWithReasonsV1>,
720}
721
722#[derive(
723    Clone,
724    Debug,
725    Serialize,
726    Deserialize,
727    PartialOrd,
728    PartialEq,
729    Eq,
730    Ord,
731    Hash,
732    Arbitrary
733)]
734pub struct CreateClusterReplicaV3 {
735    pub cluster_id: String,
736    pub cluster_name: String,
737    pub replica_id: Option<String>,
738    pub replica_name: String,
739    pub logical_size: String,
740    pub disk: bool,
741    pub billed_as: Option<String>,
742    pub internal: bool,
743    pub reason: CreateOrDropClusterReplicaReasonV1,
744    #[serde(skip_serializing_if = "Option::is_none")]
745    pub scheduling_policies: Option<SchedulingDecisionsWithReasonsV2>,
746}
747
748#[derive(
749    Clone,
750    Debug,
751    Serialize,
752    Deserialize,
753    PartialOrd,
754    PartialEq,
755    Eq,
756    Ord,
757    Hash,
758    Arbitrary
759)]
760pub struct CreateClusterReplicaV4 {
761    pub cluster_id: String,
762    pub cluster_name: String,
763    pub replica_id: Option<String>,
764    pub replica_name: String,
765    pub logical_size: String,
766    pub billed_as: Option<String>,
767    pub internal: bool,
768    pub reason: CreateOrDropClusterReplicaReasonV1,
769    #[serde(skip_serializing_if = "Option::is_none")]
770    pub scheduling_policies: Option<SchedulingDecisionsWithReasonsV2>,
771}
772
773#[derive(
774    Clone,
775    Debug,
776    Serialize,
777    Deserialize,
778    PartialOrd,
779    PartialEq,
780    Eq,
781    Ord,
782    Hash,
783    Arbitrary
784)]
785#[serde(rename_all = "kebab-case")]
786pub enum CreateOrDropClusterReplicaReasonV1 {
787    Manual,
788    Schedule,
789    System,
790    /// The cluster controller's graceful-reconfiguration strategy created the
791    /// replica while converging a cluster onto an in-flight `reconfiguration`
792    /// target (a background `ALTER CLUSTER`).
793    Reconfiguration,
794    /// The cluster controller's hydration-burst strategy created the
795    /// transient burst replica it runs while a cluster's objects are not yet
796    /// hydrated.
797    HydrationBurst,
798    /// The cluster controller dropped the replica because the cluster's
799    /// configuration no longer calls for it. NOTE: a replication-factor
800    /// decrease drop reads `retired` even though the config change itself was
801    /// user-initiated.
802    Retired,
803}
804
805/// The reason for the automated cluster scheduling to turn a cluster On or Off. Each existing
806/// policy's On/Off opinion should be recorded, along with their reasons. (Among the reasons there
807/// can be settings of the policy as well as other information about the state of the system.)
808#[derive(
809    Clone,
810    Debug,
811    Serialize,
812    Deserialize,
813    PartialOrd,
814    PartialEq,
815    Eq,
816    Ord,
817    Hash,
818    Arbitrary
819)]
820pub struct SchedulingDecisionsWithReasonsV1 {
821    /// The reason for the refresh policy for wanting to turn a cluster On or Off.
822    pub on_refresh: RefreshDecisionWithReasonV1,
823}
824
825/// The reason for the automated cluster scheduling to turn a cluster On or Off. Each existing
826/// policy's On/Off opinion should be recorded, along with their reasons. (Among the reasons there
827/// can be settings of the policy as well as other information about the state of the system.)
828#[derive(
829    Clone,
830    Debug,
831    Serialize,
832    Deserialize,
833    PartialOrd,
834    PartialEq,
835    Eq,
836    Ord,
837    Hash,
838    Arbitrary
839)]
840pub struct SchedulingDecisionsWithReasonsV2 {
841    /// The reason for the refresh policy for wanting to turn a cluster On or Off.
842    pub on_refresh: RefreshDecisionWithReasonV2,
843}
844
845#[derive(
846    Clone,
847    Debug,
848    Serialize,
849    Deserialize,
850    PartialOrd,
851    PartialEq,
852    Eq,
853    Ord,
854    Hash,
855    Arbitrary
856)]
857pub struct RefreshDecisionWithReasonV1 {
858    pub decision: SchedulingDecisionV1,
859    /// Objects that currently need a refresh on the cluster (taking into account the rehydration
860    /// time estimate).
861    pub objects_needing_refresh: Vec<String>,
862    /// The HYDRATION TIME ESTIMATE setting of the cluster.
863    pub hydration_time_estimate: String,
864}
865
866#[derive(
867    Clone,
868    Debug,
869    Serialize,
870    Deserialize,
871    PartialOrd,
872    PartialEq,
873    Eq,
874    Ord,
875    Hash,
876    Arbitrary
877)]
878pub struct RefreshDecisionWithReasonV2 {
879    pub decision: SchedulingDecisionV1,
880    /// Objects that currently need a refresh on the cluster (taking into account the rehydration
881    /// time estimate), and therefore should keep the cluster On.
882    pub objects_needing_refresh: Vec<String>,
883    /// Objects for which we estimate that they currently need Persist compaction, and therefore
884    /// should keep the cluster On.
885    pub objects_needing_compaction: Vec<String>,
886    /// The HYDRATION TIME ESTIMATE setting of the cluster.
887    pub hydration_time_estimate: String,
888}
889
890#[derive(
891    Clone,
892    Debug,
893    Serialize,
894    Deserialize,
895    PartialOrd,
896    PartialEq,
897    Eq,
898    Ord,
899    Hash,
900    Arbitrary
901)]
902#[serde(rename_all = "kebab-case")]
903pub enum SchedulingDecisionV1 {
904    On,
905    Off,
906}
907
908impl From<bool> for SchedulingDecisionV1 {
909    fn from(value: bool) -> Self {
910        match value {
911            true => SchedulingDecisionV1::On,
912            false => SchedulingDecisionV1::Off,
913        }
914    }
915}
916
917#[derive(
918    Clone,
919    Debug,
920    Serialize,
921    Deserialize,
922    PartialOrd,
923    PartialEq,
924    Eq,
925    Ord,
926    Hash,
927    Arbitrary
928)]
929pub struct CreateSourceSinkV1 {
930    pub id: String,
931    #[serde(flatten)]
932    pub name: FullNameV1,
933    pub size: Option<String>,
934}
935
936#[derive(
937    Clone,
938    Debug,
939    Serialize,
940    Deserialize,
941    PartialOrd,
942    PartialEq,
943    Eq,
944    Ord,
945    Hash,
946    Arbitrary
947)]
948pub struct CreateSourceSinkV2 {
949    pub id: String,
950    #[serde(flatten)]
951    pub name: FullNameV1,
952    pub size: Option<String>,
953    #[serde(rename = "type")]
954    pub external_type: String,
955}
956
957#[derive(
958    Clone,
959    Debug,
960    Serialize,
961    Deserialize,
962    PartialOrd,
963    PartialEq,
964    Eq,
965    Ord,
966    Hash,
967    Arbitrary
968)]
969pub struct CreateSourceSinkV3 {
970    pub id: String,
971    #[serde(flatten)]
972    pub name: FullNameV1,
973    #[serde(rename = "type")]
974    pub external_type: String,
975}
976
977#[derive(
978    Clone,
979    Debug,
980    Serialize,
981    Deserialize,
982    PartialOrd,
983    PartialEq,
984    Eq,
985    Ord,
986    Hash,
987    Arbitrary
988)]
989pub struct CreateSourceSinkV4 {
990    pub id: String,
991    pub cluster_id: Option<String>,
992    #[serde(flatten)]
993    pub name: FullNameV1,
994    #[serde(rename = "type")]
995    pub external_type: String,
996}
997
998#[derive(
999    Clone,
1000    Debug,
1001    Serialize,
1002    Deserialize,
1003    PartialOrd,
1004    PartialEq,
1005    Eq,
1006    Ord,
1007    Hash,
1008    Arbitrary
1009)]
1010pub struct CreateIndexV1 {
1011    pub id: String,
1012    pub cluster_id: String,
1013    #[serde(flatten)]
1014    pub name: FullNameV1,
1015}
1016
1017#[derive(
1018    Clone,
1019    Debug,
1020    Serialize,
1021    Deserialize,
1022    PartialOrd,
1023    PartialEq,
1024    Eq,
1025    Ord,
1026    Hash,
1027    Arbitrary
1028)]
1029pub struct CreateMaterializedViewV1 {
1030    pub id: String,
1031    pub cluster_id: String,
1032    #[serde(flatten)]
1033    pub name: FullNameV1,
1034    #[serde(skip_serializing_if = "Option::is_none")]
1035    pub replacement_target_id: Option<String>,
1036}
1037
1038#[derive(
1039    Clone,
1040    Debug,
1041    Serialize,
1042    Deserialize,
1043    PartialOrd,
1044    PartialEq,
1045    Eq,
1046    Ord,
1047    Hash,
1048    Arbitrary
1049)]
1050pub struct AlterApplyReplacementV1 {
1051    #[serde(flatten)]
1052    pub target: IdFullNameV1,
1053    pub replacement: IdFullNameV1,
1054}
1055
1056#[derive(
1057    Clone,
1058    Debug,
1059    Serialize,
1060    Deserialize,
1061    PartialOrd,
1062    PartialEq,
1063    Eq,
1064    Ord,
1065    Hash,
1066    Arbitrary
1067)]
1068pub struct AlterSourceSinkV1 {
1069    pub id: String,
1070    #[serde(flatten)]
1071    pub name: FullNameV1,
1072    pub old_size: Option<String>,
1073    pub new_size: Option<String>,
1074}
1075
1076#[derive(
1077    Clone,
1078    Debug,
1079    Serialize,
1080    Deserialize,
1081    PartialOrd,
1082    PartialEq,
1083    Eq,
1084    Ord,
1085    Hash,
1086    Arbitrary
1087)]
1088pub struct AlterSetClusterV1 {
1089    pub id: String,
1090    #[serde(flatten)]
1091    pub name: FullNameV1,
1092    pub old_cluster_id: String,
1093    pub new_cluster_id: String,
1094}
1095
1096#[derive(
1097    Clone,
1098    Debug,
1099    Serialize,
1100    Deserialize,
1101    PartialOrd,
1102    PartialEq,
1103    Eq,
1104    Ord,
1105    Hash,
1106    Arbitrary
1107)]
1108pub struct GrantRoleV1 {
1109    pub role_id: String,
1110    pub member_id: String,
1111    pub grantor_id: String,
1112}
1113
1114#[derive(
1115    Clone,
1116    Debug,
1117    Serialize,
1118    Deserialize,
1119    PartialOrd,
1120    PartialEq,
1121    Eq,
1122    Ord,
1123    Hash,
1124    Arbitrary
1125)]
1126pub struct GrantRoleV2 {
1127    pub role_id: String,
1128    pub member_id: String,
1129    pub grantor_id: String,
1130    pub executed_by: String,
1131}
1132
1133#[derive(
1134    Clone,
1135    Debug,
1136    Serialize,
1137    Deserialize,
1138    PartialOrd,
1139    PartialEq,
1140    Eq,
1141    Ord,
1142    Hash,
1143    Arbitrary
1144)]
1145pub struct RevokeRoleV1 {
1146    pub role_id: String,
1147    pub member_id: String,
1148}
1149
1150#[derive(
1151    Clone,
1152    Debug,
1153    Serialize,
1154    Deserialize,
1155    PartialOrd,
1156    PartialEq,
1157    Eq,
1158    Ord,
1159    Hash,
1160    Arbitrary
1161)]
1162pub struct RevokeRoleV2 {
1163    pub role_id: String,
1164    pub member_id: String,
1165    pub grantor_id: String,
1166    pub executed_by: String,
1167}
1168
1169#[derive(
1170    Clone,
1171    Debug,
1172    Serialize,
1173    Deserialize,
1174    PartialOrd,
1175    PartialEq,
1176    Eq,
1177    Ord,
1178    Hash,
1179    Arbitrary
1180)]
1181pub struct UpdatePrivilegeV1 {
1182    pub object_id: String,
1183    pub grantee_id: String,
1184    pub grantor_id: String,
1185    pub privileges: String,
1186}
1187
1188#[derive(
1189    Clone,
1190    Debug,
1191    Serialize,
1192    Deserialize,
1193    PartialOrd,
1194    PartialEq,
1195    Eq,
1196    Ord,
1197    Hash,
1198    Arbitrary
1199)]
1200pub struct AlterDefaultPrivilegeV1 {
1201    pub role_id: String,
1202    pub database_id: Option<String>,
1203    pub schema_id: Option<String>,
1204    pub grantee_id: String,
1205    pub privileges: String,
1206}
1207
1208#[derive(
1209    Clone,
1210    Debug,
1211    Serialize,
1212    Deserialize,
1213    PartialOrd,
1214    PartialEq,
1215    Eq,
1216    Ord,
1217    Hash,
1218    Arbitrary
1219)]
1220pub struct UpdateOwnerV1 {
1221    pub object_id: String,
1222    pub old_owner_id: String,
1223    pub new_owner_id: String,
1224}
1225
1226#[derive(
1227    Clone,
1228    Debug,
1229    Serialize,
1230    Deserialize,
1231    PartialOrd,
1232    PartialEq,
1233    Eq,
1234    Ord,
1235    Hash,
1236    Arbitrary
1237)]
1238pub struct SchemaV1 {
1239    pub id: String,
1240    pub name: String,
1241    pub database_name: String,
1242}
1243
1244#[derive(
1245    Clone,
1246    Debug,
1247    Serialize,
1248    Deserialize,
1249    PartialOrd,
1250    PartialEq,
1251    Eq,
1252    Ord,
1253    Hash,
1254    Arbitrary
1255)]
1256pub struct SchemaV2 {
1257    pub id: String,
1258    pub name: String,
1259    pub database_name: Option<String>,
1260}
1261
1262#[derive(
1263    Clone,
1264    Debug,
1265    Serialize,
1266    Deserialize,
1267    PartialOrd,
1268    PartialEq,
1269    Eq,
1270    Ord,
1271    Hash,
1272    Arbitrary
1273)]
1274pub struct RenameSchemaV1 {
1275    pub id: String,
1276    pub database_name: Option<String>,
1277    pub old_name: String,
1278    pub new_name: String,
1279}
1280
1281#[derive(
1282    Clone,
1283    Debug,
1284    Serialize,
1285    Deserialize,
1286    PartialOrd,
1287    PartialEq,
1288    Eq,
1289    Ord,
1290    Hash,
1291    Arbitrary
1292)]
1293pub struct AlterRetainHistoryV1 {
1294    pub id: String,
1295    pub old_history: Option<String>,
1296    pub new_history: Option<String>,
1297}
1298
1299#[derive(
1300    Clone,
1301    Debug,
1302    Serialize,
1303    Deserialize,
1304    PartialOrd,
1305    PartialEq,
1306    Eq,
1307    Ord,
1308    Hash,
1309    Arbitrary
1310)]
1311pub struct AlterAddColumnV1 {
1312    pub id: String,
1313    pub column: String,
1314    pub column_type: String,
1315    pub nullable: bool,
1316}
1317
1318#[derive(
1319    Clone,
1320    Debug,
1321    Serialize,
1322    Deserialize,
1323    PartialOrd,
1324    PartialEq,
1325    Eq,
1326    Ord,
1327    Hash,
1328    Arbitrary
1329)]
1330pub struct AlterSourceTimestampIntervalV1 {
1331    pub id: String,
1332    pub old_interval: Option<String>,
1333    pub new_interval: Option<String>,
1334}
1335
1336#[derive(
1337    Clone,
1338    Debug,
1339    Serialize,
1340    Deserialize,
1341    PartialOrd,
1342    PartialEq,
1343    Eq,
1344    Ord,
1345    Hash,
1346    Arbitrary
1347)]
1348pub struct UpdateItemV1 {
1349    pub id: String,
1350    #[serde(flatten)]
1351    pub name: FullNameV1,
1352}
1353
1354#[derive(
1355    Clone,
1356    Debug,
1357    Serialize,
1358    Deserialize,
1359    PartialOrd,
1360    PartialEq,
1361    Eq,
1362    Ord,
1363    Hash,
1364    Arbitrary
1365)]
1366pub struct ToNewIdV1 {
1367    pub id: String,
1368    pub new_id: String,
1369}
1370
1371#[derive(
1372    Clone,
1373    Debug,
1374    Serialize,
1375    Deserialize,
1376    PartialOrd,
1377    PartialEq,
1378    Eq,
1379    Ord,
1380    Hash,
1381    Arbitrary
1382)]
1383pub struct FromPreviousIdV1 {
1384    pub id: String,
1385    pub previous_id: String,
1386}
1387
1388impl EventDetails {
1389    pub fn as_json(&self) -> serde_json::Value {
1390        match self {
1391            EventDetails::CreateClusterReplicaV1(v) => {
1392                serde_json::to_value(v).expect("must serialize")
1393            }
1394            EventDetails::CreateClusterReplicaV2(v) => {
1395                serde_json::to_value(v).expect("must serialize")
1396            }
1397            EventDetails::CreateClusterReplicaV3(v) => {
1398                serde_json::to_value(v).expect("must serialize")
1399            }
1400            EventDetails::CreateClusterReplicaV4(v) => {
1401                serde_json::to_value(v).expect("must serialize")
1402            }
1403            EventDetails::DropClusterReplicaV1(v) => {
1404                serde_json::to_value(v).expect("must serialize")
1405            }
1406            EventDetails::DropClusterReplicaV2(v) => {
1407                serde_json::to_value(v).expect("must serialize")
1408            }
1409            EventDetails::DropClusterReplicaV3(v) => {
1410                serde_json::to_value(v).expect("must serialize")
1411            }
1412            EventDetails::IdFullNameV1(v) => serde_json::to_value(v).expect("must serialize"),
1413            EventDetails::RenameClusterV1(v) => serde_json::to_value(v).expect("must serialize"),
1414            EventDetails::RenameClusterReplicaV1(v) => {
1415                serde_json::to_value(v).expect("must serialize")
1416            }
1417            EventDetails::AlterClusterReconfigurationV1(v) => {
1418                serde_json::to_value(v).expect("must serialize")
1419            }
1420            EventDetails::ClusterHydrationBurstV1(v) => {
1421                serde_json::to_value(v).expect("must serialize")
1422            }
1423            EventDetails::RenameItemV1(v) => serde_json::to_value(v).expect("must serialize"),
1424            EventDetails::IdNameV1(v) => serde_json::to_value(v).expect("must serialize"),
1425            EventDetails::SchemaV1(v) => serde_json::to_value(v).expect("must serialize"),
1426            EventDetails::SchemaV2(v) => serde_json::to_value(v).expect("must serialize"),
1427            EventDetails::RenameSchemaV1(v) => serde_json::to_value(v).expect("must serialize"),
1428            EventDetails::CreateSourceSinkV1(v) => serde_json::to_value(v).expect("must serialize"),
1429            EventDetails::CreateSourceSinkV2(v) => serde_json::to_value(v).expect("must serialize"),
1430            EventDetails::CreateSourceSinkV3(v) => serde_json::to_value(v).expect("must serialize"),
1431            EventDetails::CreateSourceSinkV4(v) => serde_json::to_value(v).expect("must serialize"),
1432            EventDetails::CreateIndexV1(v) => serde_json::to_value(v).expect("must serialize"),
1433            EventDetails::CreateMaterializedViewV1(v) => {
1434                serde_json::to_value(v).expect("must serialize")
1435            }
1436            EventDetails::AlterApplyReplacementV1(v) => {
1437                serde_json::to_value(v).expect("must serialize")
1438            }
1439            EventDetails::AlterSourceSinkV1(v) => serde_json::to_value(v).expect("must serialize"),
1440            EventDetails::AlterSetClusterV1(v) => serde_json::to_value(v).expect("must serialize"),
1441            EventDetails::GrantRoleV1(v) => serde_json::to_value(v).expect("must serialize"),
1442            EventDetails::GrantRoleV2(v) => serde_json::to_value(v).expect("must serialize"),
1443            EventDetails::RevokeRoleV1(v) => serde_json::to_value(v).expect("must serialize"),
1444            EventDetails::RevokeRoleV2(v) => serde_json::to_value(v).expect("must serialize"),
1445            EventDetails::UpdatePrivilegeV1(v) => serde_json::to_value(v).expect("must serialize"),
1446            EventDetails::AlterDefaultPrivilegeV1(v) => {
1447                serde_json::to_value(v).expect("must serialize")
1448            }
1449            EventDetails::UpdateOwnerV1(v) => serde_json::to_value(v).expect("must serialize"),
1450            EventDetails::UpdateItemV1(v) => serde_json::to_value(v).expect("must serialize"),
1451            EventDetails::AlterRetainHistoryV1(v) => {
1452                serde_json::to_value(v).expect("must serialize")
1453            }
1454            EventDetails::AlterAddColumnV1(v) => serde_json::to_value(v).expect("must serialize"),
1455            EventDetails::AlterSourceTimestampIntervalV1(v) => {
1456                serde_json::to_value(v).expect("must serialize")
1457            }
1458            EventDetails::ToNewIdV1(v) => serde_json::to_value(v).expect("must serialize"),
1459            EventDetails::FromPreviousIdV1(v) => serde_json::to_value(v).expect("must serialize"),
1460            EventDetails::SetV1(v) => serde_json::to_value(v).expect("must serialize"),
1461            EventDetails::ResetAllV1 => serde_json::Value::Null,
1462            EventDetails::RotateKeysV1(v) => serde_json::to_value(v).expect("must serialize"),
1463            EventDetails::CreateRoleV1(v) => serde_json::to_value(v).expect("must serialize"),
1464        }
1465    }
1466}
1467
1468#[derive(
1469    Clone,
1470    Debug,
1471    Serialize,
1472    Deserialize,
1473    PartialOrd,
1474    PartialEq,
1475    Eq,
1476    Ord,
1477    Hash,
1478    Arbitrary
1479)]
1480pub struct EventV1 {
1481    pub id: u64,
1482    pub event_type: EventType,
1483    pub object_type: ObjectType,
1484    pub details: EventDetails,
1485    pub user: Option<String>,
1486    pub occurred_at: EpochMillis,
1487}
1488
1489impl EventV1 {
1490    fn new(
1491        id: u64,
1492        event_type: EventType,
1493        object_type: ObjectType,
1494        details: EventDetails,
1495        user: Option<String>,
1496        occurred_at: EpochMillis,
1497    ) -> EventV1 {
1498        EventV1 {
1499            id,
1500            event_type,
1501            object_type,
1502            details,
1503            user,
1504            occurred_at,
1505        }
1506    }
1507}
1508
1509#[derive(
1510    Clone,
1511    Debug,
1512    Serialize,
1513    Deserialize,
1514    PartialOrd,
1515    PartialEq,
1516    Eq,
1517    Ord,
1518    Hash,
1519    Arbitrary
1520)]
1521pub struct StorageUsageV1 {
1522    pub id: u64,
1523    pub shard_id: Option<String>,
1524    pub size_bytes: u64,
1525    pub collection_timestamp: EpochMillis,
1526}
1527
1528impl StorageUsageV1 {
1529    pub fn new(
1530        id: u64,
1531        shard_id: Option<String>,
1532        size_bytes: u64,
1533        collection_timestamp: EpochMillis,
1534    ) -> StorageUsageV1 {
1535        StorageUsageV1 {
1536            id,
1537            shard_id,
1538            size_bytes,
1539            collection_timestamp,
1540        }
1541    }
1542}
1543
1544/// Describes the environment's storage usage at a point in time.
1545///
1546/// This type is persisted in the catalog across restarts, so any updates to the
1547/// schema will require a new version.
1548#[derive(
1549    Clone,
1550    Debug,
1551    Serialize,
1552    Deserialize,
1553    PartialOrd,
1554    PartialEq,
1555    Eq,
1556    Ord,
1557    Hash,
1558    Arbitrary
1559)]
1560pub enum VersionedStorageUsage {
1561    V1(StorageUsageV1),
1562}
1563
1564impl VersionedStorageUsage {
1565    /// Create a new metric snapshot.
1566    /// This function must always require and produce the most
1567    /// recent variant of VersionedStorageMetrics.
1568    pub fn new(
1569        id: u64,
1570        object_id: Option<String>,
1571        size_bytes: u64,
1572        collection_timestamp: EpochMillis,
1573    ) -> Self {
1574        Self::V1(StorageUsageV1::new(
1575            id,
1576            object_id,
1577            size_bytes,
1578            collection_timestamp,
1579        ))
1580    }
1581
1582    // Implement deserialize and serialize so writers and readers don't have to
1583    // coordinate about which Serializer to use.
1584    pub fn deserialize(data: &[u8]) -> Result<Self, anyhow::Error> {
1585        Ok(serde_json::from_slice(data)?)
1586    }
1587
1588    pub fn serialize(&self) -> Vec<u8> {
1589        serde_json::to_vec(self).expect("must serialize")
1590    }
1591
1592    pub fn timestamp(&self) -> EpochMillis {
1593        match self {
1594            VersionedStorageUsage::V1(StorageUsageV1 {
1595                collection_timestamp,
1596                ..
1597            }) => *collection_timestamp,
1598        }
1599    }
1600
1601    /// Returns a globally sortable event order. All event versions must have this
1602    /// field.
1603    pub fn sortable_id(&self) -> u64 {
1604        match self {
1605            VersionedStorageUsage::V1(usage) => usage.id,
1606        }
1607    }
1608}
1609
1610#[cfg(test)]
1611mod tests {
1612    use crate::{EventDetails, EventType, EventV1, IdNameV1, ObjectType, VersionedEvent};
1613
1614    // Test all versions of events. This test hard codes bytes so that
1615    // programmers are not able to change data structures here without this test
1616    // failing. Instead of changing data structures, add new variants.
1617    #[mz_ore::test]
1618    fn test_audit_log() -> Result<(), anyhow::Error> {
1619        let cases: Vec<(VersionedEvent, &'static str)> = vec![(
1620            VersionedEvent::V1(EventV1::new(
1621                2,
1622                EventType::Drop,
1623                ObjectType::ClusterReplica,
1624                EventDetails::IdNameV1(IdNameV1 {
1625                    id: "u1".to_string(),
1626                    name: "name".into(),
1627                }),
1628                None,
1629                2,
1630            )),
1631            r#"{"V1":{"id":2,"event_type":"drop","object_type":"cluster-replica","details":{"IdNameV1":{"id":"u1","name":"name"}},"user":null,"occurred_at":2}}"#,
1632        )];
1633
1634        for (event, expected_bytes) in cases {
1635            let event_bytes = serde_json::to_vec(&event).unwrap();
1636            assert_eq!(
1637                event_bytes,
1638                expected_bytes.as_bytes(),
1639                "expected bytes {}, got {}",
1640                expected_bytes,
1641                std::str::from_utf8(&event_bytes).unwrap(),
1642            );
1643        }
1644
1645        Ok(())
1646    }
1647}
mz_audit_log/lib.rs

mz_audit_log/
lib.rs
