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