GenericEphemeralVolume: initial implementation

The implementation consists of
- identifying all places where VolumeSource.PersistentVolumeClaim has
  a special meaning and then ensuring that the same code path is taken
  for an ephemeral volume, with the ownership check
- adding a controller that produces the PVCs for each embedded
  VolumeSource.EphemeralVolume
- relaxing the PVC protection controller such that it removes
  the finalizer already before the pod is deleted (only
  if the GenericEphemeralVolume feature is enabled): this is
  needed to break a cycle where foreground deletion of the pod
  blocks on removing the PVC, which waits for deletion of the pod

The controller was derived from the endpointslices controller.

Author: pohly

cmd/kube-controller-manager/app/BUILD

@@ -74,6 +74,7 @@ go_library(
         "//pkg/controller/ttl:go_default_library",
         "//pkg/controller/ttlafterfinished:go_default_library",
         "//pkg/controller/volume/attachdetach:go_default_library",
+        "//pkg/controller/volume/ephemeral:go_default_library",
         "//pkg/controller/volume/expand:go_default_library",
         "//pkg/controller/volume/persistentvolume:go_default_library",
         "//pkg/controller/volume/persistentvolume/config:go_default_library",

cmd/kube-controller-manager/app/controllermanager.go

@@ -423,6 +423,7 @@ func NewControllerInitializers(loopMode ControllerLoopMode) map[string]InitFunc
 	controllers["pv-protection"] = startPVProtectionController
 	controllers["ttl-after-finished"] = startTTLAfterFinishedController
 	controllers["root-ca-cert-publisher"] = startRootCACertPublisher
+	controllers["ephemeral-volume"] = startEphemeralVolumeController
 
 	return controllers
 }

cmd/kube-controller-manager/app/core.go

@@ -57,6 +57,7 @@ import (
 	ttlcontroller "k8s.io/kubernetes/pkg/controller/ttl"
 	"k8s.io/kubernetes/pkg/controller/ttlafterfinished"
 	"k8s.io/kubernetes/pkg/controller/volume/attachdetach"
+	"k8s.io/kubernetes/pkg/controller/volume/ephemeral"
 	"k8s.io/kubernetes/pkg/controller/volume/expand"
 	persistentvolumecontroller "k8s.io/kubernetes/pkg/controller/volume/persistentvolume"
 	"k8s.io/kubernetes/pkg/controller/volume/pvcprotection"
@@ -373,6 +374,22 @@ func startVolumeExpandController(ctx ControllerContext) (http.Handler, bool, err
 	return nil, false, nil
 }
 
+func startEphemeralVolumeController(ctx ControllerContext) (http.Handler, bool, error) {
+	if utilfeature.DefaultFeatureGate.Enabled(features.GenericEphemeralVolume) {
+		ephemeralController, err := ephemeral.NewController(
+			ctx.ClientBuilder.ClientOrDie("ephemeral-volume-controller"),
+			ctx.InformerFactory.Core().V1().Pods(),
+			ctx.InformerFactory.Core().V1().PersistentVolumeClaims())
+		if err != nil {
+			return nil, true, fmt.Errorf("failed to start ephemeral volume controller: %v", err)
+		}
+		// TODO (before beta at the latest): make this configurable similar to the EndpointController
+		go ephemeralController.Run(1 /* int(ctx.ComponentConfig.EphemeralController.ConcurrentEphemeralVolumeSyncs) */, ctx.Stop)
+		return nil, true, nil
+	}
+	return nil, false, nil
+}
+
 func startEndpointController(ctx ControllerContext) (http.Handler, bool, error) {
 	go endpointcontroller.NewEndpointController(
 		ctx.InformerFactory.Core().V1().Pods(),
@@ -539,6 +556,7 @@ func startPVCProtectionController(ctx ControllerContext) (http.Handler, bool, er
 		ctx.InformerFactory.Core().V1().Pods(),
 		ctx.ClientBuilder.ClientOrDie("pvc-protection-controller"),
 		utilfeature.DefaultFeatureGate.Enabled(features.StorageObjectInUseProtection),
+		utilfeature.DefaultFeatureGate.Enabled(features.StorageObjectInUseProtection),
 	)
 	if err != nil {
 		return nil, true, fmt.Errorf("failed to start the pvc protection controller: %v", err)

pkg/controller/BUILD

@@ -136,6 +136,7 @@ filegroup(
         "//pkg/controller/util/node:all-srcs",
         "//pkg/controller/volume/attachdetach:all-srcs",
         "//pkg/controller/volume/common:all-srcs",
+        "//pkg/controller/volume/ephemeral:all-srcs",
         "//pkg/controller/volume/events:all-srcs",
         "//pkg/controller/volume/expand:all-srcs",
         "//pkg/controller/volume/persistentvolume:all-srcs",

pkg/controller/volume/attachdetach/attach_detach_controller.go

@@ -203,7 +203,7 @@ func NewAttachDetachController(
 
 	// This custom indexer will index pods by its PVC keys. Then we don't need
 	// to iterate all pods every time to find pods which reference given PVC.
-	if err := common.AddIndexerIfNotPresent(adc.podIndexer, common.PodPVCIndex, common.PodPVCIndexFunc); err != nil {
+	if err := common.AddPodPVCIndexerIfNotPresent(adc.podIndexer); err != nil {
 		return nil, fmt.Errorf("Could not initialize attach detach controller: %v", err)
 	}
 
@@ -425,7 +425,7 @@ func (adc *attachDetachController) populateDesiredStateOfWorld() error {
 			// The volume specs present in the ActualStateOfWorld are nil, let's replace those
 			// with the correct ones found on pods. The present in the ASW with no corresponding
 			// pod will be detached and the spec is irrelevant.
-			volumeSpec, err := util.CreateVolumeSpec(podVolume, podToAdd.Namespace, nodeName, &adc.volumePluginMgr, adc.pvcLister, adc.pvLister, adc.csiMigratedPluginManager, adc.intreeToCSITranslator)
+			volumeSpec, err := util.CreateVolumeSpec(podVolume, podToAdd, nodeName, &adc.volumePluginMgr, adc.pvcLister, adc.pvLister, adc.csiMigratedPluginManager, adc.intreeToCSITranslator)
 			if err != nil {
 				klog.Errorf(
 					"Error creating spec for volume %q, pod %q/%q: %v",

pkg/controller/volume/attachdetach/metrics/metrics.go

@@ -168,7 +168,7 @@ func (collector *attachDetachStateCollector) getVolumeInUseCount() volumeCount {
 			continue
 		}
 		for _, podVolume := range pod.Spec.Volumes {
-			volumeSpec, err := util.CreateVolumeSpec(podVolume, pod.Namespace, types.NodeName(pod.Spec.NodeName), collector.volumePluginMgr, collector.pvcLister, collector.pvLister, collector.csiMigratedPluginManager, collector.intreeToCSITranslator)
+			volumeSpec, err := util.CreateVolumeSpec(podVolume, pod, types.NodeName(pod.Spec.NodeName), collector.volumePluginMgr, collector.pvcLister, collector.pvLister, collector.csiMigratedPluginManager, collector.intreeToCSITranslator)
 			if err != nil {
 				continue
 			}

pkg/controller/volume/attachdetach/util/util.go

@@ -39,57 +39,67 @@ import (
 // A volume.Spec that refers to an in-tree plugin spec is translated to refer
 // to a migrated CSI plugin spec if all conditions for CSI migration on a node
 // for the in-tree plugin is satisfied.
-func CreateVolumeSpec(podVolume v1.Volume, podNamespace string, nodeName types.NodeName, vpm *volume.VolumePluginMgr, pvcLister corelisters.PersistentVolumeClaimLister, pvLister corelisters.PersistentVolumeLister, csiMigratedPluginManager csimigration.PluginManager, csiTranslator csimigration.InTreeToCSITranslator) (*volume.Spec, error) {
+func CreateVolumeSpec(podVolume v1.Volume, pod *v1.Pod, nodeName types.NodeName, vpm *volume.VolumePluginMgr, pvcLister corelisters.PersistentVolumeClaimLister, pvLister corelisters.PersistentVolumeLister, csiMigratedPluginManager csimigration.PluginManager, csiTranslator csimigration.InTreeToCSITranslator) (*volume.Spec, error) {
+	claimName := ""
+	readOnly := false
 	if pvcSource := podVolume.VolumeSource.PersistentVolumeClaim; pvcSource != nil {
+		claimName = pvcSource.ClaimName
+		readOnly = pvcSource.ReadOnly
+	}
+	if ephemeralSource := podVolume.VolumeSource.Ephemeral; ephemeralSource != nil && utilfeature.DefaultFeatureGate.Enabled(features.GenericEphemeralVolume) {
+		claimName = pod.Name + "-" + podVolume.Name
+		readOnly = ephemeralSource.ReadOnly
+	}
+	if claimName != "" {
 		klog.V(10).Infof(
 			"Found PVC, ClaimName: %q/%q",
-			podNamespace,
-			pvcSource.ClaimName)
+			pod.Namespace,
+			claimName)
 
 		// If podVolume is a PVC, fetch the real PV behind the claim
 		pvName, pvcUID, err := getPVCFromCacheExtractPV(
-			podNamespace, pvcSource.ClaimName, pvcLister)
+			pod.Namespace, claimName, pvcLister)
 		if err != nil {
 			return nil, fmt.Errorf(
 				"error processing PVC %q/%q: %v",
-				podNamespace,
-				pvcSource.ClaimName,
+				pod.Namespace,
+				claimName,
 				err)
 		}
 
 		klog.V(10).Infof(
 			"Found bound PV for PVC (ClaimName %q/%q pvcUID %v): pvName=%q",
-			podNamespace,
-			pvcSource.ClaimName,
+			pod.Namespace,
+			claimName,
 			pvcUID,
 			pvName)
 
 		// Fetch actual PV object
 		volumeSpec, err := getPVSpecFromCache(
-			pvName, pvcSource.ReadOnly, pvcUID, pvLister)
+			pvName, readOnly, pvcUID, pvLister)
 		if err != nil {
 			return nil, fmt.Errorf(
 				"error processing PVC %q/%q: %v",
-				podNamespace,
-				pvcSource.ClaimName,
+				pod.Namespace,
+				claimName,
 				err)
 		}
 
 		volumeSpec, err = translateInTreeSpecToCSIIfNeeded(volumeSpec, nodeName, vpm, csiMigratedPluginManager, csiTranslator)
 		if err != nil {
 			return nil, fmt.Errorf(
 				"error performing CSI migration checks and translation for PVC %q/%q: %v",
-				podNamespace,
-				pvcSource.ClaimName,
+				pod.Namespace,
+				claimName,
 				err)
 		}
 
 		klog.V(10).Infof(
 			"Extracted volumeSpec (%v) from bound PV (pvName %q) and PVC (ClaimName %q/%q pvcUID %v)",
 			volumeSpec.Name(),
 			pvName,
-			podNamespace,
-			pvcSource.ClaimName,
+			pod.Namespace,
+			claimName,
 			pvcUID)
 
 		return volumeSpec, nil
@@ -219,7 +229,7 @@ func ProcessPodVolumes(pod *v1.Pod, addVolumes bool, desiredStateOfWorld cache.D
 
 	// Process volume spec for each volume defined in pod
 	for _, podVolume := range pod.Spec.Volumes {
-		volumeSpec, err := CreateVolumeSpec(podVolume, pod.Namespace, nodeName, volumePluginMgr, pvcLister, pvLister, csiMigratedPluginManager, csiTranslator)
+		volumeSpec, err := CreateVolumeSpec(podVolume, pod, nodeName, volumePluginMgr, pvcLister, pvLister, csiMigratedPluginManager, csiTranslator)
 		if err != nil {
 			klog.V(10).Infof(
 				"Error processing volume %q for pod %q/%q: %v",

pkg/controller/volume/common/BUILD

@@ -6,7 +6,9 @@ go_library(
     importpath = "k8s.io/kubernetes/pkg/controller/volume/common",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/features:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
         "//staging/src/k8s.io/client-go/tools/cache:go_default_library",
     ],
 )

pkg/controller/volume/common/common.go

@@ -20,27 +20,46 @@ import (
 	"fmt"
 
 	v1 "k8s.io/api/core/v1"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/client-go/tools/cache"
+	"k8s.io/kubernetes/pkg/features"
 )
 
 const (
 	// PodPVCIndex is the lookup name for the index function, which is to index by pod pvcs.
 	PodPVCIndex = "pod-pvc-index"
 )
 
-// PodPVCIndexFunc returns PVC keys for given pod
-func PodPVCIndexFunc(obj interface{}) ([]string, error) {
-	pod, ok := obj.(*v1.Pod)
-	if !ok {
-		return []string{}, nil
-	}
-	keys := []string{}
-	for _, podVolume := range pod.Spec.Volumes {
-		if pvcSource := podVolume.VolumeSource.PersistentVolumeClaim; pvcSource != nil {
-			keys = append(keys, fmt.Sprintf("%s/%s", pod.Namespace, pvcSource.ClaimName))
+// PodPVCIndexFunc creates an index function that returns PVC keys (=
+// namespace/name) for given pod.  If enabled, this includes the PVCs
+// that might be created for generic ephemeral volumes.
+func PodPVCIndexFunc(genericEphemeralVolumeFeatureEnabled bool) func(obj interface{}) ([]string, error) {
+	return func(obj interface{}) ([]string, error) {
+		pod, ok := obj.(*v1.Pod)
+		if !ok {
+			return []string{}, nil
+		}
+		keys := []string{}
+		for _, podVolume := range pod.Spec.Volumes {
+			claimName := ""
+			if pvcSource := podVolume.VolumeSource.PersistentVolumeClaim; pvcSource != nil {
+				claimName = pvcSource.ClaimName
+			}
+			if ephemeralSource := podVolume.VolumeSource.Ephemeral; genericEphemeralVolumeFeatureEnabled && ephemeralSource != nil {
+				claimName = pod.Name + "-" + podVolume.Name
+			}
+			if claimName != "" {
+				keys = append(keys, fmt.Sprintf("%s/%s", pod.Namespace, claimName))
+			}
 		}
+		return keys, nil
 	}
-	return keys, nil
+}
+
+// AddPodPVCIndexerIfNotPresent adds the PodPVCIndexFunc with the current global setting for GenericEphemeralVolume.
+func AddPodPVCIndexerIfNotPresent(indexer cache.Indexer) error {
+	return AddIndexerIfNotPresent(indexer, PodPVCIndex,
+		PodPVCIndexFunc(utilfeature.DefaultFeatureGate.Enabled(features.GenericEphemeralVolume)))
 }
 
 // AddIndexerIfNotPresent adds the index function with the name into the cache indexer if not present

pkg/controller/volume/ephemeral/BUILD

@@ -0,0 +1,62 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = [
+        "controller.go",
+        "doc.go",
+    ],
+    importpath = "k8s.io/kubernetes/pkg/controller/volume/ephemeral",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/controller/volume/common:go_default_library",
+        "//pkg/controller/volume/events:go_default_library",
+        "//pkg/volume/util:go_default_library",
+        "//staging/src/k8s.io/api/core/v1:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/api/errors:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/util/runtime:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
+        "//staging/src/k8s.io/client-go/informers/core/v1:go_default_library",
+        "//staging/src/k8s.io/client-go/kubernetes:go_default_library",
+        "//staging/src/k8s.io/client-go/kubernetes/scheme:go_default_library",
+        "//staging/src/k8s.io/client-go/kubernetes/typed/core/v1:go_default_library",
+        "//staging/src/k8s.io/client-go/listers/core/v1:go_default_library",
+        "//staging/src/k8s.io/client-go/tools/cache:go_default_library",
+        "//staging/src/k8s.io/client-go/tools/record:go_default_library",
+        "//staging/src/k8s.io/client-go/util/workqueue:go_default_library",
+        "//vendor/k8s.io/klog/v2:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["controller_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//pkg/controller:go_default_library",
+        "//staging/src/k8s.io/api/core/v1:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
+        "//staging/src/k8s.io/client-go/informers:go_default_library",
+        "//staging/src/k8s.io/client-go/kubernetes/fake:go_default_library",
+        "//staging/src/k8s.io/client-go/tools/cache:go_default_library",
+        "//vendor/github.com/stretchr/testify/assert:go_default_library",
+        "//vendor/k8s.io/klog/v2:go_default_library",
+    ],
+)
+
+filegroup(
+    name = "package-srcs",
+    srcs = glob(["**"]),
+    tags = ["automanaged"],
+    visibility = ["//visibility:private"],
+)
+
+filegroup(
+    name = "all-srcs",
+    srcs = [":package-srcs"],
+    tags = ["automanaged"],
+    visibility = ["//visibility:public"],
+)