feat(kafka): add --wait flag to perform synchronous Kafka creation

go.mod

@@ -8,12 +8,14 @@ require (
 	github.com/MakeNowJust/heredoc v1.0.0
 	github.com/Nerzal/gocloak/v7 v7.11.0
 	github.com/aerogear/charmil v0.8.2
+	github.com/briandowns/spinner v1.16.0 // indirect
 	github.com/coreos/go-oidc/v3 v3.0.0
 	github.com/fatih/color v1.12.0
 	github.com/golang-jwt/jwt/v4 v4.0.0
 	github.com/google/go-cmp v0.5.6 // indirect
 	github.com/google/go-github v17.0.0+incompatible
 	github.com/google/go-querystring v1.0.0 // indirect
+	github.com/inancgumus/screen v0.0.0-20190314163918-06e984b86ed3 // indirect
 	github.com/kataras/tablewriter v0.0.0-20180708051242-e063d29b7c23 // indirect
 	github.com/landoop/tableprinter v0.0.0-20201125135848-89e81fc956e7
 	github.com/mattn/go-isatty v0.0.13
@@ -31,6 +33,7 @@ require (
 	github.com/spf13/pflag v1.0.5
 	gitlab.com/c0b/go-ordered-json v0.0.0-20201030195603-febf46534d5a
 	golang.org/x/oauth2 v0.0.0-20210819190943-2bc19b11175f
+	golang.org/x/sys v0.0.0-20210823070655-63515b42dcdf // indirect
 	golang.org/x/text v0.3.7
 	gopkg.in/yaml.v2 v2.4.0
 	k8s.io/api v0.22.1

go.sum

@@ -121,6 +121,8 @@ github.com/bketelsen/crypt v0.0.4/go.mod h1:aI6NrJ0pMGgvZKL1iVgXLnfIFJtfV+bKCoqO
 github.com/blang/semver v3.5.0+incompatible/go.mod h1:kRBLl5iJ+tD4TcOOxsy/0fnwebNt5EWlYSAyrTnjyyk=
 github.com/blang/semver v3.5.1+incompatible h1:cQNTCjp13qL8KC3Nbxr/y2Bqb63oX6wdnnjpJbkM4JQ=
 github.com/blang/semver v3.5.1+incompatible/go.mod h1:kRBLl5iJ+tD4TcOOxsy/0fnwebNt5EWlYSAyrTnjyyk=
+github.com/briandowns/spinner v1.16.0 h1:DFmp6hEaIx2QXXuqSJmtfSBSAjRmpGiKG6ip2Wm/yOs=
+github.com/briandowns/spinner v1.16.0/go.mod h1:QOuQk7x+EaDASo80FEXwlwiA+j/PPIcX3FScO+3/ZPQ=
 github.com/cenkalti/backoff/v4 v4.0.0/go.mod h1:eEew/i+1Q6OrCDZh3WiXYv3+nJwBASZ8Bog/87DQnVg=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=
@@ -417,6 +419,8 @@ github.com/imdario/mergo v0.3.9/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJ
 github.com/imdario/mergo v0.3.10/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH0dnCYA=
 github.com/imdario/mergo v0.3.12 h1:b6R2BslTbIEToALKP7LxUvijTsNI9TAe80pLWN2g/HU=
 github.com/imdario/mergo v0.3.12/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH0dnCYA=
+github.com/inancgumus/screen v0.0.0-20190314163918-06e984b86ed3 h1:fO9A67/izFYFYky7l1pDP5Dr0BTCRkaQJUG6Jm5ehsk=
+github.com/inancgumus/screen v0.0.0-20190314163918-06e984b86ed3/go.mod h1:Ey4uAp+LvIl+s5jRbOHLcZpUDnkjLBROl15fZLwPlTM=
 github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NHg9XEKhtSvM=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
 github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99/go.mod h1:1lJo3i6rXxKeerYnT8Nvf0QmHCRC1n8sfWVwXF2Frvo=
@@ -944,6 +948,8 @@ golang.org/x/sys v0.0.0-20210603081109-ebe580a85c40/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c h1:F1jZWGFhYfh0Ci55sIpILtKKK8p3i2/krTr0H1rg74I=
 golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20210823070655-63515b42dcdf h1:2ucpDCmfkl8Bd/FsLtiD653Wf96cW37s+iGx93zsu4k=
+golang.org/x/sys v0.0.0-20210823070655-63515b42dcdf/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210220032956-6a3ed077a48d/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=

pkg/cmd/kafka/create/create.go

@@ -5,27 +5,33 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"net/http"
+	"os"
+	"os/signal"
+	"time"
 
 	kafkamgmtclient "github.com/redhat-developer/app-services-sdk-go/kafkamgmt/apiv1/client"
+	"gopkg.in/yaml.v2"
 
+	"github.com/redhat-developer/app-services-cli/pkg/color"
+	"github.com/redhat-developer/app-services-cli/pkg/dump"
 	"github.com/redhat-developer/app-services-cli/pkg/localize"
 
 	"github.com/redhat-developer/app-services-cli/pkg/ams"
 	"github.com/redhat-developer/app-services-cli/pkg/cmd/flag"
 	flagutil "github.com/redhat-developer/app-services-cli/pkg/cmdutil/flags"
 	"github.com/redhat-developer/app-services-cli/pkg/connection"
+	svcstatus "github.com/redhat-developer/app-services-cli/pkg/service/status"
 
 	"github.com/redhat-developer/app-services-cli/pkg/cloudprovider/cloudproviderutil"
 	"github.com/redhat-developer/app-services-cli/pkg/cloudregion/cloudregionutil"
 
 	"github.com/AlecAivazis/survey/v2"
-	"github.com/redhat-developer/app-services-cli/pkg/dump"
 	"github.com/redhat-developer/app-services-cli/pkg/iostreams"
 	pkgKafka "github.com/redhat-developer/app-services-cli/pkg/kafka"
 	"github.com/redhat-developer/app-services-cli/pkg/logging"
 
 	"github.com/spf13/cobra"
-	"gopkg.in/yaml.v2"
 
 	"github.com/redhat-developer/app-services-cli/internal/config"
 	"github.com/redhat-developer/app-services-cli/pkg/cmd/factory"
@@ -43,6 +49,7 @@ type Options struct {
 	autoUse      bool
 
 	interactive bool
+	wait        bool
 
 	IO         *iostreams.IOStreams
 	Config     config.IConfig
@@ -111,6 +118,7 @@ func NewCreateCommand(f *factory.Factory) *cobra.Command {
 	cmd.Flags().StringVar(&opts.region, flags.FlagRegion, "", opts.localizer.MustLocalize("kafka.create.flag.cloudRegion.description"))
 	cmd.Flags().StringVarP(&opts.outputFormat, "output", "o", "json", opts.localizer.MustLocalize("kafka.common.flag.output.description"))
 	cmd.Flags().BoolVar(&opts.autoUse, "use", true, opts.localizer.MustLocalize("kafka.create.flag.autoUse.description"))
+	cmd.Flags().BoolVarP(&opts.wait, "wait", "w", false, opts.localizer.MustLocalize("kafka.create.flag.wait.description"))
 
 	_ = cmd.RegisterFlagCompletionFunc(flags.FlagProvider, func(cmd *cobra.Command, _ []string, _ string) ([]string, cobra.ShellCompDirective) {
 		return cmdutil.FetchCloudProviders(f)
@@ -174,48 +182,88 @@ func runCreate(opts *Options) error {
 		}
 	}
 
-	logger.Info(opts.localizer.MustLocalize("kafka.create.log.info.creatingKafka", localize.NewEntry("Name", payload.Name)))
-
 	api := connection.API()
 
 	a := api.Kafka().CreateKafka(context.Background())
 	a = a.KafkaRequestPayload(*payload)
 	a = a.Async(true)
 	response, httpRes, err := a.Execute()
+	defer httpRes.Body.Close()
 
-	if httpRes.StatusCode == 409 {
+	if httpRes.StatusCode == http.StatusBadRequest {
 		return errors.New(opts.localizer.MustLocalize("kafka.create.error.conflictError", localize.NewEntry("Name", payload.Name)))
 	}
 
 	if err != nil {
 		return err
 	}
 
-	logger.Info(opts.localizer.MustLocalize("kafka.create.info.successMessage", localize.NewEntry("Name", response.GetName())))
-
-	switch opts.outputFormat {
-	case dump.JSONFormat:
-		data, _ := json.MarshalIndent(response, "", cmdutil.DefaultJSONIndent)
-		_ = dump.JSON(opts.IO.Out, data)
-	case dump.YAMLFormat, dump.YMLFormat:
-		data, _ := yaml.Marshal(response)
-		_ = dump.YAML(opts.IO.Out, data)
-	}
-
 	kafkaCfg := &config.KafkaConfig{
 		ClusterID: response.GetId(),
 	}
 
 	if opts.autoUse {
 		logger.Debug("Auto-use is set, updating the current instance")
 		cfg.Services.Kafka = kafkaCfg
-		if err := opts.Config.Save(cfg); err != nil {
+		if err = opts.Config.Save(cfg); err != nil {
 			return fmt.Errorf("%v: %w", opts.localizer.MustLocalize("kafka.common.error.couldNotUseKafka"), err)
 		}
 	} else {
 		logger.Debug("Auto-use is not set, skipping updating the current instance")
 	}
 
+	nameTemplateEntry := localize.NewEntry("Name", response.GetName())
+
+	if opts.wait {
+		logger.Debug("--wait flag is enabled, waiting for Kafka to finish creating")
+		s := opts.IO.NewSpinner()
+		s.Suffix = fmt.Sprintf(" %v", opts.localizer.MustLocalize("kafka.create.log.info.creatingKafka", nameTemplateEntry))
+		s.Start()
+
+		// when there is a SIGINT, display a message informing the user that this does not cancel the creation
+		// and that it is being created in the background
+		c := make(chan os.Signal, 1)
+		signal.Notify(c, os.Interrupt)
+		go func() {
+			for range c {
+				logger.Info()
+				logger.Info(opts.localizer.MustLocalize("kafka.create.log.info.creatingKafkaSyncSigint"))
+				os.Exit(0)
+			}
+		}()
+
+		for svcstatus.IsCreating(response.GetStatus()) {
+			time.Sleep(cmdutil.DefaultPollTime)
+
+			response, httpRes, err = api.Kafka().GetKafkaById(context.Background(), response.GetId()).Execute()
+			if err != nil {
+				return err
+			}
+			defer httpRes.Body.Close()
+			logger.Debug("Checking Kafka status:", response.GetStatus())
+
+			s.Suffix = createSpinnerSuffix(opts.localizer, response.GetName(), response.GetStatus())
+
+		}
+		s.Stop()
+		logger.Info("\n")
+		logger.Info(opts.localizer.MustLocalize("kafka.create.info.successSync", nameTemplateEntry))
+	}
+
+	switch opts.outputFormat {
+	case dump.JSONFormat:
+		data, _ := json.Marshal(response)
+		_ = dump.JSON(opts.IO.Out, data)
+	case dump.YAMLFormat, dump.YMLFormat:
+		data, _ := yaml.Marshal(response)
+		_ = dump.YAML(opts.IO.Out, data)
+	}
+
+	if !opts.wait {
+		logger.Info()
+		logger.Info(opts.localizer.MustLocalize("kafka.create.info.successAsync", nameTemplateEntry))
+	}
+
 	return nil
 }
 
@@ -254,7 +302,8 @@ func promptKafkaPayload(opts *Options) (payload *kafkamgmtclient.KafkaRequestPay
 	}
 
 	// fetch all cloud available providers
-	cloudProviderResponse, _, err := api.Kafka().GetCloudProviders(context.Background()).Execute()
+	cloudProviderResponse, httpRes, err := api.Kafka().GetCloudProviders(context.Background()).Execute()
+	defer httpRes.Body.Close()
 	if err != nil {
 		return nil, err
 	}
@@ -304,3 +353,9 @@ func promptKafkaPayload(opts *Options) (payload *kafkamgmtclient.KafkaRequestPay
 
 	return payload, nil
 }
+
+func createSpinnerSuffix(localizer localize.Localizer, name string, status string) string {
+	return " " + localizer.MustLocalize("kafka.create.log.info.creationInProgress",
+		localize.NewEntry("Name", name),
+		localize.NewEntry("Status", color.Info(status)))
+}

pkg/cmdutil/constants.go

@@ -1,6 +1,10 @@
 package cmdutil
 
+import "time"
+
 const (
 	// The default indentation to use when printing data to stdout
 	DefaultJSONIndent = "    "
+	// DefaultPollTime is the default interval to wait when polling a network request
+	DefaultPollTime = time.Millisecond * 5000
 )

pkg/iostreams/iostreams.go

@@ -3,7 +3,9 @@ package iostreams
 import (
 	"io"
 	"os"
+	"time"
 
+	"github.com/briandowns/spinner"
 	"github.com/fatih/color"
 	"github.com/mattn/go-isatty"
 )
@@ -88,6 +90,11 @@ func (s *IOStreams) IsSSHSession() bool {
 	return hasClient || hasTTY
 }
 
+// NewSpinner returns a new spinner progress bar to be used in synchronous commands
+func (s *IOStreams) NewSpinner() *spinner.Spinner {
+	return spinner.New(spinner.CharSets[9], 100*time.Millisecond, spinner.WithWriter(s.ErrOut))
+}
+
 func isTerminal(f *os.File) bool {
 	return isatty.IsTerminal(f.Fd()) || isatty.IsCygwinTerminal(f.Fd())
 }

pkg/localize/locales/en/cmd/kafka_create.en.toml

@@ -38,13 +38,26 @@ one = 'Cloud Provider Region ID'
 [kafka.create.flag.autoUse.description]
 one = 'Set the new Kafka instance to the current instance'
 
+[kafka.create.flag.wait.description]
+one = 'Wait until the Kafka instance is created'
+
 [kafka.create.log.info.creatingKafka]
 description = 'Message when Kafka instance is being created'
 one = 'Creating Kafka instance "{{.Name}}"...'
 
-[kafka.create.info.successMessage]
+[kafka.create.log.info.creationInProgress]
+description = 'Message when Kafka instance is being created'
+one = 'Kafka instance "{{.Name}}" is being created. Current status: {{.Status}}.'
+
+[kafka.create.info.successAsync]
 description = 'Message to display when instance has been created'
-one = 'Kafka instance "{{.Name}}" is being provisioned. You can monitor its progress by running "rhoas status".'
+one = 'Kafka instance "{{.Name}}" is being created. To monitor its status run "rhoas status".'
+
+[kafka.create.log.info.creatingKafkaSyncSigint]
+one = 'Your Kafka instance is being created in the background. To monitor its status run "rhoas status" or "rhoas kafka describe".'
+
+[kafka.create.info.successSync]
+one = 'Kafka instance "{{.Name}}" has been created:'
 
 [kafka.create.input.name.message]
 description = 'Input title for Name'

pkg/service/status/status.go

@@ -0,0 +1,18 @@
+package svcstatus
+
+type ServiceStatus = string
+
+// accepted, preparing, provisioning, ready, failed, deprovision, deleting
+const (
+	StatusAccepted     ServiceStatus = "accepted"
+	StatusPreparing    ServiceStatus = "preparing"
+	StatusProvisioning ServiceStatus = "provisioning"
+	StatusReady        ServiceStatus = "ready"
+	StatusFailed       ServiceStatus = "failed"
+	StatusDeprovision  ServiceStatus = "deprovision"
+	StatusDeleting     ServiceStatus = "deleting"
+)
+
+func IsCreating(status string) bool {
+	return status == StatusAccepted || status == StatusPreparing || status == StatusProvisioning
+}