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

mz_audit_log/
lib.rs
