e2e storage: support creating ReplicaSet

This is useful for running a driver on a subset of all ready nodes:

- use e2enode.GetBoundedReadySchedulableNodes with a suitable
  maximum number of nodes to determine how much nodes are available
  for a test
- define pod anti-affinity in the PodTemplate:
      affinity:
        podAntiAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
          - labelSelector:
              matchLabels:
                app.kubernetes.io/instance: xxxxxxx
            topologyKey: kubernetes.io/hostname
- set the ReplicaSetSpec.Replicas value to the number of nodes

test/e2e/storage/utils/create.go

@@ -275,6 +275,7 @@ var factories = map[What]ItemFactory{
 	{"ClusterRoleBinding"}:       &clusterRoleBindingFactory{},
 	{"CSIDriver"}:                &csiDriverFactory{},
 	{"DaemonSet"}:                &daemonSetFactory{},
+	{"ReplicaSet"}:               &replicaSetFactory{},
 	{"Role"}:                     &roleFactory{},
 	{"RoleBinding"}:              &roleBindingFactory{},
 	{"Secret"}:                   &secretFactory{},
@@ -382,6 +383,14 @@ func patchItemRecursively(f *framework.Framework, driverNamespace *v1.Namespace,
 		if err := patchContainerImages(item.Spec.Template.Spec.InitContainers); err != nil {
 			return err
 		}
+	case *appsv1.ReplicaSet:
+		PatchNamespace(f, driverNamespace, &item.ObjectMeta.Namespace)
+		if err := patchContainerImages(item.Spec.Template.Spec.Containers); err != nil {
+			return err
+		}
+		if err := patchContainerImages(item.Spec.Template.Spec.InitContainers); err != nil {
+			return err
+		}
 	case *apiextensionsv1.CustomResourceDefinition:
 		// Do nothing. Patching name to all CRDs won't always be the expected behavior.
 	default:
@@ -584,6 +593,27 @@ func (*daemonSetFactory) Create(f *framework.Framework, ns *v1.Namespace, i inte
 	}, nil
 }
 
+type replicaSetFactory struct{}
+
+func (f *replicaSetFactory) New() runtime.Object {
+	return &appsv1.ReplicaSet{}
+}
+
+func (*replicaSetFactory) Create(f *framework.Framework, ns *v1.Namespace, i interface{}) (func() error, error) {
+	item, ok := i.(*appsv1.ReplicaSet)
+	if !ok {
+		return nil, errorItemNotSupported
+	}
+
+	client := f.ClientSet.AppsV1().ReplicaSets(ns.Name)
+	if _, err := client.Create(context.TODO(), item, metav1.CreateOptions{}); err != nil {
+		return nil, fmt.Errorf("create ReplicaSet: %w", err)
+	}
+	return func() error {
+		return client.Delete(context.TODO(), item.GetName(), metav1.DeleteOptions{})
+	}, nil
+}
+
 type storageClassFactory struct{}
 
 func (f *storageClassFactory) New() runtime.Object {