Merge pull request #2746 from buildkite/retry-k8s-socket

Use roko to retry k8s socket dial

internal/job/executor.go

@@ -1184,7 +1184,7 @@ func (e *Executor) startKubernetesClient(ctx context.Context, kubernetesClient *
 			return fmt.Errorf("failed to parse container id, %s", os.Getenv("BUILDKITE_CONTAINER_ID"))
 		}
 		kubernetesClient.ID = id
-		connect, err := kubernetesClient.Connect()
+		connect, err := kubernetesClient.Connect(ctx)
 		if err != nil {
 			return err
 		}

kubernetes/kubernetes.go

@@ -17,6 +17,7 @@ import (
 
 	"github.com/buildkite/agent/v3/logger"
 	"github.com/buildkite/agent/v3/process"
+	"github.com/buildkite/roko"
 )
 
 func init() {
@@ -287,11 +288,20 @@ type Client struct {
 
 var errNotConnected = errors.New("client not connected")
 
-func (c *Client) Connect() (RegisterResponse, error) {
+func (c *Client) Connect(ctx context.Context) (RegisterResponse, error) {
 	if c.SocketPath == "" {
 		c.SocketPath = defaultSocketPath
 	}
-	client, err := rpc.DialHTTP("unix", c.SocketPath)
+
+	// Because k8s might run the containers "out of order", the server socket
+	// might not exist yet. Try to connect several times.
+	r := roko.NewRetrier(
+		roko.WithMaxAttempts(30),
+		roko.WithStrategy(roko.Constant(time.Second)),
+	)
+	client, err := roko.DoFunc(ctx, r, func(*roko.Retrier) (*rpc.Client, error) {
+		return rpc.DialHTTP("unix", c.SocketPath)
+	})
 	if err != nil {
 		return RegisterResponse{}, err
 	}

kubernetes/kubernetes_test.go

@@ -171,6 +171,6 @@ func intptr(x int) *int {
 
 // helper for ignoring the response from regular client.Connect
 func connect(c *Client) error {
-	_, err := c.Connect()
+	_, err := c.Connect(context.Background())
 	return err
 }