Merge pull request #3238 from TheOriginalAlex/grpc-wait-for-ready

Added a --wait flag to helm init

cmd/helm/init.go

@@ -31,6 +31,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/yaml"
 	"k8s.io/helm/cmd/helm/installer"
 	"k8s.io/helm/pkg/getter"
+	"k8s.io/helm/pkg/helm"
 	"k8s.io/helm/pkg/helm/helmpath"
 	"k8s.io/helm/pkg/repo"
 )
@@ -79,11 +80,13 @@ type initCmd struct {
 	forceUpgrade   bool
 	skipRefresh    bool
 	out            io.Writer
+	client         helm.Interface
 	home           helmpath.Home
 	opts           installer.Options
 	kubeClient     kubernetes.Interface
 	serviceAccount string
 	maxHistory     int
+	wait           bool
 }
 
 func newInitCmd(out io.Writer) *cobra.Command {
@@ -99,6 +102,8 @@ func newInitCmd(out io.Writer) *cobra.Command {
 			}
 			i.namespace = settings.TillerNamespace
 			i.home = settings.Home
+			i.client = ensureHelmClient(i.client)
+
 			return i.run()
 		},
 	}
@@ -111,6 +116,7 @@ func newInitCmd(out io.Writer) *cobra.Command {
 	f.BoolVarP(&i.clientOnly, "client-only", "c", false, "if set does not install Tiller")
 	f.BoolVar(&i.dryRun, "dry-run", false, "do not install local or remote")
 	f.BoolVar(&i.skipRefresh, "skip-refresh", false, "do not refresh (download) the local repository cache")
+	f.BoolVar(&i.wait, "wait", false, "block until Tiller is running and ready to receive requests")
 
 	f.BoolVar(&tlsEnable, "tiller-tls", false, "install Tiller with TLS enabled")
 	f.BoolVar(&tlsVerify, "tiller-tls-verify", false, "install Tiller with TLS enabled and to verify remote certificates")
@@ -292,12 +298,18 @@ func (i *initCmd) run() error {
 				if err := installer.Upgrade(i.kubeClient, &i.opts); err != nil {
 					return fmt.Errorf("error when upgrading: %s", err)
 				}
+				if err := i.ping(); err != nil {
+					return err
+				}
 				fmt.Fprintln(i.out, "\nTiller (the Helm server-side component) has been upgraded to the current version.")
 			} else {
 				fmt.Fprintln(i.out, "Warning: Tiller is already installed in the cluster.\n"+
 					"(Use --client-only to suppress this message, or --upgrade to upgrade Tiller to the current version.)")
 			}
 		} else {
+			if err := i.ping(); err != nil {
+				return err
+			}
 			fmt.Fprintln(i.out, "\nTiller (the Helm server-side component) has been installed into your Kubernetes Cluster.")
 		}
 	} else {
@@ -308,6 +320,16 @@ func (i *initCmd) run() error {
 	return nil
 }
 
+func (i *initCmd) ping() error {
+	if i.wait {
+		if err := i.client.PingTiller(); err != nil {
+			return fmt.Errorf("could not ping Tiller: %s", err)
+		}
+	}
+
+	return nil
+}
+
 // ensureDirectories checks to see if $HELM_HOME exists.
 //
 // If $HELM_HOME does not exist, this function will create it.

docs/helm/helm_init.md

@@ -53,6 +53,7 @@ helm init
       --tiller-tls-verify        install Tiller with TLS enabled and to verify remote certificates
       --tls-ca-cert string       path to CA root certificate
       --upgrade                  upgrade if Tiller is already installed
+      --wait                     block until Tiller is running and ready to receive requests
 ```
 
 ### Options inherited from parent commands

pkg/helm/client.go

@@ -293,6 +293,12 @@ func (h *Client) RunReleaseTest(rlsName string, opts ...ReleaseTestOption) (<-ch
 	return h.test(ctx, req)
 }
 
+// PingTiller pings the Tiller pod and ensure's that it is up and runnning
+func (h *Client) PingTiller() error {
+	ctx := NewContext()
+	return h.ping(ctx)
+}
+
 // connect returns a gRPC connection to Tiller or error. The gRPC dial options
 // are constructed here.
 func (h *Client) connect(ctx context.Context) (conn *grpc.ClientConn, err error) {
@@ -467,3 +473,15 @@ func (h *Client) test(ctx context.Context, req *rls.TestReleaseRequest) (<-chan
 
 	return ch, errc
 }
+
+// Executes tiller.Ping RPC.
+func (h *Client) ping(ctx context.Context) error {
+	c, err := h.connect(ctx)
+	if err != nil {
+		return err
+	}
+	defer c.Close()
+
+	rlc := rls.NewReleaseServiceClient(c)
+	return rlc.PingTiller(ctx)
+}

pkg/helm/fake.go

@@ -184,6 +184,11 @@ func (c *FakeClient) RunReleaseTest(rlsName string, opts ...ReleaseTestOption) (
 	return results, errc
 }
 
+// PingTiller pings the Tiller pod and ensure's that it is up and runnning
+func (c *FakeClient) PingTiller() error {
+	return nil
+}
+
 // MockHookTemplate is the hook template used for all mock release objects.
 var MockHookTemplate = `apiVersion: v1
 kind: Job

pkg/helm/interface.go

@@ -35,4 +35,5 @@ type Interface interface {
 	ReleaseHistory(rlsName string, opts ...HistoryOption) (*rls.GetHistoryResponse, error)
 	GetVersion(opts ...VersionOption) (*rls.GetVersionResponse, error)
 	RunReleaseTest(rlsName string, opts ...ReleaseTestOption) (<-chan *rls.TestReleaseResponse, <-chan error)
+	PingTiller() error
 }

pkg/proto/hapi/services/tiller.pb.go

@@ -949,6 +949,8 @@ type ReleaseServiceClient interface {
 	GetHistory(ctx context.Context, in *GetHistoryRequest, opts ...grpc.CallOption) (*GetHistoryResponse, error)
 	// RunReleaseTest executes the tests defined of a named release
 	RunReleaseTest(ctx context.Context, in *TestReleaseRequest, opts ...grpc.CallOption) (ReleaseService_RunReleaseTestClient, error)
+	// PingTiller sends a test/ping signal to Tiller to ensure that it's up
+	PingTiller(ctx context.Context) error
 }
 
 type releaseServiceClient struct {
@@ -1078,6 +1080,14 @@ func (c *releaseServiceClient) RunReleaseTest(ctx context.Context, in *TestRelea
 	return x, nil
 }
 
+func (c *releaseServiceClient) PingTiller(ctx context.Context) error {
+	err := grpc.Invoke(ctx, "/hapi.services.tiller.ReleaseService/PingTiller", "Ping", nil, c.cc, grpc.FailFast(false))
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
 type ReleaseService_RunReleaseTestClient interface {
 	Recv() (*TestReleaseResponse, error)
 	grpc.ClientStream
@@ -1300,6 +1310,10 @@ func _ReleaseService_RunReleaseTest_Handler(srv interface{}, stream grpc.ServerS
 	return srv.(ReleaseServiceServer).RunReleaseTest(m, &releaseServiceRunReleaseTestServer{stream})
 }
 
+func _ReleaseService_Ping_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	return "Pong", nil
+}
+
 type ReleaseService_RunReleaseTestServer interface {
 	Send(*TestReleaseResponse) error
 	grpc.ServerStream
@@ -1349,6 +1363,10 @@ var _ReleaseService_serviceDesc = grpc.ServiceDesc{
 			MethodName: "GetHistory",
 			Handler:    _ReleaseService_GetHistory_Handler,
 		},
+		{
+			MethodName: "PingTiller",
+			Handler:    _ReleaseService_Ping_Handler,
+		},
 	},
 	Streams: []grpc.StreamDesc{
 		{