feat(service create): Wait for a service to be ready when its created

* By default wait for a service to be in condition "Ready"
* Use "--async" for returning immediately (which is the current behaviour)
* --wait-timeout to specify how long to wait for Ready condition (in seconds, default: 60s)
* In sync mode, print out the service URL as a last line (so that it can be used together with `tail -1`) to extract the service URL after the service is created.
* Refactored service create code to use smaller methods which are easier to understand

Current tests has been adapted to work with async behaviour but no
new tests has been added for the new sync mode.
This is coming in a second commit (and put into an extra test file).

Fixes knative#54

pkg/kn/commands/configuration_edit_flags.go

@@ -30,6 +30,8 @@ type ConfigurationEditFlags struct {
 	Env                        []string
 	RequestsFlags, LimitsFlags ResourceFlags
 	ForceCreate                bool
+	Async                      bool
+	WaitTimeout                 int
 }
 
 type ResourceFlags struct {
@@ -51,6 +53,8 @@ func (p *ConfigurationEditFlags) AddUpdateFlags(command *cobra.Command) {
 func (p *ConfigurationEditFlags) AddCreateFlags(command *cobra.Command) {
 	p.AddUpdateFlags(command)
 	command.Flags().BoolVar(&p.ForceCreate, "force", false, "Create service forcefully, replaces existing service if any.")
+	command.Flags().BoolVar(&p.Async, "async", false, "Don't wait for the service to become Ready.")
+	command.Flags().IntVar(&p.WaitTimeout, "wait-timeout", 60, "Wait for that many seconds before giving up when creating a service.")
 	command.MarkFlagRequired("image")
 }
 

pkg/kn/commands/service_create.go

@@ -17,14 +17,24 @@ package commands
 import (
 	"errors"
 	"fmt"
+	"github.com/knative/pkg/apis"
+	"io"
+	api_errors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/watch"
+	"time"
 
+	serving_lib "github.com/knative/client/pkg/serving"
+	serving_v1alpha1_api "github.com/knative/serving/pkg/apis/serving/v1alpha1"
+	serving_v1alpha1_client "github.com/knative/serving/pkg/client/clientset/versioned/typed/serving/v1alpha1"
 	servingv1alpha1 "github.com/knative/serving/pkg/apis/serving/v1alpha1"
 	"github.com/spf13/cobra"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
+
 func NewServiceCreateCommand(p *KnParams) *cobra.Command {
 	var editFlags ConfigurationEditFlags
 
@@ -52,66 +62,190 @@ func NewServiceCreateCommand(p *KnParams) *cobra.Command {
 
 		RunE: func(cmd *cobra.Command, args []string) (err error) {
 			if len(args) != 1 {
-				return errors.New("requires the service name.")
+				return errors.New("'service create' requires the service name given as single argument")
 			}
+			name := args[0]
 			if editFlags.Image == "" {
-				return errors.New("requires the image name to run.")
+				return errors.New("'service create' requires the image name to run provided with the --image option")
 			}
 
 			namespace, err := GetNamespace(cmd)
 			if err != nil {
 				return err
 			}
 
-			service := servingv1alpha1.Service{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      args[0],
-					Namespace: namespace,
-				},
+			service, err := constructService(cmd, editFlags, name, namespace)
+			if err != nil {
+				return err
 			}
 
-			service.Spec.DeprecatedRunLatest = &servingv1alpha1.RunLatestType{
-				Configuration: servingv1alpha1.ConfigurationSpec{
-					DeprecatedRevisionTemplate: &servingv1alpha1.RevisionTemplateSpec{
-						Spec: servingv1alpha1.RevisionSpec{
-							DeprecatedContainer: &corev1.Container{},
-						},
-					},
-				},
+			client, err := p.ServingFactory()
+			if err != nil {
+				return err
 			}
 
-			err = editFlags.Apply(&service, cmd)
+			serviceExists, err := serviceExists(client, service.Name, namespace)
 			if err != nil {
 				return err
 			}
-			client, err := p.ServingFactory()
+
+			if editFlags.ForceCreate && serviceExists {
+				err = replaceService(client, service, namespace, cmd.OutOrStdout())
+			} else {
+				err = createService(client, service, namespace, cmd.OutOrStdout())
+			}
 			if err != nil {
 				return err
 			}
-			var serviceExists bool = false
-			if editFlags.ForceCreate {
-				existingService, err := client.Services(namespace).Get(args[0], v1.GetOptions{})
-				if err == nil {
-					serviceExists = true
-					service.ResourceVersion = existingService.ResourceVersion
-					_, err = client.Services(namespace).Update(&service)
-					if err != nil {
-						return err
-					}
-					fmt.Fprintf(cmd.OutOrStdout(), "Service '%s' successfully replaced in namespace '%s'.\n", args[0], namespace)
-				}
-			}
-			if !serviceExists {
-				_, err = client.Services(namespace).Create(&service)
-				if err != nil {
+
+			if !editFlags.Async {
+				if err := waitForService(client, name, namespace, cmd.OutOrStdout(), editFlags.WaitTimeout); err != nil {
 					return err
 				}
-				fmt.Fprintf(cmd.OutOrStdout(), "Service '%s' successfully created in namespace '%s'.\n", args[0], namespace)
+				return showUrl(client, name, namespace, cmd.OutOrStdout())
 			}
+
 			return nil
 		},
 	}
 	AddNamespaceFlags(serviceCreateCommand.Flags(), false)
 	editFlags.AddCreateFlags(serviceCreateCommand)
 	return serviceCreateCommand
 }
+
+func createService(client serving_v1alpha1_client.ServingV1alpha1Interface, service *serving_v1alpha1_api.Service, namespace string, out io.Writer) error {
+	_, err := client.Services(namespace).Create(service)
+	if err != nil {
+		return err
+	}
+	fmt.Fprintf(out, "Service '%s' successfully created in namespace '%s'.\n", service.Name, namespace)
+	return nil
+}
+
+func replaceService(client serving_v1alpha1_client.ServingV1alpha1Interface, service *serving_v1alpha1_api.Service, namespace string, out io.Writer) error {
+	existingService, err := client.Services(namespace).Get(service.Name, v1.GetOptions{})
+	if err != nil {
+		return err
+	}
+	service.ResourceVersion = existingService.ResourceVersion
+	_, err = client.Services(namespace).Update(service)
+	if err != nil {
+		return err
+	}
+	fmt.Fprintf(out, "Service '%s' successfully replaced in namespace '%s'.\n", service.Name, namespace)
+	return nil
+}
+
+func serviceExists(client serving_v1alpha1_client.ServingV1alpha1Interface, name string, namespace string) (bool, error) {
+	_, err := client.Services(namespace).Get(name, v1.GetOptions{})
+	if api_errors.IsNotFound(err) {
+		return false, nil
+	}
+	if err != nil {
+		return false, err
+	}
+	return true, nil
+}
+
+// Create service struct from provided options
+func constructService(cmd *cobra.Command, editFlags ConfigurationEditFlags, name string, namespace string) (*serving_v1alpha1_api.Service,
+	error) {
+
+	service := serving_v1alpha1_api.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: namespace,
+		},
+	}
+
+	// TODO: Should it always be `runLatest` ?
+	service.Spec.DeprecatedRunLatest = &serving_v1alpha1_api.RunLatestType{
+		Configuration: serving_v1alpha1_api.ConfigurationSpec{
+			DeprecatedRevisionTemplate: &serving_v1alpha1_api.RevisionTemplateSpec{
+				Spec: serving_v1alpha1_api.RevisionSpec{
+					DeprecatedContainer: &corev1.Container{},
+				},
+			},
+		},
+	}
+
+	config, err := serving_lib.GetConfiguration(&service)
+	if err != nil {
+		return nil, err
+	}
+	err = editFlags.Apply(config, cmd)
+	if err != nil {
+		return nil, err
+	}
+	return &service, nil
+}
+
+func showUrl(client serving_v1alpha1_client.ServingV1alpha1Interface, serviceName string, namespace string, out io.Writer) error {
+	service, err := client.Services(namespace).Get(serviceName,v1.GetOptions{})
+	if err != nil {
+		return fmt.Errorf("cannot fetch service %s in namespace %s for extracting the URL", serviceName, namespace)
+	}
+	url := service.Status.URL.String()
+	if url == "" {
+		url = service.Status.DeprecatedDomain
+	}
+	fmt.Fprintln(out, "\nService URL:")
+	fmt.Fprintf(out, "%s\n", url)
+	return nil
+}
+
+// Duck type for writers having a flush
+type flusher interface {
+	Flush() error
+}
+
+func waitForService(client serving_v1alpha1_client.ServingV1alpha1Interface, serviceName string, namespace string, out io.Writer, timeout int) error {
+	// Wait for service to enter 'Ready' state, with a timeout of which is slightly larger than the provided
+	// timeout. We have our own timeout which fire after "timeout" seconds and stop the watch
+	var timeoutWatch int64 = int64(timeout + 30)
+	opts := v1.ListOptions{
+		FieldSelector: fields.OneTermEqualSelector("metadata.name", serviceName).String(),
+		TimeoutSeconds: &timeoutWatch,
+	}
+	watcher, err := client.Services(namespace).Watch(opts)
+	if err != nil {
+		return err
+	}
+	fmt.Fprintf(out, "Waiting for service %s to become ready ... ", serviceName)
+	if flusher, ok := out.(flusher); ok {
+		flusher.Flush()
+	}
+
+	if err := waitForReadyCondition(watcher, timeout); err != nil {
+		fmt.Fprintln(out)
+		return err
+	}
+	fmt.Fprintln(out, "OK")
+	return nil
+}
+
+func waitForReadyCondition(watcher watch.Interface, timeout int) error {
+	defer watcher.Stop()
+	for {
+		select {
+		case <- time.After(time.Duration(timeout) * time.Second):
+			return fmt.Errorf("timeout: Service not ready after %d seconds", timeout)
+		case event, ok := <- watcher.ResultChan():
+			if !ok || event.Object == nil {
+				return errors.New("timeout while waiting for service to become ready")
+			}
+			service := event.Object.(*serving_v1alpha1_api.Service)
+			for _, cond := range service.Status.Conditions {
+				if cond.Type == apis.ConditionReady {
+					switch cond.Status {
+					case corev1.ConditionTrue:
+						return nil
+					case corev1.ConditionFalse:
+						return fmt.Errorf("%s: %s",cond.Reason,cond.Message)
+					}
+				}
+			}
+		}
+	}
+}
+