Add http retry

cmd/pke/app/phases/kubeadm/controlplane/controlplane.go

@@ -17,10 +17,12 @@ package controlplane
 import (
 	"context"
 	"crypto/rand"
+	"crypto/tls"
 	"encoding/base64"
 	"fmt"
 	"io"
 	"io/ioutil"
+	"net"
 	"net/http"
 	"net/url"
 	"os"
@@ -30,6 +32,7 @@ import (
 	"time"
 
 	"github.com/Masterminds/semver"
+	"github.com/PuerkitoBio/rehttp"
 	"github.com/banzaicloud/pke/.gen/pipeline"
 	"github.com/banzaicloud/pke/cmd/pke/app/constants"
 	"github.com/banzaicloud/pke/cmd/pke/app/phases"
@@ -355,6 +358,13 @@ func (c *ControlPlane) Run(out io.Writer) error {
 	if c.clusterMode == "ha" {
 		// additional master node
 		if c.joinControlPlane {
+			// make sure api server stabilized operation and not restarting
+			ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
+			defer cancel()
+			if err := ensureAPIServerConnection(out, ctx, 5, c.apiServerHostPort); err != nil {
+				return err
+			}
+			// install additional master node
 			_, _ = fmt.Fprintf(out, "[%s] installing additional master node\n", c.Use())
 			return c.node.Run(out)
 		}
@@ -384,6 +394,58 @@ func (c *ControlPlane) Run(out io.Writer) error {
 	return taintRemoveNoSchedule(out, c.clusterMode, kubeConfig)
 }
 
+func ensureAPIServerConnection(out io.Writer, ctx context.Context, successTries int, apiServerHostPort string) error {
+	host, port, err := kubeadm.SplitHostPort(apiServerHostPort, "6443")
+	if err != nil {
+		return err
+	}
+	apiServerHostPort = net.JoinHostPort(host, port)
+
+	insecureTLS := &http.Transport{
+		TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
+	}
+	tl := &pipelineutil.TransportLogger{
+		RoundTripper: insecureTLS,
+		Output:       out,
+	}
+
+	tr := rehttp.NewTransport(
+		tl,
+		rehttp.RetryAll(
+			rehttp.RetryMaxRetries(4),
+			rehttp.RetryHTTPMethods(http.MethodGet),
+			rehttp.RetryStatusInterval(400, 600),
+		),
+		rehttp.ExpJitterDelay(2*time.Second, 30*time.Second),
+	)
+
+	c := &http.Client{Transport: tr}
+
+	ticker := time.NewTicker(2 * time.Second)
+	defer ticker.Stop()
+	u := "https://" + apiServerHostPort + "/version"
+	for {
+		select {
+		case <-ticker.C:
+			resp, err := c.Get(u)
+			if err != nil {
+				return err
+			}
+			_ = resp.Body.Close()
+			if resp.StatusCode/100 == 2 {
+				successTries--
+				if successTries == 0 {
+					return nil
+				}
+			} else {
+				successTries++
+			}
+		case <-ctx.Done():
+			return errors.Wrapf(ctx.Err(), "api server connection cloud not be stabilized")
+		}
+	}
+}
+
 func (c *ControlPlane) masterBootstrapParameters(cmd *cobra.Command) (err error) {
 	c.kubernetesVersion, err = cmd.Flags().GetString(constants.FlagKubernetesVersion)
 	if err != nil {

cmd/pke/app/phases/kubeadm/controlplane/controlplane_test.go

@@ -15,9 +15,11 @@
 package controlplane
 
 import (
+	"context"
 	"io/ioutil"
 	"os"
 	"testing"
+	"time"
 
 	"github.com/banzaicloud/pke/cmd/pke/app/constants"
 	"github.com/stretchr/testify/require"
@@ -64,3 +66,11 @@ func TestWriteKubeadmAmazonConfig(t *testing.T) {
 	require.NoError(t, err)
 	t.Logf("%s\n", b)
 }
+
+func TestEnsureAPIServerConnection(t *testing.T) {
+	t.SkipNow()
+	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+	defer cancel()
+	err := ensureAPIServerConnection(os.Stdout, ctx, 5, "192.168.64.11")
+	require.NoError(t, err)
+}

cmd/pke/app/util/pipeline/client.go

@@ -17,8 +17,10 @@ package pipeline
 import (
 	"context"
 	"io"
+	"net/http"
 	"time"
 
+	"github.com/PuerkitoBio/rehttp"
 	"github.com/banzaicloud/pke/.gen/pipeline"
 	"github.com/banzaicloud/pke/cmd/pke/app/constants"
 	"github.com/banzaicloud/pke/cmd/pke/app/util/validator"
@@ -35,10 +37,27 @@ func Client(out io.Writer, endpoint, token string) *pipeline.APIClient {
 		&oauth2.Token{AccessToken: token},
 	))
 	config.HTTPClient.Timeout = 30 * time.Second
-	config.HTTPClient.Transport = &transportLogger{
-		roundTripper: config.HTTPClient.Transport,
-		output:       out,
+	tl := &TransportLogger{
+		RoundTripper: config.HTTPClient.Transport,
+		Output:       out,
 	}
+	tr := rehttp.NewTransport(
+		tl,
+		rehttp.RetryAny(
+			rehttp.RetryAll(
+				rehttp.RetryMaxRetries(5),
+				rehttp.RetryHTTPMethods(http.MethodGet),
+				rehttp.RetryStatusInterval(400, 600),
+			),
+			rehttp.RetryAll(
+				rehttp.RetryMaxRetries(5),
+				rehttp.RetryHTTPMethods(http.MethodPost),
+				rehttp.RetryStatusInterval(500, 600),
+			),
+		),
+		rehttp.ExpJitterDelay(2*time.Second, 30*time.Second),
+	)
+	config.HTTPClient.Transport = tr
 
 	return pipeline.NewAPIClient(config)
 }

cmd/pke/app/util/pipeline/transport.go

@@ -23,16 +23,16 @@ import (
 	"time"
 )
 
-type transportLogger struct {
-	roundTripper http.RoundTripper
-	output       io.Writer
+type TransportLogger struct {
+	RoundTripper http.RoundTripper
+	Output       io.Writer
 }
 
-func (t *transportLogger) RoundTrip(req *http.Request) (*http.Response, error) {
+func (t *TransportLogger) RoundTrip(req *http.Request) (*http.Response, error) {
 	ctx := context.WithValue(req.Context(), "requestTS", time.Now())
 	req = req.WithContext(ctx)
 
-	_, _ = fmt.Fprintf(t.output, "%s --> %s %q\n", req.Proto, req.Method, req.URL)
+	_, _ = fmt.Fprintf(t.Output, "%s --> %s %q\n", req.Proto, req.Method, req.URL)
 
 	resp, err := t.transport().RoundTrip(req)
 	if err != nil {
@@ -41,23 +41,22 @@ func (t *transportLogger) RoundTrip(req *http.Request) (*http.Response, error) {
 
 	ctx = resp.Request.Context()
 	if ts, ok := ctx.Value("requestTS").(time.Time); ok {
-		_, _ = fmt.Fprintf(t.output, "%s <-- %d %q %s\n", resp.Proto, resp.StatusCode, resp.Request.URL, time.Now().Sub(ts))
+		_, _ = fmt.Fprintf(t.Output, "%s <-- %d %q %s\n", resp.Proto, resp.StatusCode, resp.Request.URL, time.Now().Sub(ts))
 	} else {
-		_, _ = fmt.Fprintf(t.output, "%s <-- %d %q\n", resp.Proto, resp.StatusCode, resp.Request.URL)
-		if resp.StatusCode/100 != 2 {
-			if b, err := httputil.DumpResponse(resp, true); err != nil {
-				_, _ = fmt.Fprintf(t.output, "%s\n", b)
-			}
-
+		_, _ = fmt.Fprintf(t.Output, "%s <-- %d %q\n", resp.Proto, resp.StatusCode, resp.Request.URL)
+	}
+	if resp != nil && resp.StatusCode/100 != 2 {
+		if b, err := httputil.DumpResponse(resp, true); err == nil {
+			_, _ = fmt.Fprintf(t.Output, "%s\n", b)
 		}
 	}
 
 	return resp, err
 
 }
-func (t *transportLogger) transport() http.RoundTripper {
-	if t.roundTripper != nil {
-		return t.roundTripper
+func (t *TransportLogger) transport() http.RoundTripper {
+	if t.RoundTripper != nil {
+		return t.RoundTripper
 	}
 
 	return http.DefaultTransport

go.mod

@@ -4,6 +4,7 @@ go 1.12
 
 require (
 	github.com/Masterminds/semver v1.4.2
+	github.com/PuerkitoBio/rehttp v0.0.0-20180310210549-11cf6ea5d3e9
 	github.com/antihax/optional v0.0.0-20180407024304-ca021399b1a6
 	github.com/cpuguy83/go-md2man v1.0.8 // indirect
 	github.com/ghodss/yaml v1.0.0

go.sum

@@ -23,6 +23,8 @@ github.com/Microsoft/go-winio v0.4.12/go.mod h1:VhR8bwka0BXejwEJY73c50VrPtXAaKcy
 github.com/NYTimes/gziphandler v1.1.1/go.mod h1:n/CVRwUEOgIxrgPvAQhUUr9oeUtvrhMomdKFjzJNB0c=
 github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5/go.mod h1:lmUJ/7eu/Q8D7ML55dXQrVaamCz2vxCfdQBasLZfHKk=
 github.com/PuerkitoBio/purell v1.1.0/go.mod h1:c11w/QuzBsJSee3cPx9rAFu61PvFxuPbtSwDGJws/X0=
+github.com/PuerkitoBio/rehttp v0.0.0-20180310210549-11cf6ea5d3e9 h1:VE0eMvNSQI72dADsq4gm5KpNPmt97WgqneTfaS5MWrs=
+github.com/PuerkitoBio/rehttp v0.0.0-20180310210549-11cf6ea5d3e9/go.mod h1:ItsOiHl4XeMOV3rzbZqQRjLc3QQxbE6391/9iNG7rE8=
 github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578/go.mod h1:uGdkoq3SwY9Y+13GIhn11/XLaGBb4BfwItxLd5jeuXE=
 github.com/SAP/go-hdb v0.13.2/go.mod h1:etBT+FAi1t5k3K3tf5vQTnosgYmhDkRi8jEnQqCnxF0=
 github.com/SermoDigital/jose v0.9.1/go.mod h1:ARgCUhI1MHQH+ONky/PAtmVHQrP5JlGY0F3poXOp/fA=