Repository: helm/helm
Pull request #11426 (2022-10-11 01:49:39)

Fix handling of "leader changed" errors
<!--  Thanks for sending a pull request!  Here are some tips for you:
1. Make sure to read the Contributing Guide before submitting your PR: https://github.com/helm/helm/blob/main/CONTRIBUTING.md
2. If this PR closes another issue, add 'closes #<issue number>' somewhere in the PR summary. GitHub will automatically close that issue when this PR gets merged. Alternatively, adding 'refs #<issue number>' will not close the issue, but help provide the reviewer more context.-->

**What this PR does / why we need it**:
This PR aims to fix temporary "etcdserver: leader changed" errors from kube-apiserver that is previously attempted by #11401 by adding a single retry when this kind of error is detected.

/cc @dims 

**Special notes for your reviewer**:
I also reverted the previous fix that didn't solve the issue. One commit is the revert, the other one is the new fix. It's better if you review commits separately.

**If applicable**:
- [ ] this PR contains documentation
- [ ] this PR contains unit tests
- [ ] this PR has been tested for backwards compatibility


dims: @hickeyma @technosophos @mattfarina Folks, @cenkalti tested a bunch of scenarios recreating the problem(s) with etcd with this patch.

Patch:
diff --git a/go.mod b/go.mod
index cc29497a0..8e614e085 100644
--- a/go.mod
+++ b/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
diff --git a/go.sum b/go.sum
index fe7960aa8..c9a66a641 100644
--- a/go.sum
+++ b/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=
diff --git a/pkg/cli/environment.go b/pkg/cli/environment.go
index ac3093629..ef04ed6bd 100644
--- a/pkg/cli/environment.go
+++ b/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
 		},
 	}
diff --git a/pkg/cli/roundtripper.go b/pkg/cli/roundtripper.go
new file mode 100644
index 000000000..ab90ed2e1
--- /dev/null
+++ b/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"`
+}
diff --git a/pkg/kube/wait.go b/pkg/kube/wait.go
index fd01e9bc7..8928d6745 100644
--- a/pkg/kube/wait.go
+++ b/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
 			}
 		}


Test patch:
diff --git a/pkg/kube/wait_test.go b/pkg/kube/wait_test.go
deleted file mode 100644
index 5f18e49ce..000000000
--- a/pkg/kube/wait_test.go
+++ /dev/null
@@ -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)
-		}
-	}
-}