Skip to content

Commit

Permalink
List applying EventPolicies in Brokers status (#8060)
Browse files Browse the repository at this point in the history 
List applying event policies in Brokers status
  • Loading branch information
@creydr
creydr committed Jul 4, 2024
1 parent 657c3cd commit 3264b21
Show file tree
Hide file tree
Showing 8 changed files with 283 additions and 8 deletions.
18 changes: 18 additions & 0 deletions pkg/apis/eventing/v1/broker_lifecycle.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ const (
BrokerConditionFilter apis.ConditionType = "FilterReady"
BrokerConditionAddressable apis.ConditionType = "Addressable"
BrokerConditionDeadLetterSinkResolved apis.ConditionType = "DeadLetterSinkResolved"
BrokerConditionEventPoliciesReady apis.ConditionType = "EventPoliciesReady"
)

var brokerCondSet = apis.NewLivingConditionSet(
Expand All @@ -40,6 +41,7 @@ var brokerCondSet = apis.NewLivingConditionSet(
BrokerConditionFilter,
BrokerConditionAddressable,
BrokerConditionDeadLetterSinkResolved,
BrokerConditionEventPoliciesReady,
)
var brokerCondSetLock = sync.RWMutex{}

Expand Down Expand Up @@ -118,3 +120,19 @@ func (bs *BrokerStatus) MarkDeadLetterSinkResolvedFailed(reason, messageFormat s
bs.DeliveryStatus = eventingduck.DeliveryStatus{}
bs.GetConditionSet().Manage(bs).MarkFalse(BrokerConditionDeadLetterSinkResolved, reason, messageFormat, messageA...)
}

func (bs *BrokerStatus) MarkEventPoliciesTrue() {
bs.GetConditionSet().Manage(bs).MarkTrue(BrokerConditionEventPoliciesReady)
}

func (bs *BrokerStatus) MarkEventPoliciesTrueWithReason(reason, messageFormat string, messageA ...interface{}) {
bs.GetConditionSet().Manage(bs).MarkTrueWithReason(BrokerConditionEventPoliciesReady, reason, messageFormat, messageA...)
}

func (bs *BrokerStatus) MarkEventPoliciesFailed(reason, messageFormat string, messageA ...interface{}) {
bs.GetConditionSet().Manage(bs).MarkFalse(BrokerConditionEventPoliciesReady, reason, messageFormat, messageA...)
}

func (bs *BrokerStatus) MarkEventPoliciesUnknown(reason, messageFormat string, messageA ...interface{}) {
bs.GetConditionSet().Manage(bs).MarkUnknown(BrokerConditionEventPoliciesReady, reason, messageFormat, messageA...)
}
45 changes: 45 additions & 0 deletions pkg/apis/eventing/v1/broker_lifecycle_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -193,6 +193,9 @@ func TestBrokerInitializeConditions(t *testing.T) {
}, {
Type: BrokerConditionDeadLetterSinkResolved,
Status: corev1.ConditionUnknown,
}, {
Type: BrokerConditionEventPoliciesReady,
Status: corev1.ConditionUnknown,
}, {
Type: BrokerConditionFilter,
Status: corev1.ConditionUnknown,
Expand Down Expand Up @@ -226,6 +229,9 @@ func TestBrokerInitializeConditions(t *testing.T) {
}, {
Type: BrokerConditionDeadLetterSinkResolved,
Status: corev1.ConditionUnknown,
}, {
Type: BrokerConditionEventPoliciesReady,
Status: corev1.ConditionUnknown,
}, {
Type: BrokerConditionFilter,
Status: corev1.ConditionUnknown,
Expand Down Expand Up @@ -259,6 +265,9 @@ func TestBrokerInitializeConditions(t *testing.T) {
}, {
Type: BrokerConditionDeadLetterSinkResolved,
Status: corev1.ConditionUnknown,
}, {
Type: BrokerConditionEventPoliciesReady,
Status: corev1.ConditionUnknown,
}, {
Type: BrokerConditionFilter,
Status: corev1.ConditionTrue,
Expand Down Expand Up @@ -287,6 +296,9 @@ func TestBrokerInitializeConditions(t *testing.T) {
}, {
Type: BrokerConditionDeadLetterSinkResolved,
Status: corev1.ConditionTrue,
}, {
Type: BrokerConditionEventPoliciesReady,
Status: corev1.ConditionTrue,
}, {
Type: BrokerConditionFilter,
Status: corev1.ConditionTrue,
Expand Down Expand Up @@ -316,6 +328,9 @@ func TestBrokerInitializeConditions(t *testing.T) {
}, {
Type: BrokerConditionDeadLetterSinkResolved,
Status: corev1.ConditionTrue,
}, {
Type: BrokerConditionEventPoliciesReady,
Status: corev1.ConditionTrue,
}, {
Type: BrokerConditionFilter,
Status: corev1.ConditionTrue,
Expand Down Expand Up @@ -350,6 +365,7 @@ func TestBrokerIsReady(t *testing.T) {
markTriggerChannelReady *bool
markFilterReady *bool
markDLSResolved *bool
markEventPolicyReady *bool
markAddressable *bool
address *apis.URL
markIngressSubscriptionOwned bool
Expand All @@ -361,6 +377,7 @@ func TestBrokerIsReady(t *testing.T) {
markTriggerChannelReady: &trueVal,
markFilterReady: &trueVal,
markDLSResolved: &trueVal,
markEventPolicyReady: &trueVal,
address: &apis.URL{Scheme: "http", Host: "hostname"},
markIngressSubscriptionOwned: true,
markIngressSubscriptionReady: &trueVal,
Expand All @@ -371,6 +388,7 @@ func TestBrokerIsReady(t *testing.T) {
markTriggerChannelReady: &trueVal,
markFilterReady: &trueVal,
markDLSResolved: &trueVal,
markEventPolicyReady: &trueVal,
address: &apis.URL{Scheme: "http", Host: "hostname"},
markIngressSubscriptionOwned: true,
markIngressSubscriptionReady: &trueVal,
Expand All @@ -381,6 +399,7 @@ func TestBrokerIsReady(t *testing.T) {
markTriggerChannelReady: &trueVal,
markFilterReady: &trueVal,
markDLSResolved: &trueVal,
markEventPolicyReady: &trueVal,
address: &apis.URL{Scheme: "http", Host: "hostname"},
markIngressSubscriptionOwned: true,
markIngressSubscriptionReady: &trueVal,
Expand All @@ -391,6 +410,7 @@ func TestBrokerIsReady(t *testing.T) {
markTriggerChannelReady: &falseVal,
markFilterReady: &trueVal,
markDLSResolved: &trueVal,
markEventPolicyReady: &trueVal,
address: &apis.URL{Scheme: "http", Host: "hostname"},
markIngressSubscriptionOwned: true,
markIngressSubscriptionReady: &trueVal,
Expand All @@ -401,6 +421,7 @@ func TestBrokerIsReady(t *testing.T) {
markTriggerChannelReady: &trueVal,
markFilterReady: &falseVal,
markDLSResolved: &trueVal,
markEventPolicyReady: &trueVal,
address: &apis.URL{Scheme: "http", Host: "hostname"},
markIngressSubscriptionOwned: true,
markIngressSubscriptionReady: &trueVal,
Expand All @@ -411,6 +432,7 @@ func TestBrokerIsReady(t *testing.T) {
markTriggerChannelReady: &trueVal,
markFilterReady: &trueVal,
markDLSResolved: &trueVal,
markEventPolicyReady: &trueVal,
address: nil,
markIngressSubscriptionOwned: true,
markIngressSubscriptionReady: &trueVal,
Expand All @@ -421,6 +443,7 @@ func TestBrokerIsReady(t *testing.T) {
markTriggerChannelReady: &trueVal,
markFilterReady: &trueVal,
markDLSResolved: &trueVal,
markEventPolicyReady: &trueVal,
markAddressable: nil,
address: nil,
markIngressSubscriptionOwned: true,
Expand All @@ -432,6 +455,7 @@ func TestBrokerIsReady(t *testing.T) {
markTriggerChannelReady: &trueVal,
markFilterReady: &trueVal,
markDLSResolved: &falseVal,
markEventPolicyReady: &trueVal,
address: &apis.URL{Scheme: "http", Host: "hostname"},
markIngressSubscriptionOwned: true,
markIngressSubscriptionReady: &trueVal,
Expand All @@ -442,16 +466,29 @@ func TestBrokerIsReady(t *testing.T) {
markTriggerChannelReady: &trueVal,
markFilterReady: &trueVal,
markDLSResolved: nil,
markEventPolicyReady: &trueVal,
address: &apis.URL{Scheme: "http", Host: "hostname"},
markIngressSubscriptionOwned: true,
markIngressSubscriptionReady: &trueVal,
wantReady: true,
}, {
name: "eventpolicy sad",
markIngressReady: &trueVal,
markTriggerChannelReady: &trueVal,
markFilterReady: &trueVal,
markDLSResolved: &trueVal,
markEventPolicyReady: &falseVal,
address: &apis.URL{Scheme: "http", Host: "hostname"},
markIngressSubscriptionOwned: true,
markIngressSubscriptionReady: &trueVal,
wantReady: false,
}, {
name: "all sad",
markIngressReady: &falseVal,
markTriggerChannelReady: &falseVal,
markFilterReady: &falseVal,
markDLSResolved: &falseVal,
markEventPolicyReady: &falseVal,
address: nil,
markIngressSubscriptionOwned: true,
markIngressSubscriptionReady: &falseVal,
Expand Down Expand Up @@ -488,6 +525,14 @@ func TestBrokerIsReady(t *testing.T) {
bs.MarkDeadLetterSinkNotConfigured()
}

if test.markEventPolicyReady == &trueVal {
bs.MarkEventPoliciesTrue()
} else if test.markEventPolicyReady == &falseVal {
bs.MarkEventPoliciesFailed("", "")
} else {
bs.MarkEventPoliciesUnknown("", "")
}

if test.markFilterReady != nil {
var ep *corev1.Endpoints
if *test.markFilterReady {
Expand Down
2 changes: 2 additions & 0 deletions pkg/apis/eventing/v1/test_helper.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,7 @@ func (t testHelper) ReadyBrokerStatus() *BrokerStatus {
URL: apis.HTTP("example.com"),
})
bs.MarkDeadLetterSinkResolvedSucceeded(eventingduckv1.DeliveryStatus{})
bs.MarkEventPoliciesTrue()
return bs
}

Expand All @@ -77,6 +78,7 @@ func (t testHelper) ReadyBrokerStatusWithoutDLS() *BrokerStatus {
bs.SetAddress(&duckv1.Addressable{
URL: apis.HTTP("example.com"),
})
bs.MarkEventPoliciesTrue()
bs.MarkDeadLetterSinkNotConfigured()
return bs
}
Expand Down
8 changes: 8 additions & 0 deletions pkg/reconciler/broker/broker.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@ import (
"knative.dev/eventing/pkg/auth"
clientset "knative.dev/eventing/pkg/client/clientset/versioned"
brokerreconciler "knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker"
eventingv1alpha1listers "knative.dev/eventing/pkg/client/listers/eventing/v1alpha1"
messaginglisters "knative.dev/eventing/pkg/client/listers/messaging/v1"
ducklib "knative.dev/eventing/pkg/duck"
"knative.dev/eventing/pkg/eventingtls"
Expand Down Expand Up @@ -79,6 +80,8 @@ type Reconciler struct {

// If specified, only reconcile brokers with these labels
brokerClass string

eventPolicyLister eventingv1alpha1listers.EventPolicyLister
}

// Check that our Reconciler implements Interface
Expand Down Expand Up @@ -256,6 +259,11 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, b *eventingv1.Broker) pk

b.GetConditionSet().Manage(b.GetStatus()).MarkTrue(eventingv1.BrokerConditionAddressable)

err = auth.UpdateStatusWithEventPolicies(featureFlags, &b.Status.AppliedEventPoliciesStatus, &b.Status, r.eventPolicyLister, eventingv1.SchemeGroupVersion.WithKind("Broker"), b.ObjectMeta)
if err != nil {
return fmt.Errorf("could not update broker status with EventPolicies: %v", err)
}

// So, at this point the Broker is ready and everything should be solid
// for the triggers to act upon.
return nil
Expand Down
Loading

0 comments on commit 3264b21

Please sign in to comment.