From b5378b3a5dd435e5c364ac0cfa717112ad686bd0 Mon Sep 17 00:00:00 2001 From: Cenk Alti Date: Mon, 10 Oct 2022 21:55:18 -0400 Subject: [PATCH] retry http request on temporary errors Signed-off-by: Cenk Alti --- pkg/cli/environment.go | 4 +++ pkg/cli/roundtripper.go | 77 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 81 insertions(+) create mode 100644 pkg/cli/roundtripper.go 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"` +}