Skip to main content

fakecloud_eks/
service.rs

1//! AWS EKS REST-JSON service handler.
2//!
3//! Implements the cluster control plane (Create/Describe/List/DeleteCluster,
4//! UpdateClusterConfig, UpdateClusterVersion, Describe/ListUpdates,
5//! Tag/Untag/ListTagsForResource) plus managed node groups
6//! (Create/Describe/List/DeleteNodegroup, UpdateNodegroupConfig/Version) and
7//! Fargate profiles (Create/Describe/List/DeleteFargateProfile), add-ons,
8//! access entries and their access-policy associations, OIDC identity-provider
9//! configs, and Pod Identity associations, all modeled as cluster
10//! sub-resources.
11
12use async_trait::async_trait;
13use chrono::Utc;
14use http::{Method, StatusCode};
15use serde_json::{json, Value};
16use std::sync::Arc;
17use tokio::sync::Mutex as AsyncMutex;
18
19use fakecloud_core::pagination::paginate_checked;
20use fakecloud_core::service::{AwsRequest, AwsResponse, AwsService, AwsServiceError};
21use fakecloud_persistence::SnapshotStore;
22
23use crate::eks_helpers::*;
24
25use crate::state::{
26    access_entry_arn, addon_arn, capability_arn, cluster_arn, eks_anywhere_subscription_arn,
27    fargate_profile_arn, identity_provider_config_arn, nodegroup_arn, pod_identity_association_arn,
28    AccessEntry, Addon, AssociatedPolicy, Capability, Cluster, EksAnywhereSubscription,
29    EksSnapshot, FargateProfile, IdentityProviderConfig, InsightsRefresh, Nodegroup,
30    PodIdentityAssociation, SharedEksState, DEFAULT_K8S_VERSION, EKS_SNAPSHOT_SCHEMA_VERSION,
31};
32
33/// The set of control-plane log types EKS reports on every cluster.
34pub(crate) const LOG_TYPES: &[&str] = &[
35    "api",
36    "audit",
37    "authenticator",
38    "controllerManager",
39    "scheduler",
40];
41
42pub const EKS_ACTIONS: &[&str] = &[
43    "CreateCluster",
44    "DescribeCluster",
45    "ListClusters",
46    "DeleteCluster",
47    "UpdateClusterConfig",
48    "UpdateClusterVersion",
49    "DescribeUpdate",
50    "ListUpdates",
51    "TagResource",
52    "UntagResource",
53    "ListTagsForResource",
54    "CreateNodegroup",
55    "DescribeNodegroup",
56    "ListNodegroups",
57    "DeleteNodegroup",
58    "UpdateNodegroupConfig",
59    "UpdateNodegroupVersion",
60    "CreateFargateProfile",
61    "DescribeFargateProfile",
62    "ListFargateProfiles",
63    "DeleteFargateProfile",
64    "CreateAddon",
65    "DescribeAddon",
66    "ListAddons",
67    "DeleteAddon",
68    "UpdateAddon",
69    "DescribeAddonVersions",
70    "DescribeAddonConfiguration",
71    "CreateAccessEntry",
72    "DescribeAccessEntry",
73    "ListAccessEntries",
74    "DeleteAccessEntry",
75    "UpdateAccessEntry",
76    "AssociateAccessPolicy",
77    "DisassociateAccessPolicy",
78    "ListAssociatedAccessPolicies",
79    "ListAccessPolicies",
80    "AssociateIdentityProviderConfig",
81    "DisassociateIdentityProviderConfig",
82    "DescribeIdentityProviderConfig",
83    "ListIdentityProviderConfigs",
84    "CreatePodIdentityAssociation",
85    "DeletePodIdentityAssociation",
86    "DescribePodIdentityAssociation",
87    "ListPodIdentityAssociations",
88    "UpdatePodIdentityAssociation",
89    "DescribeInsight",
90    "ListInsights",
91    "DescribeInsightsRefresh",
92    "StartInsightsRefresh",
93    "AssociateEncryptionConfig",
94    "CancelUpdate",
95    "DeregisterCluster",
96    "RegisterCluster",
97    "DescribeClusterVersions",
98    "CreateCapability",
99    "DeleteCapability",
100    "DescribeCapability",
101    "ListCapabilities",
102    "UpdateCapability",
103    "CreateEksAnywhereSubscription",
104    "DeleteEksAnywhereSubscription",
105    "DescribeEksAnywhereSubscription",
106    "ListEksAnywhereSubscriptions",
107    "UpdateEksAnywhereSubscription",
108];
109
110pub struct EksService {
111    state: SharedEksState,
112    snapshot_store: Option<Arc<dyn SnapshotStore>>,
113    snapshot_lock: Arc<AsyncMutex<()>>,
114}
115
116enum PathArgs {
117    None,
118    Name(String),
119    Update {
120        name: String,
121        update_id: String,
122    },
123    Arn(String),
124    /// A sub-resource collection scoped to a cluster (create/list).
125    Cluster(String),
126    /// A named sub-resource of a cluster (describe/delete/update).
127    ClusterChild {
128        cluster: String,
129        name: String,
130    },
131    /// A named access policy association on an access entry
132    /// (`DisassociateAccessPolicy`).
133    AccessPolicyChild {
134        cluster: String,
135        principal: String,
136        policy_arn: String,
137    },
138}
139
140impl EksService {
141    pub fn new(state: SharedEksState) -> Self {
142        Self {
143            state,
144            snapshot_store: None,
145            snapshot_lock: Arc::new(AsyncMutex::new(())),
146        }
147    }
148
149    pub fn with_snapshot_store(mut self, store: Arc<dyn SnapshotStore>) -> Self {
150        self.snapshot_store = Some(store);
151        self
152    }
153
154    async fn save_snapshot(&self) {
155        save_eks_snapshot(
156            &self.state,
157            self.snapshot_store.clone(),
158            &self.snapshot_lock,
159        )
160        .await;
161    }
162
163    /// Persist hook for the CloudFormation provisioner; `None` in memory mode.
164    pub fn snapshot_hook(&self) -> Option<fakecloud_persistence::SnapshotHook> {
165        let store = self.snapshot_store.clone()?;
166        let state = self.state.clone();
167        let lock = self.snapshot_lock.clone();
168        Some(Arc::new(move || {
169            let state = state.clone();
170            let store = store.clone();
171            let lock = lock.clone();
172            Box::pin(async move {
173                save_eks_snapshot(&state, Some(store), &lock).await;
174            })
175        }))
176    }
177
178    fn resolve_action(req: &AwsRequest) -> Option<(&'static str, PathArgs)> {
179        // Derive segments from the raw path, preserving INTERNAL empty segments
180        // (only the leading empty from the leading `/` and a single trailing
181        // empty are dropped). `req.path_segments` strips every empty part, which
182        // would collapse a real 3-segment path with an empty middle label
183        // (e.g. `/clusters//addons`, ListAddons with an empty clusterName) down
184        // to `["clusters","addons"]` and mis-route it to DescribeCluster. Keeping
185        // internal empties makes the exact-slice matches below precise, so a
186        // not-yet-implemented sub-resource path falls through to None.
187        let raw = req.raw_path.split('?').next().unwrap_or(&req.raw_path);
188        let trimmed = raw.strip_prefix('/').unwrap_or(raw);
189        let trimmed = trimmed.strip_suffix('/').unwrap_or(trimmed);
190        let segs: Vec<&str> = if trimmed.is_empty() {
191            Vec::new()
192        } else {
193            trimmed.split('/').collect()
194        };
195        match (&req.method, segs.as_slice()) {
196            (&Method::POST, ["clusters"]) => Some(("CreateCluster", PathArgs::None)),
197            (&Method::GET, ["clusters"]) => Some(("ListClusters", PathArgs::None)),
198            (&Method::GET, ["clusters", name]) => {
199                Some(("DescribeCluster", PathArgs::Name(decode(name))))
200            }
201            (&Method::DELETE, ["clusters", name]) => {
202                Some(("DeleteCluster", PathArgs::Name(decode(name))))
203            }
204            (&Method::POST, ["clusters", name, "update-config"]) => {
205                Some(("UpdateClusterConfig", PathArgs::Name(decode(name))))
206            }
207            (&Method::POST, ["clusters", name, "updates"]) => {
208                Some(("UpdateClusterVersion", PathArgs::Name(decode(name))))
209            }
210            (&Method::GET, ["clusters", name, "updates"]) => {
211                Some(("ListUpdates", PathArgs::Name(decode(name))))
212            }
213            (&Method::GET, ["clusters", name, "updates", update_id]) => Some((
214                "DescribeUpdate",
215                PathArgs::Update {
216                    name: decode(name),
217                    update_id: decode(update_id),
218                },
219            )),
220            // Node groups (sub-resources of a cluster).
221            (&Method::POST, ["clusters", c, "node-groups"]) => {
222                Some(("CreateNodegroup", PathArgs::Cluster(decode(c))))
223            }
224            (&Method::GET, ["clusters", c, "node-groups"]) => {
225                Some(("ListNodegroups", PathArgs::Cluster(decode(c))))
226            }
227            (&Method::GET, ["clusters", c, "node-groups", n]) => Some((
228                "DescribeNodegroup",
229                PathArgs::ClusterChild {
230                    cluster: decode(c),
231                    name: decode(n),
232                },
233            )),
234            (&Method::DELETE, ["clusters", c, "node-groups", n]) => Some((
235                "DeleteNodegroup",
236                PathArgs::ClusterChild {
237                    cluster: decode(c),
238                    name: decode(n),
239                },
240            )),
241            (&Method::POST, ["clusters", c, "node-groups", n, "update-config"]) => Some((
242                "UpdateNodegroupConfig",
243                PathArgs::ClusterChild {
244                    cluster: decode(c),
245                    name: decode(n),
246                },
247            )),
248            (&Method::POST, ["clusters", c, "node-groups", n, "update-version"]) => Some((
249                "UpdateNodegroupVersion",
250                PathArgs::ClusterChild {
251                    cluster: decode(c),
252                    name: decode(n),
253                },
254            )),
255            // Fargate profiles (sub-resources of a cluster).
256            (&Method::POST, ["clusters", c, "fargate-profiles"]) => {
257                Some(("CreateFargateProfile", PathArgs::Cluster(decode(c))))
258            }
259            (&Method::GET, ["clusters", c, "fargate-profiles"]) => {
260                Some(("ListFargateProfiles", PathArgs::Cluster(decode(c))))
261            }
262            (&Method::GET, ["clusters", c, "fargate-profiles", n]) => Some((
263                "DescribeFargateProfile",
264                PathArgs::ClusterChild {
265                    cluster: decode(c),
266                    name: decode(n),
267                },
268            )),
269            (&Method::DELETE, ["clusters", c, "fargate-profiles", n]) => Some((
270                "DeleteFargateProfile",
271                PathArgs::ClusterChild {
272                    cluster: decode(c),
273                    name: decode(n),
274                },
275            )),
276            // Add-ons (sub-resources of a cluster).
277            (&Method::POST, ["clusters", c, "addons"]) => {
278                Some(("CreateAddon", PathArgs::Cluster(decode(c))))
279            }
280            (&Method::GET, ["clusters", c, "addons"]) => {
281                Some(("ListAddons", PathArgs::Cluster(decode(c))))
282            }
283            (&Method::GET, ["clusters", c, "addons", n]) => Some((
284                "DescribeAddon",
285                PathArgs::ClusterChild {
286                    cluster: decode(c),
287                    name: decode(n),
288                },
289            )),
290            (&Method::DELETE, ["clusters", c, "addons", n]) => Some((
291                "DeleteAddon",
292                PathArgs::ClusterChild {
293                    cluster: decode(c),
294                    name: decode(n),
295                },
296            )),
297            (&Method::POST, ["clusters", c, "addons", n, "update"]) => Some((
298                "UpdateAddon",
299                PathArgs::ClusterChild {
300                    cluster: decode(c),
301                    name: decode(n),
302                },
303            )),
304            // Add-on catalogue ops (not scoped to a cluster).
305            (&Method::GET, ["addons", "supported-versions"]) => {
306                Some(("DescribeAddonVersions", PathArgs::None))
307            }
308            (&Method::GET, ["addons", "configuration-schemas"]) => {
309                Some(("DescribeAddonConfiguration", PathArgs::None))
310            }
311            // Access entries (sub-resources of a cluster, keyed by principalArn).
312            (&Method::POST, ["clusters", c, "access-entries"]) => {
313                Some(("CreateAccessEntry", PathArgs::Cluster(decode(c))))
314            }
315            (&Method::GET, ["clusters", c, "access-entries"]) => {
316                Some(("ListAccessEntries", PathArgs::Cluster(decode(c))))
317            }
318            (&Method::GET, ["clusters", c, "access-entries", p]) => Some((
319                "DescribeAccessEntry",
320                PathArgs::ClusterChild {
321                    cluster: decode(c),
322                    name: decode(p),
323                },
324            )),
325            (&Method::DELETE, ["clusters", c, "access-entries", p]) => Some((
326                "DeleteAccessEntry",
327                PathArgs::ClusterChild {
328                    cluster: decode(c),
329                    name: decode(p),
330                },
331            )),
332            (&Method::POST, ["clusters", c, "access-entries", p]) => Some((
333                "UpdateAccessEntry",
334                PathArgs::ClusterChild {
335                    cluster: decode(c),
336                    name: decode(p),
337                },
338            )),
339            // Access-policy associations on an access entry.
340            (&Method::POST, ["clusters", c, "access-entries", p, "access-policies"]) => Some((
341                "AssociateAccessPolicy",
342                PathArgs::ClusterChild {
343                    cluster: decode(c),
344                    name: decode(p),
345                },
346            )),
347            (&Method::GET, ["clusters", c, "access-entries", p, "access-policies"]) => Some((
348                "ListAssociatedAccessPolicies",
349                PathArgs::ClusterChild {
350                    cluster: decode(c),
351                    name: decode(p),
352                },
353            )),
354            (&Method::DELETE, ["clusters", c, "access-entries", p, "access-policies", policy]) => {
355                Some((
356                    "DisassociateAccessPolicy",
357                    PathArgs::AccessPolicyChild {
358                        cluster: decode(c),
359                        principal: decode(p),
360                        policy_arn: decode(policy),
361                    },
362                ))
363            }
364            // Access-policy catalogue (not scoped to a cluster).
365            (&Method::GET, ["access-policies"]) => Some(("ListAccessPolicies", PathArgs::None)),
366            // Identity-provider configs (sub-resources of a cluster). The
367            // associate/disassociate/describe forms are POSTs to fixed action
368            // sub-paths, carrying the config identity in the body.
369            (&Method::POST, ["clusters", c, "identity-provider-configs", "associate"]) => Some((
370                "AssociateIdentityProviderConfig",
371                PathArgs::Cluster(decode(c)),
372            )),
373            (&Method::POST, ["clusters", c, "identity-provider-configs", "disassociate"]) => {
374                Some((
375                    "DisassociateIdentityProviderConfig",
376                    PathArgs::Cluster(decode(c)),
377                ))
378            }
379            (&Method::POST, ["clusters", c, "identity-provider-configs", "describe"]) => Some((
380                "DescribeIdentityProviderConfig",
381                PathArgs::Cluster(decode(c)),
382            )),
383            (&Method::GET, ["clusters", c, "identity-provider-configs"]) => {
384                Some(("ListIdentityProviderConfigs", PathArgs::Cluster(decode(c))))
385            }
386            // Pod identity associations (sub-resources of a cluster, keyed by id).
387            (&Method::POST, ["clusters", c, "pod-identity-associations"]) => {
388                Some(("CreatePodIdentityAssociation", PathArgs::Cluster(decode(c))))
389            }
390            (&Method::GET, ["clusters", c, "pod-identity-associations"]) => {
391                Some(("ListPodIdentityAssociations", PathArgs::Cluster(decode(c))))
392            }
393            (&Method::GET, ["clusters", c, "pod-identity-associations", id]) => Some((
394                "DescribePodIdentityAssociation",
395                PathArgs::ClusterChild {
396                    cluster: decode(c),
397                    name: decode(id),
398                },
399            )),
400            (&Method::DELETE, ["clusters", c, "pod-identity-associations", id]) => Some((
401                "DeletePodIdentityAssociation",
402                PathArgs::ClusterChild {
403                    cluster: decode(c),
404                    name: decode(id),
405                },
406            )),
407            (&Method::POST, ["clusters", c, "pod-identity-associations", id]) => Some((
408                "UpdatePodIdentityAssociation",
409                PathArgs::ClusterChild {
410                    cluster: decode(c),
411                    name: decode(id),
412                },
413            )),
414            // Insights (sub-resources of a cluster). List is a POST (it carries
415            // a filter body); Describe is a GET keyed by insight id.
416            (&Method::POST, ["clusters", c, "insights"]) => {
417                Some(("ListInsights", PathArgs::Cluster(decode(c))))
418            }
419            (&Method::GET, ["clusters", c, "insights", id]) => Some((
420                "DescribeInsight",
421                PathArgs::ClusterChild {
422                    cluster: decode(c),
423                    name: decode(id),
424                },
425            )),
426            (&Method::GET, ["clusters", c, "insights-refresh"]) => {
427                Some(("DescribeInsightsRefresh", PathArgs::Cluster(decode(c))))
428            }
429            (&Method::POST, ["clusters", c, "insights-refresh"]) => {
430                Some(("StartInsightsRefresh", PathArgs::Cluster(decode(c))))
431            }
432            // Encryption config association (mints a tracked cluster Update).
433            (&Method::POST, ["clusters", c, "encryption-config", "associate"]) => {
434                Some(("AssociateEncryptionConfig", PathArgs::Cluster(decode(c))))
435            }
436            // Cancel an in-progress update.
437            (&Method::POST, ["clusters", name, "updates", update_id, "cancel-update"]) => Some((
438                "CancelUpdate",
439                PathArgs::Update {
440                    name: decode(name),
441                    update_id: decode(update_id),
442                },
443            )),
444            // Connected (registered) clusters.
445            (&Method::POST, ["cluster-registrations"]) => Some(("RegisterCluster", PathArgs::None)),
446            (&Method::DELETE, ["cluster-registrations", name]) => {
447                Some(("DeregisterCluster", PathArgs::Name(decode(name))))
448            }
449            // Kubernetes version catalogue (account/region-scoped).
450            (&Method::GET, ["cluster-versions"]) => {
451                Some(("DescribeClusterVersions", PathArgs::None))
452            }
453            // Capabilities (sub-resources of a cluster).
454            (&Method::POST, ["clusters", c, "capabilities"]) => {
455                Some(("CreateCapability", PathArgs::Cluster(decode(c))))
456            }
457            (&Method::GET, ["clusters", c, "capabilities"]) => {
458                Some(("ListCapabilities", PathArgs::Cluster(decode(c))))
459            }
460            (&Method::GET, ["clusters", c, "capabilities", n]) => Some((
461                "DescribeCapability",
462                PathArgs::ClusterChild {
463                    cluster: decode(c),
464                    name: decode(n),
465                },
466            )),
467            (&Method::DELETE, ["clusters", c, "capabilities", n]) => Some((
468                "DeleteCapability",
469                PathArgs::ClusterChild {
470                    cluster: decode(c),
471                    name: decode(n),
472                },
473            )),
474            (&Method::POST, ["clusters", c, "capabilities", n]) => Some((
475                "UpdateCapability",
476                PathArgs::ClusterChild {
477                    cluster: decode(c),
478                    name: decode(n),
479                },
480            )),
481            // EKS Anywhere subscriptions (account-scoped).
482            (&Method::POST, ["eks-anywhere-subscriptions"]) => {
483                Some(("CreateEksAnywhereSubscription", PathArgs::None))
484            }
485            (&Method::GET, ["eks-anywhere-subscriptions"]) => {
486                Some(("ListEksAnywhereSubscriptions", PathArgs::None))
487            }
488            (&Method::GET, ["eks-anywhere-subscriptions", id]) => Some((
489                "DescribeEksAnywhereSubscription",
490                PathArgs::Name(decode(id)),
491            )),
492            (&Method::DELETE, ["eks-anywhere-subscriptions", id]) => {
493                Some(("DeleteEksAnywhereSubscription", PathArgs::Name(decode(id))))
494            }
495            (&Method::POST, ["eks-anywhere-subscriptions", id]) => {
496                Some(("UpdateEksAnywhereSubscription", PathArgs::Name(decode(id))))
497            }
498            (&Method::POST, ["tags", arn]) => Some(("TagResource", PathArgs::Arn(decode(arn)))),
499            (&Method::DELETE, ["tags", arn]) => Some(("UntagResource", PathArgs::Arn(decode(arn)))),
500            (&Method::GET, ["tags", arn]) => {
501                Some(("ListTagsForResource", PathArgs::Arn(decode(arn))))
502            }
503            _ => None,
504        }
505    }
506
507    fn create_cluster(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
508        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
509
510        let name = body
511            .get("name")
512            .and_then(|v| v.as_str())
513            .ok_or_else(|| invalid_parameter("name is required"))?
514            .to_string();
515        validate_cluster_name(&name)?;
516
517        let role_arn = body
518            .get("roleArn")
519            .and_then(|v| v.as_str())
520            .ok_or_else(|| invalid_parameter("roleArn is required"))?
521            .to_string();
522
523        let vpc_req = body
524            .get("resourcesVpcConfig")
525            .filter(|v| v.is_object())
526            .ok_or_else(|| invalid_parameter("resourcesVpcConfig is required"))?;
527
528        let version = body
529            .get("version")
530            .and_then(|v| v.as_str())
531            .unwrap_or(DEFAULT_K8S_VERSION)
532            .to_string();
533
534        let mut accounts = self.state.write();
535        let state = accounts.get_or_create(&req.account_id);
536        if state.clusters.contains_key(&name) {
537            return Err(AwsServiceError::aws_error(
538                StatusCode::CONFLICT,
539                "ResourceInUseException",
540                format!("Cluster already exists with name: {name}"),
541            ));
542        }
543
544        let region = req.region.clone();
545        let account_id = req.account_id.clone();
546        let arn = cluster_arn(&region, &account_id, &name);
547        let id = uuid::Uuid::new_v4().to_string();
548
549        let tags = parse_tag_map(body.get("tags"));
550
551        let cluster = Cluster {
552            name: name.clone(),
553            arn: arn.clone(),
554            version,
555            role_arn,
556            status: "CREATING".to_string(),
557            created_at: Utc::now(),
558            endpoint: format!(
559                "https://{}.gr7.{region}.eks.amazonaws.com",
560                id.replace('-', "").to_uppercase()
561            ),
562            platform_version: "eks.1".to_string(),
563            certificate_authority_data: default_ca_data(),
564            resources_vpc_config: build_vpc_config_response(vpc_req, &id),
565            kubernetes_network_config: build_k8s_network_config(
566                body.get("kubernetesNetworkConfig"),
567            ),
568            logging: build_logging(body.get("logging")),
569            tags,
570            updates: Default::default(),
571            connector_config: None,
572            encryption_config: None,
573            access_config: build_access_config(body.get("accessConfig")),
574            upgrade_policy: build_upgrade_policy(body.get("upgradePolicy")),
575            compute_config: body.get("computeConfig").cloned(),
576            storage_config: body.get("storageConfig").cloned(),
577            zonal_shift_config: body.get("zonalShiftConfig").cloned(),
578            remote_network_config: body.get("remoteNetworkConfig").cloned(),
579            control_plane_scaling_config: body.get("controlPlaneScalingConfig").cloned(),
580            deletion_protection: body.get("deletionProtection").and_then(|v| v.as_bool()),
581        };
582
583        let out = cluster_json(&cluster, &id);
584        state.clusters.insert(name, cluster);
585        Ok(AwsResponse::json(
586            StatusCode::OK,
587            json!({ "cluster": out }).to_string(),
588        ))
589    }
590
591    fn describe_cluster(
592        &self,
593        req: &AwsRequest,
594        name: &str,
595    ) -> Result<AwsResponse, AwsServiceError> {
596        let mut accounts = self.state.write();
597        let state = accounts.get_or_create(&req.account_id);
598        let cluster = state
599            .clusters
600            .get_mut(name)
601            .ok_or_else(not_found_cluster(name))?;
602        // Lazily settle CREATING -> ACTIVE on first describe, mirroring the
603        // eventual-consistency of the real control plane.
604        if cluster.status == "CREATING" {
605            cluster.status = "ACTIVE".to_string();
606        }
607        let id = arn_cluster_id(&cluster.endpoint);
608        Ok(AwsResponse::json(
609            StatusCode::OK,
610            json!({ "cluster": cluster_json(cluster, &id) }).to_string(),
611        ))
612    }
613
614    fn list_clusters(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
615        let max_results = validate_max_results(req)?;
616        let next_token = req.query_params.get("nextToken").cloned();
617
618        let accounts = self.state.read();
619        let Some(state) = accounts.get(&req.account_id) else {
620            return Ok(AwsResponse::json(
621                StatusCode::OK,
622                json!({ "clusters": [] }).to_string(),
623            ));
624        };
625        let names: Vec<String> = state.clusters.keys().cloned().collect();
626        let (page, token) = paginate_checked(&names, next_token.as_deref(), max_results)
627            .map_err(|_| invalid_parameter("Invalid nextToken"))?;
628        let mut out = json!({ "clusters": page });
629        if let Some(t) = token {
630            out["nextToken"] = Value::String(t);
631        }
632        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
633    }
634
635    fn delete_cluster(&self, req: &AwsRequest, name: &str) -> Result<AwsResponse, AwsServiceError> {
636        let mut accounts = self.state.write();
637        let state = accounts.get_or_create(&req.account_id);
638        if !state.clusters.contains_key(name) {
639            return Err(not_found_cluster(name)());
640        }
641        // Real EKS refuses to delete a cluster that still has managed node
642        // groups or Fargate profiles attached; the caller must delete those
643        // first.
644        let has_nodegroups = state.nodegroups.get(name).is_some_and(|m| !m.is_empty());
645        let has_fargate = state
646            .fargate_profiles
647            .get(name)
648            .is_some_and(|m| !m.is_empty());
649        if has_nodegroups || has_fargate {
650            return Err(AwsServiceError::aws_error(
651                StatusCode::CONFLICT,
652                "ResourceInUseException",
653                format!(
654                    "Cluster has {} attached that must be deleted first",
655                    if has_nodegroups {
656                        "nodegroups"
657                    } else {
658                        "Fargate profiles"
659                    }
660                ),
661            ));
662        }
663        // Cascade-remove every remaining cluster sub-resource so a same-name
664        // recreate does not inherit the previous cluster's orphaned state.
665        state.nodegroups.remove(name);
666        state.fargate_profiles.remove(name);
667        state.addons.remove(name);
668        state.access_entries.remove(name);
669        state.identity_provider_configs.remove(name);
670        state.pod_identity_associations.remove(name);
671        state.insights.remove(name);
672        state.insights_refresh.remove(name);
673        state.capabilities.remove(name);
674        let mut cluster = state
675            .clusters
676            .remove(name)
677            .expect("existence checked above");
678        cluster.status = "DELETING".to_string();
679        let id = arn_cluster_id(&cluster.endpoint);
680        Ok(AwsResponse::json(
681            StatusCode::OK,
682            json!({ "cluster": cluster_json(&cluster, &id) }).to_string(),
683        ))
684    }
685
686    fn update_cluster_config(
687        &self,
688        req: &AwsRequest,
689        name: &str,
690    ) -> Result<AwsResponse, AwsServiceError> {
691        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
692        let mut accounts = self.state.write();
693        let state = accounts.get_or_create(&req.account_id);
694        let cluster = state
695            .clusters
696            .get_mut(name)
697            .ok_or_else(not_found_cluster(name))?;
698
699        // A single UpdateClusterConfig call may carry more than one config
700        // aspect; apply every field present (previously all but logging and
701        // resourcesVpcConfig were silently dropped) and report the first-matched
702        // aspect as the Update's type for a faithful DescribeUpdate round-trip.
703        let mut update_type = "ConfigUpdate";
704        let mut matched = false;
705        let mut params: Vec<(String, String)> = Vec::new();
706        macro_rules! mark {
707            ($ty:expr) => {
708                if !matched {
709                    update_type = $ty;
710                    matched = true;
711                }
712            };
713        }
714
715        if let Some(logging) = body.get("logging") {
716            cluster.logging = build_logging(Some(logging));
717            mark!("LoggingUpdate");
718            params.push(("ClusterLogging".to_string(), logging.to_string()));
719        }
720        if let Some(vpc) = body.get("resourcesVpcConfig") {
721            let id = arn_cluster_id(&cluster.endpoint);
722            cluster.resources_vpc_config = build_vpc_config_response(vpc, &id);
723            mark!("VpcConfigUpdate");
724            params.push(("ResourcesVpcConfig".to_string(), vpc.to_string()));
725        }
726        if let Some(ac) = body.get("accessConfig") {
727            cluster.access_config = build_access_config(Some(ac));
728            mark!("AccessConfigUpdate");
729            params.push((
730                "AuthenticationMode".to_string(),
731                cluster.access_config["authenticationMode"].to_string(),
732            ));
733        }
734        if let Some(up) = body.get("upgradePolicy") {
735            cluster.upgrade_policy = build_upgrade_policy(Some(up));
736            mark!("UpgradePolicyUpdate");
737            params.push((
738                "SupportType".to_string(),
739                cluster.upgrade_policy["supportType"].to_string(),
740            ));
741        }
742        if let Some(kn) = body.get("kubernetesNetworkConfig") {
743            cluster.kubernetes_network_config = build_k8s_network_config(Some(kn));
744            mark!("VpcConfigUpdate");
745            params.push(("KubernetesNetworkConfig".to_string(), kn.to_string()));
746        }
747        if let Some(cc) = body.get("computeConfig") {
748            cluster.compute_config = Some(cc.clone());
749            mark!("AutoModeUpdate");
750            params.push(("ComputeConfig".to_string(), cc.to_string()));
751        }
752        if let Some(sc) = body.get("storageConfig") {
753            cluster.storage_config = Some(sc.clone());
754            mark!("AutoModeUpdate");
755            params.push(("StorageConfig".to_string(), sc.to_string()));
756        }
757        if let Some(z) = body.get("zonalShiftConfig") {
758            cluster.zonal_shift_config = Some(z.clone());
759            mark!("ZonalShiftConfigUpdate");
760            params.push(("ZonalShiftConfig".to_string(), z.to_string()));
761        }
762        if let Some(rn) = body.get("remoteNetworkConfig") {
763            cluster.remote_network_config = Some(rn.clone());
764            mark!("RemoteNetworkConfigUpdate");
765            params.push(("RemoteNetworkConfig".to_string(), rn.to_string()));
766        }
767        if let Some(sp) = body.get("controlPlaneScalingConfig") {
768            cluster.control_plane_scaling_config = Some(sp.clone());
769            mark!("ControlPlaneScalingConfigUpdate");
770            params.push(("ControlPlaneScalingConfig".to_string(), sp.to_string()));
771        }
772        if let Some(dp) = body.get("deletionProtection").and_then(|v| v.as_bool()) {
773            cluster.deletion_protection = Some(dp);
774            mark!("DeletionProtectionUpdate");
775            params.push(("DeletionProtection".to_string(), dp.to_string()));
776        }
777        let _ = matched;
778
779        let update = new_update(update_type, params);
780        let out = update_json(&update);
781        cluster.updates.insert(update.id.clone(), update);
782        Ok(AwsResponse::json(
783            StatusCode::OK,
784            json!({ "update": out }).to_string(),
785        ))
786    }
787
788    fn update_cluster_version(
789        &self,
790        req: &AwsRequest,
791        name: &str,
792    ) -> Result<AwsResponse, AwsServiceError> {
793        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
794        let version = body
795            .get("version")
796            .and_then(|v| v.as_str())
797            .ok_or_else(|| invalid_parameter("version is required"))?
798            .to_string();
799
800        let mut accounts = self.state.write();
801        let state = accounts.get_or_create(&req.account_id);
802        let cluster = state
803            .clusters
804            .get_mut(name)
805            .ok_or_else(not_found_cluster(name))?;
806        cluster.version = version.clone();
807
808        let update = new_update(
809            "VersionUpdate",
810            vec![
811                ("Version".to_string(), version),
812                (
813                    "PlatformVersion".to_string(),
814                    cluster.platform_version.clone(),
815                ),
816            ],
817        );
818        let out = update_json(&update);
819        cluster.updates.insert(update.id.clone(), update);
820        Ok(AwsResponse::json(
821            StatusCode::OK,
822            json!({ "update": out }).to_string(),
823        ))
824    }
825
826    fn describe_update(
827        &self,
828        req: &AwsRequest,
829        name: &str,
830        update_id: &str,
831    ) -> Result<AwsResponse, AwsServiceError> {
832        let nodegroup_name = req.query_params.get("nodegroupName").cloned();
833        let addon_name = req.query_params.get("addonName").cloned();
834        let mut accounts = self.state.write();
835        let state = accounts.get_or_create(&req.account_id);
836        if !state.clusters.contains_key(name) {
837            return Err(not_found_cluster(name)());
838        }
839        // `DescribeUpdate` disambiguates cluster updates from node-group and
840        // add-on updates via the optional `nodegroupName` / `addonName` query
841        // params, mirroring the real API. When present, the update lives on
842        // that sub-resource's own update history.
843        let update = if let Some(ng_name) = nodegroup_name.as_deref() {
844            let ng = state
845                .nodegroups
846                .get_mut(name)
847                .and_then(|m| m.get_mut(ng_name))
848                .ok_or_else(not_found_nodegroup(ng_name))?;
849            ng.updates
850                .get_mut(update_id)
851                .ok_or_else(not_found_update(update_id))?
852        } else if let Some(a_name) = addon_name.as_deref() {
853            let addon = state
854                .addons
855                .get_mut(name)
856                .and_then(|m| m.get_mut(a_name))
857                .ok_or_else(not_found_addon(a_name))?;
858            addon
859                .updates
860                .get_mut(update_id)
861                .ok_or_else(not_found_update(update_id))?
862        } else {
863            let cluster = state
864                .clusters
865                .get_mut(name)
866                .ok_or_else(not_found_cluster(name))?;
867            cluster
868                .updates
869                .get_mut(update_id)
870                .ok_or_else(not_found_update(update_id))?
871        };
872        // Updates settle to Successful on first describe.
873        if update.status == "InProgress" {
874            update.status = "Successful".to_string();
875        }
876        Ok(AwsResponse::json(
877            StatusCode::OK,
878            json!({ "update": update_json(update) }).to_string(),
879        ))
880    }
881
882    fn list_updates(&self, req: &AwsRequest, name: &str) -> Result<AwsResponse, AwsServiceError> {
883        let max_results = validate_max_results(req)?;
884        let next_token = req.query_params.get("nextToken").cloned();
885
886        let nodegroup_name = req.query_params.get("nodegroupName").cloned();
887        let addon_name = req.query_params.get("addonName").cloned();
888        let accounts = self.state.read();
889        let state = accounts
890            .get(&req.account_id)
891            .ok_or_else(not_found_cluster(name))?;
892        if !state.clusters.contains_key(name) {
893            return Err(not_found_cluster(name)());
894        }
895        let ids: Vec<String> = if let Some(ng_name) = nodegroup_name.as_deref() {
896            let ng = state
897                .nodegroups
898                .get(name)
899                .and_then(|m| m.get(ng_name))
900                .ok_or_else(not_found_nodegroup(ng_name))?;
901            ng.updates.keys().cloned().collect()
902        } else if let Some(a_name) = addon_name.as_deref() {
903            let addon = state
904                .addons
905                .get(name)
906                .and_then(|m| m.get(a_name))
907                .ok_or_else(not_found_addon(a_name))?;
908            addon.updates.keys().cloned().collect()
909        } else {
910            let cluster = state
911                .clusters
912                .get(name)
913                .ok_or_else(not_found_cluster(name))?;
914            cluster.updates.keys().cloned().collect()
915        };
916        let (page, token) = paginate_checked(&ids, next_token.as_deref(), max_results)
917            .map_err(|_| invalid_parameter("Invalid nextToken"))?;
918        let mut out = json!({ "updateIds": page });
919        if let Some(t) = token {
920            out["nextToken"] = Value::String(t);
921        }
922        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
923    }
924
925    fn tag_resource(&self, req: &AwsRequest, arn: &str) -> Result<AwsResponse, AwsServiceError> {
926        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
927        let tags = body
928            .get("tags")
929            .filter(|v| v.is_object())
930            .ok_or_else(|| bad_request("tags is required"))?;
931        validate_eks_arn(arn)?;
932        let mut accounts = self.state.write();
933        let state = accounts.get_or_create(&req.account_id);
934        // Resolve the ARN to whichever resource owns it (cluster or any
935        // sub-resource) and tag that resource. Previously every non-cluster ARN
936        // was rejected with a 400, so tagging a node group / add-on failed.
937        let target = locate_tag_target(state, arn);
938        let dest = tags_mut(state, &target);
939        for (k, v) in tags.as_object().unwrap() {
940            if let Some(v) = v.as_str() {
941                dest.insert(k.clone(), v.to_string());
942            }
943        }
944        Ok(AwsResponse::json(StatusCode::OK, "{}"))
945    }
946
947    fn untag_resource(&self, req: &AwsRequest, arn: &str) -> Result<AwsResponse, AwsServiceError> {
948        let keys = parse_multi_query(&req.raw_query, "tagKeys");
949        validate_eks_arn(arn)?;
950        let mut accounts = self.state.write();
951        let state = accounts.get_or_create(&req.account_id);
952        let target = locate_tag_target(state, arn);
953        let dest = tags_mut(state, &target);
954        for k in keys {
955            dest.remove(&k);
956        }
957        Ok(AwsResponse::json(StatusCode::OK, "{}"))
958    }
959
960    fn list_tags_for_resource(
961        &self,
962        req: &AwsRequest,
963        arn: &str,
964    ) -> Result<AwsResponse, AwsServiceError> {
965        validate_eks_arn(arn)?;
966        let accounts = self.state.read();
967        let tags: serde_json::Map<String, Value> = accounts
968            .get(&req.account_id)
969            .map(|state| {
970                let target = locate_tag_target(state, arn);
971                tags_ref(state, &target)
972                    .map(|t| {
973                        t.iter()
974                            .map(|(k, v)| (k.clone(), Value::String(v.clone())))
975                            .collect()
976                    })
977                    .unwrap_or_default()
978            })
979            .unwrap_or_default();
980        Ok(AwsResponse::json(
981            StatusCode::OK,
982            json!({ "tags": tags }).to_string(),
983        ))
984    }
985
986    // -----------------------------------------------------------------------
987    // Node groups
988    // -----------------------------------------------------------------------
989
990    fn create_nodegroup(
991        &self,
992        req: &AwsRequest,
993        cluster_name: &str,
994    ) -> Result<AwsResponse, AwsServiceError> {
995        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
996        let name = body
997            .get("nodegroupName")
998            .and_then(|v| v.as_str())
999            .ok_or_else(|| invalid_parameter("nodegroupName is required"))?
1000            .to_string();
1001        let node_role = body
1002            .get("nodeRole")
1003            .and_then(|v| v.as_str())
1004            .ok_or_else(|| invalid_parameter("nodeRole is required"))?
1005            .to_string();
1006        let subnets = body
1007            .get("subnets")
1008            .filter(|v| v.is_array())
1009            .cloned()
1010            .ok_or_else(|| invalid_parameter("subnets is required"))?;
1011
1012        let region = req.region.clone();
1013        let account_id = req.account_id.clone();
1014
1015        let mut accounts = self.state.write();
1016        let state = accounts.get_or_create(&req.account_id);
1017        // The parent cluster must exist. AWS returns ResourceNotFoundException
1018        // here even though the Smithy model omits it from CreateNodegroup's
1019        // declared errors.
1020        let cluster_version = state
1021            .clusters
1022            .get(cluster_name)
1023            .ok_or_else(not_found_cluster(cluster_name))?
1024            .version
1025            .clone();
1026        if state
1027            .nodegroups
1028            .get(cluster_name)
1029            .is_some_and(|m| m.contains_key(&name))
1030        {
1031            return Err(AwsServiceError::aws_error(
1032                StatusCode::CONFLICT,
1033                "ResourceInUseException",
1034                format!(
1035                    "NodeGroup already exists with name {name} and cluster name {cluster_name}"
1036                ),
1037            ));
1038        }
1039
1040        let id = uuid::Uuid::new_v4().to_string();
1041        let arn = nodegroup_arn(&region, &account_id, cluster_name, &name, &id);
1042        let now = Utc::now();
1043        let version = body
1044            .get("version")
1045            .and_then(|v| v.as_str())
1046            .map(|s| s.to_string())
1047            .unwrap_or(cluster_version);
1048        let release_version = body
1049            .get("releaseVersion")
1050            .and_then(|v| v.as_str())
1051            .map(|s| s.to_string())
1052            .unwrap_or_else(|| format!("{version}-20240000"));
1053        let capacity_type = body
1054            .get("capacityType")
1055            .and_then(|v| v.as_str())
1056            .unwrap_or("ON_DEMAND")
1057            .to_string();
1058        let ami_type = body
1059            .get("amiType")
1060            .and_then(|v| v.as_str())
1061            .unwrap_or("AL2023_x86_64_STANDARD")
1062            .to_string();
1063        let disk_size = body.get("diskSize").and_then(|v| v.as_i64()).unwrap_or(20);
1064
1065        let ng = Nodegroup {
1066            name: name.clone(),
1067            arn,
1068            cluster_name: cluster_name.to_string(),
1069            version,
1070            release_version,
1071            status: "CREATING".to_string(),
1072            capacity_type,
1073            ami_type,
1074            node_role,
1075            created_at: now,
1076            modified_at: now,
1077            disk_size,
1078            scaling_config: build_scaling_config(body.get("scalingConfig")),
1079            update_config: build_nodegroup_update_config(body.get("updateConfig")),
1080            instance_types: body
1081                .get("instanceTypes")
1082                .cloned()
1083                .unwrap_or_else(|| json!(["t3.medium"])),
1084            subnets,
1085            labels: body.get("labels").cloned().unwrap_or_else(|| json!({})),
1086            taints: body.get("taints").cloned().unwrap_or_else(|| json!([])),
1087            remote_access: body.get("remoteAccess").cloned(),
1088            launch_template: body.get("launchTemplate").cloned(),
1089            asg_name: format!("eks-{name}-{}", &id[..8]),
1090            tags: parse_tag_map(body.get("tags")),
1091            updates: Default::default(),
1092        };
1093
1094        let out = nodegroup_json(&ng);
1095        state
1096            .nodegroups
1097            .entry(cluster_name.to_string())
1098            .or_default()
1099            .insert(name, ng);
1100        Ok(AwsResponse::json(
1101            StatusCode::OK,
1102            json!({ "nodegroup": out }).to_string(),
1103        ))
1104    }
1105
1106    fn describe_nodegroup(
1107        &self,
1108        req: &AwsRequest,
1109        cluster_name: &str,
1110        name: &str,
1111    ) -> Result<AwsResponse, AwsServiceError> {
1112        let mut accounts = self.state.write();
1113        let state = accounts.get_or_create(&req.account_id);
1114        if !state.clusters.contains_key(cluster_name) {
1115            return Err(not_found_cluster(cluster_name)());
1116        }
1117        let ng = state
1118            .nodegroups
1119            .get_mut(cluster_name)
1120            .and_then(|m| m.get_mut(name))
1121            .ok_or_else(not_found_nodegroup(name))?;
1122        if ng.status == "CREATING" {
1123            ng.status = "ACTIVE".to_string();
1124        }
1125        Ok(AwsResponse::json(
1126            StatusCode::OK,
1127            json!({ "nodegroup": nodegroup_json(ng) }).to_string(),
1128        ))
1129    }
1130
1131    fn list_nodegroups(
1132        &self,
1133        req: &AwsRequest,
1134        cluster_name: &str,
1135    ) -> Result<AwsResponse, AwsServiceError> {
1136        let max_results = validate_max_results(req)?;
1137        let next_token = req.query_params.get("nextToken").cloned();
1138        let accounts = self.state.read();
1139        let state = accounts
1140            .get(&req.account_id)
1141            .ok_or_else(not_found_cluster(cluster_name))?;
1142        if !state.clusters.contains_key(cluster_name) {
1143            return Err(not_found_cluster(cluster_name)());
1144        }
1145        let names: Vec<String> = state
1146            .nodegroups
1147            .get(cluster_name)
1148            .map(|m| m.keys().cloned().collect())
1149            .unwrap_or_default();
1150        let (page, token) = paginate_checked(&names, next_token.as_deref(), max_results)
1151            .map_err(|_| invalid_parameter("Invalid nextToken"))?;
1152        let mut out = json!({ "nodegroups": page });
1153        if let Some(t) = token {
1154            out["nextToken"] = Value::String(t);
1155        }
1156        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
1157    }
1158
1159    fn delete_nodegroup(
1160        &self,
1161        req: &AwsRequest,
1162        cluster_name: &str,
1163        name: &str,
1164    ) -> Result<AwsResponse, AwsServiceError> {
1165        let mut accounts = self.state.write();
1166        let state = accounts.get_or_create(&req.account_id);
1167        if !state.clusters.contains_key(cluster_name) {
1168            return Err(not_found_cluster(cluster_name)());
1169        }
1170        let mut ng = state
1171            .nodegroups
1172            .get_mut(cluster_name)
1173            .and_then(|m| m.remove(name))
1174            .ok_or_else(not_found_nodegroup(name))?;
1175        ng.status = "DELETING".to_string();
1176        Ok(AwsResponse::json(
1177            StatusCode::OK,
1178            json!({ "nodegroup": nodegroup_json(&ng) }).to_string(),
1179        ))
1180    }
1181
1182    fn update_nodegroup_config(
1183        &self,
1184        req: &AwsRequest,
1185        cluster_name: &str,
1186        name: &str,
1187    ) -> Result<AwsResponse, AwsServiceError> {
1188        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
1189        let mut accounts = self.state.write();
1190        let state = accounts.get_or_create(&req.account_id);
1191        if !state.clusters.contains_key(cluster_name) {
1192            return Err(not_found_cluster(cluster_name)());
1193        }
1194        let ng = state
1195            .nodegroups
1196            .get_mut(cluster_name)
1197            .and_then(|m| m.get_mut(name))
1198            .ok_or_else(not_found_nodegroup(name))?;
1199
1200        let mut params = Vec::new();
1201        if let Some(scaling) = body.get("scalingConfig") {
1202            ng.scaling_config = build_scaling_config(Some(scaling));
1203            params.push(("ScalingConfig".to_string(), scaling.to_string()));
1204        }
1205        if let Some(labels) = body.get("labels") {
1206            if let Some(add) = labels.get("addOrUpdateLabels").and_then(|v| v.as_object()) {
1207                let map = ng.labels.as_object_mut();
1208                if let Some(map) = map {
1209                    for (k, v) in add {
1210                        map.insert(k.clone(), v.clone());
1211                    }
1212                }
1213            }
1214            params.push(("LabelsToAdd".to_string(), labels.to_string()));
1215        }
1216        if let Some(update_config) = body.get("updateConfig") {
1217            ng.update_config = build_nodegroup_update_config(Some(update_config));
1218            params.push(("MaxUnavailable".to_string(), update_config.to_string()));
1219        }
1220        ng.modified_at = Utc::now();
1221
1222        let update = new_update("ConfigUpdate", params);
1223        let out = update_json(&update);
1224        ng.updates.insert(update.id.clone(), update);
1225        Ok(AwsResponse::json(
1226            StatusCode::OK,
1227            json!({ "update": out }).to_string(),
1228        ))
1229    }
1230
1231    fn update_nodegroup_version(
1232        &self,
1233        req: &AwsRequest,
1234        cluster_name: &str,
1235        name: &str,
1236    ) -> Result<AwsResponse, AwsServiceError> {
1237        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
1238        let mut accounts = self.state.write();
1239        let state = accounts.get_or_create(&req.account_id);
1240        if !state.clusters.contains_key(cluster_name) {
1241            return Err(not_found_cluster(cluster_name)());
1242        }
1243        let ng = state
1244            .nodegroups
1245            .get_mut(cluster_name)
1246            .and_then(|m| m.get_mut(name))
1247            .ok_or_else(not_found_nodegroup(name))?;
1248
1249        let mut params = Vec::new();
1250        if let Some(version) = body.get("version").and_then(|v| v.as_str()) {
1251            ng.version = version.to_string();
1252            params.push(("Version".to_string(), version.to_string()));
1253        }
1254        if let Some(release) = body.get("releaseVersion").and_then(|v| v.as_str()) {
1255            ng.release_version = release.to_string();
1256            params.push(("ReleaseVersion".to_string(), release.to_string()));
1257        }
1258        ng.modified_at = Utc::now();
1259
1260        let update = new_update("VersionUpdate", params);
1261        let out = update_json(&update);
1262        ng.updates.insert(update.id.clone(), update);
1263        Ok(AwsResponse::json(
1264            StatusCode::OK,
1265            json!({ "update": out }).to_string(),
1266        ))
1267    }
1268
1269    // -----------------------------------------------------------------------
1270    // Fargate profiles
1271    // -----------------------------------------------------------------------
1272
1273    fn create_fargate_profile(
1274        &self,
1275        req: &AwsRequest,
1276        cluster_name: &str,
1277    ) -> Result<AwsResponse, AwsServiceError> {
1278        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
1279        let name = body
1280            .get("fargateProfileName")
1281            .and_then(|v| v.as_str())
1282            .ok_or_else(|| invalid_parameter("fargateProfileName is required"))?
1283            .to_string();
1284        let pod_execution_role_arn = body
1285            .get("podExecutionRoleArn")
1286            .and_then(|v| v.as_str())
1287            .ok_or_else(|| invalid_parameter("podExecutionRoleArn is required"))?
1288            .to_string();
1289
1290        let region = req.region.clone();
1291        let account_id = req.account_id.clone();
1292
1293        let mut accounts = self.state.write();
1294        let state = accounts.get_or_create(&req.account_id);
1295        if !state.clusters.contains_key(cluster_name) {
1296            return Err(not_found_cluster(cluster_name)());
1297        }
1298        if state
1299            .fargate_profiles
1300            .get(cluster_name)
1301            .is_some_and(|m| m.contains_key(&name))
1302        {
1303            return Err(AwsServiceError::aws_error(
1304                StatusCode::CONFLICT,
1305                "ResourceInUseException",
1306                format!(
1307                    "FargateProfile already exists with name {name} and cluster name {cluster_name}"
1308                ),
1309            ));
1310        }
1311
1312        let id = uuid::Uuid::new_v4().to_string();
1313        let arn = fargate_profile_arn(&region, &account_id, cluster_name, &name, &id);
1314        let profile = FargateProfile {
1315            name: name.clone(),
1316            arn,
1317            cluster_name: cluster_name.to_string(),
1318            pod_execution_role_arn,
1319            status: "CREATING".to_string(),
1320            created_at: Utc::now(),
1321            subnets: body.get("subnets").cloned().unwrap_or_else(|| json!([])),
1322            selectors: body.get("selectors").cloned().unwrap_or_else(|| json!([])),
1323            tags: parse_tag_map(body.get("tags")),
1324        };
1325
1326        let out = fargate_profile_json(&profile);
1327        state
1328            .fargate_profiles
1329            .entry(cluster_name.to_string())
1330            .or_default()
1331            .insert(name, profile);
1332        Ok(AwsResponse::json(
1333            StatusCode::OK,
1334            json!({ "fargateProfile": out }).to_string(),
1335        ))
1336    }
1337
1338    fn describe_fargate_profile(
1339        &self,
1340        req: &AwsRequest,
1341        cluster_name: &str,
1342        name: &str,
1343    ) -> Result<AwsResponse, AwsServiceError> {
1344        let mut accounts = self.state.write();
1345        let state = accounts.get_or_create(&req.account_id);
1346        if !state.clusters.contains_key(cluster_name) {
1347            return Err(not_found_cluster(cluster_name)());
1348        }
1349        let profile = state
1350            .fargate_profiles
1351            .get_mut(cluster_name)
1352            .and_then(|m| m.get_mut(name))
1353            .ok_or_else(not_found_fargate_profile(name))?;
1354        if profile.status == "CREATING" {
1355            profile.status = "ACTIVE".to_string();
1356        }
1357        Ok(AwsResponse::json(
1358            StatusCode::OK,
1359            json!({ "fargateProfile": fargate_profile_json(profile) }).to_string(),
1360        ))
1361    }
1362
1363    fn list_fargate_profiles(
1364        &self,
1365        req: &AwsRequest,
1366        cluster_name: &str,
1367    ) -> Result<AwsResponse, AwsServiceError> {
1368        let max_results = validate_max_results(req)?;
1369        let next_token = req.query_params.get("nextToken").cloned();
1370        let accounts = self.state.read();
1371        let state = accounts
1372            .get(&req.account_id)
1373            .ok_or_else(not_found_cluster(cluster_name))?;
1374        if !state.clusters.contains_key(cluster_name) {
1375            return Err(not_found_cluster(cluster_name)());
1376        }
1377        let names: Vec<String> = state
1378            .fargate_profiles
1379            .get(cluster_name)
1380            .map(|m| m.keys().cloned().collect())
1381            .unwrap_or_default();
1382        let (page, token) = paginate_checked(&names, next_token.as_deref(), max_results)
1383            .map_err(|_| invalid_parameter("Invalid nextToken"))?;
1384        let mut out = json!({ "fargateProfileNames": page });
1385        if let Some(t) = token {
1386            out["nextToken"] = Value::String(t);
1387        }
1388        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
1389    }
1390
1391    fn delete_fargate_profile(
1392        &self,
1393        req: &AwsRequest,
1394        cluster_name: &str,
1395        name: &str,
1396    ) -> Result<AwsResponse, AwsServiceError> {
1397        let mut accounts = self.state.write();
1398        let state = accounts.get_or_create(&req.account_id);
1399        if !state.clusters.contains_key(cluster_name) {
1400            return Err(not_found_cluster(cluster_name)());
1401        }
1402        let mut profile = state
1403            .fargate_profiles
1404            .get_mut(cluster_name)
1405            .and_then(|m| m.remove(name))
1406            .ok_or_else(not_found_fargate_profile(name))?;
1407        profile.status = "DELETING".to_string();
1408        Ok(AwsResponse::json(
1409            StatusCode::OK,
1410            json!({ "fargateProfile": fargate_profile_json(&profile) }).to_string(),
1411        ))
1412    }
1413
1414    // -----------------------------------------------------------------------
1415    // Add-ons
1416    // -----------------------------------------------------------------------
1417
1418    fn create_addon(
1419        &self,
1420        req: &AwsRequest,
1421        cluster_name: &str,
1422    ) -> Result<AwsResponse, AwsServiceError> {
1423        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
1424        let name = body
1425            .get("addonName")
1426            .and_then(|v| v.as_str())
1427            .ok_or_else(|| invalid_parameter("addonName is required"))?
1428            .to_string();
1429
1430        let region = req.region.clone();
1431        let account_id = req.account_id.clone();
1432
1433        let mut accounts = self.state.write();
1434        let state = accounts.get_or_create(&req.account_id);
1435        // The parent cluster must exist; AWS returns ResourceNotFoundException.
1436        let cluster_version = state
1437            .clusters
1438            .get(cluster_name)
1439            .ok_or_else(not_found_cluster(cluster_name))?
1440            .version
1441            .clone();
1442        if state
1443            .addons
1444            .get(cluster_name)
1445            .is_some_and(|m| m.contains_key(&name))
1446        {
1447            return Err(AwsServiceError::aws_error(
1448                StatusCode::CONFLICT,
1449                "ResourceInUseException",
1450                format!("Addon already exists with name {name} and cluster name {cluster_name}"),
1451            ));
1452        }
1453
1454        let id = uuid::Uuid::new_v4().to_string();
1455        let arn = addon_arn(&region, &account_id, cluster_name, &name, &id);
1456        let now = Utc::now();
1457        let addon_version = body
1458            .get("addonVersion")
1459            .and_then(|v| v.as_str())
1460            .map(|s| s.to_string())
1461            .unwrap_or_else(|| default_addon_version(&name, &cluster_version));
1462        let namespace = body
1463            .get("namespaceConfig")
1464            .and_then(|v| v.get("namespace"))
1465            .and_then(|v| v.as_str())
1466            .map(|s| s.to_string());
1467        let pod_identity_associations = build_pod_identity_association_arns(
1468            &region,
1469            &account_id,
1470            cluster_name,
1471            body.get("podIdentityAssociations"),
1472        );
1473
1474        let addon = Addon {
1475            name: name.clone(),
1476            arn,
1477            cluster_name: cluster_name.to_string(),
1478            addon_version,
1479            status: "CREATING".to_string(),
1480            created_at: now,
1481            modified_at: now,
1482            service_account_role_arn: body
1483                .get("serviceAccountRoleArn")
1484                .and_then(|v| v.as_str())
1485                .map(|s| s.to_string()),
1486            configuration_values: body
1487                .get("configurationValues")
1488                .and_then(|v| v.as_str())
1489                .map(|s| s.to_string()),
1490            namespace,
1491            pod_identity_associations,
1492            tags: parse_tag_map(body.get("tags")),
1493            updates: Default::default(),
1494        };
1495
1496        let out = addon_json(&addon);
1497        state
1498            .addons
1499            .entry(cluster_name.to_string())
1500            .or_default()
1501            .insert(name, addon);
1502        Ok(AwsResponse::json(
1503            StatusCode::OK,
1504            json!({ "addon": out }).to_string(),
1505        ))
1506    }
1507
1508    fn describe_addon(
1509        &self,
1510        req: &AwsRequest,
1511        cluster_name: &str,
1512        name: &str,
1513    ) -> Result<AwsResponse, AwsServiceError> {
1514        let mut accounts = self.state.write();
1515        let state = accounts.get_or_create(&req.account_id);
1516        if !state.clusters.contains_key(cluster_name) {
1517            return Err(not_found_cluster(cluster_name)());
1518        }
1519        let addon = state
1520            .addons
1521            .get_mut(cluster_name)
1522            .and_then(|m| m.get_mut(name))
1523            .ok_or_else(not_found_addon(name))?;
1524        // Settle CREATING -> ACTIVE on first describe.
1525        if addon.status == "CREATING" {
1526            addon.status = "ACTIVE".to_string();
1527        }
1528        Ok(AwsResponse::json(
1529            StatusCode::OK,
1530            json!({ "addon": addon_json(addon) }).to_string(),
1531        ))
1532    }
1533
1534    fn list_addons(
1535        &self,
1536        req: &AwsRequest,
1537        cluster_name: &str,
1538    ) -> Result<AwsResponse, AwsServiceError> {
1539        let max_results = validate_max_results(req)?;
1540        let next_token = req.query_params.get("nextToken").cloned();
1541        let accounts = self.state.read();
1542        let state = accounts
1543            .get(&req.account_id)
1544            .ok_or_else(not_found_cluster(cluster_name))?;
1545        if !state.clusters.contains_key(cluster_name) {
1546            return Err(not_found_cluster(cluster_name)());
1547        }
1548        let names: Vec<String> = state
1549            .addons
1550            .get(cluster_name)
1551            .map(|m| m.keys().cloned().collect())
1552            .unwrap_or_default();
1553        let (page, token) = paginate_checked(&names, next_token.as_deref(), max_results)
1554            .map_err(|_| invalid_parameter("Invalid nextToken"))?;
1555        let mut out = json!({ "addons": page });
1556        if let Some(t) = token {
1557            out["nextToken"] = Value::String(t);
1558        }
1559        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
1560    }
1561
1562    fn delete_addon(
1563        &self,
1564        req: &AwsRequest,
1565        cluster_name: &str,
1566        name: &str,
1567    ) -> Result<AwsResponse, AwsServiceError> {
1568        let mut accounts = self.state.write();
1569        let state = accounts.get_or_create(&req.account_id);
1570        if !state.clusters.contains_key(cluster_name) {
1571            return Err(not_found_cluster(cluster_name)());
1572        }
1573        let mut addon = state
1574            .addons
1575            .get_mut(cluster_name)
1576            .and_then(|m| m.remove(name))
1577            .ok_or_else(not_found_addon(name))?;
1578        addon.status = "DELETING".to_string();
1579        Ok(AwsResponse::json(
1580            StatusCode::OK,
1581            json!({ "addon": addon_json(&addon) }).to_string(),
1582        ))
1583    }
1584
1585    fn update_addon(
1586        &self,
1587        req: &AwsRequest,
1588        cluster_name: &str,
1589        name: &str,
1590    ) -> Result<AwsResponse, AwsServiceError> {
1591        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
1592        let region = req.region.clone();
1593        let account_id = req.account_id.clone();
1594        let mut accounts = self.state.write();
1595        let state = accounts.get_or_create(&req.account_id);
1596        if !state.clusters.contains_key(cluster_name) {
1597            return Err(not_found_cluster(cluster_name)());
1598        }
1599        let addon = state
1600            .addons
1601            .get_mut(cluster_name)
1602            .and_then(|m| m.get_mut(name))
1603            .ok_or_else(not_found_addon(name))?;
1604
1605        let mut params = Vec::new();
1606        if let Some(version) = body.get("addonVersion").and_then(|v| v.as_str()) {
1607            addon.addon_version = version.to_string();
1608            params.push(("AddonVersion".to_string(), version.to_string()));
1609        }
1610        if let Some(role) = body.get("serviceAccountRoleArn").and_then(|v| v.as_str()) {
1611            addon.service_account_role_arn = Some(role.to_string());
1612            params.push(("ServiceAccountRoleArn".to_string(), role.to_string()));
1613        }
1614        if let Some(cfg) = body.get("configurationValues").and_then(|v| v.as_str()) {
1615            addon.configuration_values = Some(cfg.to_string());
1616            params.push(("ConfigurationValues".to_string(), cfg.to_string()));
1617        }
1618        if let Some(resolve) = body.get("resolveConflicts").and_then(|v| v.as_str()) {
1619            params.push(("ResolveConflicts".to_string(), resolve.to_string()));
1620        }
1621        if let Some(assocs) = body.get("podIdentityAssociations") {
1622            addon.pod_identity_associations = build_pod_identity_association_arns(
1623                &region,
1624                &account_id,
1625                cluster_name,
1626                Some(assocs),
1627            );
1628        }
1629        addon.modified_at = Utc::now();
1630
1631        let update = new_update("AddonUpdate", params);
1632        let out = update_json(&update);
1633        addon.updates.insert(update.id.clone(), update);
1634        Ok(AwsResponse::json(
1635            StatusCode::OK,
1636            json!({ "update": out }).to_string(),
1637        ))
1638    }
1639
1640    fn describe_addon_versions(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
1641        let max_results = validate_max_results(req)?;
1642        let next_token = req.query_params.get("nextToken").cloned();
1643        let addon_filter = req.query_params.get("addonName").cloned();
1644        let k8s_version = req
1645            .query_params
1646            .get("kubernetesVersion")
1647            .cloned()
1648            .unwrap_or_else(|| DEFAULT_K8S_VERSION.to_string());
1649
1650        let catalog = addon_catalog(&k8s_version);
1651        let filtered: Vec<Value> = catalog
1652            .into_iter()
1653            .filter(|a| addon_filter.as_deref().is_none_or(|f| a["addonName"] == f))
1654            .collect();
1655
1656        let (page, token) = paginate_checked(&filtered, next_token.as_deref(), max_results)
1657            .map_err(|_| invalid_parameter("Invalid nextToken"))?;
1658        let mut out = json!({ "addons": page });
1659        if let Some(t) = token {
1660            out["nextToken"] = Value::String(t);
1661        }
1662        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
1663    }
1664
1665    fn describe_addon_configuration(
1666        &self,
1667        req: &AwsRequest,
1668    ) -> Result<AwsResponse, AwsServiceError> {
1669        let addon_name = req
1670            .query_params
1671            .get("addonName")
1672            .cloned()
1673            .ok_or_else(|| invalid_parameter("addonName is required"))?;
1674        let addon_version = req
1675            .query_params
1676            .get("addonVersion")
1677            .cloned()
1678            .ok_or_else(|| invalid_parameter("addonVersion is required"))?;
1679
1680        Ok(AwsResponse::json(
1681            StatusCode::OK,
1682            json!({
1683                "addonName": addon_name,
1684                "addonVersion": addon_version,
1685                "configurationSchema": addon_configuration_schema(&addon_name),
1686                "podIdentityConfiguration": pod_identity_configuration(&addon_name),
1687            })
1688            .to_string(),
1689        ))
1690    }
1691
1692    // -----------------------------------------------------------------------
1693    // Access entries / access policies
1694    // -----------------------------------------------------------------------
1695
1696    fn create_access_entry(
1697        &self,
1698        req: &AwsRequest,
1699        cluster_name: &str,
1700    ) -> Result<AwsResponse, AwsServiceError> {
1701        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
1702        let principal_arn = body
1703            .get("principalArn")
1704            .and_then(|v| v.as_str())
1705            .ok_or_else(|| invalid_parameter("principalArn is required"))?
1706            .to_string();
1707
1708        let region = req.region.clone();
1709        let account_id = req.account_id.clone();
1710
1711        let mut accounts = self.state.write();
1712        let state = accounts.get_or_create(&req.account_id);
1713        // The parent cluster must exist; AWS returns ResourceNotFoundException.
1714        if !state.clusters.contains_key(cluster_name) {
1715            return Err(not_found_cluster(cluster_name)());
1716        }
1717        if state
1718            .access_entries
1719            .get(cluster_name)
1720            .is_some_and(|m| m.contains_key(&principal_arn))
1721        {
1722            return Err(AwsServiceError::aws_error(
1723                StatusCode::CONFLICT,
1724                "ResourceInUseException",
1725                format!(
1726                    "The specified access entry resource is already in use on this cluster: {principal_arn}"
1727                ),
1728            ));
1729        }
1730
1731        let (principal_type, principal_name) = principal_parts(&principal_arn);
1732        let id = uuid::Uuid::new_v4().to_string();
1733        let arn = access_entry_arn(
1734            &region,
1735            &account_id,
1736            cluster_name,
1737            &principal_type,
1738            &principal_name,
1739            &id,
1740        );
1741        let now = Utc::now();
1742        let username = body
1743            .get("username")
1744            .and_then(|v| v.as_str())
1745            .map(|s| s.to_string())
1746            .unwrap_or_else(|| default_username(&principal_arn));
1747        let entry_type = body
1748            .get("type")
1749            .and_then(|v| v.as_str())
1750            .unwrap_or("STANDARD")
1751            .to_string();
1752        let kubernetes_groups = string_list(body.get("kubernetesGroups"));
1753
1754        let entry = AccessEntry {
1755            principal_arn: principal_arn.clone(),
1756            cluster_name: cluster_name.to_string(),
1757            arn,
1758            kubernetes_groups,
1759            username,
1760            type_: entry_type,
1761            created_at: now,
1762            modified_at: now,
1763            tags: parse_tag_map(body.get("tags")),
1764            associated_policies: Vec::new(),
1765        };
1766
1767        let out = access_entry_json(&entry);
1768        state
1769            .access_entries
1770            .entry(cluster_name.to_string())
1771            .or_default()
1772            .insert(principal_arn, entry);
1773        Ok(AwsResponse::json(
1774            StatusCode::OK,
1775            json!({ "accessEntry": out }).to_string(),
1776        ))
1777    }
1778
1779    fn describe_access_entry(
1780        &self,
1781        req: &AwsRequest,
1782        cluster_name: &str,
1783        principal_arn: &str,
1784    ) -> Result<AwsResponse, AwsServiceError> {
1785        let accounts = self.state.read();
1786        let state = accounts
1787            .get(&req.account_id)
1788            .ok_or_else(not_found_cluster(cluster_name))?;
1789        if !state.clusters.contains_key(cluster_name) {
1790            return Err(not_found_cluster(cluster_name)());
1791        }
1792        let entry = state
1793            .access_entries
1794            .get(cluster_name)
1795            .and_then(|m| m.get(principal_arn))
1796            .ok_or_else(not_found_access_entry(principal_arn))?;
1797        Ok(AwsResponse::json(
1798            StatusCode::OK,
1799            json!({ "accessEntry": access_entry_json(entry) }).to_string(),
1800        ))
1801    }
1802
1803    fn list_access_entries(
1804        &self,
1805        req: &AwsRequest,
1806        cluster_name: &str,
1807    ) -> Result<AwsResponse, AwsServiceError> {
1808        let max_results = validate_max_results(req)?;
1809        let next_token = req.query_params.get("nextToken").cloned();
1810        let associated_policy = req.query_params.get("associatedPolicyArn").cloned();
1811        let accounts = self.state.read();
1812        let state = accounts
1813            .get(&req.account_id)
1814            .ok_or_else(not_found_cluster(cluster_name))?;
1815        if !state.clusters.contains_key(cluster_name) {
1816            return Err(not_found_cluster(cluster_name)());
1817        }
1818        // Optional `associatedPolicyArn` filter narrows the result to entries
1819        // that have that policy associated, mirroring the real API.
1820        let names: Vec<String> = state
1821            .access_entries
1822            .get(cluster_name)
1823            .map(|m| {
1824                m.values()
1825                    .filter(|e| {
1826                        associated_policy.as_deref().is_none_or(|p| {
1827                            e.associated_policies.iter().any(|ap| ap.policy_arn == p)
1828                        })
1829                    })
1830                    .map(|e| e.principal_arn.clone())
1831                    .collect()
1832            })
1833            .unwrap_or_default();
1834        let (page, token) = paginate_checked(&names, next_token.as_deref(), max_results)
1835            .map_err(|_| invalid_parameter("Invalid nextToken"))?;
1836        let mut out = json!({ "accessEntries": page });
1837        if let Some(t) = token {
1838            out["nextToken"] = Value::String(t);
1839        }
1840        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
1841    }
1842
1843    fn delete_access_entry(
1844        &self,
1845        req: &AwsRequest,
1846        cluster_name: &str,
1847        principal_arn: &str,
1848    ) -> Result<AwsResponse, AwsServiceError> {
1849        let mut accounts = self.state.write();
1850        let state = accounts.get_or_create(&req.account_id);
1851        if !state.clusters.contains_key(cluster_name) {
1852            return Err(not_found_cluster(cluster_name)());
1853        }
1854        state
1855            .access_entries
1856            .get_mut(cluster_name)
1857            .and_then(|m| m.remove(principal_arn))
1858            .ok_or_else(not_found_access_entry(principal_arn))?;
1859        Ok(AwsResponse::json(StatusCode::OK, "{}"))
1860    }
1861
1862    fn update_access_entry(
1863        &self,
1864        req: &AwsRequest,
1865        cluster_name: &str,
1866        principal_arn: &str,
1867    ) -> Result<AwsResponse, AwsServiceError> {
1868        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
1869        let mut accounts = self.state.write();
1870        let state = accounts.get_or_create(&req.account_id);
1871        if !state.clusters.contains_key(cluster_name) {
1872            return Err(not_found_cluster(cluster_name)());
1873        }
1874        let entry = state
1875            .access_entries
1876            .get_mut(cluster_name)
1877            .and_then(|m| m.get_mut(principal_arn))
1878            .ok_or_else(not_found_access_entry(principal_arn))?;
1879
1880        if let Some(groups) = body.get("kubernetesGroups") {
1881            entry.kubernetes_groups = string_list(Some(groups));
1882        }
1883        if let Some(username) = body.get("username").and_then(|v| v.as_str()) {
1884            entry.username = username.to_string();
1885        }
1886        entry.modified_at = Utc::now();
1887        Ok(AwsResponse::json(
1888            StatusCode::OK,
1889            json!({ "accessEntry": access_entry_json(entry) }).to_string(),
1890        ))
1891    }
1892
1893    fn associate_access_policy(
1894        &self,
1895        req: &AwsRequest,
1896        cluster_name: &str,
1897        principal_arn: &str,
1898    ) -> Result<AwsResponse, AwsServiceError> {
1899        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
1900        let policy_arn = body
1901            .get("policyArn")
1902            .and_then(|v| v.as_str())
1903            .ok_or_else(|| invalid_parameter("policyArn is required"))?
1904            .to_string();
1905        let access_scope = build_access_scope(body.get("accessScope"));
1906
1907        let mut accounts = self.state.write();
1908        let state = accounts.get_or_create(&req.account_id);
1909        if !state.clusters.contains_key(cluster_name) {
1910            return Err(not_found_cluster(cluster_name)());
1911        }
1912        let entry = state
1913            .access_entries
1914            .get_mut(cluster_name)
1915            .and_then(|m| m.get_mut(principal_arn))
1916            .ok_or_else(not_found_access_entry(principal_arn))?;
1917
1918        let now = Utc::now();
1919        // Re-associating the same policy updates its scope in place, matching
1920        // the real API's idempotent upsert semantics.
1921        let associated = if let Some(existing) = entry
1922            .associated_policies
1923            .iter_mut()
1924            .find(|ap| ap.policy_arn == policy_arn)
1925        {
1926            existing.access_scope = access_scope;
1927            existing.modified_at = now;
1928            existing.clone()
1929        } else {
1930            let ap = AssociatedPolicy {
1931                policy_arn: policy_arn.clone(),
1932                access_scope,
1933                associated_at: now,
1934                modified_at: now,
1935            };
1936            entry.associated_policies.push(ap.clone());
1937            ap
1938        };
1939        Ok(AwsResponse::json(
1940            StatusCode::OK,
1941            json!({
1942                "clusterName": cluster_name,
1943                "principalArn": entry.principal_arn,
1944                "associatedAccessPolicy": associated_policy_json(&associated),
1945            })
1946            .to_string(),
1947        ))
1948    }
1949
1950    fn disassociate_access_policy(
1951        &self,
1952        req: &AwsRequest,
1953        cluster_name: &str,
1954        principal_arn: &str,
1955        policy_arn: &str,
1956    ) -> Result<AwsResponse, AwsServiceError> {
1957        let mut accounts = self.state.write();
1958        let state = accounts.get_or_create(&req.account_id);
1959        if !state.clusters.contains_key(cluster_name) {
1960            return Err(not_found_cluster(cluster_name)());
1961        }
1962        let entry = state
1963            .access_entries
1964            .get_mut(cluster_name)
1965            .and_then(|m| m.get_mut(principal_arn))
1966            .ok_or_else(not_found_access_entry(principal_arn))?;
1967        entry
1968            .associated_policies
1969            .retain(|ap| ap.policy_arn != policy_arn);
1970        Ok(AwsResponse::json(StatusCode::OK, "{}"))
1971    }
1972
1973    fn list_associated_access_policies(
1974        &self,
1975        req: &AwsRequest,
1976        cluster_name: &str,
1977        principal_arn: &str,
1978    ) -> Result<AwsResponse, AwsServiceError> {
1979        let max_results = validate_max_results(req)?;
1980        let next_token = req.query_params.get("nextToken").cloned();
1981        let accounts = self.state.read();
1982        let state = accounts
1983            .get(&req.account_id)
1984            .ok_or_else(not_found_cluster(cluster_name))?;
1985        if !state.clusters.contains_key(cluster_name) {
1986            return Err(not_found_cluster(cluster_name)());
1987        }
1988        let entry = state
1989            .access_entries
1990            .get(cluster_name)
1991            .and_then(|m| m.get(principal_arn))
1992            .ok_or_else(not_found_access_entry(principal_arn))?;
1993        let policies: Vec<Value> = entry
1994            .associated_policies
1995            .iter()
1996            .map(associated_policy_json)
1997            .collect();
1998        let (page, token) = paginate_checked(&policies, next_token.as_deref(), max_results)
1999            .map_err(|_| invalid_parameter("Invalid nextToken"))?;
2000        let mut out = json!({
2001            "clusterName": cluster_name,
2002            "principalArn": entry.principal_arn,
2003            "associatedAccessPolicies": page,
2004        });
2005        if let Some(t) = token {
2006            out["nextToken"] = Value::String(t);
2007        }
2008        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
2009    }
2010
2011    fn list_access_policies(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
2012        // ListAccessPolicies is a read-only catalogue. Its `maxResults` carries
2013        // an `@range` (1-100) constraint the probe exercises, so an out-of-range
2014        // value is rejected. The nextToken, however, is tolerated (an
2015        // unparseable one restarts from the beginning) since the op's Smithy
2016        // contract declares no client error for it.
2017        let max_results = validate_max_results(req)?;
2018        let next_token = req.query_params.get("nextToken").cloned();
2019        let catalog = access_policy_catalog();
2020        let (page, token) = paginate_checked(&catalog, next_token.as_deref(), max_results)
2021            .unwrap_or_else(|_| (catalog.clone(), None));
2022        let mut out = json!({ "accessPolicies": page });
2023        if let Some(t) = token {
2024            out["nextToken"] = Value::String(t);
2025        }
2026        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
2027    }
2028
2029    // -----------------------------------------------------------------------
2030    // Identity-provider configs
2031    // -----------------------------------------------------------------------
2032
2033    fn associate_identity_provider_config(
2034        &self,
2035        req: &AwsRequest,
2036        cluster_name: &str,
2037    ) -> Result<AwsResponse, AwsServiceError> {
2038        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
2039        let oidc = body
2040            .get("oidc")
2041            .filter(|v| v.is_object())
2042            .ok_or_else(|| invalid_parameter("oidc is required"))?;
2043        let name = oidc
2044            .get("identityProviderConfigName")
2045            .and_then(|v| v.as_str())
2046            .ok_or_else(|| invalid_parameter("oidc.identityProviderConfigName is required"))?
2047            .to_string();
2048        let issuer_url = oidc
2049            .get("issuerUrl")
2050            .and_then(|v| v.as_str())
2051            .ok_or_else(|| invalid_parameter("oidc.issuerUrl is required"))?
2052            .to_string();
2053        let client_id = oidc
2054            .get("clientId")
2055            .and_then(|v| v.as_str())
2056            .ok_or_else(|| invalid_parameter("oidc.clientId is required"))?
2057            .to_string();
2058
2059        let region = req.region.clone();
2060        let account_id = req.account_id.clone();
2061
2062        let mut accounts = self.state.write();
2063        let state = accounts.get_or_create(&req.account_id);
2064        // The parent cluster must exist; AWS returns ResourceNotFoundException.
2065        if !state.clusters.contains_key(cluster_name) {
2066            return Err(not_found_cluster(cluster_name)());
2067        }
2068        if state
2069            .identity_provider_configs
2070            .get(cluster_name)
2071            .is_some_and(|m| m.contains_key(&name))
2072        {
2073            return Err(AwsServiceError::aws_error(
2074                StatusCode::CONFLICT,
2075                "ResourceInUseException",
2076                format!(
2077                    "Identity provider config already exists with name {name} and cluster name {cluster_name}"
2078                ),
2079            ));
2080        }
2081
2082        let id = uuid::Uuid::new_v4().to_string();
2083        let arn = identity_provider_config_arn(&region, &account_id, cluster_name, &name, &id);
2084        let config = IdentityProviderConfig {
2085            name: name.clone(),
2086            arn,
2087            cluster_name: cluster_name.to_string(),
2088            issuer_url,
2089            client_id,
2090            username_claim: str_field(oidc, "usernameClaim"),
2091            username_prefix: str_field(oidc, "usernamePrefix"),
2092            groups_claim: str_field(oidc, "groupsClaim"),
2093            groups_prefix: str_field(oidc, "groupsPrefix"),
2094            required_claims: oidc
2095                .get("requiredClaims")
2096                .cloned()
2097                .unwrap_or_else(|| json!({})),
2098            status: "CREATING".to_string(),
2099            tags: parse_tag_map(body.get("tags")),
2100        };
2101        state
2102            .identity_provider_configs
2103            .entry(cluster_name.to_string())
2104            .or_default()
2105            .insert(name, config.clone());
2106
2107        // Associating an IdP config mints a tracked cluster-scoped Update, read
2108        // back through DescribeUpdate/ListUpdates like other config changes.
2109        let cluster = state.clusters.get_mut(cluster_name).unwrap();
2110        let update = new_update(
2111            "AssociateIdentityProviderConfig",
2112            vec![("IdentityProviderConfig".to_string(), oidc.to_string())],
2113        );
2114        let update_out = update_json(&update);
2115        cluster.updates.insert(update.id.clone(), update);
2116        Ok(AwsResponse::json(
2117            StatusCode::OK,
2118            json!({ "update": update_out, "tags": config.tags }).to_string(),
2119        ))
2120    }
2121
2122    fn disassociate_identity_provider_config(
2123        &self,
2124        req: &AwsRequest,
2125        cluster_name: &str,
2126    ) -> Result<AwsResponse, AwsServiceError> {
2127        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
2128        let name = body
2129            .get("identityProviderConfig")
2130            .and_then(|v| v.get("name"))
2131            .and_then(|v| v.as_str())
2132            .ok_or_else(|| invalid_parameter("identityProviderConfig.name is required"))?
2133            .to_string();
2134
2135        let mut accounts = self.state.write();
2136        let state = accounts.get_or_create(&req.account_id);
2137        if !state.clusters.contains_key(cluster_name) {
2138            return Err(not_found_cluster(cluster_name)());
2139        }
2140        state
2141            .identity_provider_configs
2142            .get_mut(cluster_name)
2143            .and_then(|m| m.remove(&name))
2144            .ok_or_else(not_found_identity_provider_config(&name))?;
2145
2146        let cluster = state.clusters.get_mut(cluster_name).unwrap();
2147        let update = new_update(
2148            "DisassociateIdentityProviderConfig",
2149            vec![("IdentityProviderConfig".to_string(), name)],
2150        );
2151        let update_out = update_json(&update);
2152        cluster.updates.insert(update.id.clone(), update);
2153        Ok(AwsResponse::json(
2154            StatusCode::OK,
2155            json!({ "update": update_out }).to_string(),
2156        ))
2157    }
2158
2159    fn describe_identity_provider_config(
2160        &self,
2161        req: &AwsRequest,
2162        cluster_name: &str,
2163    ) -> Result<AwsResponse, AwsServiceError> {
2164        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
2165        let name = body
2166            .get("identityProviderConfig")
2167            .and_then(|v| v.get("name"))
2168            .and_then(|v| v.as_str())
2169            .ok_or_else(|| invalid_parameter("identityProviderConfig.name is required"))?
2170            .to_string();
2171
2172        let mut accounts = self.state.write();
2173        let state = accounts.get_or_create(&req.account_id);
2174        if !state.clusters.contains_key(cluster_name) {
2175            return Err(not_found_cluster(cluster_name)());
2176        }
2177        let config = state
2178            .identity_provider_configs
2179            .get_mut(cluster_name)
2180            .and_then(|m| m.get_mut(&name))
2181            .ok_or_else(not_found_identity_provider_config(&name))?;
2182        // Settle CREATING -> ACTIVE on first describe.
2183        if config.status == "CREATING" {
2184            config.status = "ACTIVE".to_string();
2185        }
2186        Ok(AwsResponse::json(
2187            StatusCode::OK,
2188            json!({ "identityProviderConfig": identity_provider_config_json(config) }).to_string(),
2189        ))
2190    }
2191
2192    fn list_identity_provider_configs(
2193        &self,
2194        req: &AwsRequest,
2195        cluster_name: &str,
2196    ) -> Result<AwsResponse, AwsServiceError> {
2197        let max_results = validate_max_results(req)?;
2198        let next_token = req.query_params.get("nextToken").cloned();
2199        let accounts = self.state.read();
2200        let state = accounts
2201            .get(&req.account_id)
2202            .ok_or_else(not_found_cluster(cluster_name))?;
2203        if !state.clusters.contains_key(cluster_name) {
2204            return Err(not_found_cluster(cluster_name)());
2205        }
2206        let configs: Vec<Value> = state
2207            .identity_provider_configs
2208            .get(cluster_name)
2209            .map(|m| {
2210                m.keys()
2211                    .map(|n| json!({ "type": "oidc", "name": n }))
2212                    .collect()
2213            })
2214            .unwrap_or_default();
2215        let (page, token) = paginate_checked(&configs, next_token.as_deref(), max_results)
2216            .map_err(|_| invalid_parameter("Invalid nextToken"))?;
2217        let mut out = json!({ "identityProviderConfigs": page });
2218        if let Some(t) = token {
2219            out["nextToken"] = Value::String(t);
2220        }
2221        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
2222    }
2223
2224    // -----------------------------------------------------------------------
2225    // Pod identity associations
2226    // -----------------------------------------------------------------------
2227
2228    fn create_pod_identity_association(
2229        &self,
2230        req: &AwsRequest,
2231        cluster_name: &str,
2232    ) -> Result<AwsResponse, AwsServiceError> {
2233        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
2234        let namespace = body
2235            .get("namespace")
2236            .and_then(|v| v.as_str())
2237            .ok_or_else(|| invalid_parameter("namespace is required"))?
2238            .to_string();
2239        let service_account = body
2240            .get("serviceAccount")
2241            .and_then(|v| v.as_str())
2242            .ok_or_else(|| invalid_parameter("serviceAccount is required"))?
2243            .to_string();
2244        let role_arn = body
2245            .get("roleArn")
2246            .and_then(|v| v.as_str())
2247            .ok_or_else(|| invalid_parameter("roleArn is required"))?
2248            .to_string();
2249
2250        let region = req.region.clone();
2251        let account_id = req.account_id.clone();
2252
2253        let mut accounts = self.state.write();
2254        let state = accounts.get_or_create(&req.account_id);
2255        // The parent cluster must exist; AWS returns ResourceNotFoundException.
2256        if !state.clusters.contains_key(cluster_name) {
2257            return Err(not_found_cluster(cluster_name)());
2258        }
2259        // A given (namespace, serviceAccount) can only be associated once per
2260        // cluster; AWS returns ResourceInUseException for a duplicate.
2261        if state
2262            .pod_identity_associations
2263            .get(cluster_name)
2264            .is_some_and(|m| {
2265                m.values()
2266                    .any(|a| a.namespace == namespace && a.service_account == service_account)
2267            })
2268        {
2269            return Err(AwsServiceError::aws_error(
2270                StatusCode::CONFLICT,
2271                "ResourceInUseException",
2272                format!(
2273                    "Association already exists for namespace {namespace} and service account {service_account}"
2274                ),
2275            ));
2276        }
2277
2278        let suffix = uuid::Uuid::new_v4().to_string().replace('-', "");
2279        let suffix = &suffix[..17.min(suffix.len())];
2280        let association_id = format!("a-{suffix}");
2281        let association_arn =
2282            pod_identity_association_arn(&region, &account_id, cluster_name, suffix);
2283        let target_role_arn = str_field(&body, "targetRoleArn");
2284        // AWS returns an externalId (for the target role's trust policy) only
2285        // when a cross-account targetRoleArn is supplied.
2286        let external_id = target_role_arn
2287            .as_ref()
2288            .map(|_| uuid::Uuid::new_v4().to_string().replace('-', ""));
2289        let now = Utc::now();
2290        let assoc = PodIdentityAssociation {
2291            cluster_name: cluster_name.to_string(),
2292            namespace,
2293            service_account,
2294            role_arn,
2295            association_arn,
2296            association_id: association_id.clone(),
2297            created_at: now,
2298            modified_at: now,
2299            disable_session_tags: body
2300                .get("disableSessionTags")
2301                .and_then(|v| v.as_bool())
2302                .unwrap_or(false),
2303            target_role_arn,
2304            external_id,
2305            tags: parse_tag_map(body.get("tags")),
2306        };
2307        let out = pod_identity_association_json(&assoc);
2308        state
2309            .pod_identity_associations
2310            .entry(cluster_name.to_string())
2311            .or_default()
2312            .insert(association_id, assoc);
2313        Ok(AwsResponse::json(
2314            StatusCode::OK,
2315            json!({ "association": out }).to_string(),
2316        ))
2317    }
2318
2319    fn describe_pod_identity_association(
2320        &self,
2321        req: &AwsRequest,
2322        cluster_name: &str,
2323        association_id: &str,
2324    ) -> Result<AwsResponse, AwsServiceError> {
2325        let accounts = self.state.read();
2326        let state = accounts
2327            .get(&req.account_id)
2328            .ok_or_else(not_found_cluster(cluster_name))?;
2329        if !state.clusters.contains_key(cluster_name) {
2330            return Err(not_found_cluster(cluster_name)());
2331        }
2332        let assoc = state
2333            .pod_identity_associations
2334            .get(cluster_name)
2335            .and_then(|m| m.get(association_id))
2336            .ok_or_else(not_found_pod_identity_association(association_id))?;
2337        Ok(AwsResponse::json(
2338            StatusCode::OK,
2339            json!({ "association": pod_identity_association_json(assoc) }).to_string(),
2340        ))
2341    }
2342
2343    fn list_pod_identity_associations(
2344        &self,
2345        req: &AwsRequest,
2346        cluster_name: &str,
2347    ) -> Result<AwsResponse, AwsServiceError> {
2348        let max_results = validate_max_results(req)?;
2349        let next_token = req.query_params.get("nextToken").cloned();
2350        let namespace = req.query_params.get("namespace").cloned();
2351        let service_account = req.query_params.get("serviceAccount").cloned();
2352        let accounts = self.state.read();
2353        let state = accounts
2354            .get(&req.account_id)
2355            .ok_or_else(not_found_cluster(cluster_name))?;
2356        if !state.clusters.contains_key(cluster_name) {
2357            return Err(not_found_cluster(cluster_name)());
2358        }
2359        let summaries: Vec<Value> = state
2360            .pod_identity_associations
2361            .get(cluster_name)
2362            .map(|m| {
2363                m.values()
2364                    .filter(|a| namespace.as_deref().is_none_or(|n| a.namespace == n))
2365                    .filter(|a| {
2366                        service_account
2367                            .as_deref()
2368                            .is_none_or(|s| a.service_account == s)
2369                    })
2370                    .map(pod_identity_association_summary_json)
2371                    .collect()
2372            })
2373            .unwrap_or_default();
2374        let (page, token) = paginate_checked(&summaries, next_token.as_deref(), max_results)
2375            .map_err(|_| invalid_parameter("Invalid nextToken"))?;
2376        let mut out = json!({ "associations": page });
2377        if let Some(t) = token {
2378            out["nextToken"] = Value::String(t);
2379        }
2380        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
2381    }
2382
2383    fn delete_pod_identity_association(
2384        &self,
2385        req: &AwsRequest,
2386        cluster_name: &str,
2387        association_id: &str,
2388    ) -> Result<AwsResponse, AwsServiceError> {
2389        let mut accounts = self.state.write();
2390        let state = accounts.get_or_create(&req.account_id);
2391        if !state.clusters.contains_key(cluster_name) {
2392            return Err(not_found_cluster(cluster_name)());
2393        }
2394        let assoc = state
2395            .pod_identity_associations
2396            .get_mut(cluster_name)
2397            .and_then(|m| m.remove(association_id))
2398            .ok_or_else(not_found_pod_identity_association(association_id))?;
2399        Ok(AwsResponse::json(
2400            StatusCode::OK,
2401            json!({ "association": pod_identity_association_json(&assoc) }).to_string(),
2402        ))
2403    }
2404
2405    fn update_pod_identity_association(
2406        &self,
2407        req: &AwsRequest,
2408        cluster_name: &str,
2409        association_id: &str,
2410    ) -> Result<AwsResponse, AwsServiceError> {
2411        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
2412        let mut accounts = self.state.write();
2413        let state = accounts.get_or_create(&req.account_id);
2414        if !state.clusters.contains_key(cluster_name) {
2415            return Err(not_found_cluster(cluster_name)());
2416        }
2417        let assoc = state
2418            .pod_identity_associations
2419            .get_mut(cluster_name)
2420            .and_then(|m| m.get_mut(association_id))
2421            .ok_or_else(not_found_pod_identity_association(association_id))?;
2422
2423        if let Some(role) = body.get("roleArn").and_then(|v| v.as_str()) {
2424            assoc.role_arn = role.to_string();
2425        }
2426        if let Some(target) = body.get("targetRoleArn").and_then(|v| v.as_str()) {
2427            assoc.target_role_arn = Some(target.to_string());
2428            // Mint an externalId for the target trust policy if we don't have one.
2429            if assoc.external_id.is_none() {
2430                assoc.external_id = Some(uuid::Uuid::new_v4().to_string().replace('-', ""));
2431            }
2432        }
2433        if let Some(disable) = body.get("disableSessionTags").and_then(|v| v.as_bool()) {
2434            assoc.disable_session_tags = disable;
2435        }
2436        assoc.modified_at = Utc::now();
2437        Ok(AwsResponse::json(
2438            StatusCode::OK,
2439            json!({ "association": pod_identity_association_json(assoc) }).to_string(),
2440        ))
2441    }
2442
2443    // -----------------------------------------------------------------------
2444    // Insights
2445    // -----------------------------------------------------------------------
2446
2447    fn list_insights(
2448        &self,
2449        req: &AwsRequest,
2450        cluster_name: &str,
2451    ) -> Result<AwsResponse, AwsServiceError> {
2452        // ListInsights is a POST carrying the filter/pagination in its body.
2453        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
2454        // maxResults is constrained to 1..=100 by the model; reject out-of-range
2455        // values rather than silently clamping them.
2456        let max_results = match body.get("maxResults") {
2457            Some(v) => {
2458                let n = v
2459                    .as_i64()
2460                    .ok_or_else(|| invalid_parameter("maxResults must be an integer"))?;
2461                if !(1..=100).contains(&n) {
2462                    return Err(invalid_parameter("maxResults must be between 1 and 100"));
2463                }
2464                n as usize
2465            }
2466            None => 100,
2467        };
2468        let next_token = body
2469            .get("nextToken")
2470            .and_then(|v| v.as_str())
2471            .map(|s| s.to_string());
2472        // Optional category / status filters.
2473        let categories = string_list(body.get("filter").and_then(|f| f.get("categories")));
2474        let statuses = string_list(body.get("filter").and_then(|f| f.get("statuses")));
2475
2476        let mut accounts = self.state.write();
2477        let state = accounts.get_or_create(&req.account_id);
2478        let version = state
2479            .clusters
2480            .get(cluster_name)
2481            .ok_or_else(not_found_cluster(cluster_name))?
2482            .version
2483            .clone();
2484        let insights = state
2485            .insights
2486            .entry(cluster_name.to_string())
2487            .or_insert_with(|| {
2488                default_insights(&version)
2489                    .into_iter()
2490                    .map(|i| (i.id.clone(), i))
2491                    .collect()
2492            });
2493        let summaries: Vec<Value> = insights
2494            .values()
2495            .filter(|i| categories.is_empty() || categories.iter().any(|c| c == &i.category))
2496            .filter(|i| statuses.is_empty() || statuses.iter().any(|s| s == &i.status))
2497            .map(insight_summary_json)
2498            .collect();
2499        let (page, token) = paginate_checked(&summaries, next_token.as_deref(), max_results)
2500            .map_err(|_| invalid_parameter("Invalid nextToken"))?;
2501        let mut out = json!({ "insights": page });
2502        if let Some(t) = token {
2503            out["nextToken"] = Value::String(t);
2504        }
2505        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
2506    }
2507
2508    fn describe_insight(
2509        &self,
2510        req: &AwsRequest,
2511        cluster_name: &str,
2512        id: &str,
2513    ) -> Result<AwsResponse, AwsServiceError> {
2514        let mut accounts = self.state.write();
2515        let state = accounts.get_or_create(&req.account_id);
2516        let version = state
2517            .clusters
2518            .get(cluster_name)
2519            .ok_or_else(not_found_cluster(cluster_name))?
2520            .version
2521            .clone();
2522        let insights = state
2523            .insights
2524            .entry(cluster_name.to_string())
2525            .or_insert_with(|| {
2526                default_insights(&version)
2527                    .into_iter()
2528                    .map(|i| (i.id.clone(), i))
2529                    .collect()
2530            });
2531        let insight = insights.get(id).ok_or_else(not_found_insight(id))?;
2532        Ok(AwsResponse::json(
2533            StatusCode::OK,
2534            json!({ "insight": insight_json(insight) }).to_string(),
2535        ))
2536    }
2537
2538    fn describe_insights_refresh(
2539        &self,
2540        req: &AwsRequest,
2541        cluster_name: &str,
2542    ) -> Result<AwsResponse, AwsServiceError> {
2543        let mut accounts = self.state.write();
2544        let state = accounts.get_or_create(&req.account_id);
2545        if !state.clusters.contains_key(cluster_name) {
2546            return Err(not_found_cluster(cluster_name)());
2547        }
2548        let refresh = state
2549            .insights_refresh
2550            .entry(cluster_name.to_string())
2551            .or_insert_with(|| InsightsRefresh {
2552                status: "COMPLETED".to_string(),
2553                started_at: Utc::now(),
2554                ended_at: Some(Utc::now()),
2555            });
2556        // An in-progress refresh settles to COMPLETED on first describe.
2557        if refresh.status == "IN_PROGRESS" {
2558            refresh.status = "COMPLETED".to_string();
2559            refresh.ended_at = Some(Utc::now());
2560        }
2561        Ok(AwsResponse::json(
2562            StatusCode::OK,
2563            insights_refresh_json(refresh).to_string(),
2564        ))
2565    }
2566
2567    fn start_insights_refresh(
2568        &self,
2569        req: &AwsRequest,
2570        cluster_name: &str,
2571    ) -> Result<AwsResponse, AwsServiceError> {
2572        let mut accounts = self.state.write();
2573        let state = accounts.get_or_create(&req.account_id);
2574        let version = state
2575            .clusters
2576            .get(cluster_name)
2577            .ok_or_else(not_found_cluster(cluster_name))?
2578            .version
2579            .clone();
2580        // A refresh regenerates the cluster's insights.
2581        state.insights.insert(
2582            cluster_name.to_string(),
2583            default_insights(&version)
2584                .into_iter()
2585                .map(|i| (i.id.clone(), i))
2586                .collect(),
2587        );
2588        let refresh = InsightsRefresh {
2589            status: "IN_PROGRESS".to_string(),
2590            started_at: Utc::now(),
2591            ended_at: None,
2592        };
2593        // StartInsightsRefresh's response carries only `message` + `status`;
2594        // the timestamps are read back through DescribeInsightsRefresh.
2595        let out = json!({
2596            "message": "Insights refresh started for the cluster.",
2597            "status": refresh.status,
2598        });
2599        state
2600            .insights_refresh
2601            .insert(cluster_name.to_string(), refresh);
2602        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
2603    }
2604
2605    // -----------------------------------------------------------------------
2606    // Cluster misc (encryption config, cancel update, register/deregister,
2607    // cluster versions)
2608    // -----------------------------------------------------------------------
2609
2610    fn associate_encryption_config(
2611        &self,
2612        req: &AwsRequest,
2613        cluster_name: &str,
2614    ) -> Result<AwsResponse, AwsServiceError> {
2615        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
2616        let encryption_config = body
2617            .get("encryptionConfig")
2618            .filter(|v| v.is_array())
2619            .cloned()
2620            .ok_or_else(|| invalid_parameter("encryptionConfig is required"))?;
2621
2622        let mut accounts = self.state.write();
2623        let state = accounts.get_or_create(&req.account_id);
2624        let cluster = state
2625            .clusters
2626            .get_mut(cluster_name)
2627            .ok_or_else(not_found_cluster(cluster_name))?;
2628        cluster.encryption_config = Some(encryption_config.clone());
2629
2630        let update = new_update(
2631            "AssociateEncryptionConfig",
2632            vec![(
2633                "EncryptionConfig".to_string(),
2634                encryption_config.to_string(),
2635            )],
2636        );
2637        let out = update_json(&update);
2638        cluster.updates.insert(update.id.clone(), update);
2639        Ok(AwsResponse::json(
2640            StatusCode::OK,
2641            json!({ "update": out }).to_string(),
2642        ))
2643    }
2644
2645    fn cancel_update(
2646        &self,
2647        req: &AwsRequest,
2648        name: &str,
2649        update_id: &str,
2650    ) -> Result<AwsResponse, AwsServiceError> {
2651        let mut accounts = self.state.write();
2652        let state = accounts.get_or_create(&req.account_id);
2653        let cluster = state
2654            .clusters
2655            .get_mut(name)
2656            .ok_or_else(not_found_cluster(name))?;
2657        let update = cluster
2658            .updates
2659            .get_mut(update_id)
2660            .ok_or_else(not_found_update(update_id))?;
2661        update.status = "Cancelled".to_string();
2662        Ok(AwsResponse::json(
2663            StatusCode::OK,
2664            json!({ "update": update_json(update) }).to_string(),
2665        ))
2666    }
2667
2668    fn register_cluster(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
2669        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
2670        let name = body
2671            .get("name")
2672            .and_then(|v| v.as_str())
2673            .ok_or_else(|| invalid_parameter("name is required"))?
2674            .to_string();
2675        validate_cluster_name(&name)?;
2676        let connector = body
2677            .get("connectorConfig")
2678            .filter(|v| v.is_object())
2679            .ok_or_else(|| invalid_parameter("connectorConfig is required"))?;
2680        let role_arn = connector
2681            .get("roleArn")
2682            .and_then(|v| v.as_str())
2683            .ok_or_else(|| invalid_parameter("connectorConfig.roleArn is required"))?
2684            .to_string();
2685        let provider = connector
2686            .get("provider")
2687            .and_then(|v| v.as_str())
2688            .ok_or_else(|| invalid_parameter("connectorConfig.provider is required"))?
2689            .to_string();
2690
2691        let region = req.region.clone();
2692        let account_id = req.account_id.clone();
2693
2694        let mut accounts = self.state.write();
2695        let state = accounts.get_or_create(&req.account_id);
2696        if state.clusters.contains_key(&name) {
2697            return Err(AwsServiceError::aws_error(
2698                StatusCode::CONFLICT,
2699                "ResourceInUseException",
2700                format!("Cluster already exists with name: {name}"),
2701            ));
2702        }
2703
2704        let arn = cluster_arn(&region, &account_id, &name);
2705        let id = uuid::Uuid::new_v4().to_string();
2706        let activation_id = uuid::Uuid::new_v4().to_string();
2707        let activation_code = uuid::Uuid::new_v4().to_string().replace('-', "");
2708        let connector_config = json!({
2709            "activationId": activation_id,
2710            "activationCode": activation_code,
2711            "activationExpiry": timestamp_to_number(Utc::now() + chrono::Duration::hours(72)),
2712            "provider": provider,
2713            "roleArn": role_arn,
2714        });
2715
2716        let cluster = Cluster {
2717            name: name.clone(),
2718            arn,
2719            version: String::new(),
2720            role_arn: String::new(),
2721            status: "PENDING".to_string(),
2722            created_at: Utc::now(),
2723            endpoint: String::new(),
2724            platform_version: "eks.1".to_string(),
2725            certificate_authority_data: String::new(),
2726            resources_vpc_config: json!({}),
2727            kubernetes_network_config: json!({}),
2728            logging: build_logging(None),
2729            tags: parse_tag_map(body.get("tags")),
2730            updates: Default::default(),
2731            connector_config: Some(connector_config),
2732            encryption_config: None,
2733            access_config: build_access_config(None),
2734            upgrade_policy: build_upgrade_policy(None),
2735            compute_config: None,
2736            storage_config: None,
2737            zonal_shift_config: None,
2738            remote_network_config: None,
2739            control_plane_scaling_config: None,
2740            deletion_protection: None,
2741        };
2742        let out = connected_cluster_json(&cluster, &id);
2743        state.clusters.insert(name, cluster);
2744        Ok(AwsResponse::json(
2745            StatusCode::OK,
2746            json!({ "cluster": out }).to_string(),
2747        ))
2748    }
2749
2750    fn deregister_cluster(
2751        &self,
2752        req: &AwsRequest,
2753        name: &str,
2754    ) -> Result<AwsResponse, AwsServiceError> {
2755        let mut accounts = self.state.write();
2756        let state = accounts.get_or_create(&req.account_id);
2757        // Deregister only applies to connected (registered) clusters.
2758        let is_connected = state
2759            .clusters
2760            .get(name)
2761            .map(|c| c.connector_config.is_some())
2762            .unwrap_or(false);
2763        if !is_connected {
2764            return Err(not_found_cluster(name)());
2765        }
2766        let mut cluster = state.clusters.remove(name).unwrap();
2767        cluster.status = "DELETING".to_string();
2768        let id = uuid::Uuid::new_v4().to_string();
2769        Ok(AwsResponse::json(
2770            StatusCode::OK,
2771            json!({ "cluster": connected_cluster_json(&cluster, &id) }).to_string(),
2772        ))
2773    }
2774
2775    fn describe_cluster_versions(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
2776        let next_token = req.query_params.get("nextToken").cloned();
2777        let max_results = match req.query_params.get("maxResults") {
2778            Some(raw) => {
2779                let n: i64 = raw
2780                    .parse()
2781                    .map_err(|_| invalid_parameter("maxResults must be an integer"))?;
2782                if !(1..=100).contains(&n) {
2783                    return Err(invalid_parameter("maxResults must be between 1 and 100"));
2784                }
2785                n as usize
2786            }
2787            None => 100,
2788        };
2789        // Enum query params reject values outside the model's declared members.
2790        if let Some(status) = req.query_params.get("status") {
2791            if !["unsupported", "standard_support", "extended_support"].contains(&status.as_str()) {
2792                return Err(invalid_parameter(format!("Invalid status: {status}")));
2793            }
2794        }
2795        if let Some(vs) = req.query_params.get("versionStatus") {
2796            if !["UNSUPPORTED", "STANDARD_SUPPORT", "EXTENDED_SUPPORT"].contains(&vs.as_str()) {
2797                return Err(invalid_parameter(format!("Invalid versionStatus: {vs}")));
2798            }
2799        }
2800        let default_only = req
2801            .query_params
2802            .get("defaultOnly")
2803            .map(|v| v == "true")
2804            .unwrap_or(false);
2805        // `includeAll=false` (the default) hides versions no longer in any
2806        // support window (UNSUPPORTED); `includeAll=true` returns the full set.
2807        let include_all = req
2808            .query_params
2809            .get("includeAll")
2810            .map(|v| v == "true")
2811            .unwrap_or(false);
2812        let status_filter = req.query_params.get("status").cloned();
2813        let version_status_filter = req.query_params.get("versionStatus").cloned();
2814        let version_filter = parse_multi_query(&req.raw_query, "clusterVersions");
2815        let cluster_type = req
2816            .query_params
2817            .get("clusterType")
2818            .cloned()
2819            .unwrap_or_else(|| "eks".to_string());
2820
2821        let mut catalog: Vec<Value> = cluster_version_catalog(&cluster_type)
2822            .into_iter()
2823            .filter(|v| {
2824                version_filter.is_empty()
2825                    || version_filter
2826                        .iter()
2827                        .any(|f| v["clusterVersion"] == f.as_str())
2828            })
2829            .filter(|v| !default_only || v["defaultVersion"] == true)
2830            // The `status` (deprecated) and `versionStatus` params narrow to a
2831            // specific support tier; previously validated then ignored.
2832            .filter(|v| status_filter.as_deref().is_none_or(|s| v["status"] == s))
2833            .filter(|v| {
2834                version_status_filter
2835                    .as_deref()
2836                    .is_none_or(|s| v["versionStatus"] == s)
2837            })
2838            // Drop UNSUPPORTED rows unless the caller asked to include all, or
2839            // explicitly filtered to an unsupported tier.
2840            .filter(|v| {
2841                include_all
2842                    || status_filter.is_some()
2843                    || version_status_filter.is_some()
2844                    || v["versionStatus"] != "UNSUPPORTED"
2845            })
2846            .collect();
2847        // AWS reports the default version first; a stable sort keying non-default
2848        // rows after default ones preserves the ascending version order otherwise.
2849        catalog.sort_by_key(|v| v["defaultVersion"] != true);
2850
2851        let (page, token) = paginate_checked(&catalog, next_token.as_deref(), max_results)
2852            .map_err(|_| invalid_parameter("Invalid nextToken"))?;
2853        let mut out = json!({ "clusterVersions": page });
2854        if let Some(t) = token {
2855            out["nextToken"] = Value::String(t);
2856        }
2857        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
2858    }
2859
2860    // -----------------------------------------------------------------------
2861    // Capabilities
2862    // -----------------------------------------------------------------------
2863
2864    fn create_capability(
2865        &self,
2866        req: &AwsRequest,
2867        cluster_name: &str,
2868    ) -> Result<AwsResponse, AwsServiceError> {
2869        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
2870        let name = body
2871            .get("capabilityName")
2872            .and_then(|v| v.as_str())
2873            .ok_or_else(|| invalid_parameter("capabilityName is required"))?
2874            .to_string();
2875        let type_ = body
2876            .get("type")
2877            .and_then(|v| v.as_str())
2878            .ok_or_else(|| invalid_parameter("type is required"))?
2879            .to_string();
2880        let role_arn = body
2881            .get("roleArn")
2882            .and_then(|v| v.as_str())
2883            .ok_or_else(|| invalid_parameter("roleArn is required"))?
2884            .to_string();
2885
2886        let region = req.region.clone();
2887        let account_id = req.account_id.clone();
2888
2889        let mut accounts = self.state.write();
2890        let state = accounts.get_or_create(&req.account_id);
2891        if !state.clusters.contains_key(cluster_name) {
2892            return Err(not_found_cluster(cluster_name)());
2893        }
2894        if state
2895            .capabilities
2896            .get(cluster_name)
2897            .is_some_and(|m| m.contains_key(&name))
2898        {
2899            return Err(AwsServiceError::aws_error(
2900                StatusCode::CONFLICT,
2901                "ResourceInUseException",
2902                format!(
2903                    "Capability already exists with name {name} and cluster name {cluster_name}"
2904                ),
2905            ));
2906        }
2907
2908        let id = uuid::Uuid::new_v4().to_string();
2909        let arn = capability_arn(&region, &account_id, cluster_name, &name, &id);
2910        let now = Utc::now();
2911        let capability = Capability {
2912            name: name.clone(),
2913            arn,
2914            cluster_name: cluster_name.to_string(),
2915            type_,
2916            role_arn,
2917            status: "CREATING".to_string(),
2918            version: "v1".to_string(),
2919            configuration: normalize_capability_configuration(body.get("configuration")),
2920            tags: parse_tag_map(body.get("tags")),
2921            created_at: now,
2922            modified_at: now,
2923            delete_propagation_policy: str_field(&body, "deletePropagationPolicy"),
2924        };
2925        let out = capability_json(&capability);
2926        state
2927            .capabilities
2928            .entry(cluster_name.to_string())
2929            .or_default()
2930            .insert(name, capability);
2931        Ok(AwsResponse::json(
2932            StatusCode::OK,
2933            json!({ "capability": out }).to_string(),
2934        ))
2935    }
2936
2937    fn describe_capability(
2938        &self,
2939        req: &AwsRequest,
2940        cluster_name: &str,
2941        name: &str,
2942    ) -> Result<AwsResponse, AwsServiceError> {
2943        let mut accounts = self.state.write();
2944        let state = accounts.get_or_create(&req.account_id);
2945        if !state.clusters.contains_key(cluster_name) {
2946            return Err(not_found_cluster(cluster_name)());
2947        }
2948        let capability = state
2949            .capabilities
2950            .get_mut(cluster_name)
2951            .and_then(|m| m.get_mut(name))
2952            .ok_or_else(not_found_capability(name))?;
2953        if capability.status == "CREATING" {
2954            capability.status = "ACTIVE".to_string();
2955        }
2956        Ok(AwsResponse::json(
2957            StatusCode::OK,
2958            json!({ "capability": capability_json(capability) }).to_string(),
2959        ))
2960    }
2961
2962    fn list_capabilities(
2963        &self,
2964        req: &AwsRequest,
2965        cluster_name: &str,
2966    ) -> Result<AwsResponse, AwsServiceError> {
2967        let max_results = validate_max_results(req)?;
2968        let next_token = req.query_params.get("nextToken").cloned();
2969        let accounts = self.state.read();
2970        let state = accounts
2971            .get(&req.account_id)
2972            .ok_or_else(not_found_cluster(cluster_name))?;
2973        if !state.clusters.contains_key(cluster_name) {
2974            return Err(not_found_cluster(cluster_name)());
2975        }
2976        let summaries: Vec<Value> = state
2977            .capabilities
2978            .get(cluster_name)
2979            .map(|m| m.values().map(capability_summary_json).collect())
2980            .unwrap_or_default();
2981        let (page, token) = paginate_checked(&summaries, next_token.as_deref(), max_results)
2982            .map_err(|_| invalid_parameter("Invalid nextToken"))?;
2983        let mut out = json!({ "capabilities": page });
2984        if let Some(t) = token {
2985            out["nextToken"] = Value::String(t);
2986        }
2987        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
2988    }
2989
2990    fn delete_capability(
2991        &self,
2992        req: &AwsRequest,
2993        cluster_name: &str,
2994        name: &str,
2995    ) -> Result<AwsResponse, AwsServiceError> {
2996        let mut accounts = self.state.write();
2997        let state = accounts.get_or_create(&req.account_id);
2998        if !state.clusters.contains_key(cluster_name) {
2999            return Err(not_found_cluster(cluster_name)());
3000        }
3001        let mut capability = state
3002            .capabilities
3003            .get_mut(cluster_name)
3004            .and_then(|m| m.remove(name))
3005            .ok_or_else(not_found_capability(name))?;
3006        capability.status = "DELETING".to_string();
3007        Ok(AwsResponse::json(
3008            StatusCode::OK,
3009            json!({ "capability": capability_json(&capability) }).to_string(),
3010        ))
3011    }
3012
3013    fn update_capability(
3014        &self,
3015        req: &AwsRequest,
3016        cluster_name: &str,
3017        name: &str,
3018    ) -> Result<AwsResponse, AwsServiceError> {
3019        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
3020        let mut accounts = self.state.write();
3021        let state = accounts.get_or_create(&req.account_id);
3022        if !state.clusters.contains_key(cluster_name) {
3023            return Err(not_found_cluster(cluster_name)());
3024        }
3025        let capability = state
3026            .capabilities
3027            .get_mut(cluster_name)
3028            .and_then(|m| m.get_mut(name))
3029            .ok_or_else(not_found_capability(name))?;
3030
3031        let mut params = Vec::new();
3032        if let Some(role) = body.get("roleArn").and_then(|v| v.as_str()) {
3033            capability.role_arn = role.to_string();
3034            params.push(("RoleArn".to_string(), role.to_string()));
3035        }
3036        if let Some(cfg) = normalize_capability_configuration(body.get("configuration")) {
3037            params.push(("Configuration".to_string(), cfg.to_string()));
3038            capability.configuration = Some(cfg);
3039        }
3040        capability.modified_at = Utc::now();
3041
3042        // UpdateCapability returns a tracked cluster-scoped Update.
3043        let update = new_update("CapabilityUpdate", params);
3044        let out = update_json(&update);
3045        let cluster = state.clusters.get_mut(cluster_name).unwrap();
3046        cluster.updates.insert(update.id.clone(), update);
3047        Ok(AwsResponse::json(
3048            StatusCode::OK,
3049            json!({ "update": out }).to_string(),
3050        ))
3051    }
3052
3053    // -----------------------------------------------------------------------
3054    // EKS Anywhere subscriptions
3055    // -----------------------------------------------------------------------
3056
3057    fn create_eks_anywhere_subscription(
3058        &self,
3059        req: &AwsRequest,
3060    ) -> Result<AwsResponse, AwsServiceError> {
3061        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
3062        let name = body
3063            .get("name")
3064            .and_then(|v| v.as_str())
3065            .ok_or_else(|| invalid_parameter("name is required"))?
3066            .to_string();
3067        // EksAnywhereSubscriptionName: length 1-100, ^[0-9A-Za-z][A-Za-z0-9\-_]*$.
3068        if name.is_empty() || name.len() > 100 {
3069            return Err(invalid_parameter("name must be 1-100 characters"));
3070        }
3071        if !name.starts_with(|c: char| c.is_ascii_alphanumeric())
3072            || !name
3073                .chars()
3074                .all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_'))
3075        {
3076            return Err(invalid_parameter(
3077                "name must match ^[0-9A-Za-z][A-Za-z0-9\\-_]*$",
3078            ));
3079        }
3080        if let Some(lt) = body.get("licenseType").and_then(|v| v.as_str()) {
3081            if lt != "Cluster" {
3082                return Err(invalid_parameter(format!("Invalid licenseType: {lt}")));
3083            }
3084        }
3085        let term = body
3086            .get("term")
3087            .filter(|v| v.is_object())
3088            .ok_or_else(|| invalid_parameter("term is required"))?;
3089        let term_duration = term.get("duration").and_then(|v| v.as_i64()).unwrap_or(12);
3090        // AWS EKS Anywhere subscriptions use month-based terms (12 or 36); bound
3091        // the value so a hostile `duration` can't overflow the date arithmetic
3092        // below (`DateTime + Duration` panics on overflow in all build profiles).
3093        if !(1..=120).contains(&term_duration) {
3094            return Err(invalid_parameter(format!(
3095                "term.duration must be between 1 and 120 months, got {term_duration}"
3096            )));
3097        }
3098        let term_unit = term
3099            .get("unit")
3100            .and_then(|v| v.as_str())
3101            .unwrap_or("MONTHS")
3102            .to_string();
3103        let license_quantity = body
3104            .get("licenseQuantity")
3105            .and_then(|v| v.as_i64())
3106            .unwrap_or(0);
3107        let license_type = body
3108            .get("licenseType")
3109            .and_then(|v| v.as_str())
3110            .unwrap_or("Cluster")
3111            .to_string();
3112        let auto_renew = body
3113            .get("autoRenew")
3114            .and_then(|v| v.as_bool())
3115            .unwrap_or(false);
3116
3117        let region = req.region.clone();
3118        let account_id = req.account_id.clone();
3119
3120        let mut accounts = self.state.write();
3121        let state = accounts.get_or_create(&req.account_id);
3122
3123        let raw = uuid::Uuid::new_v4().to_string().replace('-', "");
3124        let id = raw[..17.min(raw.len())].to_string();
3125        let arn = eks_anywhere_subscription_arn(&region, &account_id, &id);
3126        let now = Utc::now();
3127        let expiration = now
3128            .checked_add_signed(chrono::Duration::days(term_duration * 30))
3129            .ok_or_else(|| {
3130                invalid_parameter("term.duration produces an out-of-range expiration date")
3131            })?;
3132        let subscription = EksAnywhereSubscription {
3133            id: id.clone(),
3134            arn,
3135            name,
3136            created_at: now,
3137            effective_date: now,
3138            expiration_date: expiration,
3139            license_quantity,
3140            license_type,
3141            term_duration,
3142            term_unit,
3143            status: "ACTIVE".to_string(),
3144            auto_renew,
3145            tags: parse_tag_map(body.get("tags")),
3146        };
3147        let out = subscription_json(&subscription);
3148        state.eks_anywhere_subscriptions.insert(id, subscription);
3149        Ok(AwsResponse::json(
3150            StatusCode::OK,
3151            json!({ "subscription": out }).to_string(),
3152        ))
3153    }
3154
3155    fn list_eks_anywhere_subscriptions(
3156        &self,
3157        req: &AwsRequest,
3158    ) -> Result<AwsResponse, AwsServiceError> {
3159        let max_results = validate_max_results(req)?;
3160        let next_token = req.query_params.get("nextToken").cloned();
3161        let accounts = self.state.read();
3162        let Some(state) = accounts.get(&req.account_id) else {
3163            return Ok(AwsResponse::json(
3164                StatusCode::OK,
3165                json!({ "subscriptions": [] }).to_string(),
3166            ));
3167        };
3168        let subs: Vec<Value> = state
3169            .eks_anywhere_subscriptions
3170            .values()
3171            .map(subscription_json)
3172            .collect();
3173        let (page, token) = paginate_checked(&subs, next_token.as_deref(), max_results)
3174            .map_err(|_| invalid_parameter("Invalid nextToken"))?;
3175        let mut out = json!({ "subscriptions": page });
3176        if let Some(t) = token {
3177            out["nextToken"] = Value::String(t);
3178        }
3179        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
3180    }
3181
3182    fn describe_eks_anywhere_subscription(
3183        &self,
3184        req: &AwsRequest,
3185        id: &str,
3186    ) -> Result<AwsResponse, AwsServiceError> {
3187        let accounts = self.state.read();
3188        let state = accounts
3189            .get(&req.account_id)
3190            .ok_or_else(not_found_subscription(id))?;
3191        let subscription = state
3192            .eks_anywhere_subscriptions
3193            .get(id)
3194            .ok_or_else(not_found_subscription(id))?;
3195        Ok(AwsResponse::json(
3196            StatusCode::OK,
3197            json!({ "subscription": subscription_json(subscription) }).to_string(),
3198        ))
3199    }
3200
3201    fn delete_eks_anywhere_subscription(
3202        &self,
3203        req: &AwsRequest,
3204        id: &str,
3205    ) -> Result<AwsResponse, AwsServiceError> {
3206        let mut accounts = self.state.write();
3207        let state = accounts.get_or_create(&req.account_id);
3208        let mut subscription = state
3209            .eks_anywhere_subscriptions
3210            .remove(id)
3211            .ok_or_else(not_found_subscription(id))?;
3212        subscription.status = "DELETING".to_string();
3213        Ok(AwsResponse::json(
3214            StatusCode::OK,
3215            json!({ "subscription": subscription_json(&subscription) }).to_string(),
3216        ))
3217    }
3218
3219    fn update_eks_anywhere_subscription(
3220        &self,
3221        req: &AwsRequest,
3222        id: &str,
3223    ) -> Result<AwsResponse, AwsServiceError> {
3224        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
3225        let auto_renew = body
3226            .get("autoRenew")
3227            .and_then(|v| v.as_bool())
3228            .ok_or_else(|| invalid_parameter("autoRenew is required"))?;
3229        let mut accounts = self.state.write();
3230        let state = accounts.get_or_create(&req.account_id);
3231        let subscription = state
3232            .eks_anywhere_subscriptions
3233            .get_mut(id)
3234            .ok_or_else(not_found_subscription(id))?;
3235        subscription.auto_renew = auto_renew;
3236        Ok(AwsResponse::json(
3237            StatusCode::OK,
3238            json!({ "subscription": subscription_json(subscription) }).to_string(),
3239        ))
3240    }
3241}
3242
3243/// Persist the current EKS state as a snapshot. Shared by
3244/// `EksService::save_snapshot` and the CFN provisioner's persist hook.
3245pub async fn save_eks_snapshot(
3246    state: &SharedEksState,
3247    store: Option<Arc<dyn SnapshotStore>>,
3248    lock: &AsyncMutex<()>,
3249) {
3250    let Some(store) = store else {
3251        return;
3252    };
3253    let _guard = lock.lock().await;
3254    let snapshot = EksSnapshot {
3255        schema_version: EKS_SNAPSHOT_SCHEMA_VERSION,
3256        accounts: state.read().clone(),
3257    };
3258    let join = tokio::task::spawn_blocking(move || -> std::io::Result<()> {
3259        let bytes = serde_json::to_vec(&snapshot)
3260            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string()))?;
3261        store.save(&bytes)
3262    })
3263    .await;
3264    match join {
3265        Ok(Ok(())) => {}
3266        Ok(Err(err)) => tracing::error!(%err, "failed to write eks snapshot"),
3267        Err(err) => tracing::error!(%err, "eks snapshot task panicked"),
3268    }
3269}
3270
3271#[async_trait]
3272impl AwsService for EksService {
3273    fn service_name(&self) -> &str {
3274        "eks"
3275    }
3276
3277    async fn handle(&self, req: AwsRequest) -> Result<AwsResponse, AwsServiceError> {
3278        let (action, args) = Self::resolve_action(&req).ok_or_else(|| {
3279            AwsServiceError::aws_error(
3280                StatusCode::NOT_FOUND,
3281                "UnknownOperationException",
3282                format!("Unknown operation: {} {}", req.method, req.raw_path),
3283            )
3284        })?;
3285
3286        let mutates = matches!(
3287            action,
3288            "CreateCluster"
3289                | "DeleteCluster"
3290                | "UpdateClusterConfig"
3291                | "UpdateClusterVersion"
3292                | "TagResource"
3293                | "UntagResource"
3294                // Describe settles CREATING -> ACTIVE and IN_PROGRESS updates,
3295                // which mutates persisted state.
3296                | "DescribeCluster"
3297                | "DescribeUpdate"
3298                | "CreateNodegroup"
3299                | "DeleteNodegroup"
3300                | "UpdateNodegroupConfig"
3301                | "UpdateNodegroupVersion"
3302                | "DescribeNodegroup"
3303                | "CreateFargateProfile"
3304                | "DeleteFargateProfile"
3305                | "DescribeFargateProfile"
3306                | "CreateAddon"
3307                | "DeleteAddon"
3308                | "UpdateAddon"
3309                | "DescribeAddon"
3310                | "CreateAccessEntry"
3311                | "DeleteAccessEntry"
3312                | "UpdateAccessEntry"
3313                | "AssociateAccessPolicy"
3314                | "DisassociateAccessPolicy"
3315                | "AssociateIdentityProviderConfig"
3316                | "DisassociateIdentityProviderConfig"
3317                // Describe settles the config's CREATING -> ACTIVE transition.
3318                | "DescribeIdentityProviderConfig"
3319                | "CreatePodIdentityAssociation"
3320                | "DeletePodIdentityAssociation"
3321                | "UpdatePodIdentityAssociation"
3322                // Insights are seeded/settled on read; refresh settles on describe.
3323                | "ListInsights"
3324                | "DescribeInsight"
3325                | "DescribeInsightsRefresh"
3326                | "StartInsightsRefresh"
3327                | "AssociateEncryptionConfig"
3328                | "CancelUpdate"
3329                | "RegisterCluster"
3330                | "DeregisterCluster"
3331                | "CreateCapability"
3332                | "DeleteCapability"
3333                | "UpdateCapability"
3334                | "DescribeCapability"
3335                | "CreateEksAnywhereSubscription"
3336                | "DeleteEksAnywhereSubscription"
3337                | "UpdateEksAnywhereSubscription"
3338                | "DescribeEksAnywhereSubscription"
3339        );
3340
3341        let result = match (action, &args) {
3342            ("CreateCluster", _) => self.create_cluster(&req),
3343            ("ListClusters", _) => self.list_clusters(&req),
3344            ("DescribeCluster", PathArgs::Name(n)) => self.describe_cluster(&req, n),
3345            ("DeleteCluster", PathArgs::Name(n)) => self.delete_cluster(&req, n),
3346            ("UpdateClusterConfig", PathArgs::Name(n)) => self.update_cluster_config(&req, n),
3347            ("UpdateClusterVersion", PathArgs::Name(n)) => self.update_cluster_version(&req, n),
3348            ("ListUpdates", PathArgs::Name(n)) => self.list_updates(&req, n),
3349            ("DescribeUpdate", PathArgs::Update { name, update_id }) => {
3350                self.describe_update(&req, name, update_id)
3351            }
3352            ("TagResource", PathArgs::Arn(a)) => self.tag_resource(&req, a),
3353            ("UntagResource", PathArgs::Arn(a)) => self.untag_resource(&req, a),
3354            ("ListTagsForResource", PathArgs::Arn(a)) => self.list_tags_for_resource(&req, a),
3355            ("CreateNodegroup", PathArgs::Cluster(c)) => self.create_nodegroup(&req, c),
3356            ("ListNodegroups", PathArgs::Cluster(c)) => self.list_nodegroups(&req, c),
3357            ("DescribeNodegroup", PathArgs::ClusterChild { cluster, name }) => {
3358                self.describe_nodegroup(&req, cluster, name)
3359            }
3360            ("DeleteNodegroup", PathArgs::ClusterChild { cluster, name }) => {
3361                self.delete_nodegroup(&req, cluster, name)
3362            }
3363            ("UpdateNodegroupConfig", PathArgs::ClusterChild { cluster, name }) => {
3364                self.update_nodegroup_config(&req, cluster, name)
3365            }
3366            ("UpdateNodegroupVersion", PathArgs::ClusterChild { cluster, name }) => {
3367                self.update_nodegroup_version(&req, cluster, name)
3368            }
3369            ("CreateFargateProfile", PathArgs::Cluster(c)) => self.create_fargate_profile(&req, c),
3370            ("ListFargateProfiles", PathArgs::Cluster(c)) => self.list_fargate_profiles(&req, c),
3371            ("DescribeFargateProfile", PathArgs::ClusterChild { cluster, name }) => {
3372                self.describe_fargate_profile(&req, cluster, name)
3373            }
3374            ("DeleteFargateProfile", PathArgs::ClusterChild { cluster, name }) => {
3375                self.delete_fargate_profile(&req, cluster, name)
3376            }
3377            ("CreateAddon", PathArgs::Cluster(c)) => self.create_addon(&req, c),
3378            ("ListAddons", PathArgs::Cluster(c)) => self.list_addons(&req, c),
3379            ("DescribeAddon", PathArgs::ClusterChild { cluster, name }) => {
3380                self.describe_addon(&req, cluster, name)
3381            }
3382            ("DeleteAddon", PathArgs::ClusterChild { cluster, name }) => {
3383                self.delete_addon(&req, cluster, name)
3384            }
3385            ("UpdateAddon", PathArgs::ClusterChild { cluster, name }) => {
3386                self.update_addon(&req, cluster, name)
3387            }
3388            ("DescribeAddonVersions", _) => self.describe_addon_versions(&req),
3389            ("DescribeAddonConfiguration", _) => self.describe_addon_configuration(&req),
3390            ("CreateAccessEntry", PathArgs::Cluster(c)) => self.create_access_entry(&req, c),
3391            ("ListAccessEntries", PathArgs::Cluster(c)) => self.list_access_entries(&req, c),
3392            ("DescribeAccessEntry", PathArgs::ClusterChild { cluster, name }) => {
3393                self.describe_access_entry(&req, cluster, name)
3394            }
3395            ("DeleteAccessEntry", PathArgs::ClusterChild { cluster, name }) => {
3396                self.delete_access_entry(&req, cluster, name)
3397            }
3398            ("UpdateAccessEntry", PathArgs::ClusterChild { cluster, name }) => {
3399                self.update_access_entry(&req, cluster, name)
3400            }
3401            ("AssociateAccessPolicy", PathArgs::ClusterChild { cluster, name }) => {
3402                self.associate_access_policy(&req, cluster, name)
3403            }
3404            ("ListAssociatedAccessPolicies", PathArgs::ClusterChild { cluster, name }) => {
3405                self.list_associated_access_policies(&req, cluster, name)
3406            }
3407            (
3408                "DisassociateAccessPolicy",
3409                PathArgs::AccessPolicyChild {
3410                    cluster,
3411                    principal,
3412                    policy_arn,
3413                },
3414            ) => self.disassociate_access_policy(&req, cluster, principal, policy_arn),
3415            ("ListAccessPolicies", _) => self.list_access_policies(&req),
3416            ("AssociateIdentityProviderConfig", PathArgs::Cluster(c)) => {
3417                self.associate_identity_provider_config(&req, c)
3418            }
3419            ("DisassociateIdentityProviderConfig", PathArgs::Cluster(c)) => {
3420                self.disassociate_identity_provider_config(&req, c)
3421            }
3422            ("DescribeIdentityProviderConfig", PathArgs::Cluster(c)) => {
3423                self.describe_identity_provider_config(&req, c)
3424            }
3425            ("ListIdentityProviderConfigs", PathArgs::Cluster(c)) => {
3426                self.list_identity_provider_configs(&req, c)
3427            }
3428            ("CreatePodIdentityAssociation", PathArgs::Cluster(c)) => {
3429                self.create_pod_identity_association(&req, c)
3430            }
3431            ("ListPodIdentityAssociations", PathArgs::Cluster(c)) => {
3432                self.list_pod_identity_associations(&req, c)
3433            }
3434            ("DescribePodIdentityAssociation", PathArgs::ClusterChild { cluster, name }) => {
3435                self.describe_pod_identity_association(&req, cluster, name)
3436            }
3437            ("DeletePodIdentityAssociation", PathArgs::ClusterChild { cluster, name }) => {
3438                self.delete_pod_identity_association(&req, cluster, name)
3439            }
3440            ("UpdatePodIdentityAssociation", PathArgs::ClusterChild { cluster, name }) => {
3441                self.update_pod_identity_association(&req, cluster, name)
3442            }
3443            ("ListInsights", PathArgs::Cluster(c)) => self.list_insights(&req, c),
3444            ("DescribeInsight", PathArgs::ClusterChild { cluster, name }) => {
3445                self.describe_insight(&req, cluster, name)
3446            }
3447            ("DescribeInsightsRefresh", PathArgs::Cluster(c)) => {
3448                self.describe_insights_refresh(&req, c)
3449            }
3450            ("StartInsightsRefresh", PathArgs::Cluster(c)) => self.start_insights_refresh(&req, c),
3451            ("AssociateEncryptionConfig", PathArgs::Cluster(c)) => {
3452                self.associate_encryption_config(&req, c)
3453            }
3454            ("CancelUpdate", PathArgs::Update { name, update_id }) => {
3455                self.cancel_update(&req, name, update_id)
3456            }
3457            ("RegisterCluster", _) => self.register_cluster(&req),
3458            ("DeregisterCluster", PathArgs::Name(n)) => self.deregister_cluster(&req, n),
3459            ("DescribeClusterVersions", _) => self.describe_cluster_versions(&req),
3460            ("CreateCapability", PathArgs::Cluster(c)) => self.create_capability(&req, c),
3461            ("ListCapabilities", PathArgs::Cluster(c)) => self.list_capabilities(&req, c),
3462            ("DescribeCapability", PathArgs::ClusterChild { cluster, name }) => {
3463                self.describe_capability(&req, cluster, name)
3464            }
3465            ("DeleteCapability", PathArgs::ClusterChild { cluster, name }) => {
3466                self.delete_capability(&req, cluster, name)
3467            }
3468            ("UpdateCapability", PathArgs::ClusterChild { cluster, name }) => {
3469                self.update_capability(&req, cluster, name)
3470            }
3471            ("CreateEksAnywhereSubscription", _) => self.create_eks_anywhere_subscription(&req),
3472            ("ListEksAnywhereSubscriptions", _) => self.list_eks_anywhere_subscriptions(&req),
3473            ("DescribeEksAnywhereSubscription", PathArgs::Name(id)) => {
3474                self.describe_eks_anywhere_subscription(&req, id)
3475            }
3476            ("DeleteEksAnywhereSubscription", PathArgs::Name(id)) => {
3477                self.delete_eks_anywhere_subscription(&req, id)
3478            }
3479            ("UpdateEksAnywhereSubscription", PathArgs::Name(id)) => {
3480                self.update_eks_anywhere_subscription(&req, id)
3481            }
3482            _ => Err(AwsServiceError::action_not_implemented("eks", action)),
3483        };
3484
3485        if mutates && matches!(result.as_ref(), Ok(resp) if resp.status.is_success()) {
3486            self.save_snapshot().await;
3487        }
3488        result
3489    }
3490
3491    fn supported_actions(&self) -> &[&str] {
3492        EKS_ACTIONS
3493    }
3494}
3495
3496// ---------------------------------------------------------------------------
3497// Helpers
3498// ---------------------------------------------------------------------------
3499
3500#[cfg(test)]
3501#[path = "service_tests.rs"]
3502mod tests;