Merge pull request #29 from wack/robbie/ingress-skeleton

Plumb Lambda artifact path to Ingress.

Cargo.toml

@@ -15,7 +15,9 @@ path = "src/bin/main.rs"
 async-stream = "0.3.6"
 async-trait = "0.1.83"
 aws-config = { version = "1.1.7", features = ["behavior-version-latest"] }
+aws-sdk-apigateway = "1.50.0"
 aws-sdk-cloudwatch = "1.54.0"
+aws-sdk-lambda = "1.56.0"
 bon = "2.3.0"
 chrono = "0.4.38"
 clap = { version = "4.3", features = ["derive"] }

src/adapters/ingresses/apig.rs

@@ -0,0 +1,56 @@
+use std::path::PathBuf;
+
+use crate::utils::load_default_aws_config;
+
+use super::Ingress;
+use async_trait::async_trait;
+use miette::{IntoDiagnostic, Result};
+use tokio::{fs::File, io::AsyncReadExt};
+
+use aws_sdk_apigateway::client::Client as GatewayClient;
+use aws_sdk_lambda::client::Client as LambdaClient;
+
+/// AwsApiGateway is the Ingress implementation for AWS API Gateway + Lambda.
+/// It's responsible for creating canary deployments on API Gateway, updating their
+/// traffic and promoting them, and deploying Lambda functions.
+pub struct AwsApiGateway {
+    /// The Lambda code, loaded from a file as an array of bytes.
+    /// The AWS SDK handles the encoding from there.
+    lambda_artifact: Vec<u8>,
+    apig_client: GatewayClient,
+    lambda_client: LambdaClient,
+}
+
+impl AwsApiGateway {
+    /// Given a path to the lambda, create a new APIG Ingress.
+    pub async fn new(artifact_path: PathBuf) -> Result<Self> {
+        let artifact = read_file(artifact_path).await?;
+        // Now, configure the AWS SDKs.
+        // TODO: Extract Config into a single location so we don't have to
+        //       repeat this code every time we initialize an AWS client.
+        let config = load_default_aws_config().await;
+        let apig_client = GatewayClient::new(config);
+        let lambda_client = LambdaClient::new(config);
+        Ok(Self {
+            lambda_artifact: artifact,
+            apig_client,
+            lambda_client,
+        })
+    }
+}
+
+/// given a path to a file, load it as an array of bytes.
+async fn read_file(artifact_path: PathBuf) -> Result<Vec<u8>> {
+    let mut bytes = Vec::new();
+    // Load the lambda from file.
+    let mut artifact = File::open(artifact_path).await.into_diagnostic()?;
+    artifact.read_to_end(&mut bytes).await.into_diagnostic()?;
+    Ok(bytes)
+}
+
+#[async_trait]
+impl Ingress for AwsApiGateway {
+    async fn deploy(&mut self) -> Result<()> {
+        todo!()
+    }
+}

src/adapters/ingresses/mod.rs

@@ -34,5 +34,4 @@ impl From<MockIngress> for BoxIngress {
 /// dispatched.
 pub type BoxIngress = Box<dyn Ingress>;
 
-#[cfg(test)]
-mod tests {}
+mod apig;

src/adapters/monitors/cloudwatch.rs

@@ -1,7 +1,8 @@
 use async_trait::async_trait;
-use aws_config::BehaviorVersion;
 
-use crate::{metrics::ResponseStatusCode, stats::CategoricalObservation};
+use crate::{
+    metrics::ResponseStatusCode, stats::CategoricalObservation, utils::load_default_aws_config,
+};
 use aws_sdk_cloudwatch::client::Client as AwsClient;
 
 use super::Monitor;
@@ -12,12 +13,8 @@ pub struct CloudWatch {
 
 impl CloudWatch {
     pub async fn new() -> Self {
-        // We don't need a particular version, but we should pin to a particular
-        // behavior so it doens't accidently slip if `latest` gets updated
-        // without our knowledge.
-        let behavior = BehaviorVersion::v2024_03_28();
-        let config = aws_config::load_defaults(behavior).await;
-        let client = aws_sdk_cloudwatch::Client::new(&config);
+        let config = load_default_aws_config().await;
+        let client = aws_sdk_cloudwatch::Client::new(config);
         Self { client }
     }
 }

src/lib.rs

@@ -15,3 +15,5 @@ pub mod metrics;
 mod pipeline;
 /// Our statistics library.
 pub mod stats;
+/// For utility functions that span multiple modules and use cases.
+mod utils;

src/utils/mod.rs

@@ -0,0 +1,23 @@
+use aws_config::{BehaviorVersion, SdkConfig};
+use tokio::sync::OnceCell;
+
+/// Load AWS configuration using their standard rules. e.g. AWS_ACCESS_KEY_ID,
+/// or session profile information, etc. This function fetches the data only
+/// once, the first time it's called, and memoized the results, so all future
+/// calls with return the same information. This prevents a race condition where
+/// an external process changes an environment variable while this process is running.
+pub async fn load_default_aws_config() -> &'static SdkConfig {
+    AWS_CONFIG_CELL.get_or_init(load_config).await
+}
+
+/// Private, delegate function to be called only within a OnceCell to ensure
+/// its locked. When Rust supports async closures, we can move this into a closure
+/// to guarantee its only ever called in one place.
+async fn load_config() -> SdkConfig {
+    // We don't need a particular version, but we pin to one to ensure
+    // it doesn't accidently slip if `latest` gets updated without our knowledge.
+    let behavior = BehaviorVersion::v2024_03_28();
+    aws_config::load_defaults(behavior).await
+}
+
+static AWS_CONFIG_CELL: OnceCell<SdkConfig> = OnceCell::const_new();