mz_orchestratord/
webhook.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
10use anyhow::anyhow;
11use axum::routing::{get, post};
12use axum::{Json, Router};
13use http::StatusCode;
14use kube::core::Status;
15use kube::core::conversion::{ConversionRequest, ConversionResponse, ConversionReview};
16use kube::core::response::reason;
17
18use mz_cloud_resources::crd::materialize::{convert_v1_to_v1alpha1, convert_v1alpha1_to_v1};
19use tracing::{debug, warn};
20
21pub fn router() -> Router {
22    Router::new()
23        .route("/convert", post(post_convert))
24        .route("/healthz", get(get_health))
25}
26
27#[derive(Clone, Copy)]
28enum SupportedVersion {
29    V1alpha1,
30    V1,
31}
32
33impl TryFrom<&str> for SupportedVersion {
34    type Error = anyhow::Error;
35
36    fn try_from(value: &str) -> Result<Self, Self::Error> {
37        match value {
38            "materialize.cloud/v1alpha1" => Ok(SupportedVersion::V1alpha1),
39            "materialize.cloud/v1" => Ok(SupportedVersion::V1),
40            _ => Err(anyhow!("unexpected version: {}", value)),
41        }
42    }
43}
44
45fn version_label(v: SupportedVersion) -> &'static str {
46    match v {
47        SupportedVersion::V1alpha1 => "v1alpha1",
48        SupportedVersion::V1 => "v1",
49    }
50}
51
52fn convert(
53    desired_version: SupportedVersion,
54    value: serde_json::Value,
55) -> Result<serde_json::Value, anyhow::Error> {
56    let from_version = SupportedVersion::try_from(
57        value
58            .get("apiVersion")
59            .and_then(|version| version.as_str())
60            .ok_or_else(|| anyhow!("missing version"))?,
61    )?;
62    debug!(
63        from = version_label(from_version),
64        to = version_label(desired_version),
65        input_spec = ?value.get("spec"),
66        input_status = ?value.get("status"),
67        "conversion webhook called",
68    );
69    let result = match (from_version, desired_version) {
70        (SupportedVersion::V1alpha1, SupportedVersion::V1alpha1) => Ok(value),
71        (SupportedVersion::V1alpha1, SupportedVersion::V1) => convert_v1alpha1_to_v1(value),
72        (SupportedVersion::V1, SupportedVersion::V1alpha1) => convert_v1_to_v1alpha1(value),
73        (SupportedVersion::V1, SupportedVersion::V1) => Ok(value),
74    };
75    match &result {
76        Ok(converted) => {
77            debug!(
78                from = version_label(from_version),
79                to = version_label(desired_version),
80                output_spec = ?converted.get("spec"),
81                output_status = ?converted.get("status"),
82                "conversion webhook succeeded",
83            );
84        }
85        Err(e) => {
86            warn!(
87                from = version_label(from_version),
88                to = version_label(desired_version),
89                error = ?e,
90                "conversion webhook failed",
91            );
92        }
93    }
94    result
95}
96
97async fn post_convert(
98    Json(conversion_review): Json<ConversionReview>,
99) -> (StatusCode, Json<ConversionReview>) {
100    let Ok(request) = ConversionRequest::from_review(conversion_review) else {
101        warn!("missing request");
102        return (
103            StatusCode::UNPROCESSABLE_ENTITY,
104            Json(
105                ConversionResponse::invalid(Status::failure("missing request", reason::INVALID))
106                    .into_review(),
107            ),
108        );
109    };
110
111    let desired_version = match SupportedVersion::try_from(request.desired_api_version.as_str()) {
112        Ok(v) => v,
113        Err(e) => {
114            return (
115                StatusCode::INTERNAL_SERVER_ERROR,
116                Json(
117                    ConversionResponse::for_request(request)
118                        .failure(Status::failure(&e.to_string(), reason::BAD_REQUEST))
119                        .into_review(),
120                ),
121            );
122        }
123    };
124
125    let converted_objects: Result<Vec<serde_json::Value>, anyhow::Error> = request
126        .objects
127        .iter()
128        .cloned()
129        .map(|value| convert(desired_version, value))
130        .collect();
131    match converted_objects {
132        Ok(converted_objects) => (
133            StatusCode::OK,
134            Json(
135                ConversionResponse::for_request(request)
136                    .success(converted_objects)
137                    .into_review(),
138            ),
139        ),
140        Err(e) => {
141            warn!("error when converting: {:?}\n{:?}", &e, request.objects);
142            (
143                StatusCode::INTERNAL_SERVER_ERROR,
144                Json(
145                    ConversionResponse::for_request(request)
146                        .failure(Status::failure(&e.to_string(), reason::UNKNOWN))
147                        .into_review(),
148                ),
149            )
150        }
151    }
152}
153
154async fn get_health() -> StatusCode {
155    StatusCode::OK
156}
mz_orchestratord/webhook.rs

mz_orchestratord/
webhook.rs
