Merge branch 'main' into slf-113

.github/actions/argocd-update/action.yml

@@ -64,6 +64,8 @@ runs:
           yq -i '(.spec.source.helm.parameters.[] | select(.name == "inferenceGateway.image.tag")).value = "${{ inputs.version }}"' ${{ inputs.clone_into }}/${{ inputs.subdirectory }}/tembo-ai.yaml
           yq -i '(.spec.source.helm.parameters.[] | select(.name == "controller.image.tag")).value = "${{ inputs.version }}"' ${{ inputs.clone_into }}/${{ inputs.subdirectory }}/tembo-operator-gke-usc1.yaml
           yq -i '.spec.source.targetRevision= "${{ inputs.version }}"' ${{ inputs.clone_into }}/${{ inputs.subdirectory }}/tembo-ai.yaml
+          yq -i '(.spec.source.helm.parameters.[] | select(.name == "image.tag")).value = "${{ inputs.version }}"' ${{ inputs.clone_into }}/${{ inputs.subdirectory }}/tembo-ai-events-reporter.yaml
+
 
     - name: Update for prod deployments
       if: inputs.branch == 'prod-updates'

.github/workflows/operator.yaml

@@ -47,10 +47,10 @@ jobs:
           set -xe
           sudo apt-get update
           sudo apt-get install -y pkg-config libssl-dev git
-      - uses: azure/setup-helm@v3
-      - uses: extractions/setup-just@v1
+      - uses: azure/setup-helm@v4
+      - uses: extractions/setup-just@v2
       - name: Install kind
-        uses: helm/kind-action@v1.7.0
+        uses: helm/kind-action@v1
         with:
           install_only: true
       - uses: dtolnay/rust-toolchain@stable
@@ -79,7 +79,7 @@ jobs:
           # Start the operator in the background
           cargo run > operator-output.txt 2>&1 &
           # Run the tests
-          cargo test --jobs 1 -- --ignored --nocapture
+          cargo test -- --ignored --nocapture
       - name: Debugging information
         if: always()
         run: |

CODEOWNERS

@@ -1,10 +1,9 @@
-.github/** @sjmiller609 @nhudson @ianstanton
-charts/** @sjmiller609 @nhudson @ianstanton
-dataplane-webserver/** @sjmiller609
-tembo-operator/** @sjmiller609 @nhudson @ianstanton @ChuckHend
-conductor/** @sjmiller609 @nhudson @ianstanton @ChuckHend
-tembo-pod-init/** @nhudson
-tembo-cli/** @shahadarsh @vrmiguel @DarrenBaldwin07 @sjmiller609
-tembo-py/** @chuckhend @EvanHStanton
-tembo-stacks/** @chuckhend @jasonmp85 @shhnwz
-inference-gateway/** @chuckhend @jasonmp85 @shhnwz
+.github/** @nhudson @ianstanton @shahadarsh
+charts/**  @nhudson @ianstanton @shahadarsh
+tembo-operator/** @nhudson @ianstanton @ChuckHend
+conductor/**  @nhudson @ianstanton @ChuckHend
+tembo-pod-init/** @nhudson @ianstanton
+tembo-cli/** @shahadarsh @vrmiguel @DarrenBaldwin07 @joshuajerin
+tembo-py/** @chuckhend
+tembo-stacks/** @chuckhend @jasonmp85
+inference-gateway/** @chuckhend @jasonmp85

conductor/justfile

@@ -80,6 +80,7 @@ watch:
 	BACKUP_ARCHIVE_BUCKET=cdb-plat-use1-dev-instance-backups \
 	STORAGE_ARCHIVE_BUCKET=cdb-plat-use1-dev-instance-storage \
 	IS_CLOUD_FORMATION=false \
+	IS_GCP=false \
 	cargo watch -x run
 
 run-postgres:

conductor/src/cloud.rs

@@ -0,0 +1,62 @@
+// Add this enum at the top of your file or in a separate module
+pub struct CloudProviderBuilder {
+    gcp: bool,
+    aws: bool,
+}
+
+impl CloudProviderBuilder {
+    fn new() -> Self {
+        CloudProviderBuilder {
+            gcp: false,
+            aws: false,
+        }
+    }
+
+    pub fn gcp(mut self, value: bool) -> Self {
+        self.gcp = value;
+        self
+    }
+
+    pub fn aws(mut self, value: bool) -> Self {
+        self.aws = value;
+        self
+    }
+
+    pub fn build(self) -> CloudProvider {
+        if self.gcp {
+            CloudProvider::GCP
+        } else if self.aws {
+            CloudProvider::AWS
+        } else {
+            CloudProvider::Unknown
+        }
+    }
+}
+
+pub enum CloudProvider {
+    AWS,
+    GCP,
+    Unknown,
+}
+
+impl CloudProvider {
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            CloudProvider::AWS => "aws",
+            CloudProvider::GCP => "gcp",
+            CloudProvider::Unknown => "unknown",
+        }
+    }
+
+    pub fn prefix(&self) -> &'static str {
+        match self {
+            CloudProvider::AWS => "s3://",
+            CloudProvider::GCP => "gs://",
+            CloudProvider::Unknown => "",
+        }
+    }
+
+    pub fn builder() -> CloudProviderBuilder {
+        CloudProviderBuilder::new()
+    }
+}

conductor/src/errors.rs

@@ -53,4 +53,8 @@ pub enum ConductorError {
     /// Google Cloud Storage error
     #[error("Google Cloud Storage error: {0}")]
     GcsError(#[from] GcsError),
+
+    /// Dataplane error
+    #[error("Dataplane not found error: {0}")]
+    DataplaneError(String),
 }

conductor/src/lib.rs

@@ -1,4 +1,5 @@
 pub mod aws;
+pub mod cloud;
 pub mod errors;
 pub mod extensions;
 pub mod gcp;
@@ -7,7 +8,10 @@ pub mod monitoring;
 pub mod routes;
 pub mod types;
 
-use crate::aws::cloudformation::{AWSConfigState, CloudFormationParams};
+use crate::{
+    aws::cloudformation::{AWSConfigState, CloudFormationParams},
+    cloud::CloudProvider,
+};
 use aws_sdk_cloudformation::config::Region;
 use controller::apis::coredb_types::{CoreDB, CoreDBSpec};
 use errors::ConductorError;
@@ -18,7 +22,7 @@ use kube::api::{DeleteParams, ListParams, Patch, PatchParams};
 
 use chrono::{DateTime, SecondsFormat, Utc};
 use kube::{Api, Client, ResourceExt};
-use log::{debug, info};
+use log::{debug, info, warn};
 use serde_json::{from_str, to_string, Value};
 use std::{
     collections::hash_map::DefaultHasher,
@@ -27,6 +31,7 @@ use std::{
 
 pub type Result<T, E = ConductorError> = std::result::Result<T, E>;
 
+#[allow(clippy::too_many_arguments)]
 pub async fn generate_spec(
     org_id: &str,
     entity_name: &str,
@@ -35,18 +40,38 @@ pub async fn generate_spec(
     namespace: &str,
     backups_bucket: &str,
     spec: &CoreDBSpec,
-) -> Value {
+    cloud_provider: &CloudProvider,
+) -> Result<Value, ConductorError> {
     let mut spec = spec.clone();
-    // Add the bucket name into the backups_path if it's not already there
-    if let Some(restore) = &mut spec.restore {
-        if let Some(backups_path) = &mut restore.backups_path {
-            if !backups_path.starts_with(&format!("s3://{}", backups_bucket)) {
-                let path_suffix = backups_path.trim_start_matches("s3://");
-                *backups_path = format!("s3://{}/{}", backups_bucket, path_suffix);
+
+    match cloud_provider {
+        CloudProvider::AWS | CloudProvider::GCP => {
+            let prefix = cloud_provider.prefix();
+
+            // Format the backups_path with the correct prefix
+            if let Some(restore) = &mut spec.restore {
+                if let Some(backups_path) = &mut restore.backups_path {
+                    let clean_path = remove_known_prefixes(backups_path);
+                    if clean_path.starts_with(backups_bucket) {
+                        // If the path already includes the bucket, just add the prefix
+                        *backups_path = format!("{}{}", prefix, clean_path);
+                    } else {
+                        // If the path doesn't include the bucket, add both prefix and bucket
+                        *backups_path = format!("{}{}/{}", prefix, backups_bucket, clean_path);
+                    }
+                }
             }
         }
+        CloudProvider::Unknown => {
+            warn!(
+                "Unknown cloud provider or cloud provider is disabled, restore spec removed from Spec value",
+            );
+            // Remove the restore information if the cloud provider is unknown
+            spec.restore = None;
+        }
     }
-    serde_json::json!({
+
+    Ok(serde_json::json!({
         "apiVersion": "coredb.io/v1alpha1",
         "kind": "CoreDB",
         "metadata": {
@@ -59,7 +84,18 @@ pub async fn generate_spec(
             }
         },
         "spec": spec,
-    })
+    }))
+}
+
+// Remove known prefixes from the backup path
+fn remove_known_prefixes(path: &str) -> &str {
+    let known_prefixes = ["s3://", "gs://", "https://"];
+    for prefix in &known_prefixes {
+        if let Some(stripped) = path.strip_prefix(prefix) {
+            return stripped;
+        }
+    }
+    path
 }
 
 pub fn get_data_plane_id_from_coredb(coredb: &CoreDB) -> Result<String, Box<ConductorError>> {
@@ -636,16 +672,19 @@ mod tests {
             }),
             ..CoreDBSpec::default()
         };
+        let cloud_provider = CloudProvider::AWS;
         let result = generate_spec(
             "org-id",
             "entity-name",
             "instance-id",
-            "data-plane-id",
+            "aws_data_1_use1",
             "namespace",
             "my-bucket",
             &spec,
+            &cloud_provider,
         )
-        .await;
+        .await
+        .expect("Failed to generate spec");
         let expected_backups_path = "s3://my-bucket/coredb/coredb/org-coredb-inst-pgtrunkio-dev";
         assert_eq!(
             result["spec"]["restore"]["backupsPath"].as_str().unwrap(),
@@ -664,16 +703,19 @@ mod tests {
             }),
             ..CoreDBSpec::default()
         };
+        let cloud_provider = CloudProvider::AWS;
         let result = generate_spec(
             "org-id",
             "entity-name",
             "instance-id",
-            "data-plane-id",
+            "aws_data_1_use1",
             "namespace",
             "my-bucket",
             &spec,
+            &cloud_provider,
         )
-        .await;
+        .await
+        .expect("Failed to generate spec");
         let expected_backups_path = "s3://my-bucket/coredb/coredb/org-coredb-inst-pgtrunkio-dev";
         assert_eq!(
             result["spec"]["restore"]["backupsPath"].as_str().unwrap(),
@@ -690,16 +732,19 @@ mod tests {
             }),
             ..CoreDBSpec::default()
         };
+        let cloud_provider = CloudProvider::AWS;
         let result = generate_spec(
             "org-id",
             "entity-name",
             "instance-id",
-            "data-plane-id",
+            "aws_data_1_use1",
             "namespace",
             "my-bucket",
             &spec,
+            &cloud_provider,
         )
-        .await;
+        .await
+        .expect("Failed to generate spec");
         assert!(result["spec"]["restore"]["backupsPath"].is_null());
     }
 
@@ -709,16 +754,77 @@ mod tests {
             restore: None,
             ..CoreDBSpec::default()
         };
+        let cloud_provider = CloudProvider::AWS;
         let result = generate_spec(
             "org-id",
             "entity-name",
             "instance-id",
-            "data-plane-id",
+            "aws_data_1_use1",
             "namespace",
             "my-bucket",
             &spec,
+            &cloud_provider,
         )
-        .await;
+        .await
+        .expect("Failed to generate spec");
         assert!(result["spec"]["restore"].is_null());
     }
+
+    #[tokio::test]
+    async fn test_generate_spec_with_non_matching_gcp_bucket() {
+        let spec = CoreDBSpec {
+            restore: Some(Restore {
+                backups_path: Some("gs://v2/test-instance".to_string()),
+                ..Restore::default()
+            }),
+            ..CoreDBSpec::default()
+        };
+        let cloud_provider = CloudProvider::GCP;
+        let result = generate_spec(
+            "org-id",
+            "entity-name",
+            "instance-id",
+            "gcp_data_1_usc1",
+            "namespace",
+            "my-bucket",
+            &spec,
+            &cloud_provider,
+        )
+        .await
+        .expect("Failed to generate spec");
+        let expected_backups_path = "gs://my-bucket/v2/test-instance";
+        assert_eq!(
+            result["spec"]["restore"]["backupsPath"].as_str().unwrap(),
+            expected_backups_path
+        );
+    }
+
+    #[tokio::test]
+    async fn test_generate_spec_with_gcp_bucket() {
+        let spec = CoreDBSpec {
+            restore: Some(Restore {
+                backups_path: Some("gs://my-bucket/v2/test-instance".to_string()),
+                ..Restore::default()
+            }),
+            ..CoreDBSpec::default()
+        };
+        let cloud_provider = CloudProvider::GCP;
+        let result = generate_spec(
+            "org-id",
+            "entity-name",
+            "instance-id",
+            "gcp_data_1_usc1",
+            "namespace",
+            "my-bucket",
+            &spec,
+            &cloud_provider,
+        )
+        .await
+        .expect("Failed to generate spec");
+        let expected_backups_path = "gs://my-bucket/v2/test-instance";
+        assert_eq!(
+            result["spec"]["restore"]["backupsPath"].as_str().unwrap(),
+            expected_backups_path
+        );
+    }
 }