Merge pull request #11426 from cenkalti/leader-changed

Fix handling of "leader changed" errors

go.mod

@@ -31,11 +31,9 @@ require (
 	github.com/spf13/pflag v1.0.5
 	github.com/stretchr/testify v1.8.0
 	github.com/xeipuuv/gojsonschema v1.2.0
-	go.etcd.io/etcd/api/v3 v3.5.4
 	golang.org/x/crypto v0.0.0-20220525230936-793ad666bf5e
 	golang.org/x/term v0.0.0-20210927222741-03fcf44c2211
 	golang.org/x/text v0.3.7
-	google.golang.org/grpc v1.47.0
 	k8s.io/api v0.25.2
 	k8s.io/apiextensions-apiserver v0.25.2
 	k8s.io/apimachinery v0.25.2
@@ -151,6 +149,7 @@ require (
 	golang.org/x/time v0.0.0-20220210224613-90d013bbcef8 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
 	google.golang.org/genproto v0.0.0-20220502173005-c8bf987b8c21 // indirect
+	google.golang.org/grpc v1.47.0 // indirect
 	google.golang.org/protobuf v1.28.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect

go.sum

@@ -621,8 +621,6 @@ github.com/yvasiyarov/newrelic_platform_go v0.0.0-20140908184405-b21fdbd4370f/go
 github.com/ziutek/mymysql v1.5.4 h1:GB0qdRGsTwQSBVYuVShFBKaXSnSnYYC2d9knnE1LHFs=
 github.com/ziutek/mymysql v1.5.4/go.mod h1:LMSpPZ6DbqWFxNCHW77HeMg9I646SAhApZ/wKdgO/C0=
 go.etcd.io/etcd/api/v3 v3.5.0/go.mod h1:cbVKeC6lCfl7j/8jBhAK6aIYO9XOjdptoxU/nLQcPvs=
-go.etcd.io/etcd/api/v3 v3.5.4 h1:OHVyt3TopwtUQ2GKdd5wu3PmmipR4FTwCqoEjSyRdIc=
-go.etcd.io/etcd/api/v3 v3.5.4/go.mod h1:5GB2vv4A4AOn3yk7MftYGHkUfGtDHnEraIjym4dYz5A=
 go.etcd.io/etcd/client/pkg/v3 v3.5.0/go.mod h1:IJHfcCEKxYu1Os13ZdwCwIUTUVGYTSAM3YSwc9/Ac1g=
 go.etcd.io/etcd/client/v2 v2.305.0/go.mod h1:h9puh54ZTgAKtEbut2oe9P4L/oqKCVB6xsXlzd7alYQ=
 go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU=

pkg/cli/environment.go

@@ -24,6 +24,7 @@ package cli
 
 import (
 	"fmt"
+	"net/http"
 	"os"
 	"strconv"
 	"strings"
@@ -116,6 +117,9 @@ func New() *EnvSettings {
 		ImpersonateGroup: &env.KubeAsGroups,
 		WrapConfigFn: func(config *rest.Config) *rest.Config {
 			config.Burst = env.BurstLimit
+			config.Wrap(func(rt http.RoundTripper) http.RoundTripper {
+				return &retryingRoundTripper{wrapped: rt}
+			})
 			return config
 		},
 	}

pkg/cli/roundtripper.go

@@ -0,0 +1,77 @@
+/*
+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.
+*/
+
+package cli
+
+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.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"`
+}

pkg/kube/wait.go

@@ -22,9 +22,6 @@ import (
 	"time"
 
 	"github.com/pkg/errors"
-	"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
-	"google.golang.org/grpc/codes"
-
 	appsv1 "k8s.io/api/apps/v1"
 	appsv1beta1 "k8s.io/api/apps/v1beta1"
 	appsv1beta2 "k8s.io/api/apps/v1beta2"
@@ -35,6 +32,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
+
 	"k8s.io/apimachinery/pkg/util/wait"
 )
 
@@ -44,22 +42,6 @@ type waiter struct {
 	log     func(string, ...interface{})
 }
 
-// isServiceUnavailable helps figure out if the error is caused by etcd not being available
-// see https://pkg.go.dev/go.etcd.io/etcd/api/v3/v3rpc/rpctypes for `codes.Unavailable`
-// we use this to check if the etcdserver is not available we should retry in case
-// this is a temporary situation
-func isServiceUnavailable(err error) bool {
-	if err != nil {
-		err = rpctypes.Error(err)
-		if ev, ok := err.(rpctypes.EtcdError); ok {
-			if ev.Code() == codes.Unavailable {
-				return true
-			}
-		}
-	}
-	return false
-}
-
 // waitForResources polls to get the current status of all pods, PVCs, Services and
 // Jobs(optional) until all are ready or a timeout is reached
 func (w *waiter) waitForResources(created ResourceList) error {
@@ -72,9 +54,6 @@ func (w *waiter) waitForResources(created ResourceList) error {
 		for _, v := range created {
 			ready, err := w.c.IsReady(ctx, v)
 			if !ready || err != nil {
-				if isServiceUnavailable(err) {
-					return false, nil
-				}
 				return false, err
 			}
 		}
@@ -93,9 +72,6 @@ func (w *waiter) waitForDeletedResources(deleted ResourceList) error {
 		for _, v := range deleted {
 			err := v.Get()
 			if err == nil || !apierrors.IsNotFound(err) {
-				if isServiceUnavailable(err) {
-					return false, nil
-				}
 				return false, err
 			}
 		}

pkg/kube/wait_test.go

@@ -1,42 +0,0 @@
-/*
-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.
-*/
-
-package kube // import "helm.sh/helm/v3/pkg/kube"
-
-import (
-	"errors"
-	"testing"
-
-	"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
-)
-
-func Test_isServiceUnavailable(t *testing.T) {
-	tests := []struct {
-		err    error
-		expect bool
-	}{
-		{err: nil, expect: false},
-		{err: errors.New("random error from somewhere"), expect: false},
-		{err: rpctypes.ErrGRPCLeaderChanged, expect: true},
-		{err: errors.New("etcdserver: leader changed"), expect: true},
-	}
-
-	for _, tt := range tests {
-		if isServiceUnavailable(tt.err) != tt.expect {
-			t.Errorf("failed test for %q (expect equal: %t)", tt.err, tt.expect)
-		}
-	}
-}