PodMounter: Add Mountpoint Pod Sharing Support

Makefile

@@ -174,6 +174,14 @@ check_style:
 clean:
 	rm -rf bin/ && docker system prune
 
+# Generate files for Custom Resources (`zz_generated.deepcopy.go` and CustomResourceDefinition YAML file).
+# TODO: Wrap CRD YAML file with experimental.podMounter=true Helm flag
+# POD_ATTACHMENT_CRD_FILE ?= "./charts/aws-mountpoint-s3-csi-driver/templates/s3.csi.aws.com_mountpoints3podattachments.yaml"
+.PHONY: generate
+generate:
+	controller-gen object:headerFile="hack/boilerplate.go.txt" paths="./pkg/api/..."
+	controller-gen crd paths="./pkg/api/..." output:crd:dir=./charts/aws-mountpoint-s3-csi-driver/templates
+
 ## Binaries used in tests.
 
 TESTBIN ?= $(shell pwd)/tests/bin

charts/aws-mountpoint-s3-csi-driver/templates/s3.csi.aws.com_mountpoints3podattachments.yaml

@@ -0,0 +1,101 @@
+---
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  annotations:
+    controller-gen.kubebuilder.io/version: v0.17.3
+  name: mountpoints3podattachments.s3.csi.aws.com
+spec:
+  group: s3.csi.aws.com
+  names:
+    kind: MountpointS3PodAttachment
+    listKind: MountpointS3PodAttachmentList
+    plural: mountpoints3podattachments
+    shortNames:
+    - s3pa
+    singular: mountpoints3podattachment
+  scope: Cluster
+  versions:
+  - name: v1
+    schema:
+      openAPIV3Schema:
+        description: MountpointS3PodAttachment is the Schema for the mountpoints3podattachments
+          API.
+        properties:
+          apiVersion:
+            description: |-
+              APIVersion defines the versioned schema of this representation of an object.
+              Servers should convert recognized schemas to the latest internal value, and
+              may reject unrecognized values.
+              More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources
+            type: string
+          kind:
+            description: |-
+              Kind is a string value representing the REST resource this object represents.
+              Servers may infer this from the endpoint the client submits requests to.
+              Cannot be updated.
+              In CamelCase.
+              More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds
+            type: string
+          metadata:
+            type: object
+          spec:
+            description: MountpointS3PodAttachmentSpec defines the desired state of
+              MountpointS3PodAttachment.
+            properties:
+              authenticationSource:
+                description: Authentication source taken from volume attribute field
+                  `authenticationSource`.
+                type: string
+              mountOptions:
+                description: Comma separated mount options taken from volume.
+                type: string
+              mountpointS3PodToWorkloadPodUIDs:
+                additionalProperties:
+                  items:
+                    type: string
+                  type: array
+                description: Maps each Mountpoint S3 pod name to the list of workload
+                  pod UIDs it is attached to.
+                type: object
+              nodeName:
+                description: Name of the node.
+                type: string
+              persistentVolumeName:
+                description: Name of the Persistent Volume.
+                type: string
+              volumeID:
+                description: Volume ID.
+                type: string
+              workloadFSGroup:
+                description: Workload pod's `fsGroup` from pod security context
+                type: string
+              workloadNamespace:
+                description: 'Workload pod''s namespace. Exists only if `authenticationSource:
+                  pod`.'
+                type: string
+              workloadServiceAccountIAMRoleARN:
+                description: 'EKS IAM Role ARN from workload pod''s service account
+                  annotation (IRSA). Exists only if `authenticationSource: pod` and
+                  service account has `eks.amazonaws.com/role-arn` annotation.'
+                type: string
+              workloadServiceAccountName:
+                description: 'Workload pod''s service account name. Exists only if
+                  `authenticationSource: pod`.'
+                type: string
+            required:
+            - authenticationSource
+            - mountOptions
+            - mountpointS3PodToWorkloadPodUIDs
+            - nodeName
+            - persistentVolumeName
+            - volumeID
+            - workloadFSGroup
+            type: object
+        type: object
+    selectableFields:
+    - jsonPath: .spec.nodeName
+    served: true
+    storage: true
+    subresources:
+      status: {}

charts/aws-mountpoint-s3-csi-driver/templates/serviceaccount-csi-controller.yaml

@@ -49,8 +49,11 @@ metadata:
     {{- include "aws-mountpoint-s3-csi-driver.labels" . | nindent 4 }}
 rules:
   - apiGroups: [""]
-    resources: ["pods", "persistentvolumeclaims", "persistentvolumes"]
+    resources: ["pods", "persistentvolumeclaims", "persistentvolumes", "serviceaccounts"]
     verbs: ["get", "watch", "list"]
+  - apiGroups: ["s3.csi.aws.com"]
+    resources: ["mountpoints3podattachments"]
+    verbs: ["create", "delete", "update", "get", "watch", "list"]
 ---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1

charts/aws-mountpoint-s3-csi-driver/templates/serviceaccount-csi-node.yaml

@@ -23,8 +23,11 @@ metadata:
     app.kubernetes.io/name: aws-mountpoint-s3-csi-driver
 rules:
   - apiGroups: [""]
-    resources: ["serviceaccounts"]
+    resources: ["serviceaccounts"] # TODO: Remove once we stop supporting systemd mounts.
     verbs: ["get"]
+  - apiGroups: ["s3.csi.aws.com"]
+    resources: ["mountpoints3podattachments"]
+    verbs: ["get", "list", "watch"]
 ---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1

cmd/aws-s3-csi-controller/csicontroller/expectations.go

@@ -0,0 +1,71 @@
+package csicontroller
+
+import (
+	"sort"
+	"strings"
+	"sync"
+
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+// Expectations is a structure that manages pending expectations for Kubernetes resources.
+// It uses field filters as keys to track resources that are expected to be created
+// helping to reduce unnecessary processing and API server load.
+type Expectations struct {
+	pending sync.Map
+}
+
+// NewExpectations creates and returns a new Expectations instance.
+func NewExpectations() *Expectations {
+	return &Expectations{}
+}
+
+// SetPending marks a resource as pending based on the given field filters.
+// This is typically used when a create operation is initiated.
+func (e *Expectations) SetPending(fieldFilters client.MatchingFields) {
+	key := deriveExpectationKeyFromFilters(fieldFilters)
+	e.pending.Store(key, struct{}{})
+}
+
+// IsPending checks if a resource is marked as pending based on the given field filters.
+// Returns true if the resource is pending, false otherwise.
+func (e *Expectations) IsPending(fieldFilters client.MatchingFields) bool {
+	key := deriveExpectationKeyFromFilters(fieldFilters)
+	_, ok := e.pending.Load(key)
+	return ok
+}
+
+// Clear removes the pending mark for a resource based on the given field filters.
+// This is typically called when an expected operation has been confirmed as completed.
+func (e *Expectations) Clear(fieldFilters client.MatchingFields) {
+	key := deriveExpectationKeyFromFilters(fieldFilters)
+	e.pending.Delete(key)
+}
+
+// deriveExpectationKeyFromFilters generates a deterministic string key from a map of field filters.
+// It creates a consistent string representation of the filters by:
+// 1. Sorting the filter keys alphabetically
+// 2. Concatenating each key-value pair in the format "key=value;"
+//
+// For example, given filters {"foo": "bar", "baz": "qux"}, it will produce "baz=qux;foo=bar;"
+//
+// Parameters:
+//   - fieldFilters: A map of field names to their values used for filtering Kubernetes resources
+//
+// Returns:
+//   - A string that uniquely represents the combination of field filters
+func deriveExpectationKeyFromFilters(fieldFilters client.MatchingFields) string {
+	keys := make([]string, 0, len(fieldFilters))
+	for k := range fieldFilters {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	var sb strings.Builder
+	for _, k := range keys {
+		sb.WriteString(k)
+		sb.WriteString("=")
+		sb.WriteString(fieldFilters[k])
+		sb.WriteString(";")
+	}
+	return sb.String()
+}