add alpha asynchronous binding operation support (#1512)

* add types changes

* add binding polling queue

* add async bind conditions

* add polling handling to current reconciliation functions

* add generated files

* fix existing tests

* add polling functions

* add tests for async binding operations

* add validation

* add integration test support

* add feature gate

* rebase

* fixup int test

* address comments

* rebase

* address comments

charts/catalog/README.md

@@ -55,6 +55,7 @@ chart and their default values.
 | `controllerManager.brokerRelistInterval` | How often the controller should relist the catalogs of ready brokers; duration format (`20m`, `1h`, etc) | `24h` |
 | `useAggregator` | whether or not to set up the controller-manager to go through the main Kubernetes API server's API aggregator | `true` |
 | `rbacEnable` | If true, create & use RBAC resources | `true` |
+| `asyncBindingOperationsEnabled` | Whether or not alpha support for async binding operations is enabled | `false` |
 
 Specify each parameter using the `--set key=value[,key=value]` argument to
 `helm install`.

charts/catalog/templates/controller-manager-deployment.yaml

@@ -71,6 +71,10 @@ spec:
         - --feature-gates
         - OriginatingIdentity=true
         {{- end }}
+        {{- if .Values.asyncBindingOperationsEnabled }}
+        - --feature-gates
+        - AsyncBindingOperations=true
+        {{- end }}
         ports:
         - containerPort: 8080
         volumeMounts:

charts/catalog/values.yaml

@@ -91,3 +91,5 @@ controllerManager:
   apiserverSkipVerify: true
 # Whether the OriginatingIdentity alpha feature should be enabled
 originatingIdentityEnabled: false
+# Whether the AsyncBindingOperations alpha feature should be enabled
+asyncBindingOperationsEnabled: false

pkg/apis/servicecatalog/types.go

@@ -257,6 +257,13 @@ type ClusterServiceClassSpec struct {
 	// Bindable which overrides the value of this field.
 	Bindable bool
 
+	// Currently, this field is ALPHA: it may change or disappear at any time
+	// and its data will not be migrated.
+	//
+	// BindingRetrievable indicates whether fetching a binding via a GET on
+	// its endpoint is supported for all plans.
+	BindingRetrievable bool
+
 	// PlanUpdatable indicates whether instances provisioned from this
 	// ClusterServiceClass may change ClusterServicePlans after being provisioned.
 	PlanUpdatable bool
@@ -724,6 +731,21 @@ type ServiceBindingSpec struct {
 type ServiceBindingStatus struct {
 	Conditions []ServiceBindingCondition
 
+	// Currently, this field is ALPHA: it may change or disappear at any time
+	// and its data will not be migrated.
+	//
+	// AsyncOpInProgress is set to true if there is an ongoing async operation
+	// against this ServiceBinding in progress.
+	AsyncOpInProgress bool
+
+	// Currently, this field is ALPHA: it may change or disappear at any time
+	// and its data will not be migrated.
+	//
+	// LastOperation is the string that the broker may have returned when
+	// an async operation started, it should be sent back to the broker
+	// on poll requests as a query param.
+	LastOperation *string
+
 	// CurrentOperation is the operation the Controller is currently performing
 	// on the ServiceBinding.
 	CurrentOperation ServiceBindingOperation

pkg/apis/servicecatalog/v1beta1/types.go

@@ -259,6 +259,13 @@ type ClusterServiceClassSpec struct {
 	// this field.
 	Bindable bool `json:"bindable"`
 
+	// Currently, this field is ALPHA: it may change or disappear at any time
+	// and its data will not be migrated.
+	//
+	// BindingRetrievable indicates whether fetching a binding via a GET on
+	// its endpoint is supported for all plans.
+	BindingRetrievable bool `json:"binding_retrievable"`
+
 	// PlanUpdatable indicates whether instances provisioned from this
 	// ClusterServiceClass may change ClusterServicePlans after being
 	// provisioned.
@@ -747,6 +754,21 @@ type ServiceBindingSpec struct {
 type ServiceBindingStatus struct {
 	Conditions []ServiceBindingCondition `json:"conditions"`
 
+	// Currently, this field is ALPHA: it may change or disappear at any time
+	// and its data will not be migrated.
+	//
+	// AsyncOpInProgress is set to true if there is an ongoing async operation
+	// against this ServiceBinding in progress.
+	AsyncOpInProgress bool `json:"asyncOpInProgress"`
+
+	// Currently, this field is ALPHA: it may change or disappear at any time
+	// and its data will not be migrated.
+	//
+	// LastOperation is the string that the broker may have returned when
+	// an async operation started, it should be sent back to the broker
+	// on poll requests as a query param.
+	LastOperation *string `json:"lastOperation,omitempty"`
+
 	// CurrentOperation is the operation the Controller is currently performing
 	// on the ServiceBinding.
 	CurrentOperation ServiceBindingOperation `json:"currentOperation,omitempty"`

pkg/apis/servicecatalog/v1beta1/zz_generated.conversion.go

@@ -341,6 +341,7 @@ func autoConvert_v1beta1_ClusterServiceClassSpec_To_servicecatalog_ClusterServic
 	out.ExternalID = in.ExternalID
 	out.Description = in.Description
 	out.Bindable = in.Bindable
+	out.BindingRetrievable = in.BindingRetrievable
 	out.PlanUpdatable = in.PlanUpdatable
 	out.ExternalMetadata = (*runtime.RawExtension)(unsafe.Pointer(in.ExternalMetadata))
 	out.Tags = *(*[]string)(unsafe.Pointer(&in.Tags))
@@ -359,6 +360,7 @@ func autoConvert_servicecatalog_ClusterServiceClassSpec_To_v1beta1_ClusterServic
 	out.ExternalID = in.ExternalID
 	out.Description = in.Description
 	out.Bindable = in.Bindable
+	out.BindingRetrievable = in.BindingRetrievable
 	out.PlanUpdatable = in.PlanUpdatable
 	out.ExternalMetadata = (*runtime.RawExtension)(unsafe.Pointer(in.ExternalMetadata))
 	out.Tags = *(*[]string)(unsafe.Pointer(&in.Tags))
@@ -761,6 +763,8 @@ func Convert_servicecatalog_ServiceBindingSpec_To_v1beta1_ServiceBindingSpec(in
 
 func autoConvert_v1beta1_ServiceBindingStatus_To_servicecatalog_ServiceBindingStatus(in *ServiceBindingStatus, out *servicecatalog.ServiceBindingStatus, s conversion.Scope) error {
 	out.Conditions = *(*[]servicecatalog.ServiceBindingCondition)(unsafe.Pointer(&in.Conditions))
+	out.AsyncOpInProgress = in.AsyncOpInProgress
+	out.LastOperation = (*string)(unsafe.Pointer(in.LastOperation))
 	out.CurrentOperation = servicecatalog.ServiceBindingOperation(in.CurrentOperation)
 	out.ReconciledGeneration = in.ReconciledGeneration
 	out.OperationStartTime = (*v1.Time)(unsafe.Pointer(in.OperationStartTime))
@@ -777,6 +781,8 @@ func Convert_v1beta1_ServiceBindingStatus_To_servicecatalog_ServiceBindingStatus
 
 func autoConvert_servicecatalog_ServiceBindingStatus_To_v1beta1_ServiceBindingStatus(in *servicecatalog.ServiceBindingStatus, out *ServiceBindingStatus, s conversion.Scope) error {
 	out.Conditions = *(*[]ServiceBindingCondition)(unsafe.Pointer(&in.Conditions))
+	out.AsyncOpInProgress = in.AsyncOpInProgress
+	out.LastOperation = (*string)(unsafe.Pointer(in.LastOperation))
 	out.CurrentOperation = ServiceBindingOperation(in.CurrentOperation)
 	out.ReconciledGeneration = in.ReconciledGeneration
 	out.OperationStartTime = (*v1.Time)(unsafe.Pointer(in.OperationStartTime))

pkg/apis/servicecatalog/v1beta1/zz_generated.deepcopy.go

@@ -899,6 +899,15 @@ func (in *ServiceBindingStatus) DeepCopyInto(out *ServiceBindingStatus) {
 			(*in)[i].DeepCopyInto(&(*out)[i])
 		}
 	}
+	if in.LastOperation != nil {
+		in, out := &in.LastOperation, &out.LastOperation
+		if *in == nil {
+			*out = nil
+		} else {
+			*out = new(string)
+			**out = **in
+		}
+	}
 	if in.OperationStartTime != nil {
 		in, out := &in.OperationStartTime, &out.OperationStartTime
 		if *in == nil {

pkg/apis/servicecatalog/validation/binding.go

@@ -94,6 +94,9 @@ func validateServiceBindingStatus(status *sc.ServiceBindingStatus, fldPath *fiel
 		if status.OperationStartTime != nil {
 			allErrs = append(allErrs, field.Forbidden(fldPath.Child("operationStartTime"), "operationStartTime must not be present when currentOperation is not present"))
 		}
+		if status.AsyncOpInProgress {
+			allErrs = append(allErrs, field.Forbidden(fldPath.Child("asyncOpInProgress"), "asyncOpInProgress cannot be true when there is no currentOperation"))
+		}
 	} else {
 		if status.OperationStartTime == nil && !status.OrphanMitigationInProgress {
 			allErrs = append(allErrs, field.Required(fldPath.Child("operationStartTime"), "operationStartTime is required when currentOperation is present and no orphan mitigation in progress"))

pkg/apis/servicecatalog/zz_generated.deepcopy.go

@@ -899,6 +899,15 @@ func (in *ServiceBindingStatus) DeepCopyInto(out *ServiceBindingStatus) {
 			(*in)[i].DeepCopyInto(&(*out)[i])
 		}
 	}
+	if in.LastOperation != nil {
+		in, out := &in.LastOperation, &out.LastOperation
+		if *in == nil {
+			*out = nil
+		} else {
+			*out = new(string)
+			**out = **in
+		}
+	}
 	if in.OperationStartTime != nil {
 		in, out := &in.OperationStartTime, &out.OperationStartTime
 		if *in == nil {

pkg/controller/controller.go

@@ -31,6 +31,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 
 	corev1 "k8s.io/api/core/v1"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/record"
@@ -40,6 +41,7 @@ import (
 	servicecatalogclientset "github.com/kubernetes-incubator/service-catalog/pkg/client/clientset_generated/clientset/typed/servicecatalog/v1beta1"
 	informers "github.com/kubernetes-incubator/service-catalog/pkg/client/informers_generated/externalversions/servicecatalog/v1beta1"
 	listers "github.com/kubernetes-incubator/service-catalog/pkg/client/listers_generated/servicecatalog/v1beta1"
+	scfeatures "github.com/kubernetes-incubator/service-catalog/pkg/features"
 	pretty "github.com/kubernetes-incubator/service-catalog/pkg/pretty"
 )
 
@@ -87,7 +89,8 @@ func NewController(
 		servicePlanQueue:            workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "service-plan"),
 		instanceQueue:               workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "service-instance"),
 		bindingQueue:                workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "service-binding"),
-		pollingQueue:                workqueue.NewNamedRateLimitingQueue(workqueue.NewItemExponentialFailureRateLimiter(pollingStartInterval, pollingMaxBackoffDuration), "poller"),
+		instancePollingQueue:        workqueue.NewNamedRateLimitingQueue(workqueue.NewItemExponentialFailureRateLimiter(pollingStartInterval, pollingMaxBackoffDuration), "instance-poller"),
+		bindingPollingQueue:         workqueue.NewNamedRateLimitingQueue(workqueue.NewItemExponentialFailureRateLimiter(pollingStartInterval, pollingMaxBackoffDuration), "binding-poller"),
 	}
 
 	controller.brokerLister = brokerInformer.Lister()
@@ -156,11 +159,8 @@ type controller struct {
 	servicePlanQueue            workqueue.RateLimitingInterface
 	instanceQueue               workqueue.RateLimitingInterface
 	bindingQueue                workqueue.RateLimitingInterface
-	// pollingQueue is separate from instanceQueue because we want
-	// it to have different backoff / timeout characteristics from
-	//  a reconciling of an instance.
-	// TODO(vaikas): get rid of two queues per instance.
-	pollingQueue workqueue.RateLimitingInterface
+	instancePollingQueue        workqueue.RateLimitingInterface
+	bindingPollingQueue         workqueue.RateLimitingInterface
 }
 
 // Run runs the controller until the given stop channel can be read from.
@@ -177,7 +177,11 @@ func (c *controller) Run(workers int, stopCh <-chan struct{}) {
 		createWorker(c.servicePlanQueue, "ClusterServicePlan", maxRetries, true, c.reconcileClusterServicePlanKey, stopCh, &waitGroup)
 		createWorker(c.instanceQueue, "ServiceInstance", maxRetries, true, c.reconcileServiceInstanceKey, stopCh, &waitGroup)
 		createWorker(c.bindingQueue, "ServiceBinding", maxRetries, true, c.reconcileServiceBindingKey, stopCh, &waitGroup)
-		createWorker(c.pollingQueue, "Poller", maxRetries, false, c.requeueServiceInstanceForPoll, stopCh, &waitGroup)
+		createWorker(c.instancePollingQueue, "InstancePoller", maxRetries, false, c.requeueServiceInstanceForPoll, stopCh, &waitGroup)
+
+		if utilfeature.DefaultFeatureGate.Enabled(scfeatures.AsyncBindingOperations) {
+			createWorker(c.bindingPollingQueue, "BindingPoller", maxRetries, false, c.requeueServiceBindingForPoll, stopCh, &waitGroup)
+		}
 	}
 
 	<-stopCh
@@ -188,7 +192,8 @@ func (c *controller) Run(workers int, stopCh <-chan struct{}) {
 	c.servicePlanQueue.ShutDown()
 	c.instanceQueue.ShutDown()
 	c.bindingQueue.ShutDown()
-	c.pollingQueue.ShutDown()
+	c.instancePollingQueue.ShutDown()
+	c.bindingPollingQueue.ShutDown()
 
 	waitGroup.Wait()
 }
@@ -502,6 +507,10 @@ func convertCatalog(in *osb.CatalogResponse) ([]*v1beta1.ClusterServiceClass, []
 			},
 		}
 
+		if utilfeature.DefaultFeatureGate.Enabled(scfeatures.AsyncBindingOperations) {
+			serviceClasses[i].Spec.BindingRetrievable = svc.BindingRetrievable
+		}
+
 		if svc.Metadata != nil {
 			metadata, err := json.Marshal(svc.Metadata)
 			if err != nil {