PodMounter: Add Mountpoint Pod Sharing Support

Makefile

@@ -190,6 +190,23 @@ generate_licenses: download_go_deps
 clean:
 	rm -rf bin/ && docker system prune
 
+# Generate files for Custom Resources (`zz_generated.deepcopy.go` and CustomResourceDefinition YAML file).
+TMP_POD_ATTACHMENT_CRD_FILE ?= "./hack/s3.csi.aws.com_mountpoints3podattachments.yaml"
+# Helm CRD file needs extra `{{- if .Values.experimental.podMounter -}}` wrapper, while we don't need it for tests.
+HELM_POD_ATTACHMENT_CRD_FILE ?= "./charts/aws-mountpoint-s3-csi-driver/templates/mountpoints3podattachments-crd.yaml"
+TEST_POD_ATTACHMENT_CRD_FILE ?= "./tests/crd/mountpoints3podattachments-crd.yaml"
+.PHONY: generate
+generate:
+	controller-gen object:headerFile="hack/boilerplate.go.txt" paths="./pkg/api/..."
+	controller-gen crd paths="./pkg/api/..." output:crd:dir=./hack/
+	echo '# Auto-generated file via `make generate`. Do not edit.' > $(HELM_POD_ATTACHMENT_CRD_FILE)
+	echo '{{- if .Values.experimental.podMounter -}}' >> $(HELM_POD_ATTACHMENT_CRD_FILE)
+	cat $(TMP_POD_ATTACHMENT_CRD_FILE) >> $(HELM_POD_ATTACHMENT_CRD_FILE)
+	echo '{{- end -}}' >> $(HELM_POD_ATTACHMENT_CRD_FILE)
+	echo '# Auto-generated file via `make generate`. Do not edit.' > $(TEST_POD_ATTACHMENT_CRD_FILE)
+	cat $(TMP_POD_ATTACHMENT_CRD_FILE) >> $(TEST_POD_ATTACHMENT_CRD_FILE)
+	rm $(TMP_POD_ATTACHMENT_CRD_FILE)
+
 ## Binaries used in tests.
 
 TESTBIN ?= $(shell pwd)/tests/bin

charts/aws-mountpoint-s3-csi-driver/templates/mountpoints3podattachments-crd.yaml

@@ -0,0 +1,118 @@
+# Auto-generated file via `make generate`. Do not edit.
+{{- if .Values.experimental.podMounter -}}
+---
+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: v1beta
+    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
+              mountpointS3PodAttachments:
+                additionalProperties:
+                  items:
+                    description: WorkloadAttachment represents the attachment details
+                      of a workload pod to a Mountpoint S3 pod.
+                    properties:
+                      attachmentTime:
+                        description: AttachmentTime represents when the workload pod
+                          was attached to the Mountpoint S3 pod
+                        format: date-time
+                        type: string
+                      workloadPodUID:
+                        description: WorkloadPodUID is the unique identifier of the
+                          attached workload pod
+                        type: string
+                    required:
+                    - attachmentTime
+                    - workloadPodUID
+                    type: object
+                  type: array
+                description: Maps each Mountpoint S3 pod name to its workload attachments
+                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
+            - mountpointS3PodAttachments
+            - nodeName
+            - persistentVolumeName
+            - volumeID
+            - workloadFSGroup
+            type: object
+        type: object
+    selectableFields:
+    - jsonPath: .spec.nodeName
+    served: true
+    storage: true
+    subresources:
+      status: {}
+{{- end -}}

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

@@ -23,7 +23,7 @@ metadata:
 rules:
   - apiGroups: [""]
     resources: ["pods"]
-    verbs: ["get", "create", "watch", "delete", "list"]
+    verbs: ["get", "create", "watch", "delete", "list", "update"]
 ---
 kind: RoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
@@ -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 because in PodMounter we get IRSA Role ARN from MountpointS3PodAttachment
     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 with eventual consistency, reducing 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.WriteRune('=')
+		sb.WriteString(fieldFilters[k])
+		sb.WriteRune(';')
+	}
+	return sb.String()
+}

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

@@ -0,0 +1,105 @@
+package csicontroller
+
+import (
+	"testing"
+
+	"github.com/awslabs/aws-s3-csi-driver/pkg/util/testutil/assert"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+func TestDeriveExpectationKeyFromFilters(t *testing.T) {
+	tests := []struct {
+		name         string
+		fieldFilters client.MatchingFields
+		want         string
+	}{
+		{
+			name:         "empty filters",
+			fieldFilters: client.MatchingFields{},
+			want:         "",
+		},
+		{
+			name: "single filter",
+			fieldFilters: client.MatchingFields{
+				"key1": "value1",
+			},
+			want: "key1=value1;",
+		},
+		{
+			name: "multiple filters",
+			fieldFilters: client.MatchingFields{
+				"key2": "value2",
+				"key1": "value1",
+				"key3": "value3",
+			},
+			want: "key1=value1;key2=value2;key3=value3;",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := deriveExpectationKeyFromFilters(tt.fieldFilters)
+			assert.Equals(t, tt.want, got)
+		})
+	}
+}
+
+func TestExpectations(t *testing.T) {
+	tests := []struct {
+		name         string
+		fieldFilters client.MatchingFields
+		operations   func(*expectations)
+		wantPending  bool
+	}{
+		{
+			name: "set and check pending",
+			fieldFilters: client.MatchingFields{
+				"key1": "value1",
+			},
+			operations: func(e *expectations) {
+				e.setPending(client.MatchingFields{"key1": "value1"})
+			},
+			wantPending: true,
+		},
+		{
+			name: "set and clear pending",
+			fieldFilters: client.MatchingFields{
+				"key1": "value1",
+			},
+			operations: func(e *expectations) {
+				e.setPending(client.MatchingFields{"key1": "value1"})
+				e.clear(client.MatchingFields{"key1": "value1"})
+			},
+			wantPending: false,
+		},
+		{
+			name: "check non-existent pending",
+			fieldFilters: client.MatchingFields{
+				"key1": "value1",
+			},
+			operations:  func(e *expectations) {},
+			wantPending: false,
+		},
+		{
+			name: "multiple operations",
+			fieldFilters: client.MatchingFields{
+				"key1": "value1",
+				"key2": "value2",
+			},
+			operations: func(e *expectations) {
+				e.setPending(client.MatchingFields{"key1": "value1", "key2": "value2"})
+				e.clear(client.MatchingFields{"key1": "value1"}) // Different key, shouldn't affect the test
+			},
+			wantPending: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			e := newExpectations()
+			tt.operations(e)
+			got := e.isPending(tt.fieldFilters)
+			assert.Equals(t, tt.wantPending, got)
+		})
+	}
+}