fix: retry reconcile on transient errors during reconcile #6299
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -180,6 +180,16 @@ | |
| } | ||
| } | ||
|
|
||
| func isTransientError(err error) bool { | ||
| return kerrors.IsServerTimeout(err) || | ||
| kerrors.IsTimeout(err) || | ||
| kerrors.IsTooManyRequests(err) || | ||
| kerrors.IsServiceUnavailable(err) || | ||
| kerrors.IsStoreReadError(err) || | ||
| kerrors.IsInternalError(err) || | ||
| kerrors.IsUnexpectedServerError(err) | ||
| } | ||
|
|
||
| // Reconcile handles reconciling all resources in a single call. Any resource event should enqueue the | ||
| // same reconcile.Request containing the gateway controller name. This allows multiple resource updates to | ||
| // be handled by a single call to Reconcile. The reconcile.Request DOES NOT map to a specific resource. | ||
|
|
@@ -222,6 +232,11 @@ | |
| // This should run before processGateways and processBackendRefs | ||
| if managedGC.Spec.ParametersRef != nil && managedGC.DeletionTimestamp == nil { | ||
| if err := r.processGatewayClassParamsRef(ctx, managedGC, gwcResourceMapping, gwcResource); err != nil { | ||
| if isTransientError(err) { | ||
|
zhaohuabing marked this conversation as resolved.
Outdated
|
||
| r.log.Error(err, "transient error processing GatewayClass parametersRef", "gatewayClass", managedGC.Name) | ||
| return reconcile.Result{}, err | ||
| } | ||
|
|
||
| r.log.Error(err, fmt.Sprintf("failed processGatewayClassParamsRef for gatewayClass %s, skipping it", managedGC.Name)) | ||
| msg := fmt.Sprintf("%s: %v", status.MsgGatewayClassInvalidParams, err) | ||
| gc := status.SetGatewayClassAccepted( | ||
|
|
@@ -233,6 +248,32 @@ | |
| continue | ||
| } | ||
| } | ||
|
|
||
| // process envoy gateway secret refs | ||
| if err := r.processEnvoyProxySecretRef(ctx, gwcResource); err != nil { | ||
| if isTransientError(err) { | ||
| r.log.Error(err, "transient error processing TLS SecretRef for EnvoyProxy", "gatewayClass", managedGC.Name) | ||
| return reconcile.Result{}, err | ||
| } | ||
|
|
||
| r.log.Error(err, fmt.Sprintf("failed process TLS SecretRef for EnvoyProxy for gatewayClass %s, skipping it", managedGC.Name)) | ||
| gc := status.SetGatewayClassAccepted( | ||
| managedGC.DeepCopy(), | ||
| false, | ||
| string(gwapiv1.GatewayClassReasonAccepted), | ||
| fmt.Sprintf("%s: %v", status.MsgGatewayClassInvalidParams, err)) | ||
| r.resources.GatewayClassStatuses.Store(utils.NamespacedName(gc), &gc.Status) | ||
| continue | ||
|
Member
There was a problem hiding this comment. Skip the processing GatewayClass if failed to process its parameter - should we continue processing the resources for an invalid GatewayClass and publish them?
Contributor
There was a problem hiding this comment. What's the downside? Will any of these resources actually translate to Infra/XDS, or would they only be updated with a status that reflect problems (if they have any) ?
Member
There was a problem hiding this comment.
Yes, all resources of this GatewayClass will be translated into Infra/XDS if we continue processing the failed GatewayClass, but the resulting configuration are incorrect - for example, the
The previous change took this approach, but as this comment suggested, it's a big design decision and it may be risky to cherry-pick it into v1.4 branch. So I reverted it and track it at #6448 . We can discuss this in the next EG meeting and do it in the main branch.
Update: this PR continues processing failed GatewayClass and publish its resources - this may publish Infra IR and xDS IR with misconfiguration, but avoids taking down the whole Gateway with a single user error from an |
||
| } | ||
|
|
||
| // GatewayClass is valid so far, mark it as accepted. | ||
| gc := status.SetGatewayClassAccepted( | ||
| managedGC.DeepCopy(), | ||
| true, | ||
| string(gwapiv1.GatewayClassReasonAccepted), | ||
| status.MsgValidGatewayClass) | ||
| r.resources.GatewayClassStatuses.Store(utils.NamespacedName(gc), &gc.Status) | ||
|
|
||
| // it's safe here to append gwcResource to gwcResources | ||
| gwcResources = append(gwcResources, gwcResource) | ||
| // process global resources | ||
|
|
@@ -243,21 +284,31 @@ | |
|
|
||
| // Add all Gateways, their associated Routes, and referenced resources to the resourceTree | ||
| if err = r.processGateways(ctx, managedGC, gwcResourceMapping, gwcResource); err != nil { | ||
| if isTransientError(err) { | ||
| r.log.Error(err, "transient error processing gateways", "gatewayClass", managedGC.Name) | ||
| return reconcile.Result{}, err | ||
| } | ||
| r.log.Error(err, fmt.Sprintf("failed processGateways for gatewayClass %s, skipping it", managedGC.Name)) | ||
|
Member
There was a problem hiding this comment. If one of the resources fail, continue processing remaining resources and publish them as-is. This avoid skipping valid resources and publishing partial resources. |
||
| } | ||
|
|
||
| if r.eppCRDExists { | ||
| // Add all EnvoyPatchPolicies to the resourceTree | ||
| if err = r.processEnvoyPatchPolicies(ctx, gwcResource, gwcResourceMapping); err != nil { | ||
| if isTransientError(err) { | ||
| r.log.Error(err, "transient error processing EnvoyPatchPolicies", "gatewayClass", managedGC.Name) | ||
| return reconcile.Result{}, err | ||
| } | ||
| r.log.Error(err, fmt.Sprintf("failed processEnvoyPatchPolicies for gatewayClass %s, skipping it", managedGC.Name)) | ||
| } | ||
| } | ||
|
|
||
| if r.ctpCRDExists { | ||
| // Add all ClientTrafficPolicies and their referenced resources to the resourceTree | ||
| if err = r.processClientTrafficPolicies(ctx, gwcResource, gwcResourceMapping); err != nil { | ||
| if isTransientError(err) { | ||
| r.log.Error(err, "transient error processing ClientTrafficPolicies", "gatewayClass", managedGC.Name) | ||
| return reconcile.Result{}, err | ||
| } | ||
| r.log.Error(err, fmt.Sprintf("failed processClientTrafficPolicies for gatewayClass %s, skipping it", managedGC.Name)) | ||
| continue | ||
|
zhaohuabing marked this conversation as resolved.
Outdated
|
||
| } | ||
|
|
@@ -266,6 +317,10 @@ | |
| if r.btpCRDExists { | ||
| // Add all BackendTrafficPolicies to the resourceTree | ||
| if err = r.processBackendTrafficPolicies(ctx, gwcResource, gwcResourceMapping); err != nil { | ||
| if isTransientError(err) { | ||
| r.log.Error(err, "transient error processing BackendTrafficPolicies", "gatewayClass", managedGC.Name) | ||
| return reconcile.Result{}, err | ||
| } | ||
| r.log.Error(err, fmt.Sprintf("failed processBackendTrafficPolicies for gatewayClass %s, skipping it", managedGC.Name)) | ||
| continue | ||
| } | ||
|
|
@@ -274,49 +329,75 @@ | |
| if r.spCRDExists { | ||
| // Add all SecurityPolicies and their referenced resources to the resourceTree | ||
| if err = r.processSecurityPolicies(ctx, gwcResource, gwcResourceMapping); err != nil { | ||
| if isTransientError(err) { | ||
| r.log.Error(err, "transient error processing SecurityPolicies", "gatewayClass", managedGC.Name) | ||
| return reconcile.Result{}, err | ||
| } | ||
| r.log.Error(err, fmt.Sprintf("failed processSecurityPolicies for gatewayClass %s, skipping it", managedGC.Name)) | ||
| } | ||
| } | ||
|
|
||
| if r.bTLSPolicyCRDExists { | ||
| // Add all BackendTLSPolies to the resourceTree | ||
| if err = r.processBackendTLSPolicies(ctx, gwcResource, gwcResourceMapping); err != nil { | ||
| if isTransientError(err) { | ||
| r.log.Error(err, "transient error processing BackendTLSPolicies", "gatewayClass", managedGC.Name) | ||
| return reconcile.Result{}, err | ||
| } | ||
| r.log.Error(err, fmt.Sprintf("failed processBackendTLSPolicies for gatewayClass %s, skipping it", managedGC.Name)) | ||
| } | ||
| } | ||
|
|
||
| if r.eepCRDExists { | ||
| // Add all EnvoyExtensionPolicies and their referenced resources to the resourceTree | ||
| if err = r.processEnvoyExtensionPolicies(ctx, gwcResource, gwcResourceMapping); err != nil { | ||
| if isTransientError(err) { | ||
| r.log.Error(err, "transient error processing EnvoyExtensionPolicies", "gatewayClass", managedGC.Name) | ||
| return reconcile.Result{}, err | ||
| } | ||
| r.log.Error(err, fmt.Sprintf("failed processEnvoyExtensionPolicies for gatewayClass %s, skipping it", managedGC.Name)) | ||
| } | ||
| } | ||
|
|
||
| if err = r.processExtensionServerPolicies(ctx, gwcResource); err != nil { | ||
| if isTransientError(err) { | ||
| r.log.Error(err, "transient error processing ExtensionServerPolicies", "gatewayClass", managedGC.Name) | ||
| return reconcile.Result{}, err | ||
| } | ||
| r.log.Error(err, fmt.Sprintf("failed processExtensionServerPolicies for gatewayClass %s, skipping it", managedGC.Name)) | ||
| } | ||
|
|
||
| if r.backendCRDExists { | ||
| if err = r.processBackends(ctx, gwcResource); err != nil { | ||
| if isTransientError(err) { | ||
| r.log.Error(err, "transient error processing Backends", "gatewayClass", managedGC.Name) | ||
| return reconcile.Result{}, err | ||
| } | ||
| r.log.Error(err, fmt.Sprintf("failed processBackends for gatewayClass %s, skipping it", managedGC.Name)) | ||
| } | ||
| } | ||
|
|
||
| // Add the referenced services, ServiceImports, and EndpointSlices in | ||
| // the collected BackendRefs to the resourceTree. | ||
| // BackendRefs are referred by various Route objects and the ExtAuth in SecurityPolicies. | ||
| if err = r.processBackendRefs(ctx, gwcResource, gwcResourceMapping); err != nil { | ||
| if isTransientError(err) { | ||
| r.log.Error(err, "transient error processing BackendRefs", "gatewayClass", managedGC.Name) | ||
| return reconcile.Result{}, err | ||
| } | ||
|
|
||
| r.log.Error(err, fmt.Sprintf("failed processBackendRefs for gatewayClass %s, skipping it", managedGC.Name)) | ||
| } | ||
|
|
||
| // For this particular Gateway, and all associated objects, check whether the | ||
| // namespace exists. Add to the resourceTree. | ||
| for ns := range gwcResourceMapping.allAssociatedNamespaces { | ||
| namespace, err := r.getNamespace(ctx, ns) | ||
| if err != nil { | ||
| if isTransientError(err) { | ||
| r.log.Error(err, "transient error getting namespace", "namespace", ns, "gatewayClass", managedGC.Name) | ||
| return reconcile.Result{}, err | ||
| } | ||
| r.log.Error(err, "unable to find the namespace") | ||
| if kerrors.IsNotFound(err) { | ||
| continue | ||
|
|
@@ -335,30 +416,27 @@ | |
| } | ||
| } | ||
|
|
||
| if len(gwcResource.Gateways) == 0 { | ||
| r.log.Info("No gateways found for accepted gatewayClass") | ||
|
|
||
| // If needed, remove the finalizer from the accepted GatewayClass. | ||
| if err := r.removeFinalizer(ctx, managedGC); err != nil { | ||
| if isTransientError(err) { | ||
| r.log.Error(err, "transient error removing finalizer from gatewayClass", "gatewayClass", managedGC.Name) | ||
| return reconcile.Result{}, err | ||
| } | ||
| r.log.Error(err, fmt.Sprintf("failed to remove finalizer from gatewayClass %s", | ||
| managedGC.Name)) | ||
| } | ||
| } else { | ||
| // finalize the accepted GatewayClass. | ||
| if err := r.addFinalizer(ctx, managedGC); err != nil { | ||
| if isTransientError(err) { | ||
| r.log.Error(err, "transient error adding finalizer to gatewayClass", "gatewayClass", managedGC.Name) | ||
| return reconcile.Result{}, err | ||
| } | ||
| r.log.Error(err, fmt.Sprintf("failed adding finalizer to gatewayClass %s", | ||
| managedGC.Name)) | ||
| } | ||
| } | ||
| } | ||
|
|
@@ -373,9 +451,9 @@ | |
| return reconcile.Result{}, nil | ||
| } | ||
|
|
||
| func (r *gatewayAPIReconciler) processEnvoyProxySecretRef(ctx context.Context, gwcResource *resource.Resources) error { | ||
| if gwcResource.EnvoyProxyForGatewayClass == nil || gwcResource.EnvoyProxyForGatewayClass.Spec.BackendTLS == nil || gwcResource.EnvoyProxyForGatewayClass.Spec.BackendTLS.ClientCertificateRef == nil { | ||
| return nil | ||
| } | ||
| certRef := gwcResource.EnvoyProxyForGatewayClass.Spec.BackendTLS.ClientCertificateRef | ||
| if refsSecret(certRef) { | ||
|
|
@@ -387,11 +465,10 @@ | |
| gwcResource.EnvoyProxyForGatewayClass.Namespace, | ||
| resource.KindEnvoyProxy, | ||
| *certRef); err != nil { | ||
| return err | ||
| } | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // managedGatewayClasses returns a list of GatewayClass objects that are managed by the Envoy Gateway Controller. | ||
|
|
@@ -427,7 +504,11 @@ | |
| // - EndpointSlices | ||
| // - Backends | ||
| // - CACertificateRefs in the Backends | ||
| func (r *gatewayAPIReconciler) processBackendRefs(ctx context.Context, gwcResource *resource.Resources, resourceMappings *resourceMappings) error { | ||
| // Only transient errors are returned from this function to allow Reconcile to retry. | ||
| // All other errors are just logged and ignored - these errors result in missing referenced backend resources | ||
| // in the resource tree, which is acceptable as the Gateway API translation layer will handle them. | ||
| // The Gateway API translation layer will surface these errors in the status of the resources referencing them. | ||
| for backendRef := range resourceMappings.allAssociatedBackendRefs { | ||
| backendRefKind := gatewayapi.KindDerefOr(backendRef.Kind, resource.KindService) | ||
| r.log.Info("processing Backend", "kind", backendRefKind, "namespace", string(*backendRef.Namespace), | ||
|
|
@@ -439,6 +520,9 @@ | |
| service := new(corev1.Service) | ||
| err := r.client.Get(ctx, types.NamespacedName{Namespace: string(*backendRef.Namespace), Name: string(backendRef.Name)}, service) | ||
| if err != nil { | ||
| if isTransientError(err) { | ||
|
Contributor
There was a problem hiding this comment. is going to hard to remember to add this for every client call
Member
There was a problem hiding this comment. Yeah, we can figure out a better approach to handle this later. |
||
| return err | ||
| } | ||
| r.log.Error(err, "failed to get Service", "namespace", string(*backendRef.Namespace), | ||
| "name", string(backendRef.Name)) | ||
| } else { | ||
|
|
@@ -453,6 +537,9 @@ | |
| serviceImport := new(mcsapiv1a1.ServiceImport) | ||
| err := r.client.Get(ctx, types.NamespacedName{Namespace: string(*backendRef.Namespace), Name: string(backendRef.Name)}, serviceImport) | ||
| if err != nil { | ||
| if isTransientError(err) { | ||
| return err | ||
| } | ||
| r.log.Error(err, "failed to get ServiceImport", "namespace", string(*backendRef.Namespace), | ||
| "name", string(backendRef.Name)) | ||
| } else { | ||
|
|
@@ -471,6 +558,9 @@ | |
| backend := new(egv1a1.Backend) | ||
| err := r.client.Get(ctx, types.NamespacedName{Namespace: string(*backendRef.Namespace), Name: string(backendRef.Name)}, backend) | ||
| if err != nil { | ||
| if isTransientError(err) { | ||
| return err | ||
| } | ||
| r.log.Error(err, "failed to get Backend", "namespace", string(*backendRef.Namespace), | ||
| "name", string(backendRef.Name)) | ||
| } else { | ||
|
|
@@ -519,6 +609,9 @@ | |
| caRefNew) | ||
| } | ||
| if err != nil { | ||
| if isTransientError(err) { | ||
| return err | ||
| } | ||
| r.log.Error(err, | ||
| "failed to process CACertificateRef for Backend", | ||
| "backend", backend, "caCertificateRef", caCertRef.Name) | ||
|
|
@@ -538,6 +631,9 @@ | |
| client.InNamespace(*backendRef.Namespace), | ||
| } | ||
| if err := r.client.List(ctx, endpointSliceList, opts...); err != nil { | ||
| if isTransientError(err) { | ||
| return err | ||
| } | ||
| r.log.Error(err, "failed to get EndpointSlices", "namespace", string(*backendRef.Namespace), | ||
| backendRefKind, string(backendRef.Name)) | ||
| } else { | ||
|
|
@@ -554,6 +650,7 @@ | |
| } | ||
| } | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // processSecurityPolicyObjectRefs adds the referenced resources in SecurityPolicies | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.