Skip to content

Commit

Permalink
Retry on Gateway API resource status update
Browse files Browse the repository at this point in the history 
Co-authored-by: Kevin Pollet <pollet.kevin@gmail.com>
  • Loading branch information
@rtribotte @kevinpollet
rtribotte and kevinpollet authored Jul 11, 2024
1 parent 173a18f commit 58dcbb4
Show file tree
Hide file tree
Showing 6 changed files with 164 additions and 109 deletions.
7 changes: 0 additions & 7 deletions integration/k8s_conformance_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -207,13 +207,6 @@ func (s *K8sConformanceSuite) TestK8sGatewayAPIConformance() {
features.SupportHTTPRoutePathRewrite,
features.SupportHTTPRoutePathRedirect,
),
ExemptFeatures: sets.New(
features.SupportHTTPRouteRequestTimeout,
features.SupportHTTPRouteBackendTimeout,
features.SupportHTTPRouteResponseHeaderModification,
features.SupportHTTPRouteRequestMirror,
features.SupportHTTPRouteRequestMultipleMirrors,
),
})
require.NoError(s.T(), err)

Expand Down
240 changes: 150 additions & 90 deletions pkg/provider/kubernetes/gateway/client.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ import (
kclientset "k8s.io/client-go/kubernetes"
"k8s.io/client-go/rest"
"k8s.io/client-go/tools/clientcmd"
"k8s.io/client-go/util/retry"
gatev1 "sigs.k8s.io/gateway-api/apis/v1"
gatev1alpha2 "sigs.k8s.io/gateway-api/apis/v1alpha2"
gatev1beta1 "sigs.k8s.io/gateway-api/apis/v1beta1"
Expand Down Expand Up @@ -51,8 +52,8 @@ func (reh *resourceEventHandler) OnDelete(obj interface{}) {
// The stores can then be accessed via the Get* functions.
type Client interface {
WatchAll(namespaces []string, stopCh <-chan struct{}) (<-chan interface{}, error)
UpdateGatewayStatus(gateway *gatev1.Gateway, gatewayStatus gatev1.GatewayStatus) error
UpdateGatewayClassStatus(gatewayClass *gatev1.GatewayClass, condition metav1.Condition) error
UpdateGatewayStatus(ctx context.Context, gateway ktypes.NamespacedName, status gatev1.GatewayStatus) error
UpdateGatewayClassStatus(ctx context.Context, name string, condition metav1.Condition) error
UpdateHTTPRouteStatus(ctx context.Context, route ktypes.NamespacedName, status gatev1.HTTPRouteStatus) error
UpdateTCPRouteStatus(ctx context.Context, route ktypes.NamespacedName, status gatev1alpha2.TCPRouteStatus) error
UpdateTLSRouteStatus(ctx context.Context, route ktypes.NamespacedName, status gatev1alpha2.TLSRouteStatus) error
Expand Down Expand Up @@ -377,53 +378,76 @@ func (c *clientWrapper) ListGatewayClasses() ([]*gatev1.GatewayClass, error) {
return c.factoryGatewayClass.Gateway().V1().GatewayClasses().Lister().List(labels.Everything())
}

func (c *clientWrapper) UpdateGatewayClassStatus(gatewayClass *gatev1.GatewayClass, condition metav1.Condition) error {
gc := gatewayClass.DeepCopy()

var newConditions []metav1.Condition
for _, cond := range gc.Status.Conditions {
// No update for identical condition.
if cond.Type == condition.Type && cond.Status == condition.Status && cond.ObservedGeneration == condition.ObservedGeneration {
return nil
func (c *clientWrapper) UpdateGatewayClassStatus(ctx context.Context, name string, condition metav1.Condition) error {
err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
currentGatewayClass, err := c.factoryGatewayClass.Gateway().V1().GatewayClasses().Lister().Get(name)
if err != nil {
// We have to return err itself here (not wrapped inside another error)
// so that RetryOnConflict can identify it correctly.
return err
}

// Keep other condition types.
if cond.Type != condition.Type {
newConditions = append(newConditions, cond)
currentGatewayClass = currentGatewayClass.DeepCopy()
var newConditions []metav1.Condition
for _, cond := range currentGatewayClass.Status.Conditions {
// No update for identical condition.
if cond.Type == condition.Type && cond.Status == condition.Status && cond.ObservedGeneration == condition.ObservedGeneration {
return nil
}

// Keep other condition types.
if cond.Type != condition.Type {
newConditions = append(newConditions, cond)
}
}
}

// Append the condition to update.
newConditions = append(newConditions, condition)
gc.Status.Conditions = newConditions
// Append the condition to update.
newConditions = append(newConditions, condition)
currentGatewayClass.Status.Conditions = newConditions

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if _, err = c.csGateway.GatewayV1().GatewayClasses().UpdateStatus(ctx, currentGatewayClass, metav1.UpdateOptions{}); err != nil {
// We have to return err itself here (not wrapped inside another error)
// so that RetryOnConflict can identify it correctly.
return err
}

_, err := c.csGateway.GatewayV1().GatewayClasses().UpdateStatus(ctx, gc, metav1.UpdateOptions{})
return nil
})
if err != nil {
return fmt.Errorf("failed to update GatewayClass %q status: %w", gatewayClass.Name, err)
return fmt.Errorf("failed to update GatewayClass %q status: %w", name, err)
}

return nil
}

func (c *clientWrapper) UpdateGatewayStatus(gateway *gatev1.Gateway, gatewayStatus gatev1.GatewayStatus) error {
func (c *clientWrapper) UpdateGatewayStatus(ctx context.Context, gateway ktypes.NamespacedName, status gatev1.GatewayStatus) error {
if !c.isWatchedNamespace(gateway.Namespace) {
return fmt.Errorf("cannot update Gateway status %s/%s: namespace is not within watched namespaces", gateway.Namespace, gateway.Name)
}

if gatewayStatusEquals(gateway.Status, gatewayStatus) {
return nil
}
err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
currentGateway, err := c.factoriesGateway[c.lookupNamespace(gateway.Namespace)].Gateway().V1().Gateways().Lister().Gateways(gateway.Namespace).Get(gateway.Name)
if err != nil {
// We have to return err itself here (not wrapped inside another error)
// so that RetryOnConflict can identify it correctly.
return err
}

if gatewayStatusEquals(currentGateway.Status, status) {
return nil
}

g := gateway.DeepCopy()
g.Status = gatewayStatus
currentGateway = currentGateway.DeepCopy()
currentGateway.Status = status

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if _, err = c.csGateway.GatewayV1().Gateways(gateway.Namespace).UpdateStatus(ctx, currentGateway, metav1.UpdateOptions{}); err != nil {
// We have to return err itself here (not wrapped inside another error)
// so that RetryOnConflict can identify it correctly.
return err
}

_, err := c.csGateway.GatewayV1().Gateways(gateway.Namespace).UpdateStatus(ctx, g, metav1.UpdateOptions{})
return nil
})
if err != nil {
return fmt.Errorf("failed to update Gateway %q status: %w", gateway.Name, err)
}
Expand All @@ -436,32 +460,44 @@ func (c *clientWrapper) UpdateHTTPRouteStatus(ctx context.Context, route ktypes.
return fmt.Errorf("updating HTTPRoute status %s/%s: namespace is not within watched namespaces", route.Namespace, route.Name)
}

currentRoute, err := c.factoriesGateway[c.lookupNamespace(route.Namespace)].Gateway().V1().HTTPRoutes().Lister().HTTPRoutes(route.Namespace).Get(route.Name)
if err != nil {
return fmt.Errorf("getting HTTPRoute %s/%s: %w", route.Namespace, route.Name, err)
}
err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
currentRoute, err := c.factoriesGateway[c.lookupNamespace(route.Namespace)].Gateway().V1().HTTPRoutes().Lister().HTTPRoutes(route.Namespace).Get(route.Name)
if err != nil {
// We have to return err itself here (not wrapped inside another error)
// so that RetryOnConflict can identify it correctly.
return err
}

// TODO: keep statuses for gateways managed by other Traefik instances.
var parentStatuses []gatev1.RouteParentStatus
for _, currentParentStatus := range currentRoute.Status.Parents {
if currentParentStatus.ControllerName != controllerName {
parentStatuses = append(parentStatuses, currentParentStatus)
continue
// TODO: keep statuses for gateways managed by other Traefik instances.
var parentStatuses []gatev1.RouteParentStatus
for _, currentParentStatus := range currentRoute.Status.Parents {
if currentParentStatus.ControllerName != controllerName {
parentStatuses = append(parentStatuses, currentParentStatus)
continue
}
}
}

parentStatuses = append(parentStatuses, status.Parents...)
parentStatuses = append(parentStatuses, status.Parents...)

currentRoute = currentRoute.DeepCopy()
currentRoute.Status = gatev1.HTTPRouteStatus{
RouteStatus: gatev1.RouteStatus{
Parents: parentStatuses,
},
}
currentRoute = currentRoute.DeepCopy()
currentRoute.Status = gatev1.HTTPRouteStatus{
RouteStatus: gatev1.RouteStatus{
Parents: parentStatuses,
},
}

if _, err := c.csGateway.GatewayV1().HTTPRoutes(route.Namespace).UpdateStatus(ctx, currentRoute, metav1.UpdateOptions{}); err != nil {
return fmt.Errorf("updating HTTPRoute %s/%s status: %w", route.Namespace, route.Name, err)
if _, err = c.csGateway.GatewayV1().HTTPRoutes(route.Namespace).UpdateStatus(ctx, currentRoute, metav1.UpdateOptions{}); err != nil {
// We have to return err itself here (not wrapped inside another error)
// so that RetryOnConflict can identify it correctly.
return err
}

return nil
})
if err != nil {
return fmt.Errorf("failed to update HTTPRoute %q status: %w", route.Name, err)
}

return nil
}

Expand All @@ -470,32 +506,44 @@ func (c *clientWrapper) UpdateTCPRouteStatus(ctx context.Context, route ktypes.N
return fmt.Errorf("updating TCPRoute status %s/%s: namespace is not within watched namespaces", route.Namespace, route.Name)
}

currentRoute, err := c.factoriesGateway[c.lookupNamespace(route.Namespace)].Gateway().V1alpha2().TCPRoutes().Lister().TCPRoutes(route.Namespace).Get(route.Name)
if err != nil {
return fmt.Errorf("getting TCPRoute %s/%s: %w", route.Namespace, route.Name, err)
}
err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
currentRoute, err := c.factoriesGateway[c.lookupNamespace(route.Namespace)].Gateway().V1alpha2().TCPRoutes().Lister().TCPRoutes(route.Namespace).Get(route.Name)
if err != nil {
// We have to return err itself here (not wrapped inside another error)
// so that RetryOnConflict can identify it correctly.
return err
}

// TODO: keep statuses for gateways managed by other Traefik instances.
var parentStatuses []gatev1alpha2.RouteParentStatus
for _, currentParentStatus := range currentRoute.Status.Parents {
if currentParentStatus.ControllerName != controllerName {
parentStatuses = append(parentStatuses, currentParentStatus)
continue
// TODO: keep statuses for gateways managed by other Traefik instances.
var parentStatuses []gatev1alpha2.RouteParentStatus
for _, currentParentStatus := range currentRoute.Status.Parents {
if currentParentStatus.ControllerName != controllerName {
parentStatuses = append(parentStatuses, currentParentStatus)
continue
}
}
}

parentStatuses = append(parentStatuses, status.Parents...)
parentStatuses = append(parentStatuses, status.Parents...)

currentRoute = currentRoute.DeepCopy()
currentRoute.Status = gatev1alpha2.TCPRouteStatus{
RouteStatus: gatev1.RouteStatus{
Parents: parentStatuses,
},
}
currentRoute = currentRoute.DeepCopy()
currentRoute.Status = gatev1alpha2.TCPRouteStatus{
RouteStatus: gatev1.RouteStatus{
Parents: parentStatuses,
},
}

if _, err := c.csGateway.GatewayV1alpha2().TCPRoutes(route.Namespace).UpdateStatus(ctx, currentRoute, metav1.UpdateOptions{}); err != nil {
return fmt.Errorf("updating TCPRoute %s/%s status: %w", route.Namespace, route.Name, err)
if _, err = c.csGateway.GatewayV1alpha2().TCPRoutes(route.Namespace).UpdateStatus(ctx, currentRoute, metav1.UpdateOptions{}); err != nil {
// We have to return err itself here (not wrapped inside another error)
// so that RetryOnConflict can identify it correctly.
return err
}

return nil
})
if err != nil {
return fmt.Errorf("failed to update TCPRoute %q status: %w", route.Name, err)
}

return nil
}

Expand All @@ -504,32 +552,44 @@ func (c *clientWrapper) UpdateTLSRouteStatus(ctx context.Context, route ktypes.N
return fmt.Errorf("updating TLSRoute status %s/%s: namespace is not within watched namespaces", route.Namespace, route.Name)
}

currentRoute, err := c.factoriesGateway[c.lookupNamespace(route.Namespace)].Gateway().V1alpha2().TLSRoutes().Lister().TLSRoutes(route.Namespace).Get(route.Name)
if err != nil {
return fmt.Errorf("getting TLSRoute %s/%s: %w", route.Namespace, route.Name, err)
}
err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
currentRoute, err := c.factoriesGateway[c.lookupNamespace(route.Namespace)].Gateway().V1alpha2().TLSRoutes().Lister().TLSRoutes(route.Namespace).Get(route.Name)
if err != nil {
// We have to return err itself here (not wrapped inside another error)
// so that RetryOnConflict can identify it correctly.
return err
}

// TODO: keep statuses for gateways managed by other Traefik instances.
var parentStatuses []gatev1alpha2.RouteParentStatus
for _, currentParentStatus := range currentRoute.Status.Parents {
if currentParentStatus.ControllerName != controllerName {
parentStatuses = append(parentStatuses, currentParentStatus)
continue
// TODO: keep statuses for gateways managed by other Traefik instances.
var parentStatuses []gatev1alpha2.RouteParentStatus
for _, currentParentStatus := range currentRoute.Status.Parents {
if currentParentStatus.ControllerName != controllerName {
parentStatuses = append(parentStatuses, currentParentStatus)
continue
}
}
}

parentStatuses = append(parentStatuses, status.Parents...)
parentStatuses = append(parentStatuses, status.Parents...)

currentRoute = currentRoute.DeepCopy()
currentRoute.Status = gatev1alpha2.TLSRouteStatus{
RouteStatus: gatev1.RouteStatus{
Parents: parentStatuses,
},
}
currentRoute = currentRoute.DeepCopy()
currentRoute.Status = gatev1alpha2.TLSRouteStatus{
RouteStatus: gatev1.RouteStatus{
Parents: parentStatuses,
},
}

if _, err = c.csGateway.GatewayV1alpha2().TLSRoutes(route.Namespace).UpdateStatus(ctx, currentRoute, metav1.UpdateOptions{}); err != nil {
// We have to return err itself here (not wrapped inside another error)
// so that RetryOnConflict can identify it correctly.
return err
}

if _, err := c.csGateway.GatewayV1alpha2().TLSRoutes(route.Namespace).UpdateStatus(ctx, currentRoute, metav1.UpdateOptions{}); err != nil {
return fmt.Errorf("updating TLSRoute %s/%s status: %w", route.Namespace, route.Name, err)
return nil
})
if err != nil {
return fmt.Errorf("failed to update TLSRoute %q status: %w", route.Name, err)
}

return nil
}

Expand Down
2 changes: 1 addition & 1 deletion pkg/provider/kubernetes/gateway/httproute.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -91,7 +91,7 @@ func (p *Provider) loadHTTPRoutes(ctx context.Context, gatewayListeners []gatewa
},
}
if err := p.client.UpdateHTTPRouteStatus(ctx, ktypes.NamespacedName{Namespace: route.Namespace, Name: route.Name}, status); err != nil {
logger.Error().
logger.Warn().
Err(err).
Msg("Unable to update HTTPRoute status")
}
Expand Down
20 changes: 11 additions & 9 deletions pkg/provider/kubernetes/gateway/kubernetes.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@ import (
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels"
ktypes "k8s.io/apimachinery/pkg/types"
"k8s.io/utils/ptr"
gatev1 "sigs.k8s.io/gateway-api/apis/v1"
gatev1beta1 "sigs.k8s.io/gateway-api/apis/v1beta1"
Expand Down Expand Up @@ -317,7 +318,7 @@ func (p *Provider) loadConfigurationFromGateways(ctx context.Context) *dynamic.C

gatewayClassNames[gatewayClass.Name] = struct{}{}

err := p.client.UpdateGatewayClassStatus(gatewayClass, metav1.Condition{
err := p.client.UpdateGatewayClassStatus(ctx, gatewayClass.Name, metav1.Condition{
Type: string(gatev1.GatewayClassConditionStatusAccepted),
Status: metav1.ConditionTrue,
ObservedGeneration: gatewayClass.Generation,
Expand All @@ -327,7 +328,7 @@ func (p *Provider) loadConfigurationFromGateways(ctx context.Context) *dynamic.C
})
if err != nil {
log.Ctx(ctx).
Error().
Warn().
Err(err).
Str("gateway_class", gatewayClass.Name).
Msg("Unable to update GatewayClass status")
Expand Down Expand Up @@ -370,17 +371,18 @@ func (p *Provider) loadConfigurationFromGateways(ctx context.Context) *dynamic.C
}
}

gatewayStatus, errG := p.makeGatewayStatus(gateway, listeners, addresses)
if err = p.client.UpdateGatewayStatus(gateway, gatewayStatus); err != nil {
gatewayStatus, err := p.makeGatewayStatus(gateway, listeners, addresses)
if err != nil {
logger.Error().
Err(err).
Msg("Unable to update Gateway status")
}
if errG != nil {
logger.Error().
Err(errG).
Msg("Unable to create Gateway status")
}

if err = p.client.UpdateGatewayStatus(ctx, ktypes.NamespacedName{Name: gateway.Name, Namespace: gateway.Namespace}, gatewayStatus); err != nil {
logger.Warn().
Err(err).
Msg("Unable to update Gateway status")
}
}

return conf
Expand Down
Loading

0 comments on commit 58dcbb4

Please sign in to comment.