Skip to content

Commit 8533c4e

Browse files
reaver-flomeshreaver-flomesh
authored
[backport] fix: trigger xRoutes to reconcile when gateway/backend status changed (#435) (#436)
* fix: trigger xRoutes to reconcile when gateway status changed * fix: trigger xRoutes to reconcile when backend status changed --------- Signed-off-by: Lin Yang <[email protected]>
1 parent dec12e5 commit 8533c4e

File tree

5 files changed

+308
-0
lines changed

5 files changed

+308
-0
lines changed

‎pkg/controllers/gateway/v1/grpcroute_controller.go

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -137,6 +137,8 @@ func (r *grpcRouteReconciler) SetupWithManager(mgr ctrl.Manager) error {
137137

138138
if err := ctrl.NewControllerManagedBy(mgr).
139139
For(&gwv1.GRPCRoute{}).
140+
Watches(&gwv1.Gateway{}, handler.EnqueueRequestsFromMapFunc(r.gatewayToGRPCRoutes)).
141+
Watches(&corev1.Service{}, handler.EnqueueRequestsFromMapFunc(r.serviceToGRPCRoutes)).
140142
Watches(&gwv1alpha3.BackendTLSPolicy{}, handler.EnqueueRequestsFromMapFunc(r.backendTLSToGRPCRoutes)).
141143
Watches(&gwpav1alpha2.BackendLBPolicy{}, handler.EnqueueRequestsFromMapFunc(r.backendLBToGRPCRoutes)).
142144
Watches(&gwv1beta1.ReferenceGrant{}, handler.EnqueueRequestsFromMapFunc(r.referenceGrantToGRPCRoutes)).
@@ -147,6 +149,64 @@ func (r *grpcRouteReconciler) SetupWithManager(mgr ctrl.Manager) error {
147149
return addGRPCRouteIndexers(context.Background(), mgr)
148150
}
149151

152+
func (r *grpcRouteReconciler) gatewayToGRPCRoutes(ctx context.Context, object client.Object) []reconcile.Request {
153+
gateway, ok := object.(*gwv1.Gateway)
154+
if !ok {
155+
log.Error().Msgf("Unexpected type %T", object)
156+
return nil
157+
}
158+
159+
var requests []reconcile.Request
160+
161+
list := &gwv1.GRPCRouteList{}
162+
if err := r.fctx.Manager.GetCache().List(context.Background(), list, &client.ListOptions{
163+
FieldSelector: fields.OneTermEqualSelector(constants.GatewayGRPCRouteIndex, client.ObjectKeyFromObject(gateway).String()),
164+
}); err != nil {
165+
log.Error().Msgf("Failed to list GRPCRoutes: %v", err)
166+
return nil
167+
}
168+
169+
for _, route := range list.Items {
170+
requests = append(requests, reconcile.Request{
171+
NamespacedName: types.NamespacedName{
172+
Namespace: route.Namespace,
173+
Name: route.Name,
174+
},
175+
})
176+
}
177+
178+
return requests
179+
}
180+
181+
func (r *grpcRouteReconciler) serviceToGRPCRoutes(ctx context.Context, object client.Object) []reconcile.Request {
182+
service, ok := object.(*corev1.Service)
183+
if !ok {
184+
log.Error().Msgf("Unexpected type %T", object)
185+
return nil
186+
}
187+
188+
var requests []reconcile.Request
189+
190+
list := &gwv1.GRPCRouteList{}
191+
if err := r.fctx.Manager.GetCache().List(context.Background(), list, &client.ListOptions{
192+
FieldSelector: fields.OneTermEqualSelector(constants.BackendGRPCRouteIndex, client.ObjectKeyFromObject(service).String()),
193+
}); err != nil {
194+
log.Error().Msgf("Failed to list GRPCRoutes: %v", err)
195+
return nil
196+
}
197+
198+
for _, route := range list.Items {
199+
requests = append(requests, reconcile.Request{
200+
NamespacedName: types.NamespacedName{
201+
Namespace: route.Namespace,
202+
Name: route.Name,
203+
},
204+
})
205+
}
206+
207+
return requests
208+
}
209+
150210
func (r *grpcRouteReconciler) backendTLSToGRPCRoutes(ctx context.Context, object client.Object) []reconcile.Request {
151211
policy, ok := object.(*gwv1alpha3.BackendTLSPolicy)
152212
if !ok {

‎pkg/controllers/gateway/v1/httproute_controller.go

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -137,6 +137,8 @@ func (r *httpRouteReconciler) SetupWithManager(mgr ctrl.Manager) error {
137137

138138
if err := ctrl.NewControllerManagedBy(mgr).
139139
For(&gwv1.HTTPRoute{}).
140+
Watches(&gwv1.Gateway{}, handler.EnqueueRequestsFromMapFunc(r.gatewayToHTTPRoutes)).
141+
Watches(&corev1.Service{}, handler.EnqueueRequestsFromMapFunc(r.serviceToHTTPRoutes)).
140142
Watches(&gwv1alpha3.BackendTLSPolicy{}, handler.EnqueueRequestsFromMapFunc(r.backendTLSToHTTPRoutes)).
141143
Watches(&gwpav1alpha2.BackendLBPolicy{}, handler.EnqueueRequestsFromMapFunc(r.backendLBToHTTPRoutes)).
142144
Watches(&gwpav1alpha2.HealthCheckPolicy{}, handler.EnqueueRequestsFromMapFunc(r.healthCheckToHTTPRoutes)).
@@ -149,6 +151,64 @@ func (r *httpRouteReconciler) SetupWithManager(mgr ctrl.Manager) error {
149151
return addHTTPRouteIndexers(context.Background(), mgr)
150152
}
151153

154+
func (r *httpRouteReconciler) gatewayToHTTPRoutes(ctx context.Context, object client.Object) []reconcile.Request {
155+
gateway, ok := object.(*gwv1.Gateway)
156+
if !ok {
157+
log.Error().Msgf("Unexpected type %T", object)
158+
return nil
159+
}
160+
161+
var requests []reconcile.Request
162+
163+
list := &gwv1.HTTPRouteList{}
164+
if err := r.fctx.Manager.GetCache().List(context.Background(), list, &client.ListOptions{
165+
FieldSelector: fields.OneTermEqualSelector(constants.GatewayHTTPRouteIndex, client.ObjectKeyFromObject(gateway).String()),
166+
}); err != nil {
167+
log.Error().Msgf("Failed to list HTTPRoutes: %v", err)
168+
return nil
169+
}
170+
171+
for _, route := range list.Items {
172+
requests = append(requests, reconcile.Request{
173+
NamespacedName: types.NamespacedName{
174+
Namespace: route.Namespace,
175+
Name: route.Name,
176+
},
177+
})
178+
}
179+
180+
return requests
181+
}
182+
183+
func (r *httpRouteReconciler) serviceToHTTPRoutes(ctx context.Context, object client.Object) []reconcile.Request {
184+
service, ok := object.(*corev1.Service)
185+
if !ok {
186+
log.Error().Msgf("Unexpected type %T", object)
187+
return nil
188+
}
189+
190+
var requests []reconcile.Request
191+
192+
list := &gwv1.HTTPRouteList{}
193+
if err := r.fctx.Manager.GetCache().List(context.Background(), list, &client.ListOptions{
194+
FieldSelector: fields.OneTermEqualSelector(constants.BackendHTTPRouteIndex, client.ObjectKeyFromObject(service).String()),
195+
}); err != nil {
196+
log.Error().Msgf("Failed to list HTTPRoutes: %v", err)
197+
return nil
198+
}
199+
200+
for _, route := range list.Items {
201+
requests = append(requests, reconcile.Request{
202+
NamespacedName: types.NamespacedName{
203+
Namespace: route.Namespace,
204+
Name: route.Name,
205+
},
206+
})
207+
}
208+
209+
return requests
210+
}
211+
152212
func (r *httpRouteReconciler) backendTLSToHTTPRoutes(ctx context.Context, object client.Object) []reconcile.Request {
153213
policy, ok := object.(*gwv1alpha3.BackendTLSPolicy)
154214
if !ok {

‎pkg/controllers/gateway/v1alpha2/tcproute_controller.go

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -129,6 +129,8 @@ func (r *tcpRouteReconciler) SetupWithManager(mgr ctrl.Manager) error {
129129

130130
if err := ctrl.NewControllerManagedBy(mgr).
131131
For(&gwv1alpha2.TCPRoute{}).
132+
Watches(&gwv1.Gateway{}, handler.EnqueueRequestsFromMapFunc(r.gatewayToTCPRoutes)).
133+
Watches(&corev1.Service{}, handler.EnqueueRequestsFromMapFunc(r.serviceToTCPRoutes)).
132134
Watches(&gwv1alpha3.BackendTLSPolicy{}, handler.EnqueueRequestsFromMapFunc(r.backendTLSToTCPRoutes)).
133135
Watches(&gwv1beta1.ReferenceGrant{}, handler.EnqueueRequestsFromMapFunc(r.referenceGrantToTCPRoutes)).
134136
Complete(r); err != nil {
@@ -138,6 +140,64 @@ func (r *tcpRouteReconciler) SetupWithManager(mgr ctrl.Manager) error {
138140
return addTCPRouteIndexers(context.Background(), mgr)
139141
}
140142

143+
func (r *tcpRouteReconciler) gatewayToTCPRoutes(ctx context.Context, object client.Object) []reconcile.Request {
144+
gateway, ok := object.(*gwv1.Gateway)
145+
if !ok {
146+
log.Error().Msgf("Unexpected type %T", object)
147+
return nil
148+
}
149+
150+
var requests []reconcile.Request
151+
152+
list := &gwv1alpha2.TCPRouteList{}
153+
if err := r.fctx.Manager.GetCache().List(context.Background(), list, &client.ListOptions{
154+
FieldSelector: fields.OneTermEqualSelector(constants.GatewayTCPRouteIndex, client.ObjectKeyFromObject(gateway).String()),
155+
}); err != nil {
156+
log.Error().Msgf("Failed to list TCPRoutes: %v", err)
157+
return nil
158+
}
159+
160+
for _, route := range list.Items {
161+
requests = append(requests, reconcile.Request{
162+
NamespacedName: types.NamespacedName{
163+
Namespace: route.Namespace,
164+
Name: route.Name,
165+
},
166+
})
167+
}
168+
169+
return requests
170+
}
171+
172+
func (r *tcpRouteReconciler) serviceToTCPRoutes(ctx context.Context, object client.Object) []reconcile.Request {
173+
service, ok := object.(*corev1.Service)
174+
if !ok {
175+
log.Error().Msgf("Unexpected type %T", object)
176+
return nil
177+
}
178+
179+
var requests []reconcile.Request
180+
181+
list := &gwv1alpha2.TCPRouteList{}
182+
if err := r.fctx.Manager.GetCache().List(context.Background(), list, &client.ListOptions{
183+
FieldSelector: fields.OneTermEqualSelector(constants.BackendTCPRouteIndex, client.ObjectKeyFromObject(service).String()),
184+
}); err != nil {
185+
log.Error().Msgf("Failed to list TCPRoutes: %v", err)
186+
return nil
187+
}
188+
189+
for _, route := range list.Items {
190+
requests = append(requests, reconcile.Request{
191+
NamespacedName: types.NamespacedName{
192+
Namespace: route.Namespace,
193+
Name: route.Name,
194+
},
195+
})
196+
}
197+
198+
return requests
199+
}
200+
141201
func (r *tcpRouteReconciler) backendTLSToTCPRoutes(ctx context.Context, object client.Object) []reconcile.Request {
142202
policy, ok := object.(*gwv1alpha3.BackendTLSPolicy)
143203
if !ok {

‎pkg/controllers/gateway/v1alpha2/tlsroute_controller.go

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,12 @@ package v1alpha2
2727
import (
2828
"context"
2929

30+
corev1 "k8s.io/api/core/v1"
31+
32+
"k8s.io/apimachinery/pkg/fields"
33+
"sigs.k8s.io/controller-runtime/pkg/handler"
34+
gwv1 "sigs.k8s.io/gateway-api/apis/v1"
35+
3036
"github.com/flomesh-io/fsm/pkg/gateway/status/routes"
3137

3238
"k8s.io/apimachinery/pkg/util/sets"
@@ -123,13 +129,73 @@ func (r *tlsRouteReconciler) SetupWithManager(mgr ctrl.Manager) error {
123129

124130
if err := ctrl.NewControllerManagedBy(mgr).
125131
For(&gwv1alpha2.TLSRoute{}).
132+
Watches(&gwv1.Gateway{}, handler.EnqueueRequestsFromMapFunc(r.gatewayToTLSRoutes)).
133+
Watches(&corev1.Service{}, handler.EnqueueRequestsFromMapFunc(r.serviceToTLSRoutes)).
126134
Complete(r); err != nil {
127135
return err
128136
}
129137

130138
return addTLSRouteIndexers(context.Background(), mgr)
131139
}
132140

141+
func (r *tlsRouteReconciler) gatewayToTLSRoutes(ctx context.Context, object client.Object) []reconcile.Request {
142+
gateway, ok := object.(*gwv1.Gateway)
143+
if !ok {
144+
log.Error().Msgf("Unexpected type %T", object)
145+
return nil
146+
}
147+
148+
var requests []reconcile.Request
149+
150+
list := &gwv1alpha2.TLSRouteList{}
151+
if err := r.fctx.Manager.GetCache().List(context.Background(), list, &client.ListOptions{
152+
FieldSelector: fields.OneTermEqualSelector(constants.GatewayTLSRouteIndex, client.ObjectKeyFromObject(gateway).String()),
153+
}); err != nil {
154+
log.Error().Msgf("Failed to list TLSRoutes: %v", err)
155+
return nil
156+
}
157+
158+
for _, route := range list.Items {
159+
requests = append(requests, reconcile.Request{
160+
NamespacedName: types.NamespacedName{
161+
Namespace: route.Namespace,
162+
Name: route.Name,
163+
},
164+
})
165+
}
166+
167+
return requests
168+
}
169+
170+
func (r *tlsRouteReconciler) serviceToTLSRoutes(ctx context.Context, object client.Object) []reconcile.Request {
171+
service, ok := object.(*corev1.Service)
172+
if !ok {
173+
log.Error().Msgf("Unexpected type %T", object)
174+
return nil
175+
}
176+
177+
var requests []reconcile.Request
178+
179+
list := &gwv1alpha2.TLSRouteList{}
180+
if err := r.fctx.Manager.GetCache().List(context.Background(), list, &client.ListOptions{
181+
FieldSelector: fields.OneTermEqualSelector(constants.BackendTLSRouteIndex, client.ObjectKeyFromObject(service).String()),
182+
}); err != nil {
183+
log.Error().Msgf("Failed to list TLSRoutes: %v", err)
184+
return nil
185+
}
186+
187+
for _, route := range list.Items {
188+
requests = append(requests, reconcile.Request{
189+
NamespacedName: types.NamespacedName{
190+
Namespace: route.Namespace,
191+
Name: route.Name,
192+
},
193+
})
194+
}
195+
196+
return requests
197+
}
198+
133199
func addTLSRouteIndexers(ctx context.Context, mgr manager.Manager) error {
134200
if err := mgr.GetFieldIndexer().IndexField(ctx, &gwv1alpha2.TLSRoute{}, constants.GatewayTLSRouteIndex, func(obj client.Object) []string {
135201
tlsRoute := obj.(*gwv1alpha2.TLSRoute)

‎pkg/controllers/gateway/v1alpha2/udproute_controller.go

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,8 @@ package v1alpha2
2727
import (
2828
"context"
2929

30+
corev1 "k8s.io/api/core/v1"
31+
3032
"github.com/flomesh-io/fsm/pkg/gateway/status/routes"
3133

3234
"k8s.io/apimachinery/pkg/fields"
@@ -126,6 +128,8 @@ func (r *udpRouteReconciler) SetupWithManager(mgr ctrl.Manager) error {
126128

127129
if err := ctrl.NewControllerManagedBy(mgr).
128130
For(&gwv1alpha2.UDPRoute{}).
131+
Watches(&gwv1.Gateway{}, handler.EnqueueRequestsFromMapFunc(r.gatewayToUDPRoutes)).
132+
Watches(&corev1.Service{}, handler.EnqueueRequestsFromMapFunc(r.serviceToUDPRoutes)).
129133
Watches(&gwv1beta1.ReferenceGrant{}, handler.EnqueueRequestsFromMapFunc(r.referenceGrantToUDPRoutes)).
130134
Complete(r); err != nil {
131135
return err
@@ -134,6 +138,64 @@ func (r *udpRouteReconciler) SetupWithManager(mgr ctrl.Manager) error {
134138
return addUDPRouteIndexers(context.Background(), mgr)
135139
}
136140

141+
func (r *udpRouteReconciler) gatewayToUDPRoutes(ctx context.Context, object client.Object) []reconcile.Request {
142+
gateway, ok := object.(*gwv1.Gateway)
143+
if !ok {
144+
log.Error().Msgf("Unexpected type %T", object)
145+
return nil
146+
}
147+
148+
var requests []reconcile.Request
149+
150+
list := &gwv1alpha2.UDPRouteList{}
151+
if err := r.fctx.Manager.GetCache().List(context.Background(), list, &client.ListOptions{
152+
FieldSelector: fields.OneTermEqualSelector(constants.GatewayUDPRouteIndex, client.ObjectKeyFromObject(gateway).String()),
153+
}); err != nil {
154+
log.Error().Msgf("Failed to list UDPRoutes: %v", err)
155+
return nil
156+
}
157+
158+
for _, route := range list.Items {
159+
requests = append(requests, reconcile.Request{
160+
NamespacedName: types.NamespacedName{
161+
Namespace: route.Namespace,
162+
Name: route.Name,
163+
},
164+
})
165+
}
166+
167+
return requests
168+
}
169+
170+
func (r *udpRouteReconciler) serviceToUDPRoutes(ctx context.Context, object client.Object) []reconcile.Request {
171+
service, ok := object.(*corev1.Service)
172+
if !ok {
173+
log.Error().Msgf("Unexpected type %T", object)
174+
return nil
175+
}
176+
177+
var requests []reconcile.Request
178+
179+
list := &gwv1alpha2.UDPRouteList{}
180+
if err := r.fctx.Manager.GetCache().List(context.Background(), list, &client.ListOptions{
181+
FieldSelector: fields.OneTermEqualSelector(constants.BackendUDPRouteIndex, client.ObjectKeyFromObject(service).String()),
182+
}); err != nil {
183+
log.Error().Msgf("Failed to list UDPRoutes: %v", err)
184+
return nil
185+
}
186+
187+
for _, route := range list.Items {
188+
requests = append(requests, reconcile.Request{
189+
NamespacedName: types.NamespacedName{
190+
Namespace: route.Namespace,
191+
Name: route.Name,
192+
},
193+
})
194+
}
195+
196+
return requests
197+
}
198+
137199
func (r *udpRouteReconciler) referenceGrantToUDPRoutes(ctx context.Context, obj client.Object) []reconcile.Request {
138200
refGrant, ok := obj.(*gwv1beta1.ReferenceGrant)
139201
if !ok {

0 commit comments

Comments
 (0)