workflows

Documentation

Index

• func GetCurrentCluster(ctx context.Context) (string, error)
• func NewJobClient(options ...ClientOption) workflowsv1connect.JobServiceClient
• func NewRecurrentTaskClient(options ...ClientOption) workflowsv1connect.RecurrentTaskServiceClient
• func NewTaskClient(options ...ClientOption) workflowsv1connect.TaskServiceClient
• func SubmitSubtasks(ctx context.Context, tasks ...Task) error
• func ValidateIdentifier(identifier TaskIdentifier) error
• func WithTaskSpan(ctx context.Context, name string, f func(ctx context.Context) error) error
• func WithTaskSpanResult[Result any](ctx context.Context, name string, f func(ctx context.Context) (Result, error)) (Result, error)
• type ClientOption
  • func WithAPIKey(apiKey string) ClientOption
  • func WithConnectClientOptions(options ...connect.ClientOption) ClientOption
  • func WithHTTPClient(httpClient connect.HTTPClient) ClientOption
  • func WithURL(url string) ClientOption
• type ContextKeyTaskExecutionType
• type ExecutableTask
• type ExplicitlyIdentifiableTask
• type JobService
  • func NewJobService(client workflowsv1connect.JobServiceClient, options ...JobServiceOption) *JobService
  • func (js *JobService) Submit(ctx context.Context, jobName, clusterSlug string, maxRetries int, ...) (*workflowsv1.Job, error)
• type JobServiceOption
  • func WithJobServiceTracerName(tracerName string) JobServiceOption
  • func WithJobServiceTracerProvider(tracerProvider trace.TracerProvider) JobServiceOption
• type RecurrentTaskService
  • func NewRecurrentTaskService(client workflowsv1connect.RecurrentTaskServiceClient) *RecurrentTaskService
  • func (rs *RecurrentTaskService) CreateStorageLocation(ctx context.Context, location string, storageType workflowsv1.StorageType) (*workflowsv1.StorageLocation, error)
  • func (rs *RecurrentTaskService) DeleteStorageLocation(ctx context.Context, storageLocationID *workflowsv1.UUID) error
  • func (rs *RecurrentTaskService) GetStorageLocation(ctx context.Context, storageLocationID *workflowsv1.UUID) (*workflowsv1.StorageLocation, error)
  • func (rs *RecurrentTaskService) ListStorageLocations(ctx context.Context) ([]*workflowsv1.StorageLocation, error)
• type Task
• type TaskIdentifier
  • func NewTaskIdentifier(name, version string) TaskIdentifier
• type TaskRunner
  • func NewTaskRunner(client workflowsv1connect.TaskServiceClient, options ...TaskRunnerOption) (*TaskRunner, error)
  • func (t *TaskRunner) GetRegisteredTask(identifier TaskIdentifier) (ExecutableTask, bool)
  • func (t *TaskRunner) RegisterTask(task ExecutableTask) error
  • func (t *TaskRunner) RegisterTasks(tasks ...ExecutableTask) error
  • func (t *TaskRunner) Run(ctx context.Context)
• type TaskRunnerOption
  • func WithCluster(clusterSlug string) TaskRunnerOption
  • func WithLogExecutionTimes(logExecutionTimes bool) TaskRunnerOption
  • func WithRunnerLogger(logger *slog.Logger) TaskRunnerOption
  • func WithRunnerTracerName(tracerName string) TaskRunnerOption
  • func WithRunnerTracerProvider(tracerProvider trace.TracerProvider) TaskRunnerOption

Functions

func GetCurrentCluster

func GetCurrentCluster(ctx context.Context) (string, error)

func NewJobClient

func NewJobClient(options ...ClientOption) workflowsv1connect.JobServiceClient

func NewRecurrentTaskClient

func NewRecurrentTaskClient(options ...ClientOption) workflowsv1connect.RecurrentTaskServiceClient

func NewTaskClient

func NewTaskClient(options ...ClientOption) workflowsv1connect.TaskServiceClient

func SubmitSubtasks

func SubmitSubtasks(ctx context.Context, tasks ...Task) error

func ValidateIdentifier

func ValidateIdentifier(identifier TaskIdentifier) error

ValidateIdentifier performs client-side validation on a task identifier.

func WithTaskSpan

func WithTaskSpan(ctx context.Context, name string, f func(ctx context.Context) error) error

func WithTaskSpanResult

func WithTaskSpanResult[Result any](ctx context.Context, name string, f func(ctx context.Context) (Result, error)) (Result, error)

Types

type ClientOption

type ClientOption func(*clientConfig)

ClientOption is an interface for configuring a client. Using such options helpers is a quite common pattern in Go, as it allows for optional parameters in constructors. This concrete implementation here is inspired by how libraries such as axiom-go and connect do their configuration.

func WithAPIKey

func WithAPIKey(apiKey string) ClientOption

func WithConnectClientOptions

func WithConnectClientOptions(options ...connect.ClientOption) ClientOption

func WithHTTPClient

func WithHTTPClient(httpClient connect.HTTPClient) ClientOption

func WithURL

func WithURL(url string) ClientOption

type ContextKeyTaskExecutionType

type ContextKeyTaskExecutionType string

const ContextKeyTaskExecution ContextKeyTaskExecutionType = "x-tilebox-task-execution-object"

type ExecutableTask

type ExecutableTask interface {
	Execute(ctx context.Context) error
}

ExecutableTask is the interface for a task that can be executed, and therefore be registered with a task runner.

type ExplicitlyIdentifiableTask

type ExplicitlyIdentifiableTask interface {
	Identifier() TaskIdentifier
}

ExplicitlyIdentifiableTask is the interface for a task that provides a user-defined task identifier. The identifier is used to uniquely identify the task and specify its version. If a task is not an ExplicitlyIdentifiableTask, the task runner will generate an identifier for it using reflection.

type JobService

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

func NewJobService

func NewJobService(client workflowsv1connect.JobServiceClient, options ...JobServiceOption) *JobService

func (*JobService) Submit

func (js *JobService) Submit(ctx context.Context, jobName, clusterSlug string, maxRetries int, tasks ...Task) (*workflowsv1.Job, error)

type JobServiceOption

type JobServiceOption func(*jobServiceConfig)

func WithJobServiceTracerName

func WithJobServiceTracerName(tracerName string) JobServiceOption

func WithJobServiceTracerProvider

func WithJobServiceTracerProvider(tracerProvider trace.TracerProvider) JobServiceOption

type RecurrentTaskService

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

func NewRecurrentTaskService

func NewRecurrentTaskService(client workflowsv1connect.RecurrentTaskServiceClient) *RecurrentTaskService

func (*RecurrentTaskService) CreateStorageLocation

func (rs *RecurrentTaskService) CreateStorageLocation(ctx context.Context, location string, storageType workflowsv1.StorageType) (*workflowsv1.StorageLocation, error)

func (*RecurrentTaskService) DeleteStorageLocation

func (rs *RecurrentTaskService) DeleteStorageLocation(ctx context.Context, storageLocationID *workflowsv1.UUID) error

func (*RecurrentTaskService) GetStorageLocation

func (rs *RecurrentTaskService) GetStorageLocation(ctx context.Context, storageLocationID *workflowsv1.UUID) (*workflowsv1.StorageLocation, error)

func (*RecurrentTaskService) ListStorageLocations

func (rs *RecurrentTaskService) ListStorageLocations(ctx context.Context) ([]*workflowsv1.StorageLocation, error)

type Task

type Task interface {
}

Task is the interface for a task that can be submitted to the workflow service. It doesn't need to be identifiable or executable, but it can be both.

type TaskIdentifier

type TaskIdentifier interface {
	Name() string
	Version() string
	Display() string
}

func NewTaskIdentifier

func NewTaskIdentifier(name, version string) TaskIdentifier

type TaskRunner

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

func NewTaskRunner

func NewTaskRunner(client workflowsv1connect.TaskServiceClient, options ...TaskRunnerOption) (*TaskRunner, error)

func (*TaskRunner) GetRegisteredTask

func (t *TaskRunner) GetRegisteredTask(identifier TaskIdentifier) (ExecutableTask, bool)

func (*TaskRunner) RegisterTask

func (t *TaskRunner) RegisterTask(task ExecutableTask) error

func (*TaskRunner) RegisterTasks

func (t *TaskRunner) RegisterTasks(tasks ...ExecutableTask) error

func (*TaskRunner) Run

func (t *TaskRunner) Run(ctx context.Context)

Run runs the task runner forever, looking for new tasks to run and polling for new tasks when idle.

type TaskRunnerOption

type TaskRunnerOption func(*taskRunnerConfig)

func WithCluster

func WithCluster(clusterSlug string) TaskRunnerOption

func WithLogExecutionTimes

func WithLogExecutionTimes(logExecutionTimes bool) TaskRunnerOption

func WithRunnerLogger

func WithRunnerLogger(logger *slog.Logger) TaskRunnerOption

func WithRunnerTracerName

func WithRunnerTracerName(tracerName string) TaskRunnerOption

func WithRunnerTracerProvider

func WithRunnerTracerProvider(tracerProvider trace.TracerProvider) TaskRunnerOption