Broker is not ready until its constituent pieces are ready.

pkg/apis/eventing/v1alpha1/broker_lifecycle.go

@@ -16,7 +16,11 @@ limitations under the License.
 
 package v1alpha1
 
-import duckv1alpha1 "github.com/knative/pkg/apis/duck/v1alpha1"
+import (
+	v1 "k8s.io/api/apps/v1"
+
+	duckv1alpha1 "github.com/knative/pkg/apis/duck/v1alpha1"
+)
 
 var brokerCondSet = duckv1alpha1.NewLivingConditionSet(
 	BrokerConditionIngress,
@@ -57,44 +61,91 @@ func (bs *BrokerStatus) InitializeConditions() {
 	brokerCondSet.Manage(bs).InitializeConditions()
 }
 
-func (bs *BrokerStatus) MarkIngressReady() {
-	brokerCondSet.Manage(bs).MarkTrue(BrokerConditionIngress)
+func (bs *BrokerStatus) MarkIngressFailed(reason, format string, args ...interface{}) {
+	brokerCondSet.Manage(bs).MarkFalse(BrokerConditionIngress, reason, format, args...)
 }
 
-func (bs *BrokerStatus) MarkIngressFailed(err error) {
-	brokerCondSet.Manage(bs).MarkFalse(BrokerConditionIngress, "failed", "%v", err)
+func (bs *BrokerStatus) PropagateIngressDeploymentAvailability(d *v1.Deployment) {
+	if deploymentIsAvailable(&d.Status) {
+		brokerCondSet.Manage(bs).MarkTrue(BrokerConditionIngress)
+	} else {
+		// I don't know how to propagate the status well, so just give the name of the Deployment
+		// for now.
+		bs.MarkIngressFailed("DeploymentUnavailable", "The Deployment '%s' is unavailable.", d.Name)
+	}
 }
 
-func (bs *BrokerStatus) MarkTriggerChannelReady() {
-	brokerCondSet.Manage(bs).MarkTrue(BrokerConditionTriggerChannel)
+func (bs *BrokerStatus) MarkTriggerChannelFailed(reason, format string, args ...interface{}) {
+	brokerCondSet.Manage(bs).MarkFalse(BrokerConditionTriggerChannel, reason, format, args...)
 }
 
-func (bs *BrokerStatus) MarkTriggerChannelFailed(err error) {
-	brokerCondSet.Manage(bs).MarkFalse(BrokerConditionTriggerChannel, "failed", "%v", err)
+func (bs *BrokerStatus) PropagateTriggerChannelReadiness(cs *ChannelStatus) {
+	if cs.IsReady() {
+		brokerCondSet.Manage(bs).MarkTrue(BrokerConditionTriggerChannel)
+	} else {
+		msg := "nil"
+		if cc := chanCondSet.Manage(cs).GetCondition(ChannelConditionReady); cc != nil {
+			msg = cc.Message
+		}
+		bs.MarkTriggerChannelFailed("ChannelNotReady", "trigger Channel is not ready: %s", msg)
+	}
 }
 
-func (bs *BrokerStatus) MarkIngressChannelReady() {
-	brokerCondSet.Manage(bs).MarkTrue(BrokerConditionIngressChannel)
+func (bs *BrokerStatus) MarkIngressChannelFailed(reason, format string, args ...interface{}) {
+	brokerCondSet.Manage(bs).MarkFalse(BrokerConditionIngressChannel, reason, format, args...)
 }
 
-func (bs *BrokerStatus) MarkIngressChannelFailed(err error) {
-	brokerCondSet.Manage(bs).MarkFalse(BrokerConditionIngressChannel, "failed", "%v", err)
+func (bs *BrokerStatus) PropagateIngressChannelReadiness(cs *ChannelStatus) {
+	if cs.IsReady() {
+		brokerCondSet.Manage(bs).MarkTrue(BrokerConditionIngressChannel)
+	} else {
+		msg := "nil"
+		if cc := chanCondSet.Manage(cs).GetCondition(ChannelConditionReady); cc != nil {
+			msg = cc.Message
+		}
+		bs.MarkIngressChannelFailed("ChannelNotReady", "ingress Channel is not ready: %s", msg)
+	}
 }
 
-func (bs *BrokerStatus) MarkIngressSubscriptionReady() {
-	brokerCondSet.Manage(bs).MarkTrue(BrokerConditionIngressSubscription)
+func (bs *BrokerStatus) MarkIngressSubscriptionFailed(reason, format string, args ...interface{}) {
+	brokerCondSet.Manage(bs).MarkFalse(BrokerConditionIngressSubscription, reason, format, args...)
 }
 
-func (bs *BrokerStatus) MarkIngressSubscriptionFailed(err error) {
-	brokerCondSet.Manage(bs).MarkFalse(BrokerConditionIngressSubscription, "failed", "%v", err)
+func (bs *BrokerStatus) PropagateIngressSubscriptionReadiness(ss *SubscriptionStatus) {
+	if ss.IsReady() {
+		brokerCondSet.Manage(bs).MarkTrue(BrokerConditionIngressSubscription)
+	} else {
+		msg := "nil"
+		if sc := subCondSet.Manage(ss).GetCondition(SubscriptionConditionReady); sc != nil {
+			msg = sc.Message
+		}
+		bs.MarkIngressSubscriptionFailed("SubscriptionNotReady", "ingress Subscription is not ready: %s", msg)
+	}
 }
 
-func (bs *BrokerStatus) MarkFilterReady() {
-	brokerCondSet.Manage(bs).MarkTrue(BrokerConditionFilter)
+func (bs *BrokerStatus) MarkFilterFailed(reason, format string, args ...interface{}) {
+	brokerCondSet.Manage(bs).MarkFalse(BrokerConditionFilter, reason, format, args...)
 }
 
-func (bs *BrokerStatus) MarkFilterFailed(err error) {
-	brokerCondSet.Manage(bs).MarkFalse(BrokerConditionFilter, "failed", "%v", err)
+func (bs *BrokerStatus) PropagateFilterDeploymentAvailability(d *v1.Deployment) {
+	if deploymentIsAvailable(&d.Status) {
+		brokerCondSet.Manage(bs).MarkTrue(BrokerConditionFilter)
+	} else {
+		// I don't know how to propagate the status well, so just give the name of the Deployment
+		// for now.
+		bs.MarkFilterFailed("DeploymentUnavailable", "The Deployment '%s' is unavailable.", d.Name)
+	}
+}
+
+func deploymentIsAvailable(d *v1.DeploymentStatus) bool {
+	// Check if the Deployment is available.
+	for _, cond := range d.Conditions {
+		if cond.Type == v1.DeploymentAvailable {
+			return cond.Status == "True"
+		}
+	}
+	// Unable to find the Available condition, fail open.
+	return true
 }
 
 // SetAddress makes this Broker addressable by setting the hostname. It also

pkg/apis/eventing/v1alpha1/broker_lifecycle_test.go

@@ -17,19 +17,17 @@ limitations under the License.
 package v1alpha1
 
 import (
-	"errors"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
 	duckv1alpha1 "github.com/knative/pkg/apis/duck/v1alpha1"
+	v1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 )
 
 var (
 	trueVal  = true
 	falseVal = false
-
-	err = errors.New("foobar")
 )
 
 var (
@@ -331,39 +329,49 @@ func TestBrokerIsReady(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			bs := &BrokerStatus{}
 			if test.markIngressReady != nil {
+				var d *v1.Deployment
 				if *test.markIngressReady {
-					bs.MarkIngressReady()
+					d = availableDeployment()
 				} else {
-					bs.MarkIngressFailed(err)
+					d = unavailableDeployment()
 				}
+				bs.PropagateIngressDeploymentAvailability(d)
 			}
 			if test.markTriggerChannelReady != nil {
+				var c *ChannelStatus
 				if *test.markTriggerChannelReady {
-					bs.MarkTriggerChannelReady()
+					c = readyChannelStatus()
 				} else {
-					bs.MarkTriggerChannelFailed(err)
+					c = notReadyChannelStatus()
 				}
+				bs.PropagateTriggerChannelReadiness(c)
 			}
 			if test.markIngressChannelReady != nil {
+				var c *ChannelStatus
 				if *test.markIngressChannelReady {
-					bs.MarkIngressChannelReady()
+					c = readyChannelStatus()
 				} else {
-					bs.MarkIngressChannelFailed(err)
+					c = notReadyChannelStatus()
 				}
+				bs.PropagateIngressChannelReadiness(c)
 			}
 			if test.markIngressSubscriptionReady != nil {
+				var sub *SubscriptionStatus
 				if *test.markIngressSubscriptionReady {
-					bs.MarkIngressSubscriptionReady()
+					sub = readySubscriptionStatus()
 				} else {
-					bs.MarkIngressSubscriptionFailed(err)
+					sub = notReadySubscriptionStatus()
 				}
+				bs.PropagateIngressSubscriptionReadiness(sub)
 			}
 			if test.markFilterReady != nil {
+				var d *v1.Deployment
 				if *test.markFilterReady {
-					bs.MarkFilterReady()
+					d = availableDeployment()
 				} else {
-					bs.MarkFilterFailed(err)
+					d = unavailableDeployment()
 				}
+				bs.PropagateFilterDeploymentAvailability(d)
 			}
 			bs.SetAddress(test.address)
 
@@ -374,3 +382,54 @@ func TestBrokerIsReady(t *testing.T) {
 		})
 	}
 }
+
+func unavailableDeployment() *v1.Deployment {
+	d := &v1.Deployment{}
+	d.Name = "unavailable"
+	d.Status.Conditions = []v1.DeploymentCondition{
+		{
+			Type:   v1.DeploymentAvailable,
+			Status: "False",
+		},
+	}
+	return d
+}
+
+func availableDeployment() *v1.Deployment {
+	d := unavailableDeployment()
+	d.Name = "available"
+	d.Status.Conditions = []v1.DeploymentCondition{
+		{
+			Type:   v1.DeploymentAvailable,
+			Status: "True",
+		},
+	}
+	return d
+}
+
+func readyChannelStatus() *ChannelStatus {
+	cs := &ChannelStatus{}
+	cs.MarkProvisionerInstalled()
+	cs.MarkProvisioned()
+	cs.SetAddress("foo")
+	return cs
+}
+
+func notReadyChannelStatus() *ChannelStatus {
+	cs := readyChannelStatus()
+	cs.MarkNotProvisioned("foo", "bar")
+	return cs
+}
+
+func readySubscriptionStatus() *SubscriptionStatus {
+	ss := &SubscriptionStatus{}
+	ss.MarkChannelReady()
+	ss.MarkReferencesResolved()
+	return ss
+}
+
+func notReadySubscriptionStatus() *SubscriptionStatus {
+	ss := &SubscriptionStatus{}
+	ss.MarkReferencesResolved()
+	return ss
+}