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 cleared:
372    /// a hydrated success under either `ON TIMEOUT` action (including one that
373    /// hydrated after the deadline, since success takes precedence) or a
374    /// forced `ON TIMEOUT COMMIT` cut-over of a not-yet-hydrated target past
375    /// the deadline. The event carries the record's deadline; comparing it to
376    /// the event's occurrence time tells an in-time cut-over from a late or
377    /// forced one.
378    Finalized,
379    /// The deadline fired with the target not hydrated under `ON TIMEOUT
380    /// ROLLBACK`: the record was cleared with the realized config untouched and
381    /// the target replicas dropped, reverting to the pre-reconfiguration set.
382    /// Emitted exactly once per timeout. The clear is durable, so the
383    /// transition cannot re-fire. This event is the timeout's papertrail; the
384    /// record (and with it the abandoned target) is gone from the catalog.
385    TimedOut,
386    /// An in-flight reconfiguration was cancelled by re-targeting the record
387    /// back to the cluster's still-realized shape (the ALTER-back cancel path);
388    /// the controller drops the in-flight target replicas and clears the record.
389    Cancelled,
390}
391
392/// A cluster-level transition in a background reconfiguration's lifecycle.
393///
394/// `deadline` is the reconfiguration's active deadline as a millisecond
395/// `mz_timestamp`, recorded on every transition so an operator can correlate
396/// the transition with the originating `ALTER`: on `started` and `cancelled`
397/// the written/re-targeted record's deadline, on `timed-out` and `finalized`
398/// the just-cleared record's. On `finalized`, comparing it to the event's
399/// occurrence time distinguishes an in-time cut-over from a late or forced
400/// (`ON TIMEOUT COMMIT`) one.
401#[derive(
402    Clone,
403    Debug,
404    Serialize,
405    Deserialize,
406    PartialOrd,
407    PartialEq,
408    Eq,
409    Ord,
410    Hash,
411    Arbitrary
412)]
413pub struct AlterClusterReconfigurationV1 {
414    pub cluster_id: String,
415    pub cluster_name: String,
416    pub transition: ReconfigurationLifecycleV1,
417    pub target_size: String,
418    pub target_replication_factor: u32,
419    pub target_availability_zones: Vec<String>,
420    pub target_logging: ClusterReplicaLoggingV1,
421    pub deadline: Option<u64>,
422}
423
424/// A managed cluster's introspection-logging config, recorded on a
425/// reconfiguration event so the papertrail captures an introspection-only
426/// `ALTER` (which otherwise leaves `target_size` and
427/// `target_replication_factor` unchanged from the realized shape). Mirrors the
428/// durable `ReplicaLogging`: `log_logging` is `INTROSPECTION DEBUGGING`,
429/// `interval` is `INTROSPECTION INTERVAL` (`None` disables introspection).
430#[derive(
431    Clone,
432    Debug,
433    Serialize,
434    Deserialize,
435    PartialOrd,
436    PartialEq,
437    Eq,
438    Ord,
439    Hash,
440    Arbitrary
441)]
442pub struct ClusterReplicaLoggingV1 {
443    pub log_logging: bool,
444    pub interval: Option<Duration>,
445}
446
447#[derive(
448    Clone,
449    Debug,
450    Serialize,
451    Deserialize,
452    PartialOrd,
453    PartialEq,
454    Eq,
455    Ord,
456    Hash,
457    Arbitrary
458)]
459pub struct CreateRoleV1 {
460    pub id: String,
461    pub name: String,
462    pub auto_provision_source: Option<String>,
463}
464
465/// A transition in the lifecycle of a hydration burst (a `burst` record on a
466/// managed cluster), recorded so an operator can trace a controller-initiated
467/// burst from start to teardown.
468///
469/// The burst replica's create and drop are recorded separately, carrying
470/// [`CreateOrDropClusterReplicaReasonV1::HydrationBurst`]; this event family
471/// records the cluster-level transitions those replica lifecycle events hang off.
472#[derive(
473    Clone,
474    Debug,
475    Serialize,
476    Deserialize,
477    PartialOrd,
478    PartialEq,
479    Eq,
480    Ord,
481    Hash,
482    Arbitrary
483)]
484#[serde(rename_all = "kebab-case")]
485pub enum HydrationBurstLifecycleV1 {
486    /// A `burst` record was written: the controller is now running a burst
487    /// replica to accelerate hydration.
488    Started,
489    /// The `burst` record was cleared: the burst replica is torn down (its linger
490    /// elapsed after the steady set hydrated, or the burst is no longer warranted).
491    Finished,
492}
493
494/// A cluster-level transition in a hydration burst's lifecycle.
495#[derive(
496    Clone,
497    Debug,
498    Serialize,
499    Deserialize,
500    PartialOrd,
501    PartialEq,
502    Eq,
503    Ord,
504    Hash,
505    Arbitrary
506)]
507pub struct ClusterHydrationBurstV1 {
508    pub cluster_id: String,
509    pub cluster_name: String,
510    pub transition: HydrationBurstLifecycleV1,
511    /// The size of the burst replica the record runs.
512    pub burst_size: String,
513}
514
515#[derive(
516    Clone,
517    Debug,
518    Serialize,
519    Deserialize,
520    PartialOrd,
521    PartialEq,
522    Eq,
523    Ord,
524    Hash,
525    Arbitrary
526)]
527pub struct RenameItemV1 {
528    pub id: String,
529    pub old_name: FullNameV1,
530    pub new_name: FullNameV1,
531}
532
533#[derive(
534    Clone,
535    Debug,
536    Serialize,
537    Deserialize,
538    PartialOrd,
539    PartialEq,
540    Eq,
541    Ord,
542    Hash,
543    Arbitrary
544)]
545pub struct RenameClusterV1 {
546    pub id: String,
547    pub old_name: String,
548    pub new_name: 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 RenameClusterReplicaV1 {
564    pub cluster_id: String,
565    pub replica_id: String,
566    pub old_name: String,
567    pub new_name: String,
568}
569
570#[derive(
571    Clone,
572    Debug,
573    Serialize,
574    Deserialize,
575    PartialOrd,
576    PartialEq,
577    Eq,
578    Ord,
579    Hash,
580    Arbitrary
581)]
582pub struct DropClusterReplicaV1 {
583    pub cluster_id: String,
584    pub cluster_name: String,
585    // Events that predate v0.32.0 will not have this field set.
586    #[serde(skip_serializing_if = "Option::is_none")]
587    pub replica_id: Option<String>,
588    pub replica_name: String,
589}
590
591#[derive(
592    Clone,
593    Debug,
594    Serialize,
595    Deserialize,
596    PartialOrd,
597    PartialEq,
598    Eq,
599    Ord,
600    Hash,
601    Arbitrary
602)]
603pub struct DropClusterReplicaV2 {
604    pub cluster_id: String,
605    pub cluster_name: String,
606    pub replica_id: Option<String>,
607    pub replica_name: String,
608    pub reason: CreateOrDropClusterReplicaReasonV1,
609    #[serde(skip_serializing_if = "Option::is_none")]
610    pub scheduling_policies: Option<SchedulingDecisionsWithReasonsV1>,
611}
612
613#[derive(
614    Clone,
615    Debug,
616    Serialize,
617    Deserialize,
618    PartialOrd,
619    PartialEq,
620    Eq,
621    Ord,
622    Hash,
623    Arbitrary
624)]
625pub struct DropClusterReplicaV3 {
626    pub cluster_id: String,
627    pub cluster_name: String,
628    pub replica_id: Option<String>,
629    pub replica_name: String,
630    pub reason: CreateOrDropClusterReplicaReasonV1,
631    #[serde(skip_serializing_if = "Option::is_none")]
632    pub scheduling_policies: Option<SchedulingDecisionsWithReasonsV2>,
633}
634
635#[derive(
636    Clone,
637    Debug,
638    Serialize,
639    Deserialize,
640    PartialOrd,
641    PartialEq,
642    Eq,
643    Ord,
644    Hash,
645    Arbitrary
646)]
647pub struct CreateClusterReplicaV1 {
648    pub cluster_id: String,
649    pub cluster_name: String,
650    // Events that predate v0.32.0 will not have this field set.
651    #[serde(skip_serializing_if = "Option::is_none")]
652    pub replica_id: Option<String>,
653    pub replica_name: String,
654    pub logical_size: String,
655    pub disk: bool,
656    pub billed_as: Option<String>,
657    pub internal: bool,
658}
659
660#[derive(
661    Clone,
662    Debug,
663    Serialize,
664    Deserialize,
665    PartialOrd,
666    PartialEq,
667    Eq,
668    Ord,
669    Hash,
670    Arbitrary
671)]
672pub struct CreateClusterReplicaV2 {
673    pub cluster_id: String,
674    pub cluster_name: String,
675    pub replica_id: Option<String>,
676    pub replica_name: String,
677    pub logical_size: String,
678    pub disk: bool,
679    pub billed_as: Option<String>,
680    pub internal: bool,
681    pub reason: CreateOrDropClusterReplicaReasonV1,
682    #[serde(skip_serializing_if = "Option::is_none")]
683    pub scheduling_policies: Option<SchedulingDecisionsWithReasonsV1>,
684}
685
686#[derive(
687    Clone,
688    Debug,
689    Serialize,
690    Deserialize,
691    PartialOrd,
692    PartialEq,
693    Eq,
694    Ord,
695    Hash,
696    Arbitrary
697)]
698pub struct CreateClusterReplicaV3 {
699    pub cluster_id: String,
700    pub cluster_name: String,
701    pub replica_id: Option<String>,
702    pub replica_name: String,
703    pub logical_size: String,
704    pub disk: bool,
705    pub billed_as: Option<String>,
706    pub internal: bool,
707    pub reason: CreateOrDropClusterReplicaReasonV1,
708    #[serde(skip_serializing_if = "Option::is_none")]
709    pub scheduling_policies: Option<SchedulingDecisionsWithReasonsV2>,
710}
711
712#[derive(
713    Clone,
714    Debug,
715    Serialize,
716    Deserialize,
717    PartialOrd,
718    PartialEq,
719    Eq,
720    Ord,
721    Hash,
722    Arbitrary
723)]
724pub struct CreateClusterReplicaV4 {
725    pub cluster_id: String,
726    pub cluster_name: String,
727    pub replica_id: Option<String>,
728    pub replica_name: String,
729    pub logical_size: String,
730    pub billed_as: Option<String>,
731    pub internal: bool,
732    pub reason: CreateOrDropClusterReplicaReasonV1,
733    #[serde(skip_serializing_if = "Option::is_none")]
734    pub scheduling_policies: Option<SchedulingDecisionsWithReasonsV2>,
735}
736
737#[derive(
738    Clone,
739    Debug,
740    Serialize,
741    Deserialize,
742    PartialOrd,
743    PartialEq,
744    Eq,
745    Ord,
746    Hash,
747    Arbitrary
748)]
749#[serde(rename_all = "kebab-case")]
750pub enum CreateOrDropClusterReplicaReasonV1 {
751    Manual,
752    Schedule,
753    System,
754    /// The cluster controller's graceful-reconfiguration strategy created the
755    /// replica while converging a cluster onto an in-flight `reconfiguration`
756    /// target (a background `ALTER CLUSTER`).
757    Reconfiguration,
758    /// The cluster controller's hydration-burst strategy created the
759    /// transient burst replica it runs while a cluster's objects are not yet
760    /// hydrated.
761    HydrationBurst,
762    /// The cluster controller dropped the replica because the cluster's
763    /// configuration no longer calls for it. NOTE: a replication-factor
764    /// decrease drop reads `retired` even though the config change itself was
765    /// user-initiated.
766    Retired,
767}
768
769/// The reason for the automated cluster scheduling to turn a cluster On or Off. Each existing
770/// policy's On/Off opinion should be recorded, along with their reasons. (Among the reasons there
771/// can be settings of the policy as well as other information about the state of the system.)
772#[derive(
773    Clone,
774    Debug,
775    Serialize,
776    Deserialize,
777    PartialOrd,
778    PartialEq,
779    Eq,
780    Ord,
781    Hash,
782    Arbitrary
783)]
784pub struct SchedulingDecisionsWithReasonsV1 {
785    /// The reason for the refresh policy for wanting to turn a cluster On or Off.
786    pub on_refresh: RefreshDecisionWithReasonV1,
787}
788
789/// The reason for the automated cluster scheduling to turn a cluster On or Off. Each existing
790/// policy's On/Off opinion should be recorded, along with their reasons. (Among the reasons there
791/// can be settings of the policy as well as other information about the state of the system.)
792#[derive(
793    Clone,
794    Debug,
795    Serialize,
796    Deserialize,
797    PartialOrd,
798    PartialEq,
799    Eq,
800    Ord,
801    Hash,
802    Arbitrary
803)]
804pub struct SchedulingDecisionsWithReasonsV2 {
805    /// The reason for the refresh policy for wanting to turn a cluster On or Off.
806    pub on_refresh: RefreshDecisionWithReasonV2,
807}
808
809#[derive(
810    Clone,
811    Debug,
812    Serialize,
813    Deserialize,
814    PartialOrd,
815    PartialEq,
816    Eq,
817    Ord,
818    Hash,
819    Arbitrary
820)]
821pub struct RefreshDecisionWithReasonV1 {
822    pub decision: SchedulingDecisionV1,
823    /// Objects that currently need a refresh on the cluster (taking into account the rehydration
824    /// time estimate).
825    pub objects_needing_refresh: Vec<String>,
826    /// The HYDRATION TIME ESTIMATE setting of the cluster.
827    pub hydration_time_estimate: String,
828}
829
830#[derive(
831    Clone,
832    Debug,
833    Serialize,
834    Deserialize,
835    PartialOrd,
836    PartialEq,
837    Eq,
838    Ord,
839    Hash,
840    Arbitrary
841)]
842pub struct RefreshDecisionWithReasonV2 {
843    pub decision: SchedulingDecisionV1,
844    /// Objects that currently need a refresh on the cluster (taking into account the rehydration
845    /// time estimate), and therefore should keep the cluster On.
846    pub objects_needing_refresh: Vec<String>,
847    /// Objects for which we estimate that they currently need Persist compaction, and therefore
848    /// should keep the cluster On.
849    pub objects_needing_compaction: Vec<String>,
850    /// The HYDRATION TIME ESTIMATE setting of the cluster.
851    pub hydration_time_estimate: String,
852}
853
854#[derive(
855    Clone,
856    Debug,
857    Serialize,
858    Deserialize,
859    PartialOrd,
860    PartialEq,
861    Eq,
862    Ord,
863    Hash,
864    Arbitrary
865)]
866#[serde(rename_all = "kebab-case")]
867pub enum SchedulingDecisionV1 {
868    On,
869    Off,
870}
871
872impl From<bool> for SchedulingDecisionV1 {
873    fn from(value: bool) -> Self {
874        match value {
875            true => SchedulingDecisionV1::On,
876            false => SchedulingDecisionV1::Off,
877        }
878    }
879}
880
881#[derive(
882    Clone,
883    Debug,
884    Serialize,
885    Deserialize,
886    PartialOrd,
887    PartialEq,
888    Eq,
889    Ord,
890    Hash,
891    Arbitrary
892)]
893pub struct CreateSourceSinkV1 {
894    pub id: String,
895    #[serde(flatten)]
896    pub name: FullNameV1,
897    pub size: Option<String>,
898}
899
900#[derive(
901    Clone,
902    Debug,
903    Serialize,
904    Deserialize,
905    PartialOrd,
906    PartialEq,
907    Eq,
908    Ord,
909    Hash,
910    Arbitrary
911)]
912pub struct CreateSourceSinkV2 {
913    pub id: String,
914    #[serde(flatten)]
915    pub name: FullNameV1,
916    pub size: Option<String>,
917    #[serde(rename = "type")]
918    pub external_type: String,
919}
920
921#[derive(
922    Clone,
923    Debug,
924    Serialize,
925    Deserialize,
926    PartialOrd,
927    PartialEq,
928    Eq,
929    Ord,
930    Hash,
931    Arbitrary
932)]
933pub struct CreateSourceSinkV3 {
934    pub id: String,
935    #[serde(flatten)]
936    pub name: FullNameV1,
937    #[serde(rename = "type")]
938    pub external_type: String,
939}
940
941#[derive(
942    Clone,
943    Debug,
944    Serialize,
945    Deserialize,
946    PartialOrd,
947    PartialEq,
948    Eq,
949    Ord,
950    Hash,
951    Arbitrary
952)]
953pub struct CreateSourceSinkV4 {
954    pub id: String,
955    pub cluster_id: Option<String>,
956    #[serde(flatten)]
957    pub name: FullNameV1,
958    #[serde(rename = "type")]
959    pub external_type: String,
960}
961
962#[derive(
963    Clone,
964    Debug,
965    Serialize,
966    Deserialize,
967    PartialOrd,
968    PartialEq,
969    Eq,
970    Ord,
971    Hash,
972    Arbitrary
973)]
974pub struct CreateIndexV1 {
975    pub id: String,
976    pub cluster_id: String,
977    #[serde(flatten)]
978    pub name: FullNameV1,
979}
980
981#[derive(
982    Clone,
983    Debug,
984    Serialize,
985    Deserialize,
986    PartialOrd,
987    PartialEq,
988    Eq,
989    Ord,
990    Hash,
991    Arbitrary
992)]
993pub struct CreateMaterializedViewV1 {
994    pub id: String,
995    pub cluster_id: String,
996    #[serde(flatten)]
997    pub name: FullNameV1,
998    #[serde(skip_serializing_if = "Option::is_none")]
999    pub replacement_target_id: Option<String>,
1000}
1001
1002#[derive(
1003    Clone,
1004    Debug,
1005    Serialize,
1006    Deserialize,
1007    PartialOrd,
1008    PartialEq,
1009    Eq,
1010    Ord,
1011    Hash,
1012    Arbitrary
1013)]
1014pub struct AlterApplyReplacementV1 {
1015    #[serde(flatten)]
1016    pub target: IdFullNameV1,
1017    pub replacement: IdFullNameV1,
1018}
1019
1020#[derive(
1021    Clone,
1022    Debug,
1023    Serialize,
1024    Deserialize,
1025    PartialOrd,
1026    PartialEq,
1027    Eq,
1028    Ord,
1029    Hash,
1030    Arbitrary
1031)]
1032pub struct AlterSourceSinkV1 {
1033    pub id: String,
1034    #[serde(flatten)]
1035    pub name: FullNameV1,
1036    pub old_size: Option<String>,
1037    pub new_size: Option<String>,
1038}
1039
1040#[derive(
1041    Clone,
1042    Debug,
1043    Serialize,
1044    Deserialize,
1045    PartialOrd,
1046    PartialEq,
1047    Eq,
1048    Ord,
1049    Hash,
1050    Arbitrary
1051)]
1052pub struct AlterSetClusterV1 {
1053    pub id: String,
1054    #[serde(flatten)]
1055    pub name: FullNameV1,
1056    pub old_cluster_id: String,
1057    pub new_cluster_id: String,
1058}
1059
1060#[derive(
1061    Clone,
1062    Debug,
1063    Serialize,
1064    Deserialize,
1065    PartialOrd,
1066    PartialEq,
1067    Eq,
1068    Ord,
1069    Hash,
1070    Arbitrary
1071)]
1072pub struct GrantRoleV1 {
1073    pub role_id: String,
1074    pub member_id: String,
1075    pub grantor_id: String,
1076}
1077
1078#[derive(
1079    Clone,
1080    Debug,
1081    Serialize,
1082    Deserialize,
1083    PartialOrd,
1084    PartialEq,
1085    Eq,
1086    Ord,
1087    Hash,
1088    Arbitrary
1089)]
1090pub struct GrantRoleV2 {
1091    pub role_id: String,
1092    pub member_id: String,
1093    pub grantor_id: String,
1094    pub executed_by: String,
1095}
1096
1097#[derive(
1098    Clone,
1099    Debug,
1100    Serialize,
1101    Deserialize,
1102    PartialOrd,
1103    PartialEq,
1104    Eq,
1105    Ord,
1106    Hash,
1107    Arbitrary
1108)]
1109pub struct RevokeRoleV1 {
1110    pub role_id: String,
1111    pub member_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 RevokeRoleV2 {
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 UpdatePrivilegeV1 {
1146    pub object_id: String,
1147    pub grantee_id: String,
1148    pub grantor_id: String,
1149    pub privileges: String,
1150}
1151
1152#[derive(
1153    Clone,
1154    Debug,
1155    Serialize,
1156    Deserialize,
1157    PartialOrd,
1158    PartialEq,
1159    Eq,
1160    Ord,
1161    Hash,
1162    Arbitrary
1163)]
1164pub struct AlterDefaultPrivilegeV1 {
1165    pub role_id: String,
1166    pub database_id: Option<String>,
1167    pub schema_id: Option<String>,
1168    pub grantee_id: String,
1169    pub privileges: String,
1170}
1171
1172#[derive(
1173    Clone,
1174    Debug,
1175    Serialize,
1176    Deserialize,
1177    PartialOrd,
1178    PartialEq,
1179    Eq,
1180    Ord,
1181    Hash,
1182    Arbitrary
1183)]
1184pub struct UpdateOwnerV1 {
1185    pub object_id: String,
1186    pub old_owner_id: String,
1187    pub new_owner_id: String,
1188}
1189
1190#[derive(
1191    Clone,
1192    Debug,
1193    Serialize,
1194    Deserialize,
1195    PartialOrd,
1196    PartialEq,
1197    Eq,
1198    Ord,
1199    Hash,
1200    Arbitrary
1201)]
1202pub struct SchemaV1 {
1203    pub id: String,
1204    pub name: String,
1205    pub database_name: 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 SchemaV2 {
1221    pub id: String,
1222    pub name: String,
1223    pub database_name: Option<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 RenameSchemaV1 {
1239    pub id: String,
1240    pub database_name: Option<String>,
1241    pub old_name: String,
1242    pub new_name: String,
1243}
1244
1245#[derive(
1246    Clone,
1247    Debug,
1248    Serialize,
1249    Deserialize,
1250    PartialOrd,
1251    PartialEq,
1252    Eq,
1253    Ord,
1254    Hash,
1255    Arbitrary
1256)]
1257pub struct AlterRetainHistoryV1 {
1258    pub id: String,
1259    pub old_history: Option<String>,
1260    pub new_history: Option<String>,
1261}
1262
1263#[derive(
1264    Clone,
1265    Debug,
1266    Serialize,
1267    Deserialize,
1268    PartialOrd,
1269    PartialEq,
1270    Eq,
1271    Ord,
1272    Hash,
1273    Arbitrary
1274)]
1275pub struct AlterAddColumnV1 {
1276    pub id: String,
1277    pub column: String,
1278    pub column_type: String,
1279    pub nullable: bool,
1280}
1281
1282#[derive(
1283    Clone,
1284    Debug,
1285    Serialize,
1286    Deserialize,
1287    PartialOrd,
1288    PartialEq,
1289    Eq,
1290    Ord,
1291    Hash,
1292    Arbitrary
1293)]
1294pub struct AlterSourceTimestampIntervalV1 {
1295    pub id: String,
1296    pub old_interval: Option<String>,
1297    pub new_interval: Option<String>,
1298}
1299
1300#[derive(
1301    Clone,
1302    Debug,
1303    Serialize,
1304    Deserialize,
1305    PartialOrd,
1306    PartialEq,
1307    Eq,
1308    Ord,
1309    Hash,
1310    Arbitrary
1311)]
1312pub struct UpdateItemV1 {
1313    pub id: String,
1314    #[serde(flatten)]
1315    pub name: FullNameV1,
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 ToNewIdV1 {
1331    pub id: String,
1332    pub new_id: String,
1333}
1334
1335#[derive(
1336    Clone,
1337    Debug,
1338    Serialize,
1339    Deserialize,
1340    PartialOrd,
1341    PartialEq,
1342    Eq,
1343    Ord,
1344    Hash,
1345    Arbitrary
1346)]
1347pub struct FromPreviousIdV1 {
1348    pub id: String,
1349    pub previous_id: String,
1350}
1351
1352impl EventDetails {
1353    pub fn as_json(&self) -> serde_json::Value {
1354        match self {
1355            EventDetails::CreateClusterReplicaV1(v) => {
1356                serde_json::to_value(v).expect("must serialize")
1357            }
1358            EventDetails::CreateClusterReplicaV2(v) => {
1359                serde_json::to_value(v).expect("must serialize")
1360            }
1361            EventDetails::CreateClusterReplicaV3(v) => {
1362                serde_json::to_value(v).expect("must serialize")
1363            }
1364            EventDetails::CreateClusterReplicaV4(v) => {
1365                serde_json::to_value(v).expect("must serialize")
1366            }
1367            EventDetails::DropClusterReplicaV1(v) => {
1368                serde_json::to_value(v).expect("must serialize")
1369            }
1370            EventDetails::DropClusterReplicaV2(v) => {
1371                serde_json::to_value(v).expect("must serialize")
1372            }
1373            EventDetails::DropClusterReplicaV3(v) => {
1374                serde_json::to_value(v).expect("must serialize")
1375            }
1376            EventDetails::IdFullNameV1(v) => serde_json::to_value(v).expect("must serialize"),
1377            EventDetails::RenameClusterV1(v) => serde_json::to_value(v).expect("must serialize"),
1378            EventDetails::RenameClusterReplicaV1(v) => {
1379                serde_json::to_value(v).expect("must serialize")
1380            }
1381            EventDetails::AlterClusterReconfigurationV1(v) => {
1382                serde_json::to_value(v).expect("must serialize")
1383            }
1384            EventDetails::ClusterHydrationBurstV1(v) => {
1385                serde_json::to_value(v).expect("must serialize")
1386            }
1387            EventDetails::RenameItemV1(v) => serde_json::to_value(v).expect("must serialize"),
1388            EventDetails::IdNameV1(v) => serde_json::to_value(v).expect("must serialize"),
1389            EventDetails::SchemaV1(v) => serde_json::to_value(v).expect("must serialize"),
1390            EventDetails::SchemaV2(v) => serde_json::to_value(v).expect("must serialize"),
1391            EventDetails::RenameSchemaV1(v) => serde_json::to_value(v).expect("must serialize"),
1392            EventDetails::CreateSourceSinkV1(v) => serde_json::to_value(v).expect("must serialize"),
1393            EventDetails::CreateSourceSinkV2(v) => serde_json::to_value(v).expect("must serialize"),
1394            EventDetails::CreateSourceSinkV3(v) => serde_json::to_value(v).expect("must serialize"),
1395            EventDetails::CreateSourceSinkV4(v) => serde_json::to_value(v).expect("must serialize"),
1396            EventDetails::CreateIndexV1(v) => serde_json::to_value(v).expect("must serialize"),
1397            EventDetails::CreateMaterializedViewV1(v) => {
1398                serde_json::to_value(v).expect("must serialize")
1399            }
1400            EventDetails::AlterApplyReplacementV1(v) => {
1401                serde_json::to_value(v).expect("must serialize")
1402            }
1403            EventDetails::AlterSourceSinkV1(v) => serde_json::to_value(v).expect("must serialize"),
1404            EventDetails::AlterSetClusterV1(v) => serde_json::to_value(v).expect("must serialize"),
1405            EventDetails::GrantRoleV1(v) => serde_json::to_value(v).expect("must serialize"),
1406            EventDetails::GrantRoleV2(v) => serde_json::to_value(v).expect("must serialize"),
1407            EventDetails::RevokeRoleV1(v) => serde_json::to_value(v).expect("must serialize"),
1408            EventDetails::RevokeRoleV2(v) => serde_json::to_value(v).expect("must serialize"),
1409            EventDetails::UpdatePrivilegeV1(v) => serde_json::to_value(v).expect("must serialize"),
1410            EventDetails::AlterDefaultPrivilegeV1(v) => {
1411                serde_json::to_value(v).expect("must serialize")
1412            }
1413            EventDetails::UpdateOwnerV1(v) => serde_json::to_value(v).expect("must serialize"),
1414            EventDetails::UpdateItemV1(v) => serde_json::to_value(v).expect("must serialize"),
1415            EventDetails::AlterRetainHistoryV1(v) => {
1416                serde_json::to_value(v).expect("must serialize")
1417            }
1418            EventDetails::AlterAddColumnV1(v) => serde_json::to_value(v).expect("must serialize"),
1419            EventDetails::AlterSourceTimestampIntervalV1(v) => {
1420                serde_json::to_value(v).expect("must serialize")
1421            }
1422            EventDetails::ToNewIdV1(v) => serde_json::to_value(v).expect("must serialize"),
1423            EventDetails::FromPreviousIdV1(v) => serde_json::to_value(v).expect("must serialize"),
1424            EventDetails::SetV1(v) => serde_json::to_value(v).expect("must serialize"),
1425            EventDetails::ResetAllV1 => serde_json::Value::Null,
1426            EventDetails::RotateKeysV1(v) => serde_json::to_value(v).expect("must serialize"),
1427            EventDetails::CreateRoleV1(v) => serde_json::to_value(v).expect("must serialize"),
1428        }
1429    }
1430}
1431
1432#[derive(
1433    Clone,
1434    Debug,
1435    Serialize,
1436    Deserialize,
1437    PartialOrd,
1438    PartialEq,
1439    Eq,
1440    Ord,
1441    Hash,
1442    Arbitrary
1443)]
1444pub struct EventV1 {
1445    pub id: u64,
1446    pub event_type: EventType,
1447    pub object_type: ObjectType,
1448    pub details: EventDetails,
1449    pub user: Option<String>,
1450    pub occurred_at: EpochMillis,
1451}
1452
1453impl EventV1 {
1454    fn new(
1455        id: u64,
1456        event_type: EventType,
1457        object_type: ObjectType,
1458        details: EventDetails,
1459        user: Option<String>,
1460        occurred_at: EpochMillis,
1461    ) -> EventV1 {
1462        EventV1 {
1463            id,
1464            event_type,
1465            object_type,
1466            details,
1467            user,
1468            occurred_at,
1469        }
1470    }
1471}
1472
1473#[derive(
1474    Clone,
1475    Debug,
1476    Serialize,
1477    Deserialize,
1478    PartialOrd,
1479    PartialEq,
1480    Eq,
1481    Ord,
1482    Hash,
1483    Arbitrary
1484)]
1485pub struct StorageUsageV1 {
1486    pub id: u64,
1487    pub shard_id: Option<String>,
1488    pub size_bytes: u64,
1489    pub collection_timestamp: EpochMillis,
1490}
1491
1492impl StorageUsageV1 {
1493    pub fn new(
1494        id: u64,
1495        shard_id: Option<String>,
1496        size_bytes: u64,
1497        collection_timestamp: EpochMillis,
1498    ) -> StorageUsageV1 {
1499        StorageUsageV1 {
1500            id,
1501            shard_id,
1502            size_bytes,
1503            collection_timestamp,
1504        }
1505    }
1506}
1507
1508/// Describes the environment's storage usage at a point in time.
1509///
1510/// This type is persisted in the catalog across restarts, so any updates to the
1511/// schema will require a new version.
1512#[derive(
1513    Clone,
1514    Debug,
1515    Serialize,
1516    Deserialize,
1517    PartialOrd,
1518    PartialEq,
1519    Eq,
1520    Ord,
1521    Hash,
1522    Arbitrary
1523)]
1524pub enum VersionedStorageUsage {
1525    V1(StorageUsageV1),
1526}
1527
1528impl VersionedStorageUsage {
1529    /// Create a new metric snapshot.
1530    /// This function must always require and produce the most
1531    /// recent variant of VersionedStorageMetrics.
1532    pub fn new(
1533        id: u64,
1534        object_id: Option<String>,
1535        size_bytes: u64,
1536        collection_timestamp: EpochMillis,
1537    ) -> Self {
1538        Self::V1(StorageUsageV1::new(
1539            id,
1540            object_id,
1541            size_bytes,
1542            collection_timestamp,
1543        ))
1544    }
1545
1546    // Implement deserialize and serialize so writers and readers don't have to
1547    // coordinate about which Serializer to use.
1548    pub fn deserialize(data: &[u8]) -> Result<Self, anyhow::Error> {
1549        Ok(serde_json::from_slice(data)?)
1550    }
1551
1552    pub fn serialize(&self) -> Vec<u8> {
1553        serde_json::to_vec(self).expect("must serialize")
1554    }
1555
1556    pub fn timestamp(&self) -> EpochMillis {
1557        match self {
1558            VersionedStorageUsage::V1(StorageUsageV1 {
1559                collection_timestamp,
1560                ..
1561            }) => *collection_timestamp,
1562        }
1563    }
1564
1565    /// Returns a globally sortable event order. All event versions must have this
1566    /// field.
1567    pub fn sortable_id(&self) -> u64 {
1568        match self {
1569            VersionedStorageUsage::V1(usage) => usage.id,
1570        }
1571    }
1572}
1573
1574#[cfg(test)]
1575mod tests {
1576    use crate::{EventDetails, EventType, EventV1, IdNameV1, ObjectType, VersionedEvent};
1577
1578    // Test all versions of events. This test hard codes bytes so that
1579    // programmers are not able to change data structures here without this test
1580    // failing. Instead of changing data structures, add new variants.
1581    #[mz_ore::test]
1582    fn test_audit_log() -> Result<(), anyhow::Error> {
1583        let cases: Vec<(VersionedEvent, &'static str)> = vec![(
1584            VersionedEvent::V1(EventV1::new(
1585                2,
1586                EventType::Drop,
1587                ObjectType::ClusterReplica,
1588                EventDetails::IdNameV1(IdNameV1 {
1589                    id: "u1".to_string(),
1590                    name: "name".into(),
1591                }),
1592                None,
1593                2,
1594            )),
1595            r#"{"V1":{"id":2,"event_type":"drop","object_type":"cluster-replica","details":{"IdNameV1":{"id":"u1","name":"name"}},"user":null,"occurred_at":2}}"#,
1596        )];
1597
1598        for (event, expected_bytes) in cases {
1599            let event_bytes = serde_json::to_vec(&event).unwrap();
1600            assert_eq!(
1601                event_bytes,
1602                expected_bytes.as_bytes(),
1603                "expected bytes {}, got {}",
1604                expected_bytes,
1605                std::str::from_utf8(&event_bytes).unwrap(),
1606            );
1607        }
1608
1609        Ok(())
1610    }
1611}
mz_audit_log/lib.rs

mz_audit_log/
lib.rs
