Files
tasks/tools/tdbg/util.go
Alan WuandGitHub cf7c62c8cc Add batch workflow refresh tasks (#8793)
## What changed?
Add batch workflow refresh tasks Admin API.

Add tdbg commands to invoke `StartAdminBatchOperation`.

## Details

AdminBatchOperations are executed in the Batched System Worker, similar
to existing Frontend BatchOperations. The worker activity will fetch
pages of workflow executions to perform operations on, and periodically
heartbeat its results. For RefreshWorkflowTasks, each individual Refresh
call will go through the admin handler.

Refreshing Workflow Tasks regenerates all pending tasks of an execution
given its mutable state.

## CLI usage

```
tdbg workflow batch-refresh-tasks \
  --namespace <ns> \
  --query "WorkflowType='MyType'" \
  --reason "fixing stuck workflows" \
  [--job-id <optional-job-id>] \
  [--archetype <optional-archetype>]
```

## Why?
Unblock new Matcher migration and allow for general use case batch Admin
calls. Currently, only supports `BatchOperationRefreshWorkflowTasks`.

## How did you test it?
- [X] built
- [X] run locally and tested manually
- [X] covered by existing tests
- [X] added new unit test(s)
- [X] added new functional test(s)
2026-01-06 19:21:57 -05:00

322 lines
8.4 KiB
Go

package tdbg
import (
"context"
"encoding/json"
"fmt"
"io"
"os"
"reflect"
"regexp"
"strconv"
"strings"
"time"
"github.com/fatih/color"
"github.com/olekukonko/tablewriter"
"github.com/urfave/cli/v2"
"go.temporal.io/api/workflowservice/v1"
"go.temporal.io/server/chasm"
"go.temporal.io/server/common/codec"
"go.temporal.io/server/common/collection"
"go.temporal.io/server/common/namespace"
"google.golang.org/protobuf/proto"
)
var envKeysForUserName = []string{
"USER",
"LOGNAME",
"HOME",
}
func getCurrentUserFromEnv() string {
for _, n := range envKeysForUserName {
if len(os.Getenv(n)) > 0 {
return os.Getenv(n)
}
}
return "unknown"
}
func prettyPrintJSONObject(c *cli.Context, o interface{}) {
var b []byte
var err error
if pb, ok := o.(proto.Message); ok {
encoder := codec.NewJSONPBIndentEncoder(" ")
b, err = encoder.Encode(pb)
} else {
b, err = json.MarshalIndent(o, "", " ")
}
if err != nil {
fmt.Fprintf(c.App.ErrWriter, "Error when trying to pretty-print: %v\n%v\n", err, o)
return
}
_, _ = c.App.Writer.Write(b)
_, _ = c.App.Writer.Write([]byte("\n"))
}
func getRequiredOption(c *cli.Context, optionName string) (string, error) {
value := c.String(optionName)
if len(value) == 0 {
return "", fmt.Errorf("option is required: %s", optionName)
}
return value, nil
}
func parseTime(timeStr string, defaultValue time.Time, now time.Time) (time.Time, error) {
if len(timeStr) == 0 {
return defaultValue, nil
}
// try to parse
parsedTime, err := time.Parse(defaultDateTimeFormat, timeStr)
if err == nil {
return parsedTime, nil
}
// treat as raw unix time
resultValue, err := strconv.ParseInt(timeStr, 10, 64)
if err == nil {
return time.Unix(0, resultValue).UTC(), nil
}
// treat as time range format
parsedTime, err = parseTimeRange(timeStr, now)
if err != nil {
return time.Time{}, fmt.Errorf("cannot parse time '%s', use UTC format '2006-01-02T15:04:05', "+
"time range or raw UnixNano directly. See help for more details: %s", timeStr, err)
}
return parsedTime, nil
}
// parseTimeRange parses a given time duration string (in format X<time-duration>) and
// returns parsed timestamp given that duration in the past from current time.
// All valid values must contain a number followed by a time-duration, from the following list (long form/short form):
// - second/s
// - minute/m
// - hour/h
// - day/d
// - week/w
// - month/M
// - year/y
// For example, possible input values, and their result:
// - "3d" or "3day" --> three days --> time.Now().UTC().Add(-3 * 24 * time.Hour)
// - "2m" or "2minute" --> two minutes --> time.Now().UTC().Add(-2 * time.Minute)
// - "1w" or "1week" --> one week --> time.Now().UTC().Add(-7 * 24 * time.Hour)
// - "30s" or "30second" --> thirty seconds --> time.Now().UTC().Add(-30 * time.Second)
// Note: Duration strings are case-sensitive, and should be used as mentioned above only.
// Limitation: Value of numerical multiplier, X should be in b/w 0 - 1e6 (1 million), boundary values excluded i.e.
// 0 < X < 1e6. Also, the maximum time in the past can be 1 January 1970 00:00:00 UTC (epoch time),
// so giving "1000y" will result in epoch time.
func parseTimeRange(timeRange string, now time.Time) (time.Time, error) {
match, err := regexp.MatchString(defaultDateTimeRangeShortRE, timeRange)
if !match { // fallback on to check if it's of longer notation
_, err = regexp.MatchString(defaultDateTimeRangeLongRE, timeRange)
}
if err != nil {
return time.Time{}, err
}
re, _ := regexp.Compile(defaultDateTimeRangeNum)
idx := re.FindStringSubmatchIndex(timeRange)
if idx == nil {
return time.Time{}, fmt.Errorf("cannot parse timeRange %s", timeRange)
}
num, err := strconv.Atoi(timeRange[idx[0]:idx[1]])
if err != nil {
return time.Time{}, fmt.Errorf("cannot parse timeRange %s", timeRange)
}
if num >= 1e6 {
return time.Time{}, fmt.Errorf("invalid time-duation multiplier %d, allowed range is 0 < multiplier < 1000000", num)
}
dur, err := parseTimeDuration(timeRange[idx[1]:])
if err != nil {
return time.Time{}, fmt.Errorf("cannot parse timeRange %s", timeRange)
}
res := now.Add(time.Duration(-num) * dur) // using server's local timezone
epochTime := time.Unix(0, 0).UTC()
if res.Before(epochTime) {
res = epochTime
}
return res, nil
}
// parseTimeDuration parses the given time duration in either short or long convention
// and returns the time.Duration
// Valid values (long notation/short notation):
// - second/s
// - minute/m
// - hour/h
// - day/d
// - week/w
// - month/M
// - year/y
// NOTE: the input "duration" is case-sensitive
func parseTimeDuration(duration string) (dur time.Duration, err error) {
switch duration {
case "s", "second":
dur = time.Second
case "m", "minute":
dur = time.Minute
case "h", "hour":
dur = time.Hour
case "d", "day":
dur = day
case "w", "week":
dur = week
case "M", "month":
dur = month
case "y", "year":
dur = year
default:
err = fmt.Errorf("unknown time duration %s", duration)
}
return
}
func newContext(c *cli.Context) (context.Context, context.CancelFunc) {
return newContextWithTimeout(c, defaultContextTimeout)
}
func newContextWithTimeout(c *cli.Context, timeout time.Duration) (context.Context, context.CancelFunc) {
if c.IsSet(FlagContextTimeout) {
timeout = time.Duration(c.Int(FlagContextTimeout)) * time.Second
}
return context.WithTimeout(c.Context, timeout)
}
func StringToEnum(search string, candidates map[string]int32) (int32, error) {
if search == "" {
return 0, nil
}
var candidateNames []string
for key, value := range candidates {
if strings.EqualFold(key, search) {
return value, nil
}
candidateNames = append(candidateNames, key)
}
return 0, fmt.Errorf("could not find corresponding candidate for %s. Possible candidates: %q", search, candidateNames)
}
// paginate creates an interactive CLI mode to control the printing of items
func paginate[V any](c *cli.Context, paginationFn collection.PaginationFn[V], pageSize int) error {
more := c.Bool(FlagMore)
isTableView := !c.Bool(FlagPrintJSON)
iter := collection.NewPagingIterator(paginationFn)
var pageItems []interface{}
for iter.HasNext() {
item, err := iter.Next()
if err != nil {
return err
}
pageItems = append(pageItems, item)
if len(pageItems) == pageSize || !iter.HasNext() {
if isTableView {
if err := printTable(pageItems, c.App.Writer); err != nil {
return err
}
} else {
prettyPrintJSONObject(c, pageItems)
}
if !more || !showNextPage(c.App.Writer) {
break
}
pageItems = pageItems[:0]
}
}
return nil
}
var exportRgx = regexp.MustCompile("^[A-Z]")
func printTable(items []interface{}, writer io.Writer) error {
if len(items) == 0 {
return nil
}
e := reflect.ValueOf(items[0])
for e.Type().Kind() == reflect.Ptr {
e = e.Elem()
}
var fields []string
var exportedFields []int
t := e.Type()
for i := 0; i < e.NumField(); i++ {
if exportRgx.MatchString(t.Field(i).Name) {
fields = append(fields, t.Field(i).Name)
exportedFields = append(exportedFields, i)
}
}
table := tablewriter.NewWriter(writer)
table.SetBorder(false)
table.SetColumnSeparator("|")
table.SetHeader(fields)
table.SetHeaderLine(false)
for i := 0; i < len(items); i++ {
item := reflect.ValueOf(items[i])
for item.Type().Kind() == reflect.Ptr {
item = item.Elem()
}
var columns []string
for _, j := range exportedFields {
col := item.Field(j)
columns = append(columns, fmt.Sprintf("%v", col.Interface()))
}
table.Append(columns)
}
table.Render()
table.ClearRows()
return nil
}
func showNextPage(wr io.Writer) bool {
fmt.Fprintf(wr, "Press %s to show next page, press %s to quit: ",
color.GreenString("Enter"), color.RedString("any other key then Enter"))
var input string
_, _ = fmt.Scanln(&input)
return strings.Trim(input, " ") == ""
}
func getNamespaceID(c *cli.Context, clientFactory ClientFactory, nsName namespace.Name) (namespace.ID, error) {
wfClient := clientFactory.WorkflowClient(c)
ctx, cancel := newContext(c)
defer cancel()
nsResponse, err := wfClient.DescribeNamespace(ctx, &workflowservice.DescribeNamespaceRequest{
Namespace: nsName.String(),
})
if err != nil {
return namespace.EmptyID, err
}
return namespace.ID(nsResponse.NamespaceInfo.GetId()), nil
}
func getArchetypeWithDefault(
c *cli.Context,
defaultAchetype chasm.Archetype,
) chasm.Archetype {
archetype := c.String(FlagArchetype)
if archetype != "" {
return archetype
}
return defaultAchetype
}