fix: handle "leader changed" errors

Fixes fluxcd/flux2/#4804 by copying the solution used in helm/helm#11426

internal/kube/client.go

@@ -18,6 +18,7 @@ package kube
 
 import (
  "fmt"
+ "net/http"
  "sync"
 
  "k8s.io/apimachinery/pkg/api/meta"
@@ -130,6 +131,10 @@ func (c *MemoryRESTClientGetter) ToRESTConfig() (*rest.Config, error) {
  if c.cfg == nil {
  return nil, fmt.Errorf("MemoryRESTClientGetter has no REST config")
  }
+ // add retries to fix temporary "etcdserver: leader changed" errors from kube-apiserver
+ c.cfg.Wrap(func(rt http.RoundTripper) http.RoundTripper {
+ return &retryingRoundTripper{wrapped: rt}
+ })
  return c.cfg, nil
 }
 

internal/kube/roundtripper.go

@@ -0,0 +1,82 @@
+/*
+Copyright The Helm Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// TODO: this file was taken from https://github.com/helm/helm/blob/main/pkg/cli/roundtripper.go
+// We should be able to get rid of it once https://github.com/helm/helm/issues/13052 is addressed
+package kube
+
+import (
+ "bytes"
+ "encoding/json"
+ "io"
+ "net/http"
+ "strings"
+)
+
+type retryingRoundTripper struct {
+ wrapped http.RoundTripper
+}
+
+func (rt *retryingRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
+ return rt.roundTrip(req, 1, nil)
+}
+
+func (rt *retryingRoundTripper) roundTrip(req *http.Request, retry int, prevResp *http.Response) (*http.Response, error) {
+ if retry < 0 {
+ return prevResp, nil
+ }
+ resp, rtErr := rt.wrapped.RoundTrip(req)
+ if rtErr != nil {
+ return resp, rtErr
+ }
+ if resp.StatusCode < 500 {
+ return resp, rtErr
+ }
+ if resp.Header.Get("content-type") != "application/json" {
+ return resp, rtErr
+ }
+ b, err := io.ReadAll(resp.Body)
+ resp.Body.Close()
+ if err != nil {
+ return resp, rtErr
+ }
+
+ var ke kubernetesError
+ r := bytes.NewReader(b)
+ err = json.NewDecoder(r).Decode(&ke)
+ r.Seek(0, io.SeekStart)
+ resp.Body = io.NopCloser(r)
+ if err != nil {
+ return resp, rtErr
+ }
+ if ke.Code < 500 {
+ return resp, rtErr
+ }
+ // Matches messages like "etcdserver: leader changed"
+ if strings.HasSuffix(ke.Message, "etcdserver: leader changed") {
+ return rt.roundTrip(req, retry-1, resp)
+ }
+ // Matches messages like "rpc error: code = Unknown desc = raft proposal dropped"
+ if strings.HasSuffix(ke.Message, "raft proposal dropped") {
+ return rt.roundTrip(req, retry-1, resp)
+ }
+ return resp, rtErr
+}
+
+type kubernetesError struct {
+ Message string `json:"message"`
+ Code int `json:"code"`
+}