fixes bug where multiple nodes fight over an eip

        - introduces resource_ids to eip status
        - moves setting attachment status to be
        before an attempt at attaching
        - moves status update from patch to replace
        this should raise errors if the resource has
        changed allowing the resource_id to function
        somewhat as a lock
        - resource won't attempt to claim a resource
        with a resource id
        - fixes cargo deny errors

Author: jubrad

Cargo.lock

@@ -8,7 +8,7 @@ members = [
 
 [workspace.package]
 edition = "2021"
-rust-version = "1.74.0"
+rust-version = "1.76.0"
 
 
 # Use this section only to change the source of dependencies that might

Cargo.toml

@@ -1,3 +1,4 @@
+[graph]
 targets = [
     { triple = "aarch64-apple-darwin" },
     { triple = "aarch64-unknown-linux-gnu" },
@@ -6,7 +7,7 @@ targets = [
 ]
 
 [advisories]
-vulnerability = "deny"
+version = 2
 
 [bans]
 multiple-versions = "deny"
@@ -20,9 +21,9 @@ skip = [
     { name = "hashbrown", version = "0.12.3" },
     { name = "hashbrown", version = "0.14.0" },
     { name = "nix", version = "0.26.4" },
-    { name = "nix", version = "0.27.1" },
     { name = "ordered-float", version = "2.10.0" },
-    { name = "ordered-float", version = "3.4.0" },
+    { name = "fastrand", version = "2.0.1" },
+    { name = "regex-syntax", version = "0.6.29" },
 ]
 
 # Use `tracing` instead.
@@ -35,14 +36,14 @@ name = "env_logger"
 name = "rustls"
 
 [licenses]
+version = 2
 allow = [
     "Apache-2.0",
     "BSD-2-Clause",
     "BSD-3-Clause",
     "MIT",
     "Unicode-DFS-2016",
 ]
-copyleft = "deny"
 
 [[licenses.clarify]]
 name = "ring"

deny.toml

@@ -7,20 +7,10 @@ license = "Apache-2.0"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-aws-config = { version = "0.55.1", default-features = false, features = [
-  "native-tls",
-] }
-aws-sdk-ec2 = { version = "0.28", default-features = false, features = [
-  "native-tls",
-  "rt-tokio",
-] }
-aws-sdk-servicequotas = { version = "0.28", default-features = false, features = [
-  "native-tls",
-  "rt-tokio",
-] }
-aws-smithy-http = { version = "0.55", default-features = false, features = [
-  "rt-tokio",
-] }
+aws-config = { version = "0.101", default-features = false}
+aws-sdk-ec2 = { version = "0.38", default-features = false, features = [ "rt-tokio" ] }
+aws-sdk-servicequotas = { version = "0.38", default-features = false, features = [ "rt-tokio" ] }
+aws-smithy-http = { version = "0.59", default-features = false, features = [ "rt-tokio" ] }
 futures = { workspace = true }
 
 

eip_operator/Cargo.toml

@@ -9,7 +9,7 @@ use aws_sdk_ec2::operation::disassociate_address::DisassociateAddressError;
 use aws_sdk_ec2::operation::release_address::{ReleaseAddressError, ReleaseAddressOutput};
 use aws_sdk_ec2::types::{Address, DomainType, Filter, ResourceType, Tag, TagSpecification};
 use aws_sdk_ec2::Client as Ec2Client;
-use tracing::{debug, info, instrument};
+use tracing::{debug, error, info, instrument};
 
 pub(crate) const LEGACY_CLUSTER_NAME_TAG: &str = "eip.aws.materialize.com/cluster_name";
 
@@ -150,19 +150,24 @@ pub(crate) async fn describe_addresses_with_tag_value(
 pub(crate) async fn disassociate_eip(
     ec2_client: &Ec2Client,
     association_id: &str,
-) -> Result<(), SdkError<DisassociateAddressError>> {
+) -> Result<(), DisassociateAddressError> {
     match ec2_client
         .disassociate_address()
         .association_id(association_id)
         .send()
         .await
     {
         Ok(_) => Ok(()),
-        Err(e) if e.to_string().contains("InvalidAssociationID.NotFound") => {
-            info!(already_disassociated = true);
-            Ok(())
+        Err(e) => {
+            let e = e.into_service_error();
+            if e.meta().code() == Some("InvalidAssociationID.NotFound") {
+                info!("Association id {} already disassociated", association_id);
+                Ok(())
+            } else {
+                error!("Error disassociating {} - {:?}", association_id, e);
+                Err(e)
+            }
         }
-        Err(e) => Err(e),
     }
 }
 

eip_operator/src/aws.rs

@@ -2,7 +2,7 @@ use k8s_openapi::api::core::v1::Node;
 use kube::api::{Api, ListParams};
 use kube::Client;
 use kube_runtime::controller::Action;
-use tracing::{event, instrument, Level};
+use tracing::{event, info, instrument, Level};
 
 use eip_operator_shared::Error;
 
@@ -71,9 +71,17 @@ impl k8s_controller::Context for Context {
         if eip_description.network_interface_id != Some(eni_id.to_owned())
             || eip_description.private_ip_address != Some(node_ip.to_owned())
         {
-            crate::aws::associate_eip(&self.ec2_client, allocation_id, &eni_id, node_ip).await?;
+            if eip.status.as_ref().is_some_and(|s| s.resource_id.is_none()) {
+                crate::eip::set_status_attached(&eip_api, &eip, &eni_id, node_ip, name).await?;
+                crate::aws::associate_eip(&self.ec2_client, allocation_id, &eni_id, node_ip)
+                    .await?;
+            } else {
+                info!(
+                    "Found matching Eip {} for pod {}, but it is in use",
+                    eip_name, name
+                );
+            }
         }
-        crate::eip::set_status_attached(&eip_api, eip_name, &eni_id, node_ip).await?;
 
         Ok(None)
     }
@@ -94,7 +102,13 @@ impl k8s_controller::Context for Context {
         let eip = all_eips
             .into_iter()
             .filter(|eip| eip.attached())
-            .find(|eip| eip.matches_node(node_labels));
+            .find(|eip| {
+                eip.matches_node(node_labels)
+                    && eip
+                        .status
+                        .as_ref()
+                        .is_some_and(|s| s.resource_id == Some(node.metadata.name.clone().unwrap()))
+            });
         if let Some(eip) = eip {
             let allocation_id = eip.allocation_id().ok_or(Error::MissingAllocationId)?;
             let addresses = crate::aws::describe_address(&self.ec2_client, allocation_id)

eip_operator/src/controller/node.rs

@@ -71,7 +71,6 @@ impl k8s_controller::Context for Context {
             .into_iter()
             .find(|eip| eip.matches_pod(name))
             .ok_or_else(|| Error::NoEipResourceWithThatPodName(name.to_owned()))?;
-        let eip_name = eip.name().ok_or(Error::MissingEipName)?;
         let allocation_id = eip.allocation_id().ok_or(Error::MissingAllocationId)?;
         let eip_description = crate::aws::describe_address(&self.ec2_client, allocation_id)
             .await?
@@ -82,10 +81,10 @@ impl k8s_controller::Context for Context {
         if eip_description.network_interface_id != Some(eni_id.to_owned())
             || eip_description.private_ip_address != Some(pod_ip.to_owned())
         {
+            crate::eip::set_status_attached(&eip_api, &eip, &eni_id, pod_ip, name).await?;
             crate::aws::associate_eip(&self.ec2_client, allocation_id, &eni_id, pod_ip).await?;
+            add_dns_target_annotation(&pod_api, name, &public_ip, allocation_id).await?;
         }
-        crate::eip::set_status_attached(&eip_api, eip_name, &eni_id, pod_ip).await?;
-        add_dns_target_annotation(&pod_api, name, &public_ip, allocation_id).await?;
         Ok(None)
     }
 

eip_operator/src/controller/pod.rs

@@ -1,7 +1,7 @@
 use k8s_openapi::apiextensions_apiserver::pkg::apis::apiextensions::v1::CustomResourceDefinition;
 use kube::api::{Api, DeleteParams, ListParams, Patch, PatchParams, PostParams};
 use kube::core::crd::merge_crds;
-use kube::{Client, CustomResourceExt};
+use kube::{Client, CustomResourceExt, ResourceExt};
 use kube_runtime::wait::{await_condition, conditions};
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
@@ -121,7 +121,8 @@ pub mod v2 {
         printcolumn = r#"{"name": "PublicIP", "type": "string", "description": "Public IP address of the EIP.", "jsonPath": ".status.publicIpAddress"}"#,
         printcolumn = r#"{"name": "Selector", "type": "string", "description": "Selector for the pod or node to associate the EIP with.", "jsonPath": ".spec.selector", "priority": 1}"#,
         printcolumn = r#"{"name": "ENI", "type": "string", "description": "ID of the Elastic Network Interface of the pod.", "jsonPath": ".status.eni", "priority": 1}"#,
-        printcolumn = r#"{"name": "PrivateIP", "type": "string", "description": "Private IP address of the pod.", "jsonPath": ".status.privateIpAddress", "priority": 1}"#
+        printcolumn = r#"{"name": "PrivateIP", "type": "string", "description": "Private IP address of the pod.", "jsonPath": ".status.privateIpAddress", "priority": 1}"#,
+        printcolumn = r#"{"name": "ResourceId", "type": "string", "description": "ID of resource the EIP is attached to..", "jsonPath": ".status.resourceId", "priority": 1}"#
     )]
     pub struct EipSpec {
         pub selector: EipSelector,
@@ -205,13 +206,14 @@ pub mod v2 {
 }
 
 /// The status fields for the Eip Kubernetes custom resource.
-#[derive(Clone, Debug, Deserialize, Serialize, JsonSchema)]
+#[derive(Clone, Default, Debug, Deserialize, Serialize, JsonSchema)]
 #[serde(rename_all = "camelCase")]
 pub struct EipStatus {
     pub allocation_id: Option<String>,
     pub public_ip_address: Option<String>,
     pub eni: Option<String>,
     pub private_ip_address: Option<String>,
+    pub resource_id: Option<String>,
 }
 
 /// Registers the Eip custom resource with Kubernetes,
@@ -335,26 +337,27 @@ pub(crate) async fn set_status_created(
 #[instrument(skip(api), err)]
 pub(crate) async fn set_status_attached(
     api: &Api<Eip>,
-    name: &str,
+    eip: &Eip,
     eni: &str,
     private_ip_address: &str,
-) -> Result<Eip, kube::Error> {
+    resource_id: &str,
+) -> Result<Eip, Error> {
     event!(Level::INFO, "Updating status for attached EIP.");
-    let patch = serde_json::json!({
-        "apiVersion": Eip::version(),
-        "kind": "Eip",
-        "status": {
-            "eni": eni,
-            "privateIpAddress": private_ip_address,
-        }
-    });
-    let patch = Patch::Merge(&patch);
-    let params = PatchParams::default();
-    let result = api.patch_status(name, &params, &patch).await;
-    if result.is_ok() {
-        event!(Level::INFO, "Done updating status for attached EIP.");
-    }
-    result
+    let mut eip = eip.clone();
+    let status = eip.status.as_mut().ok_or(Error::MissingEipStatus)?;
+    status.eni = Some(eni.to_owned());
+    status.private_ip_address = Some(private_ip_address.to_owned());
+    status.resource_id = Some(resource_id.to_owned());
+    let result = api
+        .replace_status(
+            &eip.name_unchecked(),
+            &PostParams::default(),
+            serde_json::to_vec(&eip.clone())?,
+        )
+        .await?;
+    // let result = api.patch_status(name, &params, &patch).await;
+    event!(Level::INFO, "Done updating status for attached EIP.");
+    Ok(result)
 }
 
 /// Unsets the eni and privateIpAddress fields in the Eip status.
@@ -367,6 +370,7 @@ pub(crate) async fn set_status_detached(api: &Api<Eip>, name: &str) -> Result<Ei
         "status": {
             "eni": None::<String>,
             "privateIpAddress": None::<String>,
+            "resourceId": None::<String>,
         }
     });
     let patch = Patch::Merge(&patch);

eip_operator/src/eip.rs

@@ -53,7 +53,8 @@ async fn run() -> Result<(), Error> {
     let k8s_client = Client::try_default().await?;
 
     debug!("Getting ec2_client...");
-    let mut config_loader = aws_config::from_env();
+    let mut config_loader = eip_operator_shared::aws_config_loader_default();
+
     if let Ok(endpoint) = std::env::var("AWS_ENDPOINT_URL") {
         config_loader = config_loader.endpoint_url(endpoint);
     }
@@ -253,7 +254,7 @@ async fn report_eip_quota_status(
     quota_client: &ServiceQuotaClient,
 ) -> Result<(), Error> {
     let addresses_result = ec2_client.describe_addresses().send().await?;
-    let allocated = addresses_result.addresses().unwrap_or_default().len();
+    let allocated = addresses_result.addresses().len();
     let quota_result = quota_client
         .get_service_quota()
         .service_code("ec2")

eip_operator/src/main.rs

@@ -7,33 +7,31 @@ license = "Apache-2.0"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-aws-sdk-ec2 = { version = "0.28", default-features = false, features = [
-  "native-tls",
-  "rt-tokio",
-] }
-aws-sdk-servicequotas = { version = "0.28", default-features = false, features = [
-  "native-tls",
-  "rt-tokio",
-] }
-aws-smithy-http = { version = "0.55", default-features = false, features = [
-  "rt-tokio",
-] }
+aws-config = { version = "0.101", default-features = false}
+aws-sdk-ec2 = { version = "0.38", default-features = false, features = [ "rt-tokio" ] }
+aws-sdk-servicequotas = { version = "0.38", default-features = false, features = [ "rt-tokio" ] }
+aws-smithy-http = { version = "0.59", default-features = false, features = [ "rt-tokio" ] }
+aws-smithy-runtime-api = "0.101"
+aws-smithy-runtime = { version = "0.101", features = ["connector-hyper-0-14-x"] }
+hyper-tls = { version = "0.5.0" }
+
+
 futures = "0.3"
 hyper = { version = "0.14.27", features = ["http2"] }
-hyper-tls = { version = "0.5.0" }
 kube = { workspace = true }
 kube-runtime = { workspace = true }
 native-tls = { version = "0.2.11", features = ["alpn"] }
-opentelemetry = { version = "0.20", features = ["rt-tokio", "trace"] }
-opentelemetry-otlp = { version = "0.13" }
+opentelemetry = { version = "0.21", features = ["trace"] }
+opentelemetry_sdk = { version = "0.21", features = ["trace", "rt-tokio"] }
+opentelemetry-otlp = { version = "0.14" }
 serde = "1"
 serde_json = "1"
 thiserror = "1"
 tokio-native-tls = { version = "0.3.1" }
 tokio = { workspace = true }
 tonic = { version = "0.9.2", features = ["transport"] }
 tracing = "0.1"
-tracing-opentelemetry = "0.20"
+tracing-opentelemetry = "0.22"
 tracing-subscriber = { version = "0.3", features = [
   "registry",
   "env-filter",

eip_operator_shared/Cargo.toml

@@ -4,6 +4,9 @@ use std::net::AddrParseError;
 use std::str::FromStr;
 use std::time::Duration;
 
+use aws_config::{BehaviorVersion, ConfigLoader};
+use aws_smithy_runtime::client::http::hyper_014::HyperClientBuilder;
+
 use aws_sdk_ec2::error::SdkError;
 use aws_sdk_ec2::operation::allocate_address::AllocateAddressError;
 use aws_sdk_ec2::operation::associate_address::AssociateAddressError;
@@ -16,9 +19,9 @@ use aws_sdk_servicequotas::operation::get_service_quota::GetServiceQuotaError;
 use futures::Future;
 use hyper::client::HttpConnector;
 use hyper_tls::HttpsConnector;
-use opentelemetry::sdk::trace::{Config, Sampler};
-use opentelemetry::sdk::Resource as OtelResource;
 use opentelemetry::KeyValue;
+use opentelemetry_sdk::trace::{Config, Sampler};
+use opentelemetry_sdk::Resource as OtelResource;
 use tokio::time::error::Elapsed;
 use tonic::metadata::{MetadataKey, MetadataMap};
 use tonic::transport::Endpoint;
@@ -112,7 +115,7 @@ pub enum Error {
     #[error("AWS disassociate_address reported error: {source}")]
     AwsDisassociateAddress {
         #[from]
-        source: SdkError<DisassociateAddressError>,
+        source: DisassociateAddressError,
     },
     #[error("AWS release_address reported error: {source}")]
     AwsReleaseAddress {
@@ -254,7 +257,7 @@ where
                         ))
                         .with_resource(otr),
                 )
-                .install_batch(opentelemetry::runtime::Tokio)
+                .install_batch(opentelemetry_sdk::runtime::Tokio)
                 .unwrap();
             let otel_layer = tracing_opentelemetry::layer()
                 .with_tracer(tracer)
@@ -276,3 +279,8 @@ where
     };
     f().await
 }
+
+pub fn aws_config_loader_default() -> ConfigLoader {
+    aws_config::defaults(BehaviorVersion::latest())
+        .http_client(HyperClientBuilder::new().build(HttpsConnector::new()))
+}