tdbg

type BoolFlagLookup ¶ added in v1.23.0

type BoolFlagLookup interface {
	Bool(name string) bool
}

BoolFlagLookup can be satisfied by github.com/urfave/cli/v2.Context.

type ClientFactory ¶

type ClientFactory interface {
	AdminClient(c *cli.Context) adminservice.AdminServiceClient
	WorkflowClient(c *cli.Context) workflowservice.WorkflowServiceClient
}

ClientFactory is used to construct rpc clients

func NewClientFactory(opts ...ClientFactoryOption) ClientFactory

type ClientFactoryOption ¶ added in v1.23.0

type ClientFactoryOption func(params *clientFactoryParams)

ClientFactoryOption is used to configure the ClientFactory via NewClientFactory.

func WithFrontendAddress(address string) ClientFactoryOption

type DLQJobService ¶ added in v1.23.0

type DLQJobService struct {
	// contains filtered or unexported fields
}

func (ac *DLQJobService) CancelJob(c *cli.Context) error

func (ac *DLQJobService) DescribeJob(c *cli.Context) error

type DLQMessage ¶ added in v1.23.0

type DLQMessage struct {
	// MessageID is the ID of the message within the DLQ. You can use this ID as an input to the `--last_message_id`
	// flag for the `purge` and `merge` commands.
	MessageID int64 `json:"message_id"`
	// ShardID is only used for non-namespace replication tasks.
	ShardID int32 `json:"shard_id"`
	// Payload contains the parsed task metadata from the server.
	Payload *TaskPayload `json:"payload"`
}

DLQMessage is used primarily to form the JSON output of the `read` command. It's only used for v2.

type DLQService ¶ added in v1.23.0

type DLQService interface {
	ReadMessages(c *cli.Context) error
	PurgeMessages(c *cli.Context) error
	MergeMessages(c *cli.Context) error
	ListQueues(c *cli.Context) error
}

type DLQServiceProvider ¶ added in v1.23.0

type DLQServiceProvider struct {
	// contains filtered or unexported fields
}

func NewDLQServiceProvider(
	clientFactory ClientFactory,
	taskBlobEncoder TaskBlobEncoder,
	taskCategoryRegistry tasks.TaskCategoryRegistry,
	writer io.Writer,
	prompterFactory PrompterFactory,
) *DLQServiceProvider

func (p *DLQServiceProvider) GetDLQJobService() DLQJobService

func (p *DLQServiceProvider) GetDLQService(
	c *cli.Context,
) (DLQService, error)

type DLQV1Service ¶ added in v1.23.0

type DLQV1Service struct {
	// contains filtered or unexported fields
}

func NewDLQV1Service(clientFactory ClientFactory, prompter *Prompter, writer io.Writer) *DLQV1Service

func (ac *DLQV1Service) ListQueues(c *cli.Context) error

func (ac *DLQV1Service) MergeMessages(c *cli.Context) error

func (ac *DLQV1Service) PurgeMessages(c *cli.Context) error

func (ac *DLQV1Service) ReadMessages(c *cli.Context) (err error)

type DLQV2Service ¶ added in v1.23.0

type DLQV2Service struct {
	// contains filtered or unexported fields
}

DLQV2Service implements DLQService for persistence.QueueV2.

func NewDLQJobService(
	clientFactory ClientFactory,
	writer io.Writer,
) *DLQV2Service

func NewDLQV2Service(
	category tasks.Category,
	sourceCluster string,
	targetCluster string,
	clientFactory ClientFactory,
	writer io.Writer,
	prompter *Prompter,
	taskBlobEncoder TaskBlobEncoder,
) *DLQV2Service

func (ac *DLQV2Service) ListQueues(c *cli.Context) (err error)

func (ac *DLQV2Service) MergeMessages(c *cli.Context) error

func (ac *DLQV2Service) PurgeMessages(c *cli.Context) error

func (ac *DLQV2Service) ReadMessages(c *cli.Context) (err error)

type DefaultFrontendAddressProvider ¶ added in v1.23.0

type DefaultFrontendAddressProvider struct{}

DefaultFrontendAddressProvider uses FlagAddress to determine the frontend address, defaulting to DefaultFrontendAddress if FlagAddress is not set or is empty.

func (d DefaultFrontendAddressProvider) GetFrontendAddress(c *cli.Context) string

type HttpGetter ¶

type HttpGetter interface {
	Get(url string) (resp *http.Response, err error)
}

HttpGetter defines http.Client.Get(...) as an interface so we can mock it

type Option ¶ added in v1.23.0

type Option func(params *Params)

Option modifies the Params for tdbg.

type Params ¶ added in v1.23.0

type Params struct {
	// ClientFactory creates Temporal service clients for tdbg to use.
	ClientFactory ClientFactory
	// TaskCategoryRegistry is used to determine which task categories are available for tdbg to use.
	TaskCategoryRegistry tasks.TaskCategoryRegistry
	// Writer is used to write output from tdbg. The default is os.Stdout.
	Writer io.Writer
	// TaskBlobEncoder is needed for custom task serialization. The default uses PredefinedTaskBlobDeserializer.
	TaskBlobEncoder TaskBlobEncoder
}

Params which are customizable for the CLI application.

type PredefinedTaskBlobDeserializer ¶ added in v1.23.0

type PredefinedTaskBlobDeserializer struct{}

PredefinedTaskBlobDeserializer is a TaskBlobProtoDeserializer that deserializes task blobs into the predefined task categories that are used by Temporal. If your server has custom categories, you'll want to build something on top of this.

func NewPredefinedTaskBlobDeserializer() PredefinedTaskBlobDeserializer

func (d PredefinedTaskBlobDeserializer) Deserialize(categoryID int, blob *commonpb.DataBlob) (proto.Message, error)

type Prompter ¶ added in v1.23.0

type Prompter struct {
	// contains filtered or unexported fields
}

Prompter is a helper for prompting the user for confirmation.

func NewPrompter(c BoolFlagLookup, opts ...PrompterOption) *Prompter

func (p *Prompter) Prompt(msg string)

type PrompterFactory ¶ added in v1.23.0

type PrompterFactory func(c BoolFlagLookup) *Prompter

func NewPrompterFactory(opts ...PrompterOption) PrompterFactory

type PrompterOption ¶ added in v1.23.0

type PrompterOption func(*PrompterParams)

PrompterOption is used to override default PrompterParams.

type PrompterParams ¶ added in v1.23.0

type PrompterParams struct {
	// Reader defaults to os.Stdin.
	Reader io.Reader
	// Writer defaults to os.Stdout.
	Writer io.Writer
	// Exiter defaults to [os.Exit].
	Exiter func(code int)
}

PrompterParams is used to configure a new Prompter.

type ProtoTaskBlobEncoder ¶ added in v1.23.0

type ProtoTaskBlobEncoder struct {
	// contains filtered or unexported fields
}

ProtoTaskBlobEncoder is a TaskBlobEncoder that uses a TaskBlobProtoDeserializer to deserialize the blob into a proto message, and then uses protojson to marshal the proto message into a human-readable format.

func NewProtoTaskBlobEncoder(deserializer TaskBlobProtoDeserializer) *ProtoTaskBlobEncoder

func (e *ProtoTaskBlobEncoder) Encode(writer io.Writer, categoryID int, blob *commonpb.DataBlob) error

type TaskBlobEncoder ¶ added in v1.23.0

type TaskBlobEncoder interface {
	Encode(writer io.Writer, taskCategoryID int, blob *commonpb.DataBlob) error
}

TaskBlobEncoder takes a blob for a given task category and encodes it to a human-readable format. Here's a breakdown of the relationship between all the related types needed to implement a custom encoder: - NewCliApp accepts a list of Option objects. - Option objects modify Params. - Params contain a TaskBlobEncoder. - TaskBlobEncoder is implemented by ProtoTaskBlobEncoder. - ProtoTaskBlobEncoder uses protojson to marshal proto.Message objects from a TaskBlobProtoDeserializer. - TaskBlobProtoDeserializer is implemented by the stock PredefinedTaskBlobDeserializer. - PredefinedTaskBlobDeserializer deserializes commonpb.DataBlob objects into proto.Message objects.

package main

import (
	"bytes"
	"fmt"
	"io"
	"os"
	"strconv"

	commonpb "go.temporal.io/api/common/v1"
	enumspb "go.temporal.io/api/enums/v1"

	"go.temporal.io/server/service/history/tasks"
	"go.temporal.io/server/tools/tdbg"
)

var customCategory = tasks.Category{}

func main() {
	var output bytes.Buffer
	app := tdbg.NewCliApp(func(params *tdbg.Params) {
		params.Writer = &output
		stockEncoder := params.TaskBlobEncoder
		params.TaskBlobEncoder = tdbg.TaskBlobEncoderFn(func(
			writer io.Writer,
			taskCategoryID int,
			blob *commonpb.DataBlob,
		) error {
			if taskCategoryID == customCategory.ID() {
				_, err := writer.Write(append([]byte("hello, "), blob.Data...))
				return err
			}
			return stockEncoder.Encode(writer, taskCategoryID, blob)
		})
	})
	file, err := os.CreateTemp("", "*")
	if err != nil {
		panic(err)
	}
	defer func() {
		if err := os.Remove(file.Name()); err != nil {
			panic(err)
		}
	}()
	_, err = file.Write([]byte("\"world\""))
	if err != nil {
		panic(err)
	}
	err = app.Run([]string{
		"tdbg", "decode", "task",
		"--" + tdbg.FlagEncoding, enumspb.ENCODING_TYPE_JSON.String(),
		"--" + tdbg.FlagTaskCategoryID, strconv.Itoa(customCategory.ID()),
		"--" + tdbg.FlagBinaryFile, file.Name(),
	})
	if err != nil {
		panic(err)
	}

	fmt.Println(output.String())

}

Output:

hello, "world"

type TaskBlobEncoderFn ¶ added in v1.23.0

type TaskBlobEncoderFn func(writer io.Writer, taskCategoryID int, blob *commonpb.DataBlob) error

TaskBlobEncoderFn implements TaskBlobEncoder by calling a function.

func (e TaskBlobEncoderFn) Encode(writer io.Writer, taskCategoryID int, blob *commonpb.DataBlob) error

type TaskBlobProtoDeserializer ¶ added in v1.23.0

type TaskBlobProtoDeserializer interface {
	Deserialize(taskCategoryID int, blob *commonpb.DataBlob) (proto.Message, error)
}

TaskBlobProtoDeserializer is used to deserialize task blobs into proto messages. This makes it easier to create an encoder if your tasks are all backed by protos. We separate this from the encoder because we don't want the encoder to be tied to protos as the wire format.

type TaskPayload ¶ added in v1.23.0

type TaskPayload struct {
	// contains filtered or unexported fields
}

TaskPayload implements both json.Marshaler and json.Unmarshaler. This allows us to pretty-print tasks using jsonpb when serializing and then store the raw bytes of the task payload for later use when deserializing. We need to store the raw bytes instead of immediately decoding to a concrete type because that logic is dynamic and can't depend solely on the task category ID in case there are additional task categories in use.

func (p *TaskPayload) Bytes() []byte

func (p *TaskPayload) MarshalJSON() ([]byte, error)

func (p *TaskPayload) UnmarshalJSON(b []byte) error