From 4f0eb40999fbaffed9515248935734f2b1f36c7b Mon Sep 17 00:00:00 2001 From: Bowei Xu Date: Fri, 2 Mar 2018 11:50:01 -0800 Subject: [PATCH] Cadence CLI (#577) --- .travis.yml | 2 +- Makefile | 16 +- README.md | 2 +- cmd/tools/cli/main.go | 34 ++ docker/start.sh | 2 +- glide.lock | 25 +- glide.yaml | 4 + service/frontend/handler.go | 2 +- service/worker/README.md | 2 +- tools/cassandra/README.md | 17 +- tools/cli/README.md | 114 +++++ tools/cli/app.go | 71 +++ tools/cli/commands.go | 866 ++++++++++++++++++++++++++++++++++++ tools/cli/domain.go | 88 ++++ tools/cli/factory.go | 112 +++++ tools/cli/util.go | 247 ++++++++++ tools/cli/workflow.go | 307 +++++++++++++ 17 files changed, 1881 insertions(+), 30 deletions(-) create mode 100644 cmd/tools/cli/main.go create mode 100644 tools/cli/README.md create mode 100644 tools/cli/app.go create mode 100644 tools/cli/commands.go create mode 100644 tools/cli/domain.go create mode 100644 tools/cli/factory.go create mode 100644 tools/cli/util.go create mode 100644 tools/cli/workflow.go diff --git a/.travis.yml b/.travis.yml index b67d4250608..bebd5426dd6 100644 --- a/.travis.yml +++ b/.travis.yml @@ -5,7 +5,7 @@ language: go directories: - $HOME/.glide/cache go: - - 1.8 + - 1.9 addons: apt: diff --git a/Makefile b/Makefile index 223054090ad..5c35a69cb49 100644 --- a/Makefile +++ b/Makefile @@ -81,10 +81,13 @@ copyright: cmd/tools/copyright/licensegen.go cadence-cassandra-tool: vendor/glide.updated $(TOOLS_SRC) go build -i -o cadence-cassandra-tool cmd/tools/cassandra/main.go -cadence: vendor/glide.updated $(ALL_SRC) - go build -i -o cadence cmd/server/cadence.go cmd/server/server.go +cadence: vendor/glide.updated $(TOOLS_SRC) + go build -i -o cadence cmd/tools/cli/main.go -bins_nothrift: lint copyright cadence-cassandra-tool cadence +cadence-server: vendor/glide.updated $(ALL_SRC) + go build -i -o cadence-server cmd/server/cadence.go cmd/server/server.go + +bins_nothrift: lint copyright cadence-cassandra-tool cadence cadence-server bins: thriftc bins_nothrift @@ -137,6 +140,7 @@ fmt: clean: rm -f cadence rm -f cadence-cassandra-tool + rm -f cadence-server rm -Rf $(BUILD) install-schema: bins @@ -148,7 +152,7 @@ install-schema: bins ./cadence-cassandra-tool -ep 127.0.0.1 -k cadence_visibility update-schema -d ./schema/visibility/versioned start: bins - ./cadence start + ./cadence-server start install-schema-cdc: bins @echo Setting up cadence_active key space @@ -167,7 +171,7 @@ install-schema-cdc: bins ./cadence-cassandra-tool -ep 127.0.0.1 -k cadence_visibility_standby update-schema -d ./schema/visibility/versioned start-cdc-active: bins - ./cadence --zone active start + ./cadence-server --zone active start start-cdc-standby: bins - ./cadence --zone standby start \ No newline at end of file + ./cadence-server --zone standby start \ No newline at end of file diff --git a/README.md b/README.md index 1eec3c42c98..f3d75eeedc1 100644 --- a/README.md +++ b/README.md @@ -32,7 +32,7 @@ make install-schema * Start the service: ```bash -./cadence start +./cadence-server start ``` ### Using Docker diff --git a/cmd/tools/cli/main.go b/cmd/tools/cli/main.go new file mode 100644 index 00000000000..a532211401b --- /dev/null +++ b/cmd/tools/cli/main.go @@ -0,0 +1,34 @@ +// Copyright (c) 2017 Uber Technologies, Inc. +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +// THE SOFTWARE. + +package main + +import ( + "os" + + "github.com/uber/cadence/tools/cli" +) + +// Start using this CLI tool with command +// See cadence/tools/cli/README.md for usage +func main() { + app := cli.NewCliApp() + app.Run(os.Args) +} diff --git a/docker/start.sh b/docker/start.sh index 7f47dfae01b..3aecee24573 100644 --- a/docker/start.sh +++ b/docker/start.sh @@ -112,4 +112,4 @@ fi # fix up config envsubst < config/docker_template.yaml > config/docker.yaml -./cadence --root $CADENCE_HOME --env docker start --services=$SERVICES +./cadence-server --root $CADENCE_HOME --env docker start --services=$SERVICES diff --git a/glide.lock b/glide.lock index 3a02bf5a0e5..70909fef102 100644 --- a/glide.lock +++ b/glide.lock @@ -1,5 +1,5 @@ -hash: 961c13e43d7fb51e9c4e19d5fca098596754b882f54a276272717274c121a08d -updated: 2018-02-13T15:58:53.770031-08:00 +hash: 692b6e00919597293746e2c2a0b2c645fc550be9be89951aa10afafd1a4e5c6b +updated: 2018-02-20T18:14:25.806846-08:00 imports: - name: github.com/apache/thrift version: b2a4d4ae21c789b689dd162deb819665567f481c @@ -42,6 +42,8 @@ imports: - utils - name: github.com/facebookgo/clock version: 600d898af40aa09a7a93ecb9265d87b0504b6f03 +- name: github.com/fatih/color + version: 507f6050b8568533fb3f5504de8e5205fa62a114 - name: github.com/gocql/gocql version: ca7d33956650d92d29e6db7fe901e23d0f0e3359 subpackages: @@ -53,17 +55,21 @@ imports: subpackages: - gomock - name: github.com/golang/protobuf - version: c9c7427a2a70d2eb3bafa0ab2dc163e45f143317 + version: 925541529c1fa6821df4e44ce2723319eb2be768 subpackages: - proto - name: github.com/golang/snappy version: 553a641470496b2327abcac10b36396bd98e45c9 - name: github.com/hailocab/go-hostpool version: e80d13ce29ede4452c43dea11e79b9bc8a15b478 +- name: github.com/mattn/go-runewidth + version: 97311d9f7767e3d6f422ea06661bc2c7a19e8a5d - name: github.com/matttproud/golang_protobuf_extensions version: c12348ce28de40eed0136aa2b644d0ee0650e56c subpackages: - pbutil +- name: github.com/olekukonko/tablewriter + version: b8a9be070da40449e501c3c4730a889e42d87a9e - name: github.com/opentracing/opentracing-go version: 1949ddbfd147afd4d964a9f00b24eb291e0e7c38 subpackages: @@ -118,11 +124,12 @@ imports: - name: github.com/uber-common/bark version: dbf558e8a7b65e2b54e1e01c14ee0e4207a865f5 - name: github.com/uber-go/atomic - version: e682c1008ac17bf26d2e4b5ad6cdd08520ed0b22 + version: 8474b86a5a6f79c443ce4b2992817ff32cf208b8 + subpackages: + - utils - name: github.com/uber-go/kafka-client version: 01a7eeb4f9ea6e4c16a6ec5d159076c88244c947 subpackages: - - internal/backoff - internal/consumer - internal/list - internal/metrics @@ -133,7 +140,7 @@ imports: subpackages: - internal/mapstructure - name: github.com/uber-go/tally - version: 6c4631652c6aab57c64f65c2e0aaec2e9aae3a64 + version: 522328b48efad0c6034dba92bf39228694e9d31f subpackages: - m3 - m3/customtransports @@ -171,7 +178,9 @@ imports: - name: github.com/urfave/cli version: cf33a9befefdd6c6ea1a236ab6d546e797a62cbf - name: go.uber.org/atomic - version: 4e336646b2ef9fc6e47be8e21594178f98e5ebcf + version: 8474b86a5a6f79c443ce4b2992817ff32cf208b8 +- name: go.uber.org/cadence + version: 570324de7bb1008983846c97ee4d80d24e277854 - name: go.uber.org/multierr version: 3c4937480c32f4c13a875a1829af76c98ca3d40a - name: go.uber.org/net/metrics @@ -226,7 +235,7 @@ imports: - internal/exit - zapcore - name: golang.org/x/net - version: f5dfe339be1d06f81b22525fe34671ee7d2c8904 + version: cbe0f9307d0156177f9dd5dc85da1a31abc5f2fb repo: https://github.com/golang/net subpackages: - bpf diff --git a/glide.yaml b/glide.yaml index 871ce606359..c5131a701f8 100644 --- a/glide.yaml +++ b/glide.yaml @@ -23,6 +23,9 @@ import: - package: github.com/emirpasic/gods - package: github.com/davecgh/go-spew - package: github.com/urfave/cli +- package: github.com/fatih/color +- package: github.com/olekukonko/tablewriter +- package: github.com/mattn/go-runewidth - package: gopkg.in/yaml.v2 - package: gopkg.in/validator.v2 - package: golang.org/x/time @@ -31,6 +34,7 @@ import: - package: github.com/cactus/go-statsd-client subpackages: - statsd +- package: go.uber.org/cadence - package: go.uber.org/thriftrw version: ^1.6 - package: go.uber.org/yarpc diff --git a/service/frontend/handler.go b/service/frontend/handler.go index ef1336798df..ee85cc67da5 100644 --- a/service/frontend/handler.go +++ b/service/frontend/handler.go @@ -1108,7 +1108,7 @@ func (wh *WorkflowHandler) StartWorkflowExecution( return resp, nil } -// GetWorkflowExecutionHistory - retrieves the hisotry of workflow execution +// GetWorkflowExecutionHistory - retrieves the history of workflow execution func (wh *WorkflowHandler) GetWorkflowExecutionHistory( ctx context.Context, getRequest *gen.GetWorkflowExecutionHistoryRequest) (*gen.GetWorkflowExecutionHistoryResponse, error) { diff --git a/service/worker/README.md b/service/worker/README.md index 2e2074860bc..16e1520da99 100644 --- a/service/worker/README.md +++ b/service/worker/README.md @@ -32,7 +32,7 @@ bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 - ``` 4. Start Cadence development server for active zone: ``` -./cadence --zone active start +./cadence-server --zone active start ``` diff --git a/tools/cassandra/README.md b/tools/cassandra/README.md index 41c68dae462..b2f501b41d6 100644 --- a/tools/cassandra/README.md +++ b/tools/cassandra/README.md @@ -1,20 +1,16 @@ -What ----- +## What This package contains the tooling for cadence cassandra operations. -How ---- -- Run make bins +## How +- Run `make bins` - You should see an executable `cadence-cassandra-tool` -Setting up cassandra schema on a new cluster shortcut ----------------------------------------------------- +## Setting up cassandra schema on a new cluster shortcut ``` make install-schema ``` -Setting up schema on a new cluster manually ----------------------------------------------------- +## Setting up schema on a new cluster manually ``` ./cadence-cassandra-tool -ep 127.0.0.1 -k cadence setup-schema -v 0.0 -- this sets up just the schema version tables with initial version of 0.0 ./cadence-cassandra-tool -ep 127.0.0.1 -k cadence update-schema -d ./schema/cadence/versioned -- upgrades your schema to the latest version @@ -23,8 +19,7 @@ Setting up schema on a new cluster manually ./cadence-cassandra-tool -ep 127.0.0.1 -k cadence_visibility update-schema -d ./schema/visibility/versioned -- upgrades your schema to the latest version for visibility ``` -Updating schema on an existing cluster --------------------------------------- +## Updating schema on an existing cluster You can only upgrade to a new version after the initial setup done above. ``` diff --git a/tools/cli/README.md b/tools/cli/README.md new file mode 100644 index 00000000000..0d91e3e194c --- /dev/null +++ b/tools/cli/README.md @@ -0,0 +1,114 @@ +## What +Cadence CLI provide an command-line tool for users to perform various tasks on Cadence. +Users can perform operations like register, update and describe on domain; +also start workflow, show workflow history, signal workflow ... and many other tasks. + +## How +- Run `make bins` +- You should see an executable `cadence` + +## Quick Start +Run `./cadence` to view help message. There are some top level commands and global options. +Run `./cadence domain` to view help message about operations on domain +Run `./cadence workflow` to view help message about operations on workflow +(`./cadence help`, `./cadence help [domain|workflow]` will also print help messages) + +**Note:** make sure you have cadence server running before using CLI + +### Domain operation examples +- Register a new domain named "samples-domain": +``` +./cadence --domain samples-domain domain register +# OR using short alias +./cadence --do samples-domain domain re +``` +- View "samples-domain" details: +``` +./cadence --domain samples-domain domain describe +``` + +**Tips:** +to avoid repeated input global option **domain**, user can export domain-name in environment variable CADENCE_CLI_DOMAIN. +``` +export CADENCE_CLI_DOMAIN=samples-domain + +# then just run commands without --domain flag, like +./cadence domain desc +``` + +### Workflow operation examples +(The following examples assume you already export CADENCE_CLI_DOMAIN environment variable as Tips above) + +- Run workflow: Start a workflow and see it's progress, this command doesn't finish until workflow completes +``` +./cadence workflow run --tl helloWorldGroup --wt main.Workflow --et 60 -i '"cadence"' + +# view help messages for workflow run +./cadence workflow run -h +``` +Brief explanation: +To run a workflow, user must specify +1. Tasklist name (--tl), +2. Workflow type (--wt), +3. Execution start to close timeout in seconds (--et), + +Example uses [this cadence-samples workflow](https://github.com/samarabbas/cadence-samples/blob/master/cmd/samples/recipes/helloworld/helloworld_workflow.go) +and it takes a string as input, so there is the `-i '"cadence"'`. Single quote `''` is used to wrap input as json. + +**Note:** you need to start worker so that workflow can make progress. +(Run `make && ./bin/helloworld -m worker` in cadence-samples to start the worker) + +- Start workflow: +``` +./cadence workflow start --tl helloWorldGroup --wt main.Workflow --et 60 -i '"cadence"' + +# view help messages for workflow start +./cadence workflow start -h + +# for workflow with multiple input, seperate each json with space/newline like +./cadence workflow start --tl helloWorldGroup --wt main.WorkflowWith3Args --et 60 -i '"your_input_string" 123 {"Name":"my-string", "Age":12345}' +``` +Workflow `start` command is similar to `run` command and takes same flag options. But it just start the workflow and immediately return workflow_id and run_id. +User need to run `show` to view workflow history/progress. + +- Show workflow history +``` +./cadence workflow show -w 3ea6b242-b23c-4279-bb13-f215661b4717 -r 866ae14c-88cf-4f1e-980f-571e031d71b0 +# a shortcut of this is (without -w -r flag) +./cadence workflow showid 3ea6b242-b23c-4279-bb13-f215661b4717 866ae14c-88cf-4f1e-980f-571e031d71b0 + +# if run_id is not provided, it will show the latest run history for that workflow_id +./cadence workflow show -w 3ea6b242-b23c-4279-bb13-f215661b4717 +# a shortcut of this is +./cadence workflow showid 3ea6b242-b23c-4279-bb13-f215661b4717 +``` + +- List open or closed workflow executions +``` +./cadence workflow list +``` + +- Query workflow execution +``` +# use custom query type +./cadence workflow query -w -r --qt + +# use build-in query type "__stack_trace" which is supported by cadence client library +./cadence workflow query -w -r --qt __stack_trace +# a shortcut to query using __stack_trace is (without --qt flag) +./cadence workflow stack -w -r +``` + +- Signal, cancel, terminate workflow +``` +# signal +./cadence workflow signal -w -r -n -i '"signal-value"' + +# cancel +./cadence workflow cancel -w -r + +# terminate +./cadence workflow terminate -w -r --reason +``` +Terminate a running workflow execution will record WorkflowExecutionTerminated event as closing event in the history. No more decision task will be scheduled for terminated workflow execution. +Cancel a running workflow execution will record WorkflowExecutionCancelRequested event in the history, and a new decision task will be scheduled. Workflow has a chance to do some clean up work after cancellation. \ No newline at end of file diff --git a/tools/cli/app.go b/tools/cli/app.go new file mode 100644 index 00000000000..6428e7b1f7f --- /dev/null +++ b/tools/cli/app.go @@ -0,0 +1,71 @@ +// Copyright (c) 2017 Uber Technologies, Inc. +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +// THE SOFTWARE. + +package cli + +import "github.com/urfave/cli" + +const ( + // Version is the controlled version string. It should be updated every time + // before we release a new version. + Version = "0.5.0" +) + +// NewCliApp instantiates a new instance of the CLI application. +func NewCliApp() *cli.App { + app := cli.NewApp() + app.Name = "cadence" + app.Usage = "A command-line tool for cadence users" + app.Version = Version + app.Flags = []cli.Flag{ + cli.StringFlag{ + Name: FlagAddressWithAlias, + Value: "", + Usage: "host:port for cadence frontend service", + EnvVar: "CADENCE_CLI_ADDRESS", + }, + cli.StringFlag{ + Name: FlagDomainWithAlias, + Usage: "cadence workflow domain", + EnvVar: "CADENCE_CLI_DOMAIN", + }, + } + app.Commands = []cli.Command{ + { + Name: "domain", + Aliases: []string{"d"}, + Usage: "Operate cadence domain", + Subcommands: newDomainCommands(), + }, + { + Name: "workflow", + Aliases: []string{"wf"}, + Usage: "Operate cadence workflow", + Subcommands: newWorkflowCommands(), + }, + } + + // set builder if not customized + if cBuilder == nil { + SetBuilder(NewBuilder()) + } + + return app +} diff --git a/tools/cli/commands.go b/tools/cli/commands.go new file mode 100644 index 00000000000..2e0358f78a0 --- /dev/null +++ b/tools/cli/commands.go @@ -0,0 +1,866 @@ +// Copyright (c) 2017 Uber Technologies, Inc. +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +// THE SOFTWARE. + +package cli + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "io/ioutil" + "os" + "runtime/debug" + "strconv" + "time" + + "github.com/fatih/color" + "github.com/olekukonko/tablewriter" + "github.com/pborman/uuid" + "github.com/uber/cadence/common" + "github.com/urfave/cli" + + "go.uber.org/cadence/.gen/go/cadence/workflowserviceclient" + s "go.uber.org/cadence/.gen/go/shared" + "go.uber.org/cadence/client" +) + +/** +Flags used to specify cli command line arguments +*/ +const ( + FlagAddress = "address" + FlagAddressWithAlias = FlagAddress + ", ad" + FlagDomain = "domain" + FlagDomainWithAlias = FlagDomain + ", do" + FlagWorkflowID = "workflow_id" + FlagWorkflowIDWithAlias = FlagWorkflowID + ", wid, w" + FlagRunID = "run_id" + FlagRunIDWithAlias = FlagRunID + ", rid, r" + FlagTaskList = "tasklist" + FlagTaskListWithAlias = FlagTaskList + ", tl" + FlagWorkflowType = "workflow_type" + FlagWorkflowTypeWithAlias = FlagWorkflowType + ", wt" + FlagExecutionTimeout = "execution_timeout" + FlagExecutionTimeoutWithAlias = FlagExecutionTimeout + ", et" + FlagDecisionTimeout = "decision_timeout" + FlagDecisionTimeoutWithAlias = FlagDecisionTimeout + ", dt" + FlagContextTimeout = "context_timeout" + FlagContextTimeoutWithAlias = FlagContextTimeout + ", ct" + FlagInput = "input" + FlagInputWithAlias = FlagInput + ", i" + FlagInputFile = "input_file" + FlagInputFileWithAlias = FlagInputFile + ", if" + FlagReason = "reason" + FlagReasonWithAlias = FlagReason + ", re" + FlagOpen = "open" + FlagOpenWithAlias = FlagOpen + ", op" + FlagPageSize = "pagesize" + FlagPageSizeWithAlias = FlagPageSize + ", ps" + FlagEarliestTime = "earliest_time" + FlagEarliestTimeWithAlias = FlagEarliestTime + ", et" + FlagLatestTime = "latest_time" + FlagLatestTimeWithAlias = FlagLatestTime + ", lt" + FlagPrintRawTime = "print_raw_time" + FlagPrintRawTimeWithAlias = FlagPrintRawTime + ", prt" + FlagDescription = "description" + FlagDescriptionWithAlias = FlagDescription + ", desc" + FlagOwnerEmail = "owner_email" + FlagOwnerEmailWithAlias = FlagOwnerEmail + ", oe" + FlagRetentionDays = "retention_days" + FlagRetentionDaysWithAlias = FlagRetentionDays + ", rd" + FlagEmitMetric = "emit_metric" + FlagEmitMetricWithAlias = FlagEmitMetric + ", em" + FlagName = "name" + FlagNameWithAlias = FlagName + ", n" + FlagOutputFilename = "output_filename" + FlagOutputFilenameWithAlias = FlagOutputFilename + ", of" + FlagQueryType = "query_type" + FlagQueryTypeWithAlias = FlagQueryType + ", qt" +) + +const ( + localHostPort = "127.0.0.1:7933" + + maxOutputStringLength = 200 // max length for output string + + defaultTimeFormat = time.RFC3339 // used for converting UnixNano to string like 2018-02-15T16:16:36-08:00 + defaultDomainRetentionDays = 3 + defaultContextTimeoutForLongPoll = 2 * time.Minute + defaultDecisionTimeoutInSeconds = 10 +) + +// For color output to terminal +var ( + colorRed = color.New(color.FgRed).SprintFunc() + colorMagenta = color.New(color.FgMagenta).SprintFunc() + colorGreen = color.New(color.FgGreen).SprintFunc() +) + +// cBuilder is used to create cadence clients +// To provide customized builder, call SetBuilder() before call NewCliApp() +var cBuilder WorkflowClientBuilderInterface + +// SetBuilder can be used to inject customized builder of cadence clients +func SetBuilder(builder WorkflowClientBuilderInterface) { + cBuilder = builder +} + +// ErrorAndExit print easy to understand error msg first then error detail in a new line +func ErrorAndExit(msg string, err error) { + fmt.Printf("%s %s\n%s %+v\n", colorRed("Error:"), msg, colorMagenta("Error Details:"), err) + os.Exit(1) +} + +// ExitIfError exit while err is not nil and print the calling stack also +func ExitIfError(err error) { + const stacksEnv = `CADENCE_CLI_SHOW_STACKS` + if err != nil { + fmt.Fprintln(os.Stderr, err.Error()) + if os.Getenv(stacksEnv) != `` { + debug.PrintStack() + } else { + fmt.Fprintf(os.Stderr, "('export %s=1' to see stack traces)\n", stacksEnv) + } + os.Exit(1) + } +} + +// RegisterDomain register a domain +func RegisterDomain(c *cli.Context) { + domainClient := getDomainClient(c) + domain := getRequiredGlobalOption(c, FlagDomain) + + description := c.String(FlagDescription) + ownerEmail := c.String(FlagOwnerEmail) + retentionDays := defaultDomainRetentionDays + if c.IsSet(FlagRetentionDays) { + retentionDays = c.Int(FlagRetentionDays) + } + emitMetric := false + var err error + if c.IsSet(FlagEmitMetric) { + emitMetric, err = strconv.ParseBool(c.String(FlagEmitMetric)) + if err != nil { + fmt.Printf("Register Domain failed: %v.\n", err.Error()) + return + } + } + + request := &s.RegisterDomainRequest{ + Name: common.StringPtr(domain), + Description: common.StringPtr(description), + OwnerEmail: common.StringPtr(ownerEmail), + WorkflowExecutionRetentionPeriodInDays: common.Int32Ptr(int32(retentionDays)), + EmitMetric: common.BoolPtr(emitMetric), + } + + ctx, cancel := newContext() + defer cancel() + err = domainClient.Register(ctx, request) + if err != nil { + if _, ok := err.(*s.DomainAlreadyExistsError); !ok { + fmt.Printf("Operation failed: %v.\n", err.Error()) + } else { + fmt.Printf("Domain %s already registered.\n", domain) + } + } else { + fmt.Printf("Domain %s succeesfully registered.\n", domain) + } +} + +// UpdateDomain updates a domain +func UpdateDomain(c *cli.Context) { + domainClient := getDomainClient(c) + domain := getRequiredGlobalOption(c, FlagDomain) + + ctx, cancel := newContext() + defer cancel() + info, config, err := domainClient.Describe(ctx, domain) + if err != nil { + if _, ok := err.(*s.EntityNotExistsError); !ok { + fmt.Printf("Operation failed: %v.\n", err.Error()) + } else { + fmt.Printf("Domain %s does not exist.\n", domain) + } + } + + description := info.GetDescription() + if c.IsSet(FlagDescription) { + description = c.String(FlagDescription) + } + ownerEmail := info.GetOwnerEmail() + if c.IsSet(FlagOwnerEmail) { + ownerEmail = c.String(FlagOwnerEmail) + } + retentionDays := config.GetWorkflowExecutionRetentionPeriodInDays() + if c.IsSet(FlagRetentionDays) { + retentionDays = int32(c.Int(FlagRetentionDays)) + } + emitMetric := config.GetEmitMetric() + if c.IsSet(FlagEmitMetric) { + emitMetric, err = strconv.ParseBool(c.String(FlagEmitMetric)) + if err != nil { + ErrorAndExit("Update Domain failed", err) + } + } + + updateInfo := &s.UpdateDomainInfo{ + Description: common.StringPtr(description), + OwnerEmail: common.StringPtr(ownerEmail), + } + updateConfig := &s.DomainConfiguration{ + WorkflowExecutionRetentionPeriodInDays: common.Int32Ptr(int32(retentionDays)), + EmitMetric: common.BoolPtr(emitMetric), + } + + err = domainClient.Update(ctx, domain, updateInfo, updateConfig) + if err != nil { + if _, ok := err.(*s.EntityNotExistsError); !ok { + fmt.Printf("Operation failed: %v.\n", err.Error()) + } else { + fmt.Printf("Domain %s does not exist.\n", domain) + } + } else { + fmt.Printf("Domain %s succeesfully updated.\n", domain) + } +} + +// DescribeDomain updates a domain +func DescribeDomain(c *cli.Context) { + domainClient := getDomainClient(c) + domain := getRequiredGlobalOption(c, FlagDomain) + + ctx, cancel := newContext() + defer cancel() + info, config, err := domainClient.Describe(ctx, domain) + if err != nil { + if _, ok := err.(*s.EntityNotExistsError); !ok { + fmt.Printf("Operation failed: %v.\n", err.Error()) + } else { + fmt.Printf("Domain %s does not exist.\n", domain) + } + } else { + fmt.Printf("Name:%v, Description:%v, OwnerEmail:%v, Status:%v, RetentionInDays:%v, EmitMetrics:%v\n", + info.GetName(), + info.GetDescription(), + info.GetOwnerEmail(), + info.GetStatus(), + config.GetWorkflowExecutionRetentionPeriodInDays(), + config.GetEmitMetric()) + } +} + +// ShowHistory shows the history of given workflow execution based on workflowID and runID. +func ShowHistory(c *cli.Context) { + wid := getRequiredOption(c, FlagWorkflowID) + rid := c.String(FlagRunID) + showHistoryHelper(c, wid, rid) +} + +// ShowHistoryWithWID shows the history of given workflow with workflow_id +func ShowHistoryWithWID(c *cli.Context) { + if !c.Args().Present() { + ExitIfError(errors.New("workflow_id is required")) + } + wid := c.Args().First() + rid := "" + if c.NArg() >= 2 { + rid = c.Args().Get(1) + } + showHistoryHelper(c, wid, rid) +} + +func showHistoryHelper(c *cli.Context, wid, rid string) { + wfClient := getWorkflowClient(c) + + printRawTime := c.Bool(FlagPrintRawTime) + outputFileName := c.String(FlagOutputFilename) + + ctx, cancel := newContext() + defer cancel() + history, err := GetHistory(ctx, wfClient, wid, rid) + if err != nil { + ExitIfError(err) + } + + for _, e := range history.Events { + if printRawTime { + fmt.Printf("%d, %d, %s\n", e.GetEventId(), e.GetTimestamp(), HistoryEventToString(e)) + } else { + fmt.Printf("%d, %s, %s\n", e.GetEventId(), convertTime(e.GetTimestamp()), HistoryEventToString(e)) + } + } + + if outputFileName != "" { + serializer := &JSONHistorySerializer{} + data, err := serializer.Serialize(history) + if err != nil { + ExitIfError(err) + } + if err := ioutil.WriteFile(outputFileName, data, 0777); err != nil { + ExitIfError(err) + } + } +} + +// StartWorkflow starts a new workflow execution +func StartWorkflow(c *cli.Context) { + // using service client instead of cadence.Client because we need to directly pass the json blob as input. + serviceClient := getWorkflowServiceClient(c) + + domain := getRequiredGlobalOption(c, FlagDomain) + tasklist := getRequiredOption(c, FlagTaskList) + workflowType := getRequiredOption(c, FlagWorkflowType) + et := c.Int(FlagExecutionTimeout) + if et == 0 { + ExitIfError(errors.New(FlagExecutionTimeout + " is required")) + } + dt := c.Int(FlagDecisionTimeout) + wid := c.String(FlagWorkflowID) + if len(wid) == 0 { + wid = uuid.New() + } + + input := processJSONInput(c) + + tcCtx, cancel := newContext() + defer cancel() + + resp, err := serviceClient.StartWorkflowExecution(tcCtx, &s.StartWorkflowExecutionRequest{ + RequestId: common.StringPtr(uuid.New()), + Domain: common.StringPtr(domain), + WorkflowId: common.StringPtr(wid), + WorkflowType: &s.WorkflowType{ + Name: common.StringPtr(workflowType), + }, + TaskList: &s.TaskList{ + Name: common.StringPtr(tasklist), + }, + Input: []byte(input), + ExecutionStartToCloseTimeoutSeconds: common.Int32Ptr(int32(et)), + TaskStartToCloseTimeoutSeconds: common.Int32Ptr(int32(dt)), + Identity: common.StringPtr(getCliIdentity()), + }) + + if err != nil { + ErrorAndExit("Failed to create workflow", err) + } else { + fmt.Printf("Started Workflow Id: %s, run Id: %s\n", wid, resp.GetRunId()) + } +} + +// RunWorkflow starts a new workflow execution and print workflow progress and result +func RunWorkflow(c *cli.Context) { + // using service client instead of cadence.Client because we need to directly pass the json blob as input. + serviceClient := getWorkflowServiceClient(c) + + domain := getRequiredGlobalOption(c, FlagDomain) + tasklist := getRequiredOption(c, FlagTaskList) + workflowType := getRequiredOption(c, FlagWorkflowType) + et := c.Int(FlagExecutionTimeout) + if et == 0 { + ExitIfError(errors.New(FlagExecutionTimeout + " is required")) + } + dt := c.Int(FlagDecisionTimeout) + wid := c.String(FlagWorkflowID) + if len(wid) == 0 { + wid = uuid.New() + } + + input := processJSONInput(c) + + contextTimeout := defaultContextTimeoutForLongPoll + if c.IsSet(FlagContextTimeout) { + contextTimeout = time.Duration(c.Int(FlagContextTimeout)) * time.Second + } + tcCtx, cancel := newContextForLongPoll(contextTimeout) + defer cancel() + + resp, err := serviceClient.StartWorkflowExecution(tcCtx, &s.StartWorkflowExecutionRequest{ + RequestId: common.StringPtr(uuid.New()), + Domain: common.StringPtr(domain), + WorkflowId: common.StringPtr(wid), + WorkflowType: &s.WorkflowType{ + Name: common.StringPtr(workflowType), + }, + TaskList: &s.TaskList{ + Name: common.StringPtr(tasklist), + }, + Input: []byte(input), + ExecutionStartToCloseTimeoutSeconds: common.Int32Ptr(int32(et)), + TaskStartToCloseTimeoutSeconds: common.Int32Ptr(int32(dt)), + Identity: common.StringPtr(getCliIdentity()), + }) + + if err != nil { + ErrorAndExit("Failed to create workflow", err) + } + + // print execution summary + fmt.Println(colorMagenta("Running execution:")) + table := tablewriter.NewWriter(os.Stdout) + executionData := [][]string{ + {"Workflow Id", wid}, + {"Run Id", resp.GetRunId()}, + {"Type", workflowType}, + {"Domain", domain}, + {"Task List", tasklist}, + {"Args", truncate(input)}, // in case of large input + } + table.SetBorder(false) + table.SetColumnSeparator(":") + table.AppendBulk(executionData) // Add Bulk Data + table.Render() + + // print workflow progress + fmt.Println(colorMagenta("Progress:")) + wfClient := getWorkflowClient(c) + timeElapse := 1 + isTimeElapseExist := false + doneChan := make(chan bool) + var lastEvent *s.HistoryEvent // used for print result of this run + ticker := time.NewTicker(time.Second).C + + go func() { + iter := wfClient.GetWorkflowHistory(tcCtx, wid, resp.GetRunId(), true, s.HistoryEventFilterTypeAllEvent) + for iter.HasNext() { + event, err := iter.Next() + if err != nil { + ErrorAndExit("Unable to read event", err) + } + if isTimeElapseExist { + removePrevious2LinesFromTerminal() + isTimeElapseExist = false + } + fmt.Printf(" %d, %s, %v\n", event.GetEventId(), convertTime(event.GetTimestamp()), event.GetEventType()) + lastEvent = event + } + doneChan <- true + }() + + for { + select { + case <-ticker: + if isTimeElapseExist { + removePrevious2LinesFromTerminal() + } + fmt.Printf("\nTime elapse: %ds\n", timeElapse) + isTimeElapseExist = true + timeElapse++ + case <-doneChan: // print result of this run + fmt.Println(colorMagenta("\nResult:")) + fmt.Printf(" Run Time: %d seconds\n", timeElapse) + printRunStatus(lastEvent) + return + } + } +} + +// TerminateWorkflow terminates a workflow execution +func TerminateWorkflow(c *cli.Context) { + wfClient := getWorkflowClient(c) + + wid := getRequiredOption(c, FlagWorkflowID) + rid := c.String(FlagRunID) + reason := c.String(FlagReason) + + ctx, cancel := newContext() + defer cancel() + err := wfClient.TerminateWorkflow(ctx, wid, rid, reason, nil) + + if err != nil { + ErrorAndExit("Terminate workflow failed", err) + } else { + fmt.Println("Terminate workflow succeed.") + } +} + +// CancelWorkflow cancels a workflow execution +func CancelWorkflow(c *cli.Context) { + wfClient := getWorkflowClient(c) + + wid := getRequiredOption(c, FlagWorkflowID) + rid := c.String(FlagRunID) + + ctx, cancel := newContext() + defer cancel() + err := wfClient.CancelWorkflow(ctx, wid, rid) + + if err != nil { + ErrorAndExit("Cancel workflow failed", err) + } else { + fmt.Println("Cancel workflow succeed.") + } +} + +// SignalWorkflow signals a workflow execution +func SignalWorkflow(c *cli.Context) { + // using service client instead of cadence.Client because we need to directly pass the json blob as input. + serviceClient := getWorkflowServiceClient(c) + + domain := getRequiredGlobalOption(c, FlagDomain) + wid := getRequiredOption(c, FlagWorkflowID) + rid := c.String(FlagRunID) + name := getRequiredOption(c, FlagName) + input := processJSONInput(c) + + tcCtx, cancel := newContext() + defer cancel() + err := serviceClient.SignalWorkflowExecution(tcCtx, &s.SignalWorkflowExecutionRequest{ + Domain: common.StringPtr(domain), + WorkflowExecution: &s.WorkflowExecution{ + WorkflowId: common.StringPtr(wid), + RunId: getPtrOrNilIfEmpty(rid), + }, + SignalName: common.StringPtr(name), + Input: []byte(input), + Identity: common.StringPtr(getCliIdentity()), + }) + + if err != nil { + ErrorAndExit("Signal workflow failed", err) + } else { + fmt.Println("Signal workflow succeed.") + } +} + +// QueryWorkflow query workflow execution +func QueryWorkflow(c *cli.Context) { + getRequiredGlobalOption(c, FlagDomain) // for pre-check and alert if not provided + getRequiredOption(c, FlagWorkflowID) + queryType := getRequiredOption(c, FlagQueryType) + + queryWorkflowHelper(c, queryType) +} + +// QueryWorkflowUsingStackTrace query workflow execution using __stack_trace as query type +func QueryWorkflowUsingStackTrace(c *cli.Context) { + queryWorkflowHelper(c, "__stack_trace") +} + +func queryWorkflowHelper(c *cli.Context, queryType string) { + // using service client instead of cadence.Client because we need to directly pass the json blob as input. + serviceClient := getWorkflowServiceClient(c) + + domain := getRequiredGlobalOption(c, FlagDomain) + wid := getRequiredOption(c, FlagWorkflowID) + rid := c.String(FlagRunID) + input := processJSONInput(c) + + tcCtx, cancel := newContext() + defer cancel() + queryRequest := &s.QueryWorkflowRequest{ + Domain: common.StringPtr(domain), + Execution: &s.WorkflowExecution{ + WorkflowId: common.StringPtr(wid), + RunId: getPtrOrNilIfEmpty(rid), + }, + Query: &s.WorkflowQuery{ + QueryType: common.StringPtr(queryType), + }, + } + if input != "" { + queryRequest.Query.QueryArgs = []byte(input) + } + queryResponse, err := serviceClient.QueryWorkflow(tcCtx, queryRequest) + if err != nil { + ErrorAndExit("Query failed", err) + return + } + + // assume it is json encoded + fmt.Printf("Query result as JSON:\n%v\n", string(queryResponse.QueryResult)) +} + +// ListWorkflow list workflow executions based on filters +func ListWorkflow(c *cli.Context) { + wfClient := getWorkflowClient(c) + + queryOpen := c.Bool(FlagOpen) + pageSize := c.Int(FlagPageSize) + earliestTime := parseTime(c.String(FlagEarliestTime), 0) + latestTime := parseTime(c.String(FlagLatestTime), time.Now().UnixNano()) + workflowID := c.String(FlagWorkflowID) + workflowType := c.String(FlagWorkflowType) + printRawTime := c.Bool(FlagPrintRawTime) + + if len(workflowID) > 0 && len(workflowType) > 0 { + ExitIfError(errors.New("you can filter on workflow_id or workflow_type, but not on both")) + } + + reader := bufio.NewReader(os.Stdin) + var result []*s.WorkflowExecutionInfo + var nextPageToken []byte + for { + if queryOpen { + result, nextPageToken = listOpenWorkflow(wfClient, pageSize, earliestTime, latestTime, workflowID, workflowType, nextPageToken) + } else { + result, nextPageToken = listClosedWorkflow(wfClient, pageSize, earliestTime, latestTime, workflowID, workflowType, nextPageToken) + } + + for _, e := range result { + fmt.Printf("%s, -w %s -r %s", e.Type.GetName(), e.Execution.GetWorkflowId(), e.Execution.GetRunId()) + if printRawTime { + fmt.Printf(" [%d, %d]\n", e.GetStartTime(), e.GetCloseTime()) + } else { + fmt.Printf(" [%s, %s]\n", convertTime(e.GetStartTime()), convertTime(e.GetCloseTime())) + } + } + + if len(result) < pageSize { + break + } + + fmt.Println("Press C then Enter to show more result, press any other key then Enter to quit: ") + input, _ := reader.ReadString('\n') + c := []byte(input)[0] + if c == 'C' || c == 'c' { + continue + } else { + break + } + } +} + +func listOpenWorkflow(client client.Client, pageSize int, earliestTime, latestTime int64, workflowID, workflowType string, nextPageToken []byte) ([]*s.WorkflowExecutionInfo, []byte) { + request := &s.ListOpenWorkflowExecutionsRequest{ + MaximumPageSize: common.Int32Ptr(int32(pageSize)), + NextPageToken: nextPageToken, + StartTimeFilter: &s.StartTimeFilter{ + EarliestTime: common.Int64Ptr(earliestTime), + LatestTime: common.Int64Ptr(latestTime), + }, + } + if len(workflowID) > 0 { + request.ExecutionFilter = &s.WorkflowExecutionFilter{WorkflowId: common.StringPtr(workflowID)} + } + if len(workflowType) > 0 { + request.TypeFilter = &s.WorkflowTypeFilter{Name: common.StringPtr(workflowType)} + } + + ctx, cancel := newContext() + defer cancel() + response, err := client.ListOpenWorkflow(ctx, request) + if err != nil { + ExitIfError(err) + } + return response.Executions, response.NextPageToken +} + +func listClosedWorkflow(client client.Client, pageSize int, earliestTime, latestTime int64, workflowID, workflowType string, nextPageToken []byte) ([]*s.WorkflowExecutionInfo, []byte) { + request := &s.ListClosedWorkflowExecutionsRequest{ + MaximumPageSize: common.Int32Ptr(int32(pageSize)), + NextPageToken: nextPageToken, + StartTimeFilter: &s.StartTimeFilter{ + EarliestTime: common.Int64Ptr(earliestTime), + LatestTime: common.Int64Ptr(latestTime), + }, + } + if len(workflowID) > 0 { + request.ExecutionFilter = &s.WorkflowExecutionFilter{WorkflowId: common.StringPtr(workflowID)} + } + if len(workflowType) > 0 { + request.TypeFilter = &s.WorkflowTypeFilter{Name: common.StringPtr(workflowType)} + } + + ctx, cancel := newContext() + defer cancel() + response, err := client.ListClosedWorkflow(ctx, request) + if err != nil { + ExitIfError(err) + } + return response.Executions, response.NextPageToken +} + +func getDomainClient(c *cli.Context) client.DomainClient { + service, err := cBuilder.BuildServiceClient(c) + if err != nil { + ExitIfError(err) + } + + domainClient, err := client.NewDomainClient(service, &client.Options{}), nil + if err != nil { + ExitIfError(err) + } + return domainClient +} + +func getWorkflowClient(c *cli.Context) client.Client { + domain := getRequiredGlobalOption(c, FlagDomain) + + service, err := cBuilder.BuildServiceClient(c) + if err != nil { + ExitIfError(err) + } + + wfClient, err := client.NewClient(service, domain, &client.Options{}), nil + if err != nil { + ExitIfError(err) + } + + return wfClient +} + +func getWorkflowServiceClient(c *cli.Context) workflowserviceclient.Interface { + client, err := cBuilder.BuildServiceClient(c) + if err != nil { + ExitIfError(err) + } + + return client +} + +func getRequiredOption(c *cli.Context, optionName string) string { + value := c.String(optionName) + if len(value) == 0 { + ExitIfError(fmt.Errorf("%s is required", optionName)) + } + return value +} + +func getRequiredGlobalOption(c *cli.Context, optionName string) string { + value := c.GlobalString(optionName) + if len(value) == 0 { + ExitIfError(fmt.Errorf("%s is required", optionName)) + } + return value +} + +func convertTime(unixNano int64) string { + t2 := time.Unix(0, unixNano) + return t2.Format(defaultTimeFormat) +} + +func parseTime(timeStr string, defaultValue int64) int64 { + if len(timeStr) == 0 { + return defaultValue + } + + // try to parse + parsedTime, err := time.Parse(defaultTimeFormat, timeStr) + if err == nil { + return parsedTime.UnixNano() + } + + // treat as raw time + resultValue, err := strconv.ParseInt(timeStr, 10, 64) + if err != nil { + ExitIfError(fmt.Errorf("cannot parse time '%s', use UTC format '2006-01-02T15:04:05Z' or raw UnixNano directly. Error: %v", timeStr, err)) + } + + return resultValue +} + +func getPtrOrNilIfEmpty(value string) *string { + if value == "" { + return nil + } + return common.StringPtr(value) +} + +func getCliIdentity() string { + hostName, err := os.Hostname() + if err != nil { + hostName = "UnKnown" + } + return fmt.Sprintf("cadence-cli@%s", hostName) +} + +func newContext() (context.Context, context.CancelFunc) { + return context.WithTimeout(context.Background(), time.Second*5) +} + +func newContextForLongPoll(timeout time.Duration) (context.Context, context.CancelFunc) { + return context.WithTimeout(context.Background(), timeout) +} + +// process and validate input provided through cmd or file +func processJSONInput(c *cli.Context) string { + var input string + if c.IsSet(FlagInput) { + input = c.String(FlagInput) + } else if c.IsSet(FlagInputFile) { + inputFile := c.String(FlagInputFile) + data, err := ioutil.ReadFile(inputFile) + if err != nil { + ErrorAndExit("Error reading input file", err) + } + input = string(data) + } + if input != "" { + if err := validateJSONs(input); err != nil { + ErrorAndExit("Input is not valid JSON, or JSONs concatenated with spaces/newlines.", err) + } + } + return input +} + +// validate whether str is a valid json or multi valid json concatenated with spaces/newlines +func validateJSONs(str string) error { + input := []byte(str) + dec := json.NewDecoder(bytes.NewReader(input)) + for { + _, err := dec.Token() + if err == io.EOF { + return nil // End of input, valid JSON + } + if err != nil { + return err // Invalid input + } + } +} + +func truncate(str string) string { + if len(str) > maxOutputStringLength { + return str[:maxOutputStringLength] + } + return str +} + +// this only works for ANSI terminal, which means remove existing lines won't work if users redirect to file +// ref: https://en.wikipedia.org/wiki/ANSI_escape_code +func removePrevious2LinesFromTerminal() { + fmt.Printf("\033[1A") + fmt.Printf("\033[2K") + fmt.Printf("\033[1A") + fmt.Printf("\033[2K") +} + +func printRunStatus(event *s.HistoryEvent) { + switch event.GetEventType() { + case s.EventTypeWorkflowExecutionCompleted: + fmt.Printf(" Status: %s\n", colorGreen("COMPLETED")) + fmt.Printf(" Output: %s\n", string(event.WorkflowExecutionCompletedEventAttributes.Result)) + case s.EventTypeWorkflowExecutionFailed: + fmt.Printf(" Status: %s\n", colorRed("FAILED")) + fmt.Printf(" Reason: %s\n", event.WorkflowExecutionFailedEventAttributes.GetReason()) + fmt.Printf(" Detail: %s\n", string(event.WorkflowExecutionFailedEventAttributes.Details)) + case s.EventTypeWorkflowExecutionTimedOut: + fmt.Printf(" Status: %s\n", colorRed("TIMEOUT")) + fmt.Printf(" Timeout Type: %s\n", event.WorkflowExecutionTimedOutEventAttributes.GetTimeoutType()) + case s.EventTypeWorkflowExecutionCanceled: + fmt.Printf(" Status: %s\n", colorRed("CANCELED")) + fmt.Printf(" Detail: %s\n", string(event.WorkflowExecutionCanceledEventAttributes.Details)) + } +} diff --git a/tools/cli/domain.go b/tools/cli/domain.go new file mode 100644 index 00000000000..6e14587c281 --- /dev/null +++ b/tools/cli/domain.go @@ -0,0 +1,88 @@ +// Copyright (c) 2017 Uber Technologies, Inc. +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +// THE SOFTWARE. + +package cli + +import "github.com/urfave/cli" + +func newDomainCommands() []cli.Command { + return []cli.Command{ + { + Name: "register", + Aliases: []string{"re"}, + Usage: "Register workflow domain", + Flags: []cli.Flag{ + cli.StringFlag{ + Name: FlagDescriptionWithAlias, + Usage: "Domain description", + }, + cli.StringFlag{ + Name: FlagOwnerEmailWithAlias, + Usage: "Owner email", + }, + cli.StringFlag{ + Name: FlagRetentionDaysWithAlias, + Usage: "Workflow execution retention in days", + }, + cli.StringFlag{ + Name: FlagEmitMetricWithAlias, + Usage: "Flag to emit metric", + }, + }, + Action: func(c *cli.Context) { + RegisterDomain(c) + }, + }, + { + Name: "update", + Aliases: []string{"up", "u"}, + Usage: "Update existing workflow domain", + Flags: []cli.Flag{ + cli.StringFlag{ + Name: FlagDescriptionWithAlias, + Usage: "Domain description", + }, + cli.StringFlag{ + Name: FlagOwnerEmailWithAlias, + Usage: "Owner email", + }, + cli.StringFlag{ + Name: FlagRetentionDaysWithAlias, + Usage: "Workflow execution retention in days", + }, + cli.StringFlag{ + Name: FlagEmitMetricWithAlias, + Usage: "Flag to emit metric", + }, + }, + Action: func(c *cli.Context) { + UpdateDomain(c) + }, + }, + { + Name: "describe", + Aliases: []string{"desc"}, + Usage: "Describe existing workflow domain", + Action: func(c *cli.Context) { + DescribeDomain(c) + }, + }, + } +} diff --git a/tools/cli/factory.go b/tools/cli/factory.go new file mode 100644 index 00000000000..3b4f5e9b7cf --- /dev/null +++ b/tools/cli/factory.go @@ -0,0 +1,112 @@ +// Copyright (c) 2017 Uber Technologies, Inc. +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +// THE SOFTWARE. + +package cli + +import ( + "errors" + + "github.com/urfave/cli" + "go.uber.org/cadence/.gen/go/cadence/workflowserviceclient" + "go.uber.org/yarpc" + "go.uber.org/yarpc/transport/tchannel" + "go.uber.org/zap" +) + +const ( + _cadenceClientName = "cadence-client" + _cadenceFrontendService = "cadence-frontend" +) + +// WorkflowClientBuilderInterface is an interface to build client to cadence service. +// User can customize builder by implementing this interface, and call SetBulder after initialize cli. +// (See cadence/cmd/tools/cli/main.go for example) +// The customized builder may have more processing on Env, Address and other info. +type WorkflowClientBuilderInterface interface { + BuildServiceClient(c *cli.Context) (workflowserviceclient.Interface, error) +} + +// WorkflowClientBuilder build client to cadence service +type WorkflowClientBuilder struct { + hostPort string + dispatcher *yarpc.Dispatcher + logger *zap.Logger +} + +// NewBuilder creates a new WorkflowClientBuilder +func NewBuilder() *WorkflowClientBuilder { + logger, err := zap.NewDevelopment() + if err != nil { + panic(err) + } + + return &WorkflowClientBuilder{ + logger: logger, + } +} + +// BuildServiceClient builds a rpc service client to cadence service +func (b *WorkflowClientBuilder) BuildServiceClient(c *cli.Context) (workflowserviceclient.Interface, error) { + b.hostPort = localHostPort + if addr := c.GlobalString(FlagAddress); addr != "" { + b.hostPort = addr + } + + if err := b.build(); err != nil { + return nil, err + } + + if b.dispatcher == nil { + b.logger.Fatal("No RPC dispatcher provided to create a connection to Cadence Service") + } + + return workflowserviceclient.New(b.dispatcher.ClientConfig(_cadenceFrontendService)), nil +} + +func (b *WorkflowClientBuilder) build() error { + if b.dispatcher != nil { + return nil + } + + if len(b.hostPort) == 0 { + return errors.New("HostPort is empty") + } + + ch, err := tchannel.NewChannelTransport( + tchannel.ServiceName(_cadenceClientName)) + if err != nil { + b.logger.Fatal("Failed to create transport channel", zap.Error(err)) + } + + b.dispatcher = yarpc.NewDispatcher(yarpc.Config{ + Name: _cadenceClientName, + Outbounds: yarpc.Outbounds{ + _cadenceFrontendService: {Unary: ch.NewSingleOutbound(b.hostPort)}, + }, + }) + + if b.dispatcher != nil { + if err := b.dispatcher.Start(); err != nil { + b.logger.Fatal("Failed to create outbound transport channel: %v", zap.Error(err)) + } + } + + return nil +} diff --git a/tools/cli/util.go b/tools/cli/util.go new file mode 100644 index 00000000000..493d52d7ce6 --- /dev/null +++ b/tools/cli/util.go @@ -0,0 +1,247 @@ +// Copyright (c) 2017 Uber Technologies, Inc. +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +// THE SOFTWARE. + +package cli + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "reflect" + + s "go.uber.org/cadence/.gen/go/shared" + "go.uber.org/cadence/client" +) + +// JSONHistorySerializer is used to encode history event in JSON +type JSONHistorySerializer struct{} + +// Serialize serializes history. +func (j *JSONHistorySerializer) Serialize(h *s.History) ([]byte, error) { + return json.Marshal(h.Events) +} + +// Deserialize deserializes history +func (j *JSONHistorySerializer) Deserialize(data []byte) (*s.History, error) { + var events []*s.HistoryEvent + err := json.Unmarshal(data, &events) + if err != nil { + return nil, err + } + return &s.History{Events: events}, nil +} + +// GetHistory helper method to iterate over all pages and return complete list of history events +func GetHistory(ctx context.Context, workflowClient client.Client, workflowID, runID string) (*s.History, error) { + iter := workflowClient.GetWorkflowHistory(ctx, workflowID, runID, false, + s.HistoryEventFilterTypeAllEvent) + events := []*s.HistoryEvent{} + for iter.HasNext() { + event, err := iter.Next() + if err != nil { + return nil, err + } + events = append(events, event) + } + + history := &s.History{} + history.Events = events + return history, nil +} + +// HistoryEventToString convert HistoryEvent to string +func HistoryEventToString(e *s.HistoryEvent) string { + var data interface{} + switch e.GetEventType() { + case s.EventTypeWorkflowExecutionStarted: + data = e.WorkflowExecutionStartedEventAttributes + + case s.EventTypeWorkflowExecutionCompleted: + data = e.WorkflowExecutionCompletedEventAttributes + + case s.EventTypeWorkflowExecutionFailed: + data = e.WorkflowExecutionFailedEventAttributes + + case s.EventTypeWorkflowExecutionTimedOut: + data = e.WorkflowExecutionTimedOutEventAttributes + + case s.EventTypeDecisionTaskScheduled: + data = e.DecisionTaskScheduledEventAttributes + + case s.EventTypeDecisionTaskStarted: + data = e.DecisionTaskStartedEventAttributes + + case s.EventTypeDecisionTaskCompleted: + data = e.DecisionTaskCompletedEventAttributes + + case s.EventTypeDecisionTaskTimedOut: + data = e.DecisionTaskTimedOutEventAttributes + + case s.EventTypeActivityTaskScheduled: + data = e.ActivityTaskScheduledEventAttributes + + case s.EventTypeActivityTaskStarted: + data = e.ActivityTaskStartedEventAttributes + + case s.EventTypeActivityTaskCompleted: + data = e.ActivityTaskCompletedEventAttributes + + case s.EventTypeActivityTaskFailed: + data = e.ActivityTaskFailedEventAttributes + + case s.EventTypeActivityTaskTimedOut: + data = e.ActivityTaskTimedOutEventAttributes + + case s.EventTypeActivityTaskCancelRequested: + data = e.ActivityTaskCancelRequestedEventAttributes + + case s.EventTypeRequestCancelActivityTaskFailed: + data = e.RequestCancelActivityTaskFailedEventAttributes + + case s.EventTypeActivityTaskCanceled: + data = e.ActivityTaskCanceledEventAttributes + + case s.EventTypeTimerStarted: + data = e.TimerStartedEventAttributes + + case s.EventTypeTimerFired: + data = e.TimerFiredEventAttributes + + case s.EventTypeCancelTimerFailed: + data = e.CancelTimerFailedEventAttributes + + case s.EventTypeTimerCanceled: + data = e.TimerCanceledEventAttributes + + case s.EventTypeWorkflowExecutionCancelRequested: + data = e.WorkflowExecutionCancelRequestedEventAttributes + + case s.EventTypeWorkflowExecutionCanceled: + data = e.WorkflowExecutionCanceledEventAttributes + + case s.EventTypeRequestCancelExternalWorkflowExecutionInitiated: + data = e.RequestCancelExternalWorkflowExecutionInitiatedEventAttributes + + case s.EventTypeRequestCancelExternalWorkflowExecutionFailed: + data = e.RequestCancelExternalWorkflowExecutionFailedEventAttributes + + case s.EventTypeExternalWorkflowExecutionCancelRequested: + data = e.ExternalWorkflowExecutionCancelRequestedEventAttributes + + case s.EventTypeMarkerRecorded: + data = e.MarkerRecordedEventAttributes + + case s.EventTypeWorkflowExecutionSignaled: + data = e.WorkflowExecutionSignaledEventAttributes + + case s.EventTypeWorkflowExecutionTerminated: + data = e.WorkflowExecutionTerminatedEventAttributes + + case s.EventTypeWorkflowExecutionContinuedAsNew: + data = e.WorkflowExecutionContinuedAsNewEventAttributes + + case s.EventTypeStartChildWorkflowExecutionInitiated: + data = e.StartChildWorkflowExecutionInitiatedEventAttributes + + case s.EventTypeStartChildWorkflowExecutionFailed: + data = e.StartChildWorkflowExecutionFailedEventAttributes + + case s.EventTypeChildWorkflowExecutionStarted: + data = e.ChildWorkflowExecutionStartedEventAttributes + + case s.EventTypeChildWorkflowExecutionCompleted: + data = e.ChildWorkflowExecutionCompletedEventAttributes + + case s.EventTypeChildWorkflowExecutionFailed: + data = e.ChildWorkflowExecutionFailedEventAttributes + + case s.EventTypeChildWorkflowExecutionCanceled: + data = e.ChildWorkflowExecutionCanceledEventAttributes + + case s.EventTypeChildWorkflowExecutionTimedOut: + data = e.ChildWorkflowExecutionTimedOutEventAttributes + + case s.EventTypeChildWorkflowExecutionTerminated: + data = e.ChildWorkflowExecutionTerminatedEventAttributes + + case s.EventTypeSignalExternalWorkflowExecutionInitiated: + data = e.SignalExternalWorkflowExecutionInitiatedEventAttributes + + case s.EventTypeSignalExternalWorkflowExecutionFailed: + data = e.SignalExternalWorkflowExecutionFailedEventAttributes + + case s.EventTypeExternalWorkflowExecutionSignaled: + data = e.ExternalWorkflowExecutionSignaledEventAttributes + + default: + data = e + } + + return e.GetEventType().String() + ": " + anyToString(data) +} + +func anyToString(d interface{}) string { + v := reflect.ValueOf(d) + switch v.Kind() { + case reflect.Ptr: + return anyToString(v.Elem().Interface()) + case reflect.Struct: + var buf bytes.Buffer + t := reflect.TypeOf(d) + buf.WriteString("(") + for i := 0; i < v.NumField(); i++ { + f := v.Field(i) + if f.Kind() == reflect.Invalid { + continue + } + fieldValue := valueToString(f) + if len(fieldValue) == 0 { + continue + } + if buf.Len() > 1 { + buf.WriteString(", ") + } + buf.WriteString(fmt.Sprintf("%s:%s", t.Field(i).Name, fieldValue)) + } + buf.WriteString(")") + return buf.String() + default: + return fmt.Sprint(d) + } +} + +func valueToString(v reflect.Value) string { + switch v.Kind() { + case reflect.Ptr: + return valueToString(v.Elem()) + case reflect.Struct: + return anyToString(v.Interface()) + case reflect.Invalid: + return "" + case reflect.Slice: + if v.Type().Elem().Kind() == reflect.Uint8 { + return fmt.Sprintf("[%v]", string(v.Bytes())) + } + return fmt.Sprintf("[len=%d]", v.Len()) + default: + return fmt.Sprint(v.Interface()) + } +} diff --git a/tools/cli/workflow.go b/tools/cli/workflow.go new file mode 100644 index 00000000000..c6ab9e3fd11 --- /dev/null +++ b/tools/cli/workflow.go @@ -0,0 +1,307 @@ +// Copyright (c) 2017 Uber Technologies, Inc. +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +// THE SOFTWARE. + +package cli + +import "github.com/urfave/cli" + +func newWorkflowCommands() []cli.Command { + return []cli.Command{ + { + Name: "show", + Usage: "show workflow history", + Flags: []cli.Flag{ + cli.StringFlag{ + Name: FlagWorkflowIDWithAlias, + Usage: "WorkflowID", + }, + cli.StringFlag{ + Name: FlagRunIDWithAlias, + Usage: "RunID", + }, + cli.BoolFlag{ + Name: FlagPrintRawTimeWithAlias, + Usage: "Print raw time stamp", + }, + cli.StringFlag{ + Name: FlagOutputFilenameWithAlias, + Usage: "Serialize history event to a file", + }, + }, + Action: func(c *cli.Context) { + ShowHistory(c) + }, + }, + { + Name: "showid", + Usage: "show workflow history with given workflow_id and optional run_id (a shortcut of `show -w -r `)", + Description: "cadence workflow showid . workflow_id is required; run_id is optional", + Action: func(c *cli.Context) { + ShowHistoryWithWID(c) + }, + }, + { + Name: "start", + Usage: "start a new workflow execution", + Flags: []cli.Flag{ + cli.StringFlag{ + Name: FlagTaskListWithAlias, + Usage: "TaskList", + }, + cli.StringFlag{ + Name: FlagWorkflowIDWithAlias, + Usage: "WorkflowID", + }, + cli.StringFlag{ + Name: FlagWorkflowTypeWithAlias, + Usage: "WorkflowTypeName", + }, + cli.IntFlag{ + Name: FlagExecutionTimeoutWithAlias, + Usage: "Execution start to close timeout in seconds", + }, + cli.IntFlag{ + Name: FlagDecisionTimeoutWithAlias, + Value: defaultDecisionTimeoutInSeconds, + Usage: "Decision task start to close timeout in seconds", + }, + cli.StringFlag{ + Name: FlagInputWithAlias, + Usage: "Optional input for the workflow, in JSON format. If there are multiple parameters, concatenate them and separate by space.", + }, + cli.StringFlag{ + Name: FlagInputFileWithAlias, + Usage: "Optional input for the workflow from JSON file. If there are multiple JSON, concatenate them and separate by space or newline. " + + "Input from file will be overwrite by input from command line", + }, + }, + Action: func(c *cli.Context) { + StartWorkflow(c) + }, + }, + { + Name: "run", + Usage: "start a new workflow execution and get workflow progress", + Flags: []cli.Flag{ + cli.StringFlag{ + Name: FlagTaskListWithAlias, + Usage: "TaskList", + }, + cli.StringFlag{ + Name: FlagWorkflowIDWithAlias, + Usage: "WorkflowID", + }, + cli.StringFlag{ + Name: FlagWorkflowTypeWithAlias, + Usage: "WorkflowTypeName", + }, + cli.IntFlag{ + Name: FlagExecutionTimeoutWithAlias, + Usage: "Execution start to close timeout in seconds", + }, + cli.IntFlag{ + Name: FlagDecisionTimeoutWithAlias, + Value: defaultDecisionTimeoutInSeconds, + Usage: "Decision task start to close timeout in seconds", + }, + cli.IntFlag{ + Name: FlagContextTimeoutWithAlias, + Usage: "Optional timeout for start command context in seconds, default value is 120", + }, + cli.StringFlag{ + Name: FlagInputWithAlias, + Usage: "Optional input for the workflow, in JSON format. If there are multiple parameters, concatenate them and separate by space.", + }, + cli.StringFlag{ + Name: FlagInputFileWithAlias, + Usage: "Optional input for the workflow from JSON file. If there are multiple JSON, concatenate them and separate by space or newline. " + + "Input from file will be overwrite by input from command line", + }, + }, + Action: func(c *cli.Context) { + RunWorkflow(c) + }, + }, + { + Name: "cancel", + Aliases: []string{"c"}, + Usage: "cancel a workflow execution", + Flags: []cli.Flag{ + cli.StringFlag{ + Name: FlagWorkflowIDWithAlias, + Usage: "WorkflowID", + }, + cli.StringFlag{ + Name: FlagRunIDWithAlias, + Usage: "RunID", + }, + }, + Action: func(c *cli.Context) { + CancelWorkflow(c) + }, + }, + { + Name: "signal", + Aliases: []string{"s"}, + Usage: "signal a workflow execution", + Flags: []cli.Flag{ + cli.StringFlag{ + Name: FlagWorkflowIDWithAlias, + Usage: "WorkflowID", + }, + cli.StringFlag{ + Name: FlagRunIDWithAlias, + Usage: "RunID", + }, + cli.StringFlag{ + Name: FlagNameWithAlias, + Usage: "SignalName", + }, + cli.StringFlag{ + Name: FlagInputWithAlias, + Usage: "Input for the signal, in JSON format.", + }, + cli.StringFlag{ + Name: FlagInputFileWithAlias, + Usage: "Input for the signal from JSON file.", + }, + }, + Action: func(c *cli.Context) { + SignalWorkflow(c) + }, + }, + { + Name: "terminate", + Aliases: []string{"term"}, + Usage: "terminate a new workflow execution", + Flags: []cli.Flag{ + cli.StringFlag{ + Name: FlagWorkflowIDWithAlias, + Usage: "WorkflowID", + }, + cli.StringFlag{ + Name: FlagRunIDWithAlias, + Usage: "RunID", + }, + cli.StringFlag{ + Name: FlagReasonWithAlias, + Usage: "The reason you want to terminate the workflow", + }, + }, + Action: func(c *cli.Context) { + TerminateWorkflow(c) + }, + }, + { + Name: "list", + Usage: "list open or closed workflow executions", + Flags: []cli.Flag{ + cli.BoolFlag{ + Name: FlagOpenWithAlias, + Usage: "list for open workflow executions, default is to list for closed ones", + }, + cli.IntFlag{ + Name: FlagPageSizeWithAlias, + Value: 10, + Usage: "Result page size", + }, + cli.StringFlag{ + Name: FlagEarliestTimeWithAlias, + Usage: "EarliestTime of start time, supported formats are '2006-01-02T15:04:05Z07:00' and raw UnixNano", + }, + cli.StringFlag{ + Name: FlagLatestTimeWithAlias, + Usage: "LatestTime of start time, supported formats are '2006-01-02T15:04:05Z07:00' and raw UnixNano", + }, + cli.StringFlag{ + Name: FlagWorkflowIDWithAlias, + Usage: "WorkflowID", + }, + cli.StringFlag{ + Name: FlagWorkflowTypeWithAlias, + Usage: "WorkflowTypeName", + }, + cli.BoolFlag{ + Name: FlagPrintRawTimeWithAlias, + Usage: "Print raw time stamp", + }, + }, + Action: func(c *cli.Context) { + ListWorkflow(c) + }, + }, + { + Name: "query", + Usage: "query workflow execution", + Flags: []cli.Flag{ + cli.StringFlag{ + Name: FlagWorkflowIDWithAlias, + Usage: "WorkflowID", + }, + cli.StringFlag{ + Name: FlagRunIDWithAlias, + Usage: "RunID", + }, + cli.StringFlag{ + Name: FlagQueryTypeWithAlias, + Usage: "The query type you want to run", + }, + cli.StringFlag{ + Name: FlagInputWithAlias, + Usage: "Optional input for the query, in JSON format. If there are multiple parameters, concatenate them and separate by space.", + }, + cli.StringFlag{ + Name: FlagInputFileWithAlias, + Usage: "Optional input for the query from JSON file. If there are multiple JSON, concatenate them and separate by space or newline. " + + "Input from file will be overwrite by input from command line", + }, + }, + Action: func(c *cli.Context) { + QueryWorkflow(c) + }, + }, + { + Name: "stack", + Usage: "query workflow execution with __stack_trace as query type", + Flags: []cli.Flag{ + cli.StringFlag{ + Name: FlagWorkflowIDWithAlias, + Usage: "WorkflowID", + }, + cli.StringFlag{ + Name: FlagRunIDWithAlias, + Usage: "RunID", + }, + cli.StringFlag{ + Name: FlagInputWithAlias, + Usage: "Optional input for the query, in JSON format. If there are multiple parameters, concatenate them and separate by space.", + }, + cli.StringFlag{ + Name: FlagInputFileWithAlias, + Usage: "Optional input for the query from JSON file. If there are multiple JSON, concatenate them and separate by space or newline. " + + "Input from file will be overwrite by input from command line", + }, + }, + Action: func(c *cli.Context) { + QueryWorkflowUsingStackTrace(c) + }, + }, + } +}