diff --git a/internal/commandsgen/docs.go b/internal/commandsgen/docs.go index d699c23f6..5bfdcbd56 100644 --- a/internal/commandsgen/docs.go +++ b/internal/commandsgen/docs.go @@ -104,7 +104,7 @@ func (w *docWriter) writeCommand(c *Command) { w.fileMap[fileName].WriteString("title: Temporal CLI " + fileName + " command reference\n") w.fileMap[fileName].WriteString("sidebar_label: " + fileName + "\n") w.fileMap[fileName].WriteString("description: " + c.Docs.DescriptionHeader + "\n") - w.fileMap[fileName].WriteString("toc_max_heading_level: 4\n") + w.fileMap[fileName].WriteString("toc_max_heading_level: 6\n") w.fileMap[fileName].WriteString("keywords:\n") for _, keyword := range c.Docs.Keywords { @@ -217,7 +217,7 @@ func (w *docWriter) writeSplitCommand(c *Command, splitRoot string) { buf.WriteString("title: Temporal CLI " + c.FullName + " command reference\n") buf.WriteString("sidebar_label: " + leafName + "\n") buf.WriteString("description: " + c.Docs.DescriptionHeader + "\n") - buf.WriteString("toc_max_heading_level: 4\n") + buf.WriteString("toc_max_heading_level: 6\n") buf.WriteString("keywords:\n") for _, keyword := range c.Docs.Keywords { buf.WriteString(" - " + keyword + "\n") diff --git a/internal/temporalcli/commands.gen.go b/internal/temporalcli/commands.gen.go index 2074c14d8..eb961410f 100644 --- a/internal/temporalcli/commands.gen.go +++ b/internal/temporalcli/commands.gen.go @@ -122,6 +122,54 @@ func (v *DeploymentVersionOptions) BuildFlags(f *pflag.FlagSet) { _ = cobra.MarkFlagRequired(f, "build-id") } +type ScalingGroupOptions struct { + ScalingGroup string + FlagSet *pflag.FlagSet +} + +func (v *ScalingGroupOptions) BuildFlags(f *pflag.FlagSet) { + v.FlagSet = f + f.StringVar(&v.ScalingGroup, "scaling-group", "default", "Key of the scaling group.") +} + +type AwsAssumeRoleOptions struct { + AssumeRoleArn string + AssumeRoleExternalId string + SkipRoleAndExternalId bool + FlagSet *pflag.FlagSet +} + +func (v *AwsAssumeRoleOptions) BuildFlags(f *pflag.FlagSet) { + v.FlagSet = f + f.StringVar(&v.AssumeRoleArn, "assume-role-arn", "", "AWS IAM role ARN that the Temporal server will assume when calling the compute provider that spawns a new Worker in this Worker Deployment Version. Required when setting the compute provider, and must be omitted when --skip-role-and-external-id is passed.") + f.StringVar(&v.AssumeRoleExternalId, "assume-role-external-id", "", "Temporal server will enforce that the AWS IAM trust policy associated with the AWS IAM role specified in --assume-role-arn has an aws:ExternalId condition that matches the supplied value. Required when setting the compute provider, and must be omitted when --skip-role-and-external-id is passed.") + f.BoolVar(&v.SkipRoleAndExternalId, "skip-role-and-external-id", false, "When setting the compute provider, --assume-role-arn and --assume-role-external-id are required unless this flag is passed, in which case both must be omitted.") +} + +type NoSyncScalerOptions struct { + ScaleUpBacklogThreshold int + ScaleUpCooloffMs int + MaxWorkerLifetimeMs int + MetricsPollIntervalMs int + ScaleUpDispatchRateEpsilon float32 + ScaleUpDispatchRateEpsilonConfirmMs int + SuppressScaleUpMs int + SuppressPollIntervalMs int + FlagSet *pflag.FlagSet +} + +func (v *NoSyncScalerOptions) BuildFlags(f *pflag.FlagSet) { + v.FlagSet = f + f.IntVar(&v.ScaleUpBacklogThreshold, "scale-up-backlog-threshold", 0, "At each Task Queue metrics check (see --metrics-poll-interval-ms), start a new Worker if the backlog is larger than this number of tasks. Defaults to 0 when unset. A task that arrives while no Worker is waiting for it still starts a Worker, subject to --scale-up-cooloff-ms.") + f.IntVar(&v.ScaleUpCooloffMs, "scale-up-cooloff-ms", 0, "Minimum time between two Worker starts, in milliseconds. A value of 0 disables the wait. Defaults to 100 when unset.") + f.IntVar(&v.MaxWorkerLifetimeMs, "max-worker-lifetime-ms", 0, "While there is a backlog, start a new Worker at least this often, in milliseconds, even within --scale-up-cooloff-ms. A value of 0 disables this. Defaults to 600000 (10 minutes) when unset.") + f.IntVar(&v.MetricsPollIntervalMs, "metrics-poll-interval-ms", 0, "How often Task Queue metrics are checked to decide whether to start a Worker, in milliseconds. Must be at least 10000, and at least --scale-up-cooloff-ms. Defaults to 60000 (1 minute) when unset.") + f.Float32Var(&v.ScaleUpDispatchRateEpsilon, "scale-up-dispatch-rate-epsilon", 0, "Band around the dispatch rate, as a fraction of it, of at most 0.1. While a Task Queue type has a backlog and its dispatch rate stays within this band, Worker starts for it are paused. Starts from --max-worker-lifetime-ms are not paused. Defaults to 0 when unset, which disables pausing.") + f.IntVar(&v.ScaleUpDispatchRateEpsilonConfirmMs, "scale-up-dispatch-rate-epsilon-confirm-ms", 0, "How long the dispatch rate must stay within --scale-up-dispatch-rate-epsilon before Worker starts are paused, in milliseconds. Must be longer than --metrics-poll-interval-ms. Only valid with --scale-up-dispatch-rate-epsilon. Defaults to 90000 when unset.") + f.IntVar(&v.SuppressScaleUpMs, "suppress-scale-up-ms", 0, "How long Worker starts stay paused each time the dispatch rate is found within --scale-up-dispatch-rate-epsilon, in milliseconds. Must be longer than --suppress-poll-interval-ms. Only valid with --scale-up-dispatch-rate-epsilon. Defaults to 120000 when unset.") + f.IntVar(&v.SuppressPollIntervalMs, "suppress-poll-interval-ms", 0, "How often Task Queue metrics are checked while Worker starts are paused, in milliseconds. Only valid with --scale-up-dispatch-rate-epsilon. Defaults to 90000 when unset.") +} + type DeploymentVersionOrUnversionedOptions struct { DeploymentName string BuildId string @@ -3729,6 +3777,7 @@ func NewTemporalWorkerDeploymentCommand(cctx *CommandContext, parent *TemporalWo s.Command.AddCommand(&NewTemporalWorkerDeploymentSetRampingVersionCommand(cctx, &s).Command) s.Command.AddCommand(&NewTemporalWorkerDeploymentUpdateVersionComputeConfigCommand(cctx, &s).Command) s.Command.AddCommand(&NewTemporalWorkerDeploymentUpdateVersionMetadataCommand(cctx, &s).Command) + s.Command.AddCommand(&NewTemporalWorkerDeploymentVersionCommand(cctx, &s).Command) return &s } @@ -4206,6 +4255,286 @@ func NewTemporalWorkerDeploymentUpdateVersionMetadataCommand(cctx *CommandContex return &s } +type TemporalWorkerDeploymentVersionCommand struct { + Parent *TemporalWorkerDeploymentCommand + Command cobra.Command +} + +func NewTemporalWorkerDeploymentVersionCommand(cctx *CommandContext, parent *TemporalWorkerDeploymentCommand) *TemporalWorkerDeploymentVersionCommand { + var s TemporalWorkerDeploymentVersionCommand + s.Parent = parent + s.Command.Use = "version" + s.Command.Short = "Create, describe, or delete a Worker Deployment Version, and manage its compute configuration (Experimental)" + if hasHighlighting { + s.Command.Long = "Version commands create, describe, and delete a single Worker\nDeployment Version, and manage its compute configuration:\n\n\x1b[1mtemporal worker deployment version [command] [options]\x1b[0m\n\nNote: This is an experimental feature and may change in the future." + } else { + s.Command.Long = "Version commands create, describe, and delete a single Worker\nDeployment Version, and manage its compute configuration:\n\n```\ntemporal worker deployment version [command] [options]\n```\n\nNote: This is an experimental feature and may change in the future." + } + s.Command.Args = cobra.NoArgs + s.Command.AddCommand(&NewTemporalWorkerDeploymentVersionComputeConfigCommand(cctx, &s).Command) + s.Command.AddCommand(&NewTemporalWorkerDeploymentVersionCreateCommand(cctx, &s).Command) + s.Command.AddCommand(&NewTemporalWorkerDeploymentVersionDeleteCommand(cctx, &s).Command) + s.Command.AddCommand(&NewTemporalWorkerDeploymentVersionDescribeCommand(cctx, &s).Command) + return &s +} + +type TemporalWorkerDeploymentVersionComputeConfigCommand struct { + Parent *TemporalWorkerDeploymentVersionCommand + Command cobra.Command +} + +func NewTemporalWorkerDeploymentVersionComputeConfigCommand(cctx *CommandContext, parent *TemporalWorkerDeploymentVersionCommand) *TemporalWorkerDeploymentVersionComputeConfigCommand { + var s TemporalWorkerDeploymentVersionComputeConfigCommand + s.Parent = parent + s.Command.Use = "compute-config" + s.Command.Short = "Set, update, or remove compute configuration for a Version (Experimental)" + if hasHighlighting { + s.Command.Long = "A Worker Deployment Version's compute configuration is made of one\nor more scaling groups. Each scaling group is identified by a key\nand is made of a compute provider that starts Workers and a scaler\nthat decides when to start them. Scaling groups set from the CLI\nserve every Task Queue type. Each compute provider has its own\ncommand, such as \x1b[1maws-lambda\x1b[0m, to create and update scaling groups:\n\n\x1b[1mtemporal worker deployment version compute-config [command] [options]\x1b[0m\n\nNote: This is an experimental feature and may change in the future." + } else { + s.Command.Long = "A Worker Deployment Version's compute configuration is made of one\nor more scaling groups. Each scaling group is identified by a key\nand is made of a compute provider that starts Workers and a scaler\nthat decides when to start them. Scaling groups set from the CLI\nserve every Task Queue type. Each compute provider has its own\ncommand, such as `aws-lambda`, to create and update scaling groups:\n\n```\ntemporal worker deployment version compute-config [command] [options]\n```\n\nNote: This is an experimental feature and may change in the future." + } + s.Command.Args = cobra.NoArgs + s.Command.AddCommand(&NewTemporalWorkerDeploymentVersionComputeConfigAwsLambdaCommand(cctx, &s).Command) + s.Command.AddCommand(&NewTemporalWorkerDeploymentVersionComputeConfigDescribeCommand(cctx, &s).Command) + s.Command.AddCommand(&NewTemporalWorkerDeploymentVersionComputeConfigRemoveCommand(cctx, &s).Command) + return &s +} + +type TemporalWorkerDeploymentVersionComputeConfigAwsLambdaCommand struct { + Parent *TemporalWorkerDeploymentVersionComputeConfigCommand + Command cobra.Command +} + +func NewTemporalWorkerDeploymentVersionComputeConfigAwsLambdaCommand(cctx *CommandContext, parent *TemporalWorkerDeploymentVersionComputeConfigCommand) *TemporalWorkerDeploymentVersionComputeConfigAwsLambdaCommand { + var s TemporalWorkerDeploymentVersionComputeConfigAwsLambdaCommand + s.Parent = parent + s.Command.Use = "aws-lambda" + s.Command.Short = "Create or update AWS Lambda compute configuration (Experimental)" + if hasHighlighting { + s.Command.Long = "AWS Lambda commands set the compute configuration of a Worker Deployment\nVersion for an AWS Lambda function that spawns a Worker in the Worker\nDeployment:\n\n\x1b[1mtemporal worker deployment version compute-config aws-lambda [command] [options]\x1b[0m\n\nNote: This is an experimental feature and may change in the future." + } else { + s.Command.Long = "AWS Lambda commands set the compute configuration of a Worker Deployment\nVersion for an AWS Lambda function that spawns a Worker in the Worker\nDeployment:\n\n```\ntemporal worker deployment version compute-config aws-lambda [command] [options]\n```\n\nNote: This is an experimental feature and may change in the future." + } + s.Command.Args = cobra.NoArgs + s.Command.AddCommand(&NewTemporalWorkerDeploymentVersionComputeConfigAwsLambdaCreateCommand(cctx, &s).Command) + s.Command.AddCommand(&NewTemporalWorkerDeploymentVersionComputeConfigAwsLambdaUpdateCommand(cctx, &s).Command) + return &s +} + +type TemporalWorkerDeploymentVersionComputeConfigAwsLambdaCreateCommand struct { + Parent *TemporalWorkerDeploymentVersionComputeConfigAwsLambdaCommand + Command cobra.Command + DeploymentVersionOptions + ScalingGroupOptions + AwsAssumeRoleOptions + NoSyncScalerOptions + FunctionArn string +} + +func NewTemporalWorkerDeploymentVersionComputeConfigAwsLambdaCreateCommand(cctx *CommandContext, parent *TemporalWorkerDeploymentVersionComputeConfigAwsLambdaCommand) *TemporalWorkerDeploymentVersionComputeConfigAwsLambdaCreateCommand { + var s TemporalWorkerDeploymentVersionComputeConfigAwsLambdaCreateCommand + s.Parent = parent + s.Command.DisableFlagsInUseLine = true + s.Command.Use = "create [flags]" + s.Command.Short = "Create a Version with AWS Lambda compute configuration (Experimental)" + if hasHighlighting { + s.Command.Long = "Create a new Worker Deployment Version with compute configuration for an\nAWS Lambda function that spawns a Worker in the Worker Deployment:\n\n\x1b[1mtemporal worker deployment version compute-config aws-lambda create \\\n --deployment-name YourDeploymentName \\\n --build-id YourBuildID \\\n --function-arn YourLambdaFunctionARN \\\n --assume-role-arn YourAssumeRoleARN \\\n --assume-role-external-id YourExternalID\x1b[0m\n\nTo tune when Workers are started, pass any of the scaler flags (the\n--scale-up-* and --suppress-* flags, --max-worker-lifetime-ms, and\n--metrics-poll-interval-ms). Each one that is not set keeps the\nserver default:\n\n\x1b[1mtemporal worker deployment version compute-config aws-lambda create \\\n --deployment-name YourDeploymentName \\\n --build-id YourBuildID \\\n --function-arn YourLambdaFunctionARN \\\n --assume-role-arn YourAssumeRoleARN \\\n --assume-role-external-id YourExternalID \\\n --scale-up-backlog-threshold 5 \\\n --scale-up-cooloff-ms 500 \\\n --max-worker-lifetime-ms 300000\x1b[0m\n\nThe Worker Deployment must already exist; create it with\n\x1b[1mtemporal worker deployment create\x1b[0m.\n\nIf a Worker Deployment Version with the supplied BuildID already exists,\nthis command will return an error. To update its compute configuration,\nuse \x1b[1mtemporal worker deployment version compute-config aws-lambda update\x1b[0m.\n\nNote: This is an experimental feature and may change in the future." + } else { + s.Command.Long = "Create a new Worker Deployment Version with compute configuration for an\nAWS Lambda function that spawns a Worker in the Worker Deployment:\n\n```\ntemporal worker deployment version compute-config aws-lambda create \\\n --deployment-name YourDeploymentName \\\n --build-id YourBuildID \\\n --function-arn YourLambdaFunctionARN \\\n --assume-role-arn YourAssumeRoleARN \\\n --assume-role-external-id YourExternalID\n```\n\nTo tune when Workers are started, pass any of the scaler flags (the\n--scale-up-* and --suppress-* flags, --max-worker-lifetime-ms, and\n--metrics-poll-interval-ms). Each one that is not set keeps the\nserver default:\n\n```\ntemporal worker deployment version compute-config aws-lambda create \\\n --deployment-name YourDeploymentName \\\n --build-id YourBuildID \\\n --function-arn YourLambdaFunctionARN \\\n --assume-role-arn YourAssumeRoleARN \\\n --assume-role-external-id YourExternalID \\\n --scale-up-backlog-threshold 5 \\\n --scale-up-cooloff-ms 500 \\\n --max-worker-lifetime-ms 300000\n```\n\nThe Worker Deployment must already exist; create it with\n`temporal worker deployment create`.\n\nIf a Worker Deployment Version with the supplied BuildID already exists,\nthis command will return an error. To update its compute configuration,\nuse `temporal worker deployment version compute-config aws-lambda update`.\n\nNote: This is an experimental feature and may change in the future." + } + s.Command.Args = cobra.NoArgs + s.Command.Flags().StringVar(&s.FunctionArn, "function-arn", "", "Qualified (contains version suffix) or unqualified AWS Lambda function ARN to invoke when there are no active pollers for task queue targets in the Worker Deployment. Required.") + _ = cobra.MarkFlagRequired(s.Command.Flags(), "function-arn") + s.DeploymentVersionOptions.BuildFlags(s.Command.Flags()) + s.ScalingGroupOptions.BuildFlags(s.Command.Flags()) + s.AwsAssumeRoleOptions.BuildFlags(s.Command.Flags()) + s.NoSyncScalerOptions.BuildFlags(s.Command.Flags()) + s.Command.Run = func(c *cobra.Command, args []string) { + if err := s.run(cctx, args); err != nil { + cctx.Options.Fail(err) + } + } + return &s +} + +type TemporalWorkerDeploymentVersionComputeConfigAwsLambdaUpdateCommand struct { + Parent *TemporalWorkerDeploymentVersionComputeConfigAwsLambdaCommand + Command cobra.Command + DeploymentVersionOptions + ScalingGroupOptions + AwsAssumeRoleOptions + NoSyncScalerOptions + FunctionArn string +} + +func NewTemporalWorkerDeploymentVersionComputeConfigAwsLambdaUpdateCommand(cctx *CommandContext, parent *TemporalWorkerDeploymentVersionComputeConfigAwsLambdaCommand) *TemporalWorkerDeploymentVersionComputeConfigAwsLambdaUpdateCommand { + var s TemporalWorkerDeploymentVersionComputeConfigAwsLambdaUpdateCommand + s.Parent = parent + s.Command.DisableFlagsInUseLine = true + s.Command.Use = "update [flags]" + s.Command.Short = "Update AWS Lambda compute configuration for a Version (Experimental)" + if hasHighlighting { + s.Command.Long = "Update a scaling group in the compute configuration of a Worker\nDeployment Version, for an AWS Lambda function that spawns a Worker in\nthe Worker Deployment. Only the scaling group named by --scaling-group\n(\"default\" when unset) changes; other scaling groups are not affected.\n\nTo update the AWS Lambda function and the AWS IAM role associated with\nan existing Worker Deployment Version:\n\n\x1b[1mtemporal worker deployment version compute-config aws-lambda update \\\n --deployment-name YourDeploymentName \\\n --build-id YourBuildID \\\n --function-arn YourLambdaFunctionARN \\\n --assume-role-arn YourAssumeRoleARN \\\n --assume-role-external-id YourExternalID\x1b[0m\n\nTo update only the scaler of an existing scaling group, supply the\nscaler flags without the provider flags. The scaler is replaced as a\nwhole, so scaler flags that are not passed go back to their\ndefaults:\n\n\x1b[1mtemporal worker deployment version compute-config aws-lambda update \\\n --deployment-name YourDeploymentName \\\n --build-id YourBuildID \\\n --scale-up-backlog-threshold 5 \\\n --max-worker-lifetime-ms 300000\x1b[0m\n\nThe provider flags are --function-arn, --assume-role-arn,\n--assume-role-external-id, and --skip-role-and-external-id. The\nscaler flags are the --scale-up-* and --suppress-* flags,\n--max-worker-lifetime-ms, and --metrics-poll-interval-ms. The\nprovider flags are only needed when changing the compute provider,\nand then must be passed together: --function-arn with either both\nrole flags or --skip-role-and-external-id. Changing the provider\nalso resets the scaler: scaler flags passed with it are applied, and\nthe rest, or all of them if none are passed, go back to their\ndefaults.\n\nIf a Worker Deployment Version with the supplied BuildID does not exist,\nthis command will return an error. To create it, use\n\x1b[1mtemporal worker deployment version compute-config aws-lambda create\x1b[0m.\n\nReturns an error if all compute configuration fields are empty.\n\nNote: This is an experimental feature and may change in the future." + } else { + s.Command.Long = "Update a scaling group in the compute configuration of a Worker\nDeployment Version, for an AWS Lambda function that spawns a Worker in\nthe Worker Deployment. Only the scaling group named by --scaling-group\n(\"default\" when unset) changes; other scaling groups are not affected.\n\nTo update the AWS Lambda function and the AWS IAM role associated with\nan existing Worker Deployment Version:\n\n```\ntemporal worker deployment version compute-config aws-lambda update \\\n --deployment-name YourDeploymentName \\\n --build-id YourBuildID \\\n --function-arn YourLambdaFunctionARN \\\n --assume-role-arn YourAssumeRoleARN \\\n --assume-role-external-id YourExternalID\n```\n\nTo update only the scaler of an existing scaling group, supply the\nscaler flags without the provider flags. The scaler is replaced as a\nwhole, so scaler flags that are not passed go back to their\ndefaults:\n\n```\ntemporal worker deployment version compute-config aws-lambda update \\\n --deployment-name YourDeploymentName \\\n --build-id YourBuildID \\\n --scale-up-backlog-threshold 5 \\\n --max-worker-lifetime-ms 300000\n```\n\nThe provider flags are --function-arn, --assume-role-arn,\n--assume-role-external-id, and --skip-role-and-external-id. The\nscaler flags are the --scale-up-* and --suppress-* flags,\n--max-worker-lifetime-ms, and --metrics-poll-interval-ms. The\nprovider flags are only needed when changing the compute provider,\nand then must be passed together: --function-arn with either both\nrole flags or --skip-role-and-external-id. Changing the provider\nalso resets the scaler: scaler flags passed with it are applied, and\nthe rest, or all of them if none are passed, go back to their\ndefaults.\n\nIf a Worker Deployment Version with the supplied BuildID does not exist,\nthis command will return an error. To create it, use\n`temporal worker deployment version compute-config aws-lambda create`.\n\nReturns an error if all compute configuration fields are empty.\n\nNote: This is an experimental feature and may change in the future." + } + s.Command.Args = cobra.NoArgs + s.Command.Flags().StringVar(&s.FunctionArn, "function-arn", "", "Qualified (contains version suffix) or unqualified AWS Lambda function ARN to invoke when there are no active pollers for task queue targets in the Worker Deployment. Required when changing the compute provider.") + s.DeploymentVersionOptions.BuildFlags(s.Command.Flags()) + s.ScalingGroupOptions.BuildFlags(s.Command.Flags()) + s.AwsAssumeRoleOptions.BuildFlags(s.Command.Flags()) + s.NoSyncScalerOptions.BuildFlags(s.Command.Flags()) + s.Command.Run = func(c *cobra.Command, args []string) { + if err := s.run(cctx, args); err != nil { + cctx.Options.Fail(err) + } + } + return &s +} + +type TemporalWorkerDeploymentVersionComputeConfigDescribeCommand struct { + Parent *TemporalWorkerDeploymentVersionComputeConfigCommand + Command cobra.Command + DeploymentVersionOptions +} + +func NewTemporalWorkerDeploymentVersionComputeConfigDescribeCommand(cctx *CommandContext, parent *TemporalWorkerDeploymentVersionComputeConfigCommand) *TemporalWorkerDeploymentVersionComputeConfigDescribeCommand { + var s TemporalWorkerDeploymentVersionComputeConfigDescribeCommand + s.Parent = parent + s.Command.DisableFlagsInUseLine = true + s.Command.Use = "describe [flags]" + s.Command.Short = "Show the compute configuration of a Version (Experimental)" + if hasHighlighting { + s.Command.Long = "Show the scaling groups in the compute configuration of a Worker\nDeployment Version, with their compute provider and scaler settings:\n\n\x1b[1mtemporal worker deployment version compute-config describe \\\n --deployment-name YourDeploymentName \\\n --build-id YourBuildID\x1b[0m\n\nIf a Worker Deployment Version with the supplied BuildID does not exist,\nthis command will return an error.\n\nNote: This is an experimental feature and may change in the future." + } else { + s.Command.Long = "Show the scaling groups in the compute configuration of a Worker\nDeployment Version, with their compute provider and scaler settings:\n\n```\ntemporal worker deployment version compute-config describe \\\n --deployment-name YourDeploymentName \\\n --build-id YourBuildID\n```\n\nIf a Worker Deployment Version with the supplied BuildID does not exist,\nthis command will return an error.\n\nNote: This is an experimental feature and may change in the future." + } + s.Command.Args = cobra.NoArgs + s.DeploymentVersionOptions.BuildFlags(s.Command.Flags()) + s.Command.Run = func(c *cobra.Command, args []string) { + if err := s.run(cctx, args); err != nil { + cctx.Options.Fail(err) + } + } + return &s +} + +type TemporalWorkerDeploymentVersionComputeConfigRemoveCommand struct { + Parent *TemporalWorkerDeploymentVersionComputeConfigCommand + Command cobra.Command + DeploymentVersionOptions + ScalingGroup string + Yes bool +} + +func NewTemporalWorkerDeploymentVersionComputeConfigRemoveCommand(cctx *CommandContext, parent *TemporalWorkerDeploymentVersionComputeConfigCommand) *TemporalWorkerDeploymentVersionComputeConfigRemoveCommand { + var s TemporalWorkerDeploymentVersionComputeConfigRemoveCommand + s.Parent = parent + s.Command.DisableFlagsInUseLine = true + s.Command.Use = "remove [flags]" + s.Command.Short = "Remove compute configuration from a Version (Experimental)" + if hasHighlighting { + s.Command.Long = "Remove the compute configuration of a Worker Deployment Version. The\ncommand lists its scaling groups and asks for confirmation first:\n\n\x1b[1mtemporal worker deployment version compute-config remove \\\n --deployment-name YourDeploymentName \\\n --build-id YourBuildID\x1b[0m\n\nRemove one scaling group from the compute configuration, such as the one\nnamed \"default\":\n\n\x1b[1mtemporal worker deployment version compute-config remove \\\n --deployment-name YourDeploymentName \\\n --build-id YourBuildID \\\n --scaling-group default\x1b[0m\n\nIf the Worker Deployment Version or the scaling group does not exist,\nthis command will return an error.\n\nNote: This is an experimental feature and may change in the future." + } else { + s.Command.Long = "Remove the compute configuration of a Worker Deployment Version. The\ncommand lists its scaling groups and asks for confirmation first:\n\n```\ntemporal worker deployment version compute-config remove \\\n --deployment-name YourDeploymentName \\\n --build-id YourBuildID\n```\n\nRemove one scaling group from the compute configuration, such as the one\nnamed \"default\":\n\n```\ntemporal worker deployment version compute-config remove \\\n --deployment-name YourDeploymentName \\\n --build-id YourBuildID \\\n --scaling-group default\n```\n\nIf the Worker Deployment Version or the scaling group does not exist,\nthis command will return an error.\n\nNote: This is an experimental feature and may change in the future." + } + s.Command.Args = cobra.NoArgs + s.Command.Flags().StringVar(&s.ScalingGroup, "scaling-group", "", "Key of the scaling group to remove. If unset, all scaling groups are removed.") + s.Command.Flags().BoolVarP(&s.Yes, "yes", "y", false, "Don't prompt to confirm removing all scaling groups.") + s.DeploymentVersionOptions.BuildFlags(s.Command.Flags()) + s.Command.Run = func(c *cobra.Command, args []string) { + if err := s.run(cctx, args); err != nil { + cctx.Options.Fail(err) + } + } + return &s +} + +type TemporalWorkerDeploymentVersionCreateCommand struct { + Parent *TemporalWorkerDeploymentVersionCommand + Command cobra.Command + DeploymentVersionOptions +} + +func NewTemporalWorkerDeploymentVersionCreateCommand(cctx *CommandContext, parent *TemporalWorkerDeploymentVersionCommand) *TemporalWorkerDeploymentVersionCreateCommand { + var s TemporalWorkerDeploymentVersionCreateCommand + s.Parent = parent + s.Command.DisableFlagsInUseLine = true + s.Command.Use = "create [flags]" + s.Command.Short = "Create a new Worker Deployment Version (Experimental)" + if hasHighlighting { + s.Command.Long = "Create a new Worker Deployment Version:\n\n\x1b[1mtemporal worker deployment version create \\\n --deployment-name YourDeploymentName \\\n --build-id YourBuildID\x1b[0m\n\nThe Worker Deployment must already exist; create it with\n\x1b[1mtemporal worker deployment create\x1b[0m.\n\nIf a Worker Deployment Version with the supplied BuildID already exists,\nthis command will return an error.\n\nNote: This is an experimental feature and may change in the future." + } else { + s.Command.Long = "Create a new Worker Deployment Version:\n\n```\ntemporal worker deployment version create \\\n --deployment-name YourDeploymentName \\\n --build-id YourBuildID\n```\n\nThe Worker Deployment must already exist; create it with\n`temporal worker deployment create`.\n\nIf a Worker Deployment Version with the supplied BuildID already exists,\nthis command will return an error.\n\nNote: This is an experimental feature and may change in the future." + } + s.Command.Args = cobra.NoArgs + s.DeploymentVersionOptions.BuildFlags(s.Command.Flags()) + s.Command.Run = func(c *cobra.Command, args []string) { + if err := s.run(cctx, args); err != nil { + cctx.Options.Fail(err) + } + } + return &s +} + +type TemporalWorkerDeploymentVersionDeleteCommand struct { + Parent *TemporalWorkerDeploymentVersionCommand + Command cobra.Command + DeploymentVersionOptions + SkipDrainage bool +} + +func NewTemporalWorkerDeploymentVersionDeleteCommand(cctx *CommandContext, parent *TemporalWorkerDeploymentVersionCommand) *TemporalWorkerDeploymentVersionDeleteCommand { + var s TemporalWorkerDeploymentVersionDeleteCommand + s.Parent = parent + s.Command.DisableFlagsInUseLine = true + s.Command.Use = "delete [flags]" + s.Command.Short = "Delete a Worker Deployment Version (Experimental)" + if hasHighlighting { + s.Command.Long = "Remove a Worker Deployment Version given its fully-qualified identifier.\nThis is rarely needed during normal operation\nsince unused Versions are eventually garbage collected.\nThe client can delete a Version only when all of the following conditions\nare met:\n - It is not the Current or Ramping Version for this Deployment.\n - It has no active pollers, i.e., none of the task queues in the\n Version have pollers.\n - It is not draining. This requirement can be ignored with the option\n\x1b[1m--skip-drainage\x1b[0m.\n\x1b[1mtemporal worker deployment version delete [options]\x1b[0m\n\nFor example, skipping the drainage restriction:\n\n\x1b[1mtemporal worker deployment version delete \\\n --deployment-name YourDeploymentName \\\n --build-id YourBuildID \\\n --skip-drainage\x1b[0m\n\nNote: This is an experimental feature and may change in the future." + } else { + s.Command.Long = "Remove a Worker Deployment Version given its fully-qualified identifier.\nThis is rarely needed during normal operation\nsince unused Versions are eventually garbage collected.\nThe client can delete a Version only when all of the following conditions\nare met:\n - It is not the Current or Ramping Version for this Deployment.\n - It has no active pollers, i.e., none of the task queues in the\n Version have pollers.\n - It is not draining. This requirement can be ignored with the option\n`--skip-drainage`.\n```\ntemporal worker deployment version delete [options]\n```\n\nFor example, skipping the drainage restriction:\n\n```\ntemporal worker deployment version delete \\\n --deployment-name YourDeploymentName \\\n --build-id YourBuildID \\\n --skip-drainage\n```\n\nNote: This is an experimental feature and may change in the future." + } + s.Command.Args = cobra.NoArgs + s.Command.Flags().BoolVar(&s.SkipDrainage, "skip-drainage", false, "Ignore the deletion requirement of not draining.") + s.DeploymentVersionOptions.BuildFlags(s.Command.Flags()) + s.Command.Run = func(c *cobra.Command, args []string) { + if err := s.run(cctx, args); err != nil { + cctx.Options.Fail(err) + } + } + return &s +} + +type TemporalWorkerDeploymentVersionDescribeCommand struct { + Parent *TemporalWorkerDeploymentVersionCommand + Command cobra.Command + DeploymentVersionOptions + ReportTaskQueueStats bool +} + +func NewTemporalWorkerDeploymentVersionDescribeCommand(cctx *CommandContext, parent *TemporalWorkerDeploymentVersionCommand) *TemporalWorkerDeploymentVersionDescribeCommand { + var s TemporalWorkerDeploymentVersionDescribeCommand + s.Parent = parent + s.Command.DisableFlagsInUseLine = true + s.Command.Use = "describe [flags]" + s.Command.Short = "Show properties of a Worker Deployment Version (Experimental)" + if hasHighlighting { + s.Command.Long = "Describe properties of a Worker Deployment Version, such as the task\nqueues polled by workers in this Deployment Version, or drainage\ninformation required to safely decommission workers, or user-provided\nmetadata, or its creation/modification time.\n\n\x1b[1mtemporal worker deployment version describe [options]\x1b[0m\n\nFor example, to describe a deployment version in a deployment\n\x1b[1mYourDeploymentName\x1b[0m, with Build ID \x1b[1mYourBuildID\x1b[0m, and in the default\nnamespace:\n\n\x1b[1mtemporal worker deployment version describe \\\n --deployment-name YourDeploymentName \\\n --build-id YourBuildID\x1b[0m\n\nNote: This is an experimental feature and may change in the future." + } else { + s.Command.Long = "Describe properties of a Worker Deployment Version, such as the task\nqueues polled by workers in this Deployment Version, or drainage\ninformation required to safely decommission workers, or user-provided\nmetadata, or its creation/modification time.\n\n```\ntemporal worker deployment version describe [options]\n```\n\nFor example, to describe a deployment version in a deployment\n`YourDeploymentName`, with Build ID `YourBuildID`, and in the default\nnamespace:\n\n```\ntemporal worker deployment version describe \\\n --deployment-name YourDeploymentName \\\n --build-id YourBuildID\n```\n\nNote: This is an experimental feature and may change in the future." + } + s.Command.Args = cobra.NoArgs + s.Command.Flags().BoolVar(&s.ReportTaskQueueStats, "report-task-queue-stats", false, "Report stats for task queues that are present in this version.") + s.DeploymentVersionOptions.BuildFlags(s.Command.Flags()) + s.Command.Run = func(c *cobra.Command, args []string) { + if err := s.run(cctx, args); err != nil { + cctx.Options.Fail(err) + } + } + return &s +} + type TemporalWorkerDescribeCommand struct { Parent *TemporalWorkerCommand Command cobra.Command diff --git a/internal/temporalcli/commands.worker.deployment.version.aws_lambda.go b/internal/temporalcli/commands.worker.deployment.version.aws_lambda.go new file mode 100644 index 000000000..d812d06b3 --- /dev/null +++ b/internal/temporalcli/commands.worker.deployment.version.aws_lambda.go @@ -0,0 +1,75 @@ +package temporalcli + +import ( + "fmt" + + computepb "go.temporal.io/api/compute/v1" + "go.temporal.io/sdk/converter" +) + +func (c *TemporalWorkerDeploymentVersionComputeConfigAwsLambdaCreateCommand) run(cctx *CommandContext, args []string) error { + provider, err := awsLambdaProvider(c.FunctionArn, &c.AwsAssumeRoleOptions) + if err != nil { + return err + } + scalerDetails, err := c.NoSyncScalerOptions.details() + if err != nil { + return err + } + sg, err := noSyncScalingGroup(&c.ScalingGroupOptions, provider, scalerDetails) + if err != nil { + return err + } + return createVersionWithScalingGroup(cctx, c.Parent.Parent.Parent.Parent.Parent, c.deploymentVersion(), c.ScalingGroup, sg, "aws-lambda") +} + +func (c *TemporalWorkerDeploymentVersionComputeConfigAwsLambdaUpdateCommand) run(cctx *CommandContext, args []string) error { + provider, err := c.provider() + if err != nil { + return err + } + scalerDetails, err := c.NoSyncScalerOptions.details() + if err != nil { + return err + } + update, err := noSyncScalingGroupUpdate(&c.ScalingGroupOptions, provider, scalerDetails) + if err != nil { + return err + } + return updateVersionScalingGroup(cctx, c.Parent.Parent.Parent.Parent.Parent, c.deploymentVersion(), c.ScalingGroup, update, "aws-lambda") +} + +// provider returns the AWS Lambda provider to set, or nil to leave it unchanged. +// The provider flags are all or nothing, because the provider's details are +// replaced as a whole. +func (c *TemporalWorkerDeploymentVersionComputeConfigAwsLambdaUpdateCommand) provider() (*computepb.ComputeProvider, error) { + functionArnSet := c.Command.Flags().Changed("function-arn") + if !functionArnSet && !c.AwsAssumeRoleOptions.passed() { + return nil, nil + } + if !functionArnSet { + return nil, fmt.Errorf("--function-arn is required when changing the provider") + } + return awsLambdaProvider(c.FunctionArn, &c.AwsAssumeRoleOptions) +} + +// The AWS Lambda compute provider. +const awsLambdaProviderType = "aws-lambda" + +// awsLambdaProvider builds the AWS Lambda compute provider from the provider +// flags. Provider config keys: +// https://github.com/temporalio/temporal-auto-scaled-workers/blob/main/wci/workflow/compute_provider/aws_lambda.go +func awsLambdaProvider(functionArn string, role *AwsAssumeRoleOptions) (*computepb.ComputeProvider, error) { + if functionArn == "" { + return nil, fmt.Errorf("--function-arn cannot be empty") + } + details := map[string]any{"arn": functionArn} + if err := role.setDetails(details); err != nil { + return nil, err + } + payload, err := converter.GetDefaultDataConverter().ToPayload(details) + if err != nil { + return nil, err + } + return &computepb.ComputeProvider{Type: awsLambdaProviderType, Details: payload}, nil +} diff --git a/internal/temporalcli/commands.worker.deployment.version.aws_lambda_test.go b/internal/temporalcli/commands.worker.deployment.version.aws_lambda_test.go new file mode 100644 index 000000000..8b2730e08 --- /dev/null +++ b/internal/temporalcli/commands.worker.deployment.version.aws_lambda_test.go @@ -0,0 +1,345 @@ +package temporalcli_test + +import ( + "github.com/google/uuid" + computepb "go.temporal.io/api/compute/v1" + enumspb "go.temporal.io/api/enums/v1" + "go.temporal.io/api/serviceerror" + "google.golang.org/protobuf/proto" +) + +func lambdaComputeConfigArgs(verb, address, deploymentName string, extra ...string) []string { + return append([]string{ + "worker", "deployment", "version", "compute-config", "aws-lambda", verb, + "--address", address, + "--deployment-name", deploymentName, + "--build-id", "build-17", + "--function-arn", "arn:aws:lambda:us-east-1:123:function:payments-worker:17", + "--assume-role-arn", "arn:aws:iam::123:role/temporal-invoke-use1", + "--assume-role-external-id", "ext-id", + }, extra...) +} + +func (s *SharedServerSuite) TestComputeConfig_AwsLambda_Update() { + fake := s.newFakeComputeConfigServer() + deploymentName := uuid.NewString() + lambdaArgs := lambdaComputeConfigArgs("update", s.Address(), deploymentName, + "--scaling-group", "aws-us-east-1-wf", + ) + + // The Version exists but the scaling group does not, so it is created. + fake.setComputeConfig(nil) + res := s.Execute(lambdaArgs...) + s.NoError(res.Err) + s.Contains(res.Stdout.String(), "Successfully updated worker deployment version compute config") + s.Empty(fake.takeCreates()) + updates := fake.takeUpdates() + s.Len(updates, 1) + req := updates[0] + s.Equal(deploymentName, req.GetDeploymentVersion().GetDeploymentName()) + s.Equal("build-17", req.GetDeploymentVersion().GetBuildId()) + s.Empty(req.GetRemoveComputeConfigScalingGroups()) + s.Len(req.GetComputeConfigScalingGroups(), 1) + update := req.GetComputeConfigScalingGroups()["aws-us-east-1-wf"] + s.NotNil(update) + s.Equal([]string{"provider.type", "provider.details", "scaler.type", "scaler.details"}, update.GetUpdateMask().GetPaths()) + sg := update.GetScalingGroup() + // Scaling groups set from the CLI serve every Task Queue type. + s.Empty(sg.GetTaskQueueTypes()) + s.Equal("aws-lambda", sg.GetProvider().GetType()) + s.Equal(map[string]any{ + "arn": "arn:aws:lambda:us-east-1:123:function:payments-worker:17", + "role": "arn:aws:iam::123:role/temporal-invoke-use1", + "role_external_id": "ext-id", + }, s.decodePayloadMap(sg.GetProvider().GetDetails())) + s.Equal("no-sync", sg.GetScaler().GetType()) + s.Nil(sg.GetScaler().GetDetails()) + + // Running the same command again sends the same update again. + res = s.Execute(lambdaArgs...) + s.NoError(res.Err) + updates = fake.takeUpdates() + s.Len(updates, 1) + s.True(proto.Equal(update, updates[0].GetComputeConfigScalingGroups()["aws-us-east-1-wf"])) + + // A different function ARN replaces the provider. + res = s.Execute(append(lambdaArgs, "--function-arn", "arn:aws:lambda:us-east-1:123:function:payments-worker:18")...) + s.NoError(res.Err) + s.Contains(res.Stdout.String(), "Successfully updated worker deployment version compute config") + updates = fake.takeUpdates() + s.Len(updates, 1) + update = updates[0].GetComputeConfigScalingGroups()["aws-us-east-1-wf"] + s.Equal([]string{"provider.type", "provider.details", "scaler.type", "scaler.details"}, update.GetUpdateMask().GetPaths()) + s.Equal( + "arn:aws:lambda:us-east-1:123:function:payments-worker:18", + s.decodePayloadMap(update.GetScalingGroup().GetProvider().GetDetails())["arn"], + ) + s.Empty(fake.takeCreates()) +} + +func (s *SharedServerSuite) TestComputeConfig_AwsLambda_Create() { + fake := s.newFakeComputeConfigServer() + fake.versionMissing = true + deploymentName := uuid.NewString() + + res := s.Execute(lambdaComputeConfigArgs("create", s.Address(), deploymentName)...) + s.NoError(res.Err) + s.Contains(res.Stdout.String(), "Successfully created worker deployment version") + s.Empty(fake.takeUpdates()) + creates := fake.takeCreates() + s.Len(creates, 1) + s.Equal(deploymentName, creates[0].GetDeploymentVersion().GetDeploymentName()) + s.Equal("build-17", creates[0].GetDeploymentVersion().GetBuildId()) + s.NotEmpty(creates[0].GetRequestId()) + sgs := creates[0].GetComputeConfig().GetScalingGroups() + s.Len(sgs, 1) + s.Equal("aws-lambda", sgs["default"].GetProvider().GetType()) + s.Equal("no-sync", sgs["default"].GetScaler().GetType()) + + // The Version now exists, so creating it again fails and points to update. + res = s.Execute(lambdaComputeConfigArgs("create", s.Address(), deploymentName)...) + s.ErrorContains(res.Err, "error creating worker deployment version: Worker Deployment Version already exists; "+ + "to set its compute config, use `temporal worker deployment version compute-config aws-lambda update`") + s.Len(fake.takeCreates(), 1) + s.Empty(fake.takeUpdates()) + + // Update sets the scaling group on the existing Version. + res = s.Execute(lambdaComputeConfigArgs("update", s.Address(), deploymentName)...) + s.NoError(res.Err) + s.Contains(res.Stdout.String(), "Successfully updated worker deployment version compute config") + s.Empty(fake.takeCreates()) + s.Len(fake.takeUpdates(), 1) +} + +func (s *SharedServerSuite) TestComputeConfig_AwsLambda_UpdateMissingVersion() { + fake := s.newFakeComputeConfigServer() + fake.versionMissing = true + + res := s.Execute(lambdaComputeConfigArgs("update", s.Address(), uuid.NewString())...) + s.ErrorContains(res.Err, "error updating worker deployment version compute config: build ID 'build-17' not found in Worker Deployment; "+ + "to create it, use `temporal worker deployment version compute-config aws-lambda create`") + s.Empty(fake.takeCreates()) + s.Len(fake.takeUpdates(), 1) +} + +func (s *SharedServerSuite) TestComputeConfig_AwsLambda_DeploymentMissing() { + fake := s.newFakeComputeConfigServer() + fake.versionMissing = true + fake.createErr = serviceerror.NewNotFound("Worker Deployment not found") + deploymentName := uuid.NewString() + + res := s.Execute(lambdaComputeConfigArgs("create", s.Address(), deploymentName)...) + s.ErrorContains(res.Err, "error creating worker deployment version: Worker Deployment not found") + s.Len(fake.takeCreates(), 1) + s.Empty(fake.takeUpdates()) +} + +func (s *SharedServerSuite) TestComputeConfig_AwsLambda_DefaultScalingGroup() { + fake := s.newFakeComputeConfigServer() + fake.versionMissing = true + res := s.Execute( + "worker", "deployment", "version", "compute-config", "aws-lambda", "create", + "--address", s.Address(), + "--deployment-name", uuid.NewString(), + "--build-id", "build-17", + "--function-arn", "arn:aws:lambda:us-east-1:123:function:F", + "--skip-role-and-external-id", + ) + s.NoError(res.Err) + creates := fake.takeCreates() + s.Len(creates, 1) + sg := creates[0].GetComputeConfig().GetScalingGroups()["default"] + s.NotNil(sg) + s.Empty(sg.GetTaskQueueTypes()) + s.Equal(map[string]any{"arn": "arn:aws:lambda:us-east-1:123:function:F"}, s.decodePayloadMap(sg.GetProvider().GetDetails())) +} + +func (s *SharedServerSuite) TestComputeConfig_AwsLambda_UpdateError() { + fake := s.newFakeComputeConfigServer() + fake.setComputeConfig(nil) + fake.updateErr = serviceerror.NewInvalidArgument("invalid compute config") + + res := s.Execute(lambdaComputeConfigArgs("update", s.Address(), uuid.NewString())...) + s.ErrorContains(res.Err, "error updating worker deployment version compute config: invalid compute config") +} + +func (s *SharedServerSuite) TestComputeConfig_AwsLambda_ScalerSettings() { + fake := s.newFakeComputeConfigServer() + fake.versionMissing = true + deploymentName := uuid.NewString() + + // Every scaling flag sets the scaler details key of the same name. + res := s.Execute(lambdaComputeConfigArgs("create", s.Address(), deploymentName, + "--scale-up-backlog-threshold", "5", + "--scale-up-cooloff-ms", "500", + "--max-worker-lifetime-ms", "300000", + "--metrics-poll-interval-ms", "30000", + "--scale-up-dispatch-rate-epsilon", "0.05", + "--scale-up-dispatch-rate-epsilon-confirm-ms", "90000", + "--suppress-scale-up-ms", "120000", + "--suppress-poll-interval-ms", "60000", + )...) + s.NoError(res.Err) + creates := fake.takeCreates() + s.Len(creates, 1) + scaler := creates[0].GetComputeConfig().GetScalingGroups()["default"].GetScaler() + s.Equal("no-sync", scaler.GetType()) + s.Equal(map[string]any{ + "scale_up_backlog_threshold": float64(5), + "scale_up_cooloff_ms": float64(500), + "max_worker_lifetime_ms": float64(300000), + "metrics_poll_interval_ms": float64(30000), + "scale_up_dispatch_rate_epsilon": 0.05, + "scale_up_dispatch_rate_epsilon_confirm_ms": float64(90000), + "suppress_scale_up_ms": float64(120000), + "suppress_poll_interval_ms": float64(60000), + }, s.decodePayloadMap(scaler.GetDetails())) + + // Only the flags that are set are sent; the others keep the server default. + res = s.Execute(lambdaComputeConfigArgs("update", s.Address(), deploymentName, + "--max-worker-lifetime-ms", "0", + )...) + s.NoError(res.Err) + updates := fake.takeUpdates() + s.Len(updates, 1) + update := updates[0].GetComputeConfigScalingGroups()["default"] + s.Contains(update.GetUpdateMask().GetPaths(), "scaler.details") + s.Equal(map[string]any{"max_worker_lifetime_ms": float64(0)}, + s.decodePayloadMap(update.GetScalingGroup().GetScaler().GetDetails())) +} + +func (s *SharedServerSuite) TestComputeConfig_AwsLambda_PartialUpdate() { + fake := s.newFakeComputeConfigServer() + fake.setComputeConfig(twoScalingGroupComputeConfig()) + deploymentName := uuid.NewString() + args := func(extra ...string) []string { + return append([]string{ + "worker", "deployment", "version", "compute-config", "aws-lambda", "update", + "--address", s.Address(), + "--deployment-name", deploymentName, + "--build-id", "build-17", + }, extra...) + } + onlyUpdate := func(key string) *computepb.ComputeConfigScalingGroupUpdate { + updates := fake.takeUpdates() + s.Len(updates, 1) + s.Len(updates[0].GetComputeConfigScalingGroups(), 1) + return updates[0].GetComputeConfigScalingGroups()[key] + } + stored := func(key string) *computepb.ComputeConfigScalingGroup { + return fake.computeConfig.GetScalingGroups()[key] + } + + // Scaling flags on their own change only the scaling settings. The ones that + // are not passed go back to the server default. + res := s.Execute(args("--scaling-group", "aws-us-west-2-ac", "--scale-up-cooloff-ms", "500")...) + s.NoError(res.Err) + s.Contains(res.Stdout.String(), "Successfully updated worker deployment version compute config") + update := onlyUpdate("aws-us-west-2-ac") + s.Equal([]string{"scaler.details"}, update.GetUpdateMask().GetPaths()) + s.Nil(update.GetScalingGroup().GetProvider()) + sg := stored("aws-us-west-2-ac") + s.Equal("aws-lambda", sg.GetProvider().GetType()) + s.Equal("no-sync", sg.GetScaler().GetType()) + s.Equal(map[string]any{"scale_up_cooloff_ms": float64(500)}, s.decodePayloadMap(sg.GetScaler().GetDetails())) + s.Equal([]enumspb.TaskQueueType{enumspb.TASK_QUEUE_TYPE_ACTIVITY}, sg.GetTaskQueueTypes()) + + // The provider flags replace the provider and reset the scaling settings to + // the scaling flags passed with them, which here are none. + res = s.Execute(args("--scaling-group", "aws-us-west-2-ac", + "--function-arn", "arn:aws:lambda:us-west-2:123:function:F", "--skip-role-and-external-id")...) + s.NoError(res.Err) + s.Equal([]string{"provider.type", "provider.details", "scaler.type", "scaler.details"}, + onlyUpdate("aws-us-west-2-ac").GetUpdateMask().GetPaths()) + sg = stored("aws-us-west-2-ac") + s.Equal(map[string]any{"arn": "arn:aws:lambda:us-west-2:123:function:F"}, s.decodePayloadMap(sg.GetProvider().GetDetails())) + s.Equal("no-sync", sg.GetScaler().GetType()) + s.Nil(sg.GetScaler().GetDetails()) + s.Equal([]enumspb.TaskQueueType{enumspb.TASK_QUEUE_TYPE_ACTIVITY}, sg.GetTaskQueueTypes()) + + // The provider flags are passed together, and something must be passed. + for _, tc := range []struct { + name string + args []string + errText string + }{ + { + name: "role without function ARN", + args: args("--assume-role-arn", "role", "--assume-role-external-id", "ext-id"), + errText: "--function-arn is required when changing the provider", + }, + { + name: "function ARN without role", + args: args("--function-arn", "arn"), + errText: "--assume-role-arn and --assume-role-external-id are required unless --skip-role-and-external-id is passed", + }, + { + name: "empty function ARN", + args: args("--function-arn", "", "--skip-role-and-external-id"), + errText: "--function-arn cannot be empty", + }, + { + name: "nothing to update", + args: args("--scaling-group", "aws-us-east-1-wf"), + errText: "no compute configuration provided to update", + }, + } { + res := s.Execute(tc.args...) + s.EqualError(res.Err, tc.errText, tc.name) + } + s.Empty(fake.takeUpdates()) +} + +func (s *SharedServerSuite) TestComputeConfig_Provider_Errors() { + fake := s.newFakeComputeConfigServer() + lambda := []string{ + "worker", "deployment", "version", "compute-config", "aws-lambda", "create", + "--address", s.Address(), + "--deployment-name", uuid.NewString(), + "--build-id", "build-17", + } + + for _, tc := range []struct { + name string + args []string + errText string + }{ + { + name: "missing function ARN", + args: append(append([]string{}, lambda...), "--skip-role-and-external-id"), + errText: `required flag(s) "function-arn" not set`, + }, + { + name: "empty function ARN", + args: append(append([]string{}, lambda...), "--function-arn", "", "--skip-role-and-external-id"), + errText: "--function-arn cannot be empty", + }, + { + name: "missing assume role", + args: append(append([]string{}, lambda...), "--function-arn", "arn"), + errText: "--assume-role-arn and --assume-role-external-id are required unless --skip-role-and-external-id is passed", + }, + { + name: "missing external ID", + args: append(append([]string{}, lambda...), "--function-arn", "arn", "--assume-role-arn", "role"), + errText: "--assume-role-arn and --assume-role-external-id are required unless --skip-role-and-external-id is passed", + }, + { + name: "assume role with skip", + args: append(append([]string{}, lambda...), + "--function-arn", "arn", "--assume-role-arn", "role", "--skip-role-and-external-id"), + errText: "must be omitted when --skip-role-and-external-id is passed", + }, + { + name: "empty scaling group", + args: append(append([]string{}, lambda...), + "--function-arn", "arn", "--skip-role-and-external-id", "--scaling-group", ""), + errText: "--scaling-group cannot be empty", + }, + } { + res := s.Execute(tc.args...) + s.ErrorContains(res.Err, tc.errText, tc.name) + } + s.Empty(fake.takeCreates()) + s.Empty(fake.takeUpdates()) +} diff --git a/internal/temporalcli/commands.worker.deployment.version.go b/internal/temporalcli/commands.worker.deployment.version.go new file mode 100644 index 000000000..2ae6b2015 --- /dev/null +++ b/internal/temporalcli/commands.worker.deployment.version.go @@ -0,0 +1,498 @@ +package temporalcli + +import ( + "errors" + "fmt" + "maps" + "slices" + "strconv" + "strings" + + "github.com/fatih/color" + "github.com/google/uuid" + "github.com/temporalio/cli/internal/printer" + commonpb "go.temporal.io/api/common/v1" + computepb "go.temporal.io/api/compute/v1" + deploymentpb "go.temporal.io/api/deployment/v1" + "go.temporal.io/api/serviceerror" + "go.temporal.io/api/workflowservice/v1" + "go.temporal.io/sdk/client" + "go.temporal.io/sdk/converter" + "google.golang.org/protobuf/types/known/fieldmaskpb" +) + +func describeVersionComputeConfig( + cctx *CommandContext, + cl workflowservice.WorkflowServiceClient, + namespace string, + version *deploymentpb.WorkerDeploymentVersion, +) (*computepb.ComputeConfig, error) { + resp, err := cl.DescribeWorkerDeploymentVersion(cctx, &workflowservice.DescribeWorkerDeploymentVersionRequest{ + Namespace: namespace, + DeploymentVersion: version, + }) + if err != nil { + return nil, fmt.Errorf("error describing worker deployment version: %w", err) + } + return resp.GetWorkerDeploymentVersionInfo().GetComputeConfig(), nil +} + +func (o *DeploymentVersionOptions) deploymentVersion() *deploymentpb.WorkerDeploymentVersion { + return &deploymentpb.WorkerDeploymentVersion{ + DeploymentName: o.DeploymentName, + BuildId: o.BuildId, + } +} + +func (c *TemporalWorkerDeploymentVersionCreateCommand) run(cctx *CommandContext, args []string) error { + worker := c.Parent.Parent.Parent + cl, err := dialClient(cctx, &worker.ClientOptions) + if err != nil { + return err + } + defer cl.Close() + + _, err = cl.WorkflowService().CreateWorkerDeploymentVersion(cctx, &workflowservice.CreateWorkerDeploymentVersionRequest{ + Namespace: worker.Namespace, + DeploymentVersion: c.deploymentVersion(), + Identity: worker.Identity, + RequestId: uuid.NewString(), + }) + if err != nil { + return fmt.Errorf("error creating worker deployment version: %w", err) + } + cctx.Printer.Println("Successfully created worker deployment version") + return nil +} + +func (c *TemporalWorkerDeploymentVersionDescribeCommand) run(cctx *CommandContext, args []string) error { + worker := c.Parent.Parent.Parent + cl, err := dialClient(cctx, &worker.ClientOptions) + if err != nil { + return err + } + defer cl.Close() + + // Use raw gRPC instead of SDK's DeploymentClient to avoid circular dependency + // with SDK release that exposes TaskQueuesInfos in DescribeVersionResponse + resp, err := cl.WorkflowService().DescribeWorkerDeploymentVersion(cctx, &workflowservice.DescribeWorkerDeploymentVersionRequest{ + Namespace: worker.Namespace, + DeploymentVersion: c.deploymentVersion(), + ReportTaskQueueStats: c.ReportTaskQueueStats, + }) + if err != nil { + return fmt.Errorf("error describing worker deployment version: %w", err) + } + return printWorkerDeploymentVersionInfoProto(cctx, resp.GetWorkerDeploymentVersionInfo(), resp.GetVersionTaskQueues(), "Worker Deployment Version:", printVersionInfoOptions{ + showStats: c.ReportTaskQueueStats, + }) +} + +func (c *TemporalWorkerDeploymentVersionComputeConfigDescribeCommand) run(cctx *CommandContext, args []string) error { + worker := c.Parent.Parent.Parent.Parent + cl, err := dialClient(cctx, &worker.ClientOptions) + if err != nil { + return err + } + defer cl.Close() + + cc, err := describeVersionComputeConfig(cctx, cl.WorkflowService(), worker.Namespace, c.deploymentVersion()) + if err != nil { + return err + } + described := formatComputeConfigDescription(cc) + if cctx.JSONOutput { + return cctx.Printer.PrintStructured(described, printer.StructuredOptions{}) + } + if len(described.ScalingGroups) == 0 { + cctx.Printer.Println("Worker deployment version has no compute config") + return nil + } + + type scalingGroupCard struct { + ScalingGroup string + Provider string `cli:",cardOmitEmpty"` + Scaler string `cli:",cardOmitEmpty"` + } + keys := slices.Sorted(maps.Keys(described.ScalingGroups)) + cards := make([]scalingGroupCard, 0, len(keys)) + for _, key := range keys { + sg := described.ScalingGroups[key] + card := scalingGroupCard{ScalingGroup: key} + if sg.Provider != nil { + card.Provider = sg.Provider.Type + } + if sg.Scaler != nil { + card.Scaler = scalerSummary(sg.Scaler) + } + cards = append(cards, card) + } + cctx.Printer.Println(color.MagentaString("Compute Config Scaling Groups:")) + return cctx.Printer.PrintStructured(cards, printer.StructuredOptions{}) +} + +func (c *TemporalWorkerDeploymentVersionDeleteCommand) run(cctx *CommandContext, args []string) error { + worker := c.Parent.Parent.Parent + cl, err := dialClient(cctx, &worker.ClientOptions) + if err != nil { + return err + } + defer cl.Close() + + dHandle := cl.WorkerDeploymentClient().GetHandle(c.DeploymentName) + _, err = dHandle.DeleteVersion(cctx, client.WorkerDeploymentDeleteVersionOptions{ + BuildID: c.BuildId, + SkipDrainage: c.SkipDrainage, + Identity: worker.Identity, + }) + if err != nil { + return fmt.Errorf("error deleting worker deployment version: %w", err) + } + + cctx.Printer.Println("Successfully deleted worker deployment version") + return nil +} + +func (c *TemporalWorkerDeploymentVersionComputeConfigRemoveCommand) run(cctx *CommandContext, args []string) error { + removeOne := c.Command.Flags().Changed("scaling-group") + if removeOne && c.ScalingGroup == "" { + return fmt.Errorf("--scaling-group cannot be empty") + } + + worker := c.Parent.Parent.Parent.Parent + cl, err := dialClient(cctx, &worker.ClientOptions) + if err != nil { + return err + } + defer cl.Close() + + // The scaling groups are read first because the API removes them by key and + // silently ignores keys that do not exist. + cc, err := describeVersionComputeConfig(cctx, cl.WorkflowService(), worker.Namespace, c.deploymentVersion()) + if err != nil { + return err + } + var keys []string + if removeOne { + if _, ok := cc.GetScalingGroups()[c.ScalingGroup]; !ok { + return fmt.Errorf("scaling group %q not found; run `temporal worker deployment version compute-config describe` to see the scaling groups of this worker deployment version", c.ScalingGroup) + } + keys = []string{c.ScalingGroup} + } else { + keys = slices.Sorted(maps.Keys(cc.GetScalingGroups())) + if len(keys) == 0 { + cctx.Printer.Println("Worker deployment version has no compute config to remove") + return nil + } + prompt := fmt.Sprintf("Remove all %d scaling groups (%s) from Worker Deployment Version %q with Build ID %q? y/N", + len(keys), strings.Join(keys, ", "), c.DeploymentName, c.BuildId) + if len(keys) == 1 { + prompt = fmt.Sprintf("Remove scaling group %q from Worker Deployment Version %q with Build ID %q? y/N", + keys[0], c.DeploymentName, c.BuildId) + } + yes, err := cctx.promptYes(prompt, c.Yes) + if err != nil { + return err + } else if !yes { + return fmt.Errorf("user denied confirmation") + } + } + _, err = cl.WorkflowService().UpdateWorkerDeploymentVersionComputeConfig(cctx, &workflowservice.UpdateWorkerDeploymentVersionComputeConfigRequest{ + Namespace: worker.Namespace, + DeploymentVersion: c.deploymentVersion(), + RemoveComputeConfigScalingGroups: keys, + Identity: worker.Identity, + RequestId: uuid.NewString(), + }) + if err != nil { + return fmt.Errorf("error removing worker deployment version compute config: %w", err) + } + cctx.Printer.Println("Successfully removed worker deployment version compute config") + return nil +} + +// scalingGroup builds the provider-independent part of a scaling group from +// the --scaling-group flag. Scaling groups set from the CLI serve every Task +// Queue type. +func (o *ScalingGroupOptions) scalingGroup() (*computepb.ComputeConfigScalingGroup, error) { + if o.ScalingGroup == "" { + return nil, fmt.Errorf("--scaling-group cannot be empty") + } + return &computepb.ComputeConfigScalingGroup{}, nil +} + +// update builds the provider-independent part of a scaling group update: the +// fields whose flags are passed, and their update mask paths. +func (o *ScalingGroupOptions) update() (*computepb.ComputeConfigScalingGroup, []string, error) { + sg, err := o.scalingGroup() + if err != nil { + return nil, nil, err + } + return sg, nil, nil +} + +// createVersionWithScalingGroup creates a Worker Deployment Version with one +// scaling group. provider is the compute-config subcommand of the scaling +// group's compute provider, used in the error hint. +func createVersionWithScalingGroup( + cctx *CommandContext, + worker *TemporalWorkerCommand, + version *deploymentpb.WorkerDeploymentVersion, + key string, + sg *computepb.ComputeConfigScalingGroup, + provider string, +) error { + cl, err := dialClient(cctx, &worker.ClientOptions) + if err != nil { + return err + } + defer cl.Close() + + _, err = cl.WorkflowService().CreateWorkerDeploymentVersion(cctx, &workflowservice.CreateWorkerDeploymentVersionRequest{ + Namespace: worker.Namespace, + DeploymentVersion: version, + ComputeConfig: &computepb.ComputeConfig{ + ScalingGroups: map[string]*computepb.ComputeConfigScalingGroup{key: sg}, + }, + Identity: worker.Identity, + RequestId: uuid.NewString(), + }) + var alreadyExists *serviceerror.AlreadyExists + if errors.As(err, &alreadyExists) { + return fmt.Errorf("error creating worker deployment version: %w; to set its compute config, use `temporal worker deployment version compute-config %s update`", err, provider) + } + if err != nil { + return fmt.Errorf("error creating worker deployment version: %w", err) + } + cctx.Printer.Println("Successfully created worker deployment version") + return nil +} + +// updateVersionScalingGroup updates one scaling group of an existing Worker +// Deployment Version. provider is the compute-config subcommand of the scaling +// group's compute provider, used in the error hint. +func updateVersionScalingGroup( + cctx *CommandContext, + worker *TemporalWorkerCommand, + version *deploymentpb.WorkerDeploymentVersion, + key string, + update *computepb.ComputeConfigScalingGroupUpdate, + provider string, +) error { + cl, err := dialClient(cctx, &worker.ClientOptions) + if err != nil { + return err + } + defer cl.Close() + + _, err = cl.WorkflowService().UpdateWorkerDeploymentVersionComputeConfig(cctx, &workflowservice.UpdateWorkerDeploymentVersionComputeConfigRequest{ + Namespace: worker.Namespace, + DeploymentVersion: version, + ComputeConfigScalingGroups: map[string]*computepb.ComputeConfigScalingGroupUpdate{ + key: update, + }, + Identity: worker.Identity, + RequestId: uuid.NewString(), + }) + var notFound *serviceerror.NotFound + if errors.As(err, ¬Found) { + return fmt.Errorf("error updating worker deployment version compute config: %w; to create it, use `temporal worker deployment version compute-config %s create`", err, provider) + } + if err != nil { + return fmt.Errorf("error updating worker deployment version compute config: %w", err) + } + cctx.Printer.Println("Successfully updated worker deployment version compute config") + return nil +} + +// passed reports whether any of the role flags is passed. +func (o *AwsAssumeRoleOptions) passed() bool { + return slices.ContainsFunc([]string{"assume-role-arn", "assume-role-external-id", "skip-role-and-external-id"}, o.FlagSet.Changed) +} + +// The worker controller's no-sync scaler, which starts Workers by invoking them, +// such as AWS Lambda functions. +const noSyncScalerType = "no-sync" + +// noSyncScalingGroup builds a scaling group for a compute provider that uses the +// no-sync scaler. +func noSyncScalingGroup( + group *ScalingGroupOptions, + provider *computepb.ComputeProvider, + scalerDetails *commonpb.Payload, +) (*computepb.ComputeConfigScalingGroup, error) { + sg, err := group.scalingGroup() + if err != nil { + return nil, err + } + sg.Provider = provider + sg.Scaler = &computepb.ComputeScaler{Type: noSyncScalerType, Details: scalerDetails} + return sg, nil +} + +// noSyncScalingGroupUpdate builds an update of a scaling group for a compute +// provider that uses the no-sync scaler, from the flags that are passed. The +// provider is nil when its flags are not passed. +func noSyncScalingGroupUpdate( + group *ScalingGroupOptions, + provider *computepb.ComputeProvider, + scalerDetails *commonpb.Payload, +) (*computepb.ComputeConfigScalingGroupUpdate, error) { + sg, paths, err := group.update() + if err != nil { + return nil, err + } + switch { + case provider != nil: + // Provider (re)configuration: rebuild the provider and scaler type, + // and keep scaler.details consistent with the (possibly new) type. + sg.Provider = provider + sg.Scaler = &computepb.ComputeScaler{Type: noSyncScalerType, Details: scalerDetails} + paths = append(paths, "provider.type", "provider.details", "scaler.type", "scaler.details") + case scalerDetails != nil: + // Scaler-only update: change scaler.details without touching the + // provider or scaler type. + sg.Scaler = &computepb.ComputeScaler{Details: scalerDetails} + paths = append(paths, "scaler.details") + } + if len(paths) == 0 { + return nil, fmt.Errorf("no compute configuration provided to update") + } + return &computepb.ComputeConfigScalingGroupUpdate{ + ScalingGroup: sg, + UpdateMask: &fieldmaskpb.FieldMask{Paths: paths}, + }, nil +} + +// setDetails sets the role that the server assumes in an AWS compute provider's +// details. The AWS compute providers read it from the same config keys. +func (o *AwsAssumeRoleOptions) setDetails(details map[string]any) error { + if o.SkipRoleAndExternalId { + if o.AssumeRoleArn != "" || o.AssumeRoleExternalId != "" { + return fmt.Errorf("--assume-role-arn and --assume-role-external-id must be omitted when --skip-role-and-external-id is passed") + } + return nil + } + if o.AssumeRoleArn == "" || o.AssumeRoleExternalId == "" { + return fmt.Errorf("--assume-role-arn and --assume-role-external-id are required unless --skip-role-and-external-id is passed") + } + details["role"] = o.AssumeRoleArn + details["role_external_id"] = o.AssumeRoleExternalId + return nil +} + +// details builds the details of the worker controller's no-sync scaler, used by +// compute providers that start Workers by invoking them, such as AWS Lambda, +// from the scaling flags that are set. Each flag is named after the config key +// it sets, with dashes for underscores. It returns nil when none are set, so the +// server defaults apply. The settings are validated by the server. Config keys: +// https://github.com/temporalio/temporal-auto-scaled-workers/blob/main/wci/workflow/scaling_algorithm/no_sync_match.go +func (o *NoSyncScalerOptions) details() (*commonpb.Payload, error) { + details := map[string]any{} + for flag, value := range map[string]any{ + "scale-up-backlog-threshold": o.ScaleUpBacklogThreshold, + "scale-up-cooloff-ms": o.ScaleUpCooloffMs, + "max-worker-lifetime-ms": o.MaxWorkerLifetimeMs, + "metrics-poll-interval-ms": o.MetricsPollIntervalMs, + "scale-up-dispatch-rate-epsilon": o.ScaleUpDispatchRateEpsilon, + "scale-up-dispatch-rate-epsilon-confirm-ms": o.ScaleUpDispatchRateEpsilonConfirmMs, + "suppress-scale-up-ms": o.SuppressScaleUpMs, + "suppress-poll-interval-ms": o.SuppressPollIntervalMs, + } { + if o.FlagSet.Changed(flag) { + details[strings.ReplaceAll(flag, "-", "_")] = value + } + } + if len(details) == 0 { + return nil, nil + } + return converter.GetDefaultDataConverter().ToPayload(details) +} + +type formattedComputeConfigDescription struct { + ScalingGroups map[string]formattedScalingGroupDescription `json:"scalingGroups"` +} + +type formattedScalingGroupDescription struct { + Provider *formattedComputeConfigProvider `json:"provider,omitempty"` + Scaler *formattedScalerDescription `json:"scaler,omitempty"` +} + +type formattedScalerDescription struct { + Type string `json:"type"` + ScaleUpBacklogThreshold *int64 `json:"scaleUpBacklogThreshold,omitempty"` + ScaleUpCooloffMs *int64 `json:"scaleUpCooloffMs,omitempty"` + MaxWorkerLifetimeMs *int64 `json:"maxWorkerLifetimeMs,omitempty"` + MetricsPollIntervalMs *int64 `json:"metricsPollIntervalMs,omitempty"` + ScaleUpDispatchRateEpsilon *float64 `json:"scaleUpDispatchRateEpsilon,omitempty"` + ScaleUpDispatchRateEpsilonConfirmMs *int64 `json:"scaleUpDispatchRateEpsilonConfirmMs,omitempty"` + SuppressScaleUpMs *int64 `json:"suppressScaleUpMs,omitempty"` + SuppressPollIntervalMs *int64 `json:"suppressPollIntervalMs,omitempty"` +} + +func formatComputeConfigDescription(cc *computepb.ComputeConfig) formattedComputeConfigDescription { + out := formattedComputeConfigDescription{ScalingGroups: map[string]formattedScalingGroupDescription{}} + for name, msgSG := range cc.GetScalingGroups() { + var sg formattedScalingGroupDescription + if p := msgSG.GetProvider(); p != nil { + sg.Provider = &formattedComputeConfigProvider{Type: p.GetType()} + } + if s := msgSG.GetScaler(); s != nil { + sg.Scaler = &formattedScalerDescription{Type: s.GetType()} + decodeNoSyncScalerSettings(s, sg.Scaler) + } + out.ScalingGroups[name] = sg + } + return out +} + +// scalerSummary returns a scaler's type followed by the settings that are set, +// such as "no-sync (scale_up_backlog_threshold 2, max_worker_lifetime_ms 300000)". +func scalerSummary(fs *formattedScalerDescription) string { + var parts []string + addInt := func(key string, v *int64) { + if v != nil { + parts = append(parts, fmt.Sprintf("%s %d", key, *v)) + } + } + addInt("scale_up_backlog_threshold", fs.ScaleUpBacklogThreshold) + addInt("scale_up_cooloff_ms", fs.ScaleUpCooloffMs) + addInt("max_worker_lifetime_ms", fs.MaxWorkerLifetimeMs) + addInt("metrics_poll_interval_ms", fs.MetricsPollIntervalMs) + if v := fs.ScaleUpDispatchRateEpsilon; v != nil { + parts = append(parts, "scale_up_dispatch_rate_epsilon "+strconv.FormatFloat(*v, 'g', -1, 64)) + } + addInt("scale_up_dispatch_rate_epsilon_confirm_ms", fs.ScaleUpDispatchRateEpsilonConfirmMs) + addInt("suppress_scale_up_ms", fs.SuppressScaleUpMs) + addInt("suppress_poll_interval_ms", fs.SuppressPollIntervalMs) + if len(parts) == 0 { + return fs.Type + } + return fs.Type + " (" + strings.Join(parts, ", ") + ")" +} + +func decodeNoSyncScalerSettings(s *computepb.ComputeScaler, fs *formattedScalerDescription) { + if s.GetType() != noSyncScalerType || s.GetDetails() == nil { + return + } + var m map[string]any + if err := converter.GetDefaultDataConverter().FromPayload(s.GetDetails(), &m); err != nil { + return + } + count := func(key string) *int64 { + if v, ok := scalerCountFromMap(m, key); ok { + return &v + } + return nil + } + fs.ScaleUpBacklogThreshold = count("scale_up_backlog_threshold") + fs.ScaleUpCooloffMs = count("scale_up_cooloff_ms") + fs.MaxWorkerLifetimeMs = count("max_worker_lifetime_ms") + fs.MetricsPollIntervalMs = count("metrics_poll_interval_ms") + if v, ok := scalerFloatFromMap(m, "scale_up_dispatch_rate_epsilon"); ok { + fs.ScaleUpDispatchRateEpsilon = &v + } + fs.ScaleUpDispatchRateEpsilonConfirmMs = count("scale_up_dispatch_rate_epsilon_confirm_ms") + fs.SuppressScaleUpMs = count("suppress_scale_up_ms") + fs.SuppressPollIntervalMs = count("suppress_poll_interval_ms") +} diff --git a/internal/temporalcli/commands.worker.deployment.version_test.go b/internal/temporalcli/commands.worker.deployment.version_test.go new file mode 100644 index 000000000..3beee4870 --- /dev/null +++ b/internal/temporalcli/commands.worker.deployment.version_test.go @@ -0,0 +1,521 @@ +package temporalcli_test + +import ( + "context" + "encoding/json" + "fmt" + "maps" + "slices" + "strings" + "sync" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + commonpb "go.temporal.io/api/common/v1" + computepb "go.temporal.io/api/compute/v1" + deploymentpb "go.temporal.io/api/deployment/v1" + enumspb "go.temporal.io/api/enums/v1" + "go.temporal.io/api/serviceerror" + taskqueuepb "go.temporal.io/api/taskqueue/v1" + "go.temporal.io/api/workflowservice/v1" + "go.temporal.io/sdk/converter" + "google.golang.org/grpc" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/durationpb" + "google.golang.org/protobuf/types/known/timestamppb" +) + +// fakeComputeConfigServer intercepts the describe, create, and update compute +// config RPCs: describe returns the configured version, and create and update +// requests are captured instead of being sent. Updates are applied the way the +// worker controller applies them: a new scaling group is stored whole, and an +// existing one only gets the fields in the update mask. This lets the tests +// cover multiple scaling groups without real cloud resources behind the server. +type fakeComputeConfigServer struct { + mu sync.Mutex + // versionMissing makes describe return NotFound. + versionMissing bool + computeConfig *computepb.ComputeConfig + // createErr is returned by create. Otherwise create fails with + // AlreadyExists if the version exists, and makes the version exist with the + // created compute config if not. + createErr error + // updateErr is returned by update. + updateErr error + // describeTemplate, if set, is the describe response before the compute + // config is filled in. + describeTemplate *workflowservice.DescribeWorkerDeploymentVersionResponse + creates []*workflowservice.CreateWorkerDeploymentVersionRequest + updates []*workflowservice.UpdateWorkerDeploymentVersionComputeConfigRequest +} + +func (f *fakeComputeConfigServer) setComputeConfig(cc *computepb.ComputeConfig) { + f.mu.Lock() + defer f.mu.Unlock() + f.versionMissing = false + f.computeConfig = cc +} + +func (f *fakeComputeConfigServer) takeCreates() []*workflowservice.CreateWorkerDeploymentVersionRequest { + f.mu.Lock() + defer f.mu.Unlock() + creates := f.creates + f.creates = nil + return creates +} + +func (f *fakeComputeConfigServer) takeUpdates() []*workflowservice.UpdateWorkerDeploymentVersionComputeConfigRequest { + f.mu.Lock() + defer f.mu.Unlock() + updates := f.updates + f.updates = nil + return updates +} + +func (f *fakeComputeConfigServer) store(key string, sg *computepb.ComputeConfigScalingGroup) { + sg = proto.Clone(sg).(*computepb.ComputeConfigScalingGroup) + if f.computeConfig == nil { + f.computeConfig = &computepb.ComputeConfig{} + } + if f.computeConfig.ScalingGroups == nil { + f.computeConfig.ScalingGroups = map[string]*computepb.ComputeConfigScalingGroup{} + } + f.computeConfig.ScalingGroups[key] = sg +} + +// update applies one scaling group update like the worker controller does. +func (f *fakeComputeConfigServer) update(key string, update *computepb.ComputeConfigScalingGroupUpdate) error { + existing, ok := f.computeConfig.GetScalingGroups()[key] + if !ok { + f.store(key, update.GetScalingGroup()) + return nil + } + merged := proto.Clone(existing).(*computepb.ComputeConfigScalingGroup) + src := update.GetScalingGroup() + for _, path := range update.GetUpdateMask().GetPaths() { + switch path { + case "provider.type", "provider.details": + if merged.Provider == nil { + merged.Provider = &computepb.ComputeProvider{} + } + if path == "provider.type" { + merged.Provider.Type = src.GetProvider().GetType() + } else { + merged.Provider.Details = src.GetProvider().GetDetails() + } + case "scaler.type", "scaler.details": + if src.GetScaler() == nil { + return serviceerror.NewInvalidArgumentf("%s set in field mask, but not provided", path) + } + if merged.Scaler == nil { + merged.Scaler = &computepb.ComputeScaler{} + } + if path == "scaler.type" { + merged.Scaler.Type = src.GetScaler().GetType() + } else { + merged.Scaler.Details = src.GetScaler().GetDetails() + } + default: + return serviceerror.NewInvalidArgumentf("unsupported field mask path: %q", path) + } + } + f.store(key, merged) + return nil +} + +func (f *fakeComputeConfigServer) interceptor( + ctx context.Context, method string, req, reply any, + cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption, +) error { + switch r := req.(type) { + case *workflowservice.DescribeWorkerDeploymentVersionRequest: + f.mu.Lock() + defer f.mu.Unlock() + if f.versionMissing { + return serviceerror.ToStatus(serviceerror.NewNotFound("Worker Deployment Version not found")).Err() + } + resp := &workflowservice.DescribeWorkerDeploymentVersionResponse{} + if f.describeTemplate != nil { + resp = proto.Clone(f.describeTemplate).(*workflowservice.DescribeWorkerDeploymentVersionResponse) + } + if resp.WorkerDeploymentVersionInfo == nil { + resp.WorkerDeploymentVersionInfo = &deploymentpb.WorkerDeploymentVersionInfo{} + } + resp.WorkerDeploymentVersionInfo.DeploymentVersion = r.GetDeploymentVersion() + if f.computeConfig != nil { + resp.WorkerDeploymentVersionInfo.ComputeConfig = proto.Clone(f.computeConfig).(*computepb.ComputeConfig) + } + proto.Merge(reply.(proto.Message), resp) + return nil + case *workflowservice.CreateWorkerDeploymentVersionRequest: + f.mu.Lock() + defer f.mu.Unlock() + f.creates = append(f.creates, r) + if f.createErr != nil { + return serviceerror.ToStatus(f.createErr).Err() + } + if !f.versionMissing { + return serviceerror.ToStatus(serviceerror.NewAlreadyExists("Worker Deployment Version already exists")).Err() + } + f.versionMissing = false + f.computeConfig = nil + for key, sg := range r.GetComputeConfig().GetScalingGroups() { + f.store(key, sg) + } + return nil + case *workflowservice.UpdateWorkerDeploymentVersionComputeConfigRequest: + f.mu.Lock() + defer f.mu.Unlock() + f.updates = append(f.updates, r) + if f.updateErr != nil { + return serviceerror.ToStatus(f.updateErr).Err() + } + if f.versionMissing { + return serviceerror.ToStatus(serviceerror.NewNotFound("build ID 'build-17' not found in Worker Deployment")).Err() + } + for key, update := range r.GetComputeConfigScalingGroups() { + if err := f.update(key, update); err != nil { + return serviceerror.ToStatus(err).Err() + } + } + for _, key := range r.GetRemoveComputeConfigScalingGroups() { + delete(f.computeConfig.GetScalingGroups(), key) + } + return nil + } + return invoker(ctx, method, req, reply, cc, opts...) +} + +func (s *SharedServerSuite) newFakeComputeConfigServer() *fakeComputeConfigServer { + f := &fakeComputeConfigServer{} + s.CommandHarness.Options.AdditionalClientGRPCDialOptions = append( + s.CommandHarness.Options.AdditionalClientGRPCDialOptions, + grpc.WithChainUnaryInterceptor(f.interceptor), + ) + return f +} + +func (s *SharedServerSuite) decodePayloadMap(p *commonpb.Payload) map[string]any { + var m map[string]any + s.NoError(converter.GetDefaultDataConverter().FromPayload(p, &m)) + return m +} + +func twoScalingGroupComputeConfig() *computepb.ComputeConfig { + scalerDetails, _ := converter.GetDefaultDataConverter().ToPayload(map[string]any{ + "scale_up_backlog_threshold": 2, + "max_worker_lifetime_ms": 300000, + "scale_up_dispatch_rate_epsilon": 0.05, + }) + return &computepb.ComputeConfig{ + ScalingGroups: map[string]*computepb.ComputeConfigScalingGroup{ + "aws-us-west-2-ac": { + TaskQueueTypes: []enumspb.TaskQueueType{enumspb.TASK_QUEUE_TYPE_ACTIVITY}, + Provider: &computepb.ComputeProvider{Type: "aws-lambda"}, + Scaler: &computepb.ComputeScaler{Type: "no-sync", Details: scalerDetails}, + }, + "aws-us-east-1-wf": { + TaskQueueTypes: []enumspb.TaskQueueType{enumspb.TASK_QUEUE_TYPE_WORKFLOW}, + Provider: &computepb.ComputeProvider{Type: "aws-lambda"}, + Scaler: &computepb.ComputeScaler{Type: "no-sync"}, + }, + }, + } +} + +func (s *SharedServerSuite) TestComputeConfig_Describe() { + fake := s.newFakeComputeConfigServer() + describeArgs := []string{ + "worker", "deployment", "version", "compute-config", "describe", + "--address", s.Address(), + "--deployment-name", uuid.NewString(), + "--build-id", "build-17", + } + + // No compute config. + res := s.Execute(describeArgs...) + s.NoError(res.Err) + s.Contains(res.Stdout.String(), "Worker deployment version has no compute config") + res = s.Execute(append(describeArgs, "-o", "json")...) + s.NoError(res.Err) + s.JSONEq(`{"scalingGroups": {}}`, res.Stdout.String()) + + fake.setComputeConfig(twoScalingGroupComputeConfig()) + + // One card per scaling group, sorted by key. The Scaler line lists the + // settings that are set. + res = s.Execute(describeArgs...) + s.NoError(res.Err) + _, groups, found := strings.Cut(res.Stdout.String(), "Compute Config Scaling Groups:") + s.True(found) + cards := strings.Split(strings.TrimSpace(groups), "\n\n") + s.Len(cards, 2) + // Like describe-version, the Task Queue types are not shown. + s.NotContains(groups, "TaskQueueTypes") + s.ContainsOnSameLine(cards[0], "ScalingGroup", "aws-us-east-1-wf") + s.ContainsOnSameLine(cards[0], "Provider", "aws-lambda") + s.ContainsOnSameLine(cards[0], "Scaler", "no-sync") + s.NotContains(cards[0], "(") + s.ContainsOnSameLine(cards[1], "ScalingGroup", "aws-us-west-2-ac") + s.ContainsOnSameLine(cards[1], "Scaler", + "no-sync (scale_up_backlog_threshold 2, max_worker_lifetime_ms 300000, scale_up_dispatch_rate_epsilon 0.05)") + + res = s.Execute(append(describeArgs, "-o", "json")...) + s.NoError(res.Err) + var described struct { + ScalingGroups map[string]struct { + Provider struct { + Type string `json:"type"` + } `json:"provider"` + Scaler struct { + Type string `json:"type"` + ScaleUpBacklogThreshold *int64 `json:"scaleUpBacklogThreshold"` + MaxWorkerLifetimeMs *int64 `json:"maxWorkerLifetimeMs"` + ScaleUpDispatchRateEpsilon *float64 `json:"scaleUpDispatchRateEpsilon"` + ScaleUpCooloffMs *int64 `json:"scaleUpCooloffMs"` + } `json:"scaler"` + } `json:"scalingGroups"` + } + s.NotContains(res.Stdout.String(), "taskQueueTypes") + s.NoError(json.Unmarshal(res.Stdout.Bytes(), &described)) + s.Len(described.ScalingGroups, 2) + wf := described.ScalingGroups["aws-us-east-1-wf"] + s.Equal("aws-lambda", wf.Provider.Type) + s.Equal("no-sync", wf.Scaler.Type) + ac := described.ScalingGroups["aws-us-west-2-ac"] + s.Equal("aws-lambda", ac.Provider.Type) + s.Equal("no-sync", ac.Scaler.Type) + // The no-sync settings that are set are shown; the others are left out. + s.Equal(int64(2), *ac.Scaler.ScaleUpBacklogThreshold) + s.Equal(int64(300000), *ac.Scaler.MaxWorkerLifetimeMs) + s.Equal(0.05, *ac.Scaler.ScaleUpDispatchRateEpsilon) + s.Nil(ac.Scaler.ScaleUpCooloffMs) + s.Nil(wf.Scaler.ScaleUpBacklogThreshold) + s.Empty(fake.takeUpdates()) +} + +func (s *SharedServerSuite) TestComputeConfig_Remove() { + fake := s.newFakeComputeConfigServer() + deploymentName := uuid.NewString() + removeArgs := []string{ + "worker", "deployment", "version", "compute-config", "remove", + "--address", s.Address(), + "--deployment-name", deploymentName, + "--build-id", "build-17", + } + + // With no scaling groups there is nothing to confirm or remove. + res := s.Execute(removeArgs...) + s.NoError(res.Err) + s.Contains(res.Stdout.String(), "Worker deployment version has no compute config to remove") + s.Empty(fake.takeUpdates()) + + // Only the named scaling group is removed, without a prompt. + fake.setComputeConfig(twoScalingGroupComputeConfig()) + res = s.Execute(append(removeArgs, "--scaling-group", "aws-us-west-2-ac")...) + s.NoError(res.Err) + s.Contains(res.Stdout.String(), "Successfully removed worker deployment version compute config") + updates := fake.takeUpdates() + s.Len(updates, 1) + s.Equal([]string{"aws-us-west-2-ac"}, updates[0].GetRemoveComputeConfigScalingGroups()) + s.Empty(updates[0].GetComputeConfigScalingGroups()) + s.Equal([]string{"aws-us-east-1-wf"}, slices.Collect(maps.Keys(fake.computeConfig.GetScalingGroups()))) + + // Removing a scaling group that does not exist fails without an update. + res = s.Execute(append(removeArgs, "--scaling-group", "aws-us-west-2-ac")...) + s.ErrorContains(res.Err, `scaling group "aws-us-west-2-ac" not found; run `+ + "`temporal worker deployment version compute-config describe`") + s.Empty(fake.takeUpdates()) + + // Without --scaling-group, every scaling group is removed after the prompt + // lists them. + fake.setComputeConfig(twoScalingGroupComputeConfig()) + s.CommandHarness.Stdin.WriteString("n\n") + res = s.Execute(removeArgs...) + s.ErrorContains(res.Err, "user denied confirmation") + s.Contains(res.Stdout.String(), fmt.Sprintf( + `Remove all 2 scaling groups (aws-us-east-1-wf, aws-us-west-2-ac) from Worker Deployment Version %q with Build ID "build-17"? y/N`, + deploymentName)) + s.Empty(fake.takeUpdates()) + + s.CommandHarness.Stdin.WriteString("y\n") + res = s.Execute(removeArgs...) + s.NoError(res.Err) + s.Contains(res.Stdout.String(), "Successfully removed worker deployment version compute config") + updates = fake.takeUpdates() + s.Len(updates, 1) + s.Equal([]string{"aws-us-east-1-wf", "aws-us-west-2-ac"}, updates[0].GetRemoveComputeConfigScalingGroups()) + s.Empty(updates[0].GetComputeConfigScalingGroups()) + + // With a single scaling group, the prompt names it. + single := twoScalingGroupComputeConfig() + delete(single.ScalingGroups, "aws-us-west-2-ac") + fake.setComputeConfig(single) + s.CommandHarness.Stdin.WriteString("n\n") + res = s.Execute(removeArgs...) + s.ErrorContains(res.Err, "user denied confirmation") + s.Contains(res.Stdout.String(), fmt.Sprintf( + `Remove scaling group "aws-us-east-1-wf" from Worker Deployment Version %q with Build ID "build-17"? y/N`, + deploymentName)) + s.Empty(fake.takeUpdates()) + + // --yes skips the prompt, and JSON output requires it. + fake.setComputeConfig(twoScalingGroupComputeConfig()) + res = s.Execute(append(removeArgs, "-o", "json")...) + s.ErrorContains(res.Err, "must bypass prompts when using JSON output") + s.Empty(fake.takeUpdates()) + res = s.Execute(append(removeArgs, "--yes")...) + s.NoError(res.Err) + s.Len(fake.takeUpdates(), 1) +} + +func (s *SharedServerSuite) TestComputeConfig_Remove_EmptyScalingGroup() { + fake := s.newFakeComputeConfigServer() + res := s.Execute( + "worker", "deployment", "version", "compute-config", "remove", + "--address", s.Address(), + "--deployment-name", uuid.NewString(), + "--build-id", "build-17", + "--scaling-group", "", + ) + s.ErrorContains(res.Err, "--scaling-group cannot be empty") + s.Empty(fake.takeUpdates()) +} + +// TestComputeConfig_NonExistentDeployment runs against the real server, where +// neither the Worker Deployment nor the Version exists. +func (s *SharedServerSuite) TestComputeConfig_NonExistentDeployment() { + deploymentName := uuid.NewString() + res := s.Execute( + "worker", "deployment", "version", "compute-config", "aws-lambda", "create", + "--address", s.Address(), + "--deployment-name", deploymentName, + "--build-id", "non-existing", + "--function-arn", "arn:aws:lambda:us-east-1:123:function:F", + "--skip-role-and-external-id", + ) + s.ErrorContains(res.Err, "error creating worker deployment version") + s.ErrorContains(res.Err, deploymentName) + + res = s.Execute( + "worker", "deployment", "version", "compute-config", "aws-lambda", "update", + "--address", s.Address(), + "--deployment-name", deploymentName, + "--build-id", "non-existing", + "--function-arn", "arn:aws:lambda:us-east-1:123:function:F", + "--skip-role-and-external-id", + ) + s.ErrorContains(res.Err, "to create it, use `temporal worker deployment version compute-config aws-lambda create`") + + res = s.Execute( + "worker", "deployment", "version", "compute-config", "remove", + "--address", s.Address(), + "--deployment-name", deploymentName, + "--build-id", "non-existing", + ) + s.ErrorContains(res.Err, "error describing worker deployment version") +} + +// TestVersion_Describe_MatchesDescribeVersion checks that version describe prints +// exactly what describe-version prints. +func (s *SharedServerSuite) TestVersion_Describe_MatchesDescribeVersion() { + fake := s.newFakeComputeConfigServer() + at := func(minute int) *timestamppb.Timestamp { + return timestamppb.New(time.Date(2026, 1, 2, 3, minute, 0, 0, time.UTC)) + } + stats := func(backlog int64) *taskqueuepb.TaskQueueStats { + return &taskqueuepb.TaskQueueStats{ + ApproximateBacklogCount: backlog, + ApproximateBacklogAge: durationpb.New(90 * time.Second), + TasksAddRate: 2.5, + TasksDispatchRate: 1.5, + } + } + owner, err := converter.GetDefaultDataConverter().ToPayload("payments-team") + s.NoError(err) + fake.describeTemplate = &workflowservice.DescribeWorkerDeploymentVersionResponse{ + WorkerDeploymentVersionInfo: &deploymentpb.WorkerDeploymentVersionInfo{ + CreateTime: at(1), + RoutingChangedTime: at(2), + CurrentSinceTime: at(3), + RampingSinceTime: at(4), + RampPercentage: 25, + DrainageInfo: &deploymentpb.VersionDrainageInfo{ + Status: enumspb.VERSION_DRAINAGE_STATUS_DRAINING, + LastChangedTime: at(5), + LastCheckedTime: at(6), + }, + Metadata: &deploymentpb.VersionMetadata{Entries: map[string]*commonpb.Payload{"owner": owner}}, + }, + VersionTaskQueues: []*workflowservice.DescribeWorkerDeploymentVersionResponse_VersionTaskQueue{ + { + Name: "tq-a", + Type: enumspb.TASK_QUEUE_TYPE_WORKFLOW, + Stats: stats(7), + StatsByPriorityKey: map[int32]*taskqueuepb.TaskQueueStats{1: stats(4), 3: stats(3)}, + }, + {Name: "tq-b", Type: enumspb.TASK_QUEUE_TYPE_ACTIVITY, Stats: stats(0)}, + }, + } + fake.setComputeConfig(twoScalingGroupComputeConfig()) + + for _, statsFlags := range [][]string{nil, {"--report-task-queue-stats"}} { + for _, outputFlags := range [][]string{nil, {"-o", "json"}} { + flags := append([]string{"--address", s.Address(), "--deployment-name", "d", "--build-id", "b"}, statsFlags...) + flags = append(flags, outputFlags...) + existing := s.Execute(append([]string{"worker", "deployment", "describe-version"}, flags...)...) + s.NoError(existing.Err) + described := s.Execute(append([]string{"worker", "deployment", "version", "describe"}, flags...)...) + s.NoError(described.Err) + if outputFlags != nil { + s.JSONEq(existing.Stdout.String(), described.Stdout.String(), "flags %v", flags) + } else { + s.Equal(existing.Stdout.String(), described.Stdout.String(), "flags %v", flags) + } + } + } + s.Empty(fake.takeUpdates()) +} + +// TestVersion_CreateDescribeDelete runs against the real server, with a Version +// that has no compute config. +func (s *SharedServerSuite) TestVersion_CreateDescribeDelete() { + deploymentName := uuid.NewString() + version := []string{ + "--address", s.Address(), + "--deployment-name", deploymentName, + "--build-id", "build-1", + } + + // The Worker Deployment must exist first. + res := s.Execute(append([]string{"worker", "deployment", "version", "create"}, version...)...) + s.ErrorContains(res.Err, "error creating worker deployment version") + + res = s.Execute("worker", "deployment", "create", "--address", s.Address(), "--name", deploymentName) + s.NoError(res.Err) + s.EventuallyWithT(func(t *assert.CollectT) { + res := s.Execute(append([]string{"worker", "deployment", "version", "create"}, version...)...) + assert.NoError(t, res.Err) + assert.Contains(t, res.Stdout.String(), "Successfully created worker deployment version") + }, 30*time.Second, 100*time.Millisecond) + + res = s.Execute(append([]string{"worker", "deployment", "version", "describe", "-o", "json"}, version...)...) + s.NoError(res.Err) + var described map[string]any + s.NoError(json.Unmarshal(res.Stdout.Bytes(), &described)) + s.Equal(deploymentName, described["deploymentName"]) + s.Equal("build-1", described["BuildID"]) + s.NotContains(described, "computeConfig") + + // Creating the same Version again fails. + res = s.Execute(append([]string{"worker", "deployment", "version", "create"}, version...)...) + s.ErrorContains(res.Err, "already exists") + + res = s.Execute(append([]string{"worker", "deployment", "version", "delete"}, version...)...) + s.NoError(res.Err) + s.Contains(res.Stdout.String(), "Successfully deleted worker deployment version") + + res = s.Execute(append([]string{"worker", "deployment", "version", "describe"}, version...)...) + s.ErrorContains(res.Err, "error describing worker deployment version") +} diff --git a/internal/temporalcli/commands.yaml b/internal/temporalcli/commands.yaml index e363e1fda..bc2e358a0 100644 --- a/internal/temporalcli/commands.yaml +++ b/internal/temporalcli/commands.yaml @@ -1145,6 +1145,8 @@ commands: - worker deployment delete-version - worker deployment update-version-metadata - worker deployment update-version-compute-config + - worker deployment version + - worker deployment version compute-config - worker deployment manager-identity - name: temporal worker deployment create @@ -1851,6 +1853,309 @@ commands: Removes any compute configuration associated with this Worker Deployment Version. + - name: temporal worker deployment version + summary: Create, describe, or delete a Worker Deployment Version, and manage its compute configuration (Experimental) + description: | + Version commands create, describe, and delete a single Worker + Deployment Version, and manage its compute configuration: + + ``` + temporal worker deployment version [command] [options] + ``` + + Note: This is an experimental feature and may change in the future. + + - name: temporal worker deployment version create + summary: Create a new Worker Deployment Version (Experimental) + description: | + Create a new Worker Deployment Version: + + ``` + temporal worker deployment version create \ + --deployment-name YourDeploymentName \ + --build-id YourBuildID + ``` + + The Worker Deployment must already exist; create it with + `temporal worker deployment create`. + + If a Worker Deployment Version with the supplied BuildID already exists, + this command will return an error. + + Note: This is an experimental feature and may change in the future. + option-sets: + - deployment-version + + - name: temporal worker deployment version describe + summary: Show properties of a Worker Deployment Version (Experimental) + description: | + Describe properties of a Worker Deployment Version, such as the task + queues polled by workers in this Deployment Version, or drainage + information required to safely decommission workers, or user-provided + metadata, or its creation/modification time. + + ``` + temporal worker deployment version describe [options] + ``` + + For example, to describe a deployment version in a deployment + `YourDeploymentName`, with Build ID `YourBuildID`, and in the default + namespace: + + ``` + temporal worker deployment version describe \ + --deployment-name YourDeploymentName \ + --build-id YourBuildID + ``` + + Note: This is an experimental feature and may change in the future. + option-sets: + - deployment-version + options: + - name: report-task-queue-stats + type: bool + description: Report stats for task queues that are present in this version. + + - name: temporal worker deployment version delete + summary: Delete a Worker Deployment Version (Experimental) + description: | + Remove a Worker Deployment Version given its fully-qualified identifier. + This is rarely needed during normal operation + since unused Versions are eventually garbage collected. + The client can delete a Version only when all of the following conditions + are met: + - It is not the Current or Ramping Version for this Deployment. + - It has no active pollers, i.e., none of the task queues in the + Version have pollers. + - It is not draining. This requirement can be ignored with the option + `--skip-drainage`. + ``` + temporal worker deployment version delete [options] + ``` + + For example, skipping the drainage restriction: + + ``` + temporal worker deployment version delete \ + --deployment-name YourDeploymentName \ + --build-id YourBuildID \ + --skip-drainage + ``` + + Note: This is an experimental feature and may change in the future. + option-sets: + - deployment-version + options: + - name: skip-drainage + type: bool + description: Ignore the deletion requirement of not draining. + + - name: temporal worker deployment version compute-config + summary: Set, update, or remove compute configuration for a Version (Experimental) + description: | + A Worker Deployment Version's compute configuration is made of one + or more scaling groups. Each scaling group is identified by a key + and is made of a compute provider that starts Workers and a scaler + that decides when to start them. Scaling groups set from the CLI + serve every Task Queue type. Each compute provider has its own + command, such as `aws-lambda`, to create and update scaling groups: + + ``` + temporal worker deployment version compute-config [command] [options] + ``` + + Note: This is an experimental feature and may change in the future. + + - name: temporal worker deployment version compute-config describe + summary: Show the compute configuration of a Version (Experimental) + description: | + Show the scaling groups in the compute configuration of a Worker + Deployment Version, with their compute provider and scaler settings: + + ``` + temporal worker deployment version compute-config describe \ + --deployment-name YourDeploymentName \ + --build-id YourBuildID + ``` + + If a Worker Deployment Version with the supplied BuildID does not exist, + this command will return an error. + + Note: This is an experimental feature and may change in the future. + option-sets: + - deployment-version + + - name: temporal worker deployment version compute-config aws-lambda + summary: Create or update AWS Lambda compute configuration (Experimental) + description: | + AWS Lambda commands set the compute configuration of a Worker Deployment + Version for an AWS Lambda function that spawns a Worker in the Worker + Deployment: + + ``` + temporal worker deployment version compute-config aws-lambda [command] [options] + ``` + + Note: This is an experimental feature and may change in the future. + + - name: temporal worker deployment version compute-config aws-lambda create + summary: Create a Version with AWS Lambda compute configuration (Experimental) + description: | + Create a new Worker Deployment Version with compute configuration for an + AWS Lambda function that spawns a Worker in the Worker Deployment: + + ``` + temporal worker deployment version compute-config aws-lambda create \ + --deployment-name YourDeploymentName \ + --build-id YourBuildID \ + --function-arn YourLambdaFunctionARN \ + --assume-role-arn YourAssumeRoleARN \ + --assume-role-external-id YourExternalID + ``` + + To tune when Workers are started, pass any of the scaler flags (the + --scale-up-* and --suppress-* flags, --max-worker-lifetime-ms, and + --metrics-poll-interval-ms). Each one that is not set keeps the + server default: + + ``` + temporal worker deployment version compute-config aws-lambda create \ + --deployment-name YourDeploymentName \ + --build-id YourBuildID \ + --function-arn YourLambdaFunctionARN \ + --assume-role-arn YourAssumeRoleARN \ + --assume-role-external-id YourExternalID \ + --scale-up-backlog-threshold 5 \ + --scale-up-cooloff-ms 500 \ + --max-worker-lifetime-ms 300000 + ``` + + The Worker Deployment must already exist; create it with + `temporal worker deployment create`. + + If a Worker Deployment Version with the supplied BuildID already exists, + this command will return an error. To update its compute configuration, + use `temporal worker deployment version compute-config aws-lambda update`. + + Note: This is an experimental feature and may change in the future. + option-sets: + - deployment-version + - scaling-group + - aws-assume-role + - no-sync-scaler + options: + - name: function-arn + type: string + required: true + description: | + Qualified (contains version suffix) or unqualified AWS Lambda + function ARN to invoke when there are no active pollers for task + queue targets in the Worker Deployment. + + - name: temporal worker deployment version compute-config aws-lambda update + summary: Update AWS Lambda compute configuration for a Version (Experimental) + description: | + Update a scaling group in the compute configuration of a Worker + Deployment Version, for an AWS Lambda function that spawns a Worker in + the Worker Deployment. Only the scaling group named by --scaling-group + ("default" when unset) changes; other scaling groups are not affected. + + To update the AWS Lambda function and the AWS IAM role associated with + an existing Worker Deployment Version: + + ``` + temporal worker deployment version compute-config aws-lambda update \ + --deployment-name YourDeploymentName \ + --build-id YourBuildID \ + --function-arn YourLambdaFunctionARN \ + --assume-role-arn YourAssumeRoleARN \ + --assume-role-external-id YourExternalID + ``` + + To update only the scaler of an existing scaling group, supply the + scaler flags without the provider flags. The scaler is replaced as a + whole, so scaler flags that are not passed go back to their + defaults: + + ``` + temporal worker deployment version compute-config aws-lambda update \ + --deployment-name YourDeploymentName \ + --build-id YourBuildID \ + --scale-up-backlog-threshold 5 \ + --max-worker-lifetime-ms 300000 + ``` + + The provider flags are --function-arn, --assume-role-arn, + --assume-role-external-id, and --skip-role-and-external-id. The + scaler flags are the --scale-up-* and --suppress-* flags, + --max-worker-lifetime-ms, and --metrics-poll-interval-ms. The + provider flags are only needed when changing the compute provider, + and then must be passed together: --function-arn with either both + role flags or --skip-role-and-external-id. Changing the provider + also resets the scaler: scaler flags passed with it are applied, and + the rest, or all of them if none are passed, go back to their + defaults. + + If a Worker Deployment Version with the supplied BuildID does not exist, + this command will return an error. To create it, use + `temporal worker deployment version compute-config aws-lambda create`. + + Returns an error if all compute configuration fields are empty. + + Note: This is an experimental feature and may change in the future. + option-sets: + - deployment-version + - scaling-group + - aws-assume-role + - no-sync-scaler + options: + - name: function-arn + type: string + description: | + Qualified (contains version suffix) or unqualified AWS Lambda + function ARN to invoke when there are no active pollers for task + queue targets in the Worker Deployment. Required when changing the + compute provider. + + - name: temporal worker deployment version compute-config remove + summary: Remove compute configuration from a Version (Experimental) + description: | + Remove the compute configuration of a Worker Deployment Version. The + command lists its scaling groups and asks for confirmation first: + + ``` + temporal worker deployment version compute-config remove \ + --deployment-name YourDeploymentName \ + --build-id YourBuildID + ``` + + Remove one scaling group from the compute configuration, such as the one + named "default": + + ``` + temporal worker deployment version compute-config remove \ + --deployment-name YourDeploymentName \ + --build-id YourBuildID \ + --scaling-group default + ``` + + If the Worker Deployment Version or the scaling group does not exist, + this command will return an error. + + Note: This is an experimental feature and may change in the future. + option-sets: + - deployment-version + options: + - name: scaling-group + type: string + description: | + Key of the scaling group to remove. If unset, all scaling groups are + removed. + - name: yes + short: y + type: bool + description: Don't prompt to confirm removing all scaling groups. + - name: temporal worker deployment manager-identity summary: Manager Identity commands change the `ManagerIdentity` of a Worker Deployment description: | @@ -5452,6 +5757,95 @@ option-sets: Build ID of the Worker Deployment Version. required: true + - name: scaling-group + options: + - name: scaling-group + type: string + default: default + description: | + Key of the scaling group. + + - name: aws-assume-role + options: + - name: assume-role-arn + type: string + description: | + AWS IAM role ARN that the Temporal server will assume when calling + the compute provider that spawns a new Worker in this Worker + Deployment Version. Required when setting the compute provider, and + must be omitted when --skip-role-and-external-id is passed. + - name: assume-role-external-id + type: string + description: | + Temporal server will enforce that the AWS IAM trust policy associated + with the AWS IAM role specified in --assume-role-arn has an + aws:ExternalId condition that matches the supplied value. Required + when setting the compute provider, and must be omitted when + --skip-role-and-external-id is passed. + - name: skip-role-and-external-id + type: bool + description: | + When setting the compute provider, --assume-role-arn and + --assume-role-external-id are required unless this flag is passed, + in which case both must be omitted. + + - name: no-sync-scaler + options: + - name: scale-up-backlog-threshold + type: int + description: | + At each Task Queue metrics check (see --metrics-poll-interval-ms), + start a new Worker if the backlog is larger than this number of + tasks. Defaults to 0 when unset. A task that arrives while no Worker + is waiting for it still starts a Worker, subject to + --scale-up-cooloff-ms. + - name: scale-up-cooloff-ms + type: int + description: | + Minimum time between two Worker starts, in milliseconds. A value of + 0 disables the wait. Defaults to 100 when unset. + - name: max-worker-lifetime-ms + type: int + description: | + While there is a backlog, start a new Worker at least this often, in + milliseconds, even within --scale-up-cooloff-ms. A value of 0 + disables this. Defaults to 600000 (10 minutes) when unset. + - name: metrics-poll-interval-ms + type: int + description: | + How often Task Queue metrics are checked to decide whether to start + a Worker, in milliseconds. Must be at least 10000, and at least + --scale-up-cooloff-ms. Defaults to 60000 (1 minute) when unset. + - name: scale-up-dispatch-rate-epsilon + type: float + description: | + Band around the dispatch rate, as a fraction of it, of at most 0.1. + While a Task Queue type has a backlog and its dispatch rate stays + within this band, Worker starts for it are paused. Starts from + --max-worker-lifetime-ms are not paused. Defaults to 0 when unset, + which disables pausing. + - name: scale-up-dispatch-rate-epsilon-confirm-ms + type: int + description: | + How long the dispatch rate must stay within + --scale-up-dispatch-rate-epsilon before Worker starts are paused, in + milliseconds. Must be longer than --metrics-poll-interval-ms. Only + valid with --scale-up-dispatch-rate-epsilon. Defaults to 90000 when + unset. + - name: suppress-scale-up-ms + type: int + description: | + How long Worker starts stay paused each time the dispatch rate is + found within --scale-up-dispatch-rate-epsilon, in milliseconds. Must + be longer than --suppress-poll-interval-ms. Only valid with + --scale-up-dispatch-rate-epsilon. Defaults to 120000 when unset. + - name: suppress-poll-interval-ms + type: int + description: | + How often Task Queue metrics are checked while Worker starts are + paused, in milliseconds. Only valid with + --scale-up-dispatch-rate-epsilon. Defaults to 90000 when unset. + - name: deployment-version-or-unversioned options: - name: deployment-name