task

package
v0.10.0-rc.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 24, 2026 License: PostgreSQL Imports: 13 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrTaskCanceled = errors.New("task was canceled")
	ErrTaskFailed   = errors.New("task failed")
)
View Source 
var ErrTaskNotFound = errors.New("task not found")

Functions

func Provide 

func Provide(i *do.Injector)

Types

type LogEntry 

type LogEntry struct {
	Timestamp time.Time
	Message   string
	Fields    map[string]any
}

type Options 

type Options struct {
	Scope               Scope     `json:"scope"`
	ParentID            uuid.UUID `json:"parent_id"`
	DatabaseID          string    `json:"database_id"`
	NodeName            string    `json:"node_name"`
	InstanceID          string    `json:"instance_id"`
	HostID              string    `json:"host_id"`
	Type                Type      `json:"type"`
	WorkflowInstanceID  string    `json:"workflow_id"`
	WorkflowExecutionID string    `json:"workflow_execution_id"`
}

func (Options) EntityID added in v0.7.0 

func (o Options) EntityID() string

type Scope added in v0.7.0 

type Scope string
const (
	ScopeDatabase Scope = "database"
	ScopeHost     Scope = "host"
)

func (Scope) String added in v0.7.0 

func (s Scope) String() string

type Service 

type Service struct {
	Store *Store
	// contains filtered or unexported fields
}

func NewService 

func NewService(store *Store) *Service

func (*Service) AddLogEntry 

func (s *Service) AddLogEntry(ctx context.Context, scope Scope, entityID string, taskID uuid.UUID, entry LogEntry) error

func (*Service) CreateTask 

func (s *Service) CreateTask(ctx context.Context, opts Options) (*Task, error)

func (*Service) DeleteAllTaskLogs 

func (s *Service) DeleteAllTaskLogs(ctx context.Context, scope Scope, entityID string) error

func (*Service) DeleteAllTasks 

func (s *Service) DeleteAllTasks(ctx context.Context, scope Scope, entityID string) error

func (*Service) DeleteTask 

func (s *Service) DeleteTask(ctx context.Context, scope Scope, entityID string, taskID uuid.UUID) error

func (*Service) DeleteTaskLogs 

func (s *Service) DeleteTaskLogs(ctx context.Context, scope Scope, entityID string, taskID uuid.UUID) error

func (*Service) GetTask 

func (s *Service) GetTask(ctx context.Context, scope Scope, entityID string, taskID uuid.UUID) (*Task, error)

func (*Service) GetTaskLog 

func (s *Service) GetTaskLog(ctx context.Context, scope Scope, entityID string, taskID uuid.UUID, options TaskLogOptions) (*TaskLog, error)

func (*Service) GetTasks 

func (s *Service) GetTasks(ctx context.Context, scope Scope, entityID string, options TaskListOptions) ([]*Task, error)

func (*Service) NewWatcher 

func (s *Service) NewWatcher(_ context.Context, scope Scope, entityID string, taskID uuid.UUID) (*Watcher, error)

NewWatcher creates a Watcher that monitors the given task and closes its Done channel when the task reaches a terminal state or is deleted. Multiple callers watching the same task share a single etcd watch stream. The caller must call Close on the returned Watcher when done with it.

func (*Service) UpdateTask 

func (s *Service) UpdateTask(ctx context.Context, task *Task) error

type SortOrder 

type SortOrder string
const (
	SortAscend  SortOrder = "ascend"
	SortDescend SortOrder = "descend"
)

func (SortOrder) String 

func (s SortOrder) String() string

type Status 

type Status string
const (
	StatusPending   Status = "pending"
	StatusRunning   Status = "running"
	StatusCompleted Status = "completed"
	StatusFailed    Status = "failed"
	StatusCanceled  Status = "canceled"
	StatusCanceling Status = "canceling"
	StatusUnknown   Status = "unknown"
)

func (Status) String 

func (s Status) String() string

type Store 

type Store struct {
	Task           *TaskStore
	TaskLogMessage *TaskLogEntryStore
	// contains filtered or unexported fields
}

func NewStore 

func NewStore(client *clientv3.Client, root string) *Store

func (*Store) Txn 

func (s *Store) Txn(ops ...storage.TxnOperation) storage.Txn

type StoredTask 

type StoredTask struct {
	storage.StoredValue
	Task *Task `json:"task"`
}

type StoredTaskLogEntry 

type StoredTaskLogEntry struct {
	storage.StoredValue
	Scope      Scope          `json:"scope"`
	EntityID   string         `json:"entity_id"`
	DatabaseID string         `json:"database_id"`
	TaskID     uuid.UUID      `json:"task_id"`
	EntryID    uuid.UUID      `json:"entry_id"`
	Timestamp  time.Time      `json:"timestamp"`
	Message    string         `json:"message"`
	Fields     map[string]any `json:"fields"`
}

type Task 

type Task struct {
	Scope               Scope     `json:"scope"`
	EntityID            string    `json:"entity_id"`
	ParentID            uuid.UUID `json:"parent_id"`
	DatabaseID          string    `json:"database_id"`
	NodeName            string    `json:"node_name"`
	InstanceID          string    `json:"instance_id"`
	HostID              string    `json:"host_id"`
	TaskID              uuid.UUID `json:"task_id"`
	CreatedAt           time.Time `json:"created_at"`
	CompletedAt         time.Time `json:"completed_at"`
	Type                Type      `json:"type"`
	WorkflowInstanceID  string    `json:"workflow_id"`
	WorkflowExecutionID string    `json:"workflow_execution_id"`
	Status              Status    `json:"status"`
	Error               string    `json:"error"`
}

func NewTask 

func NewTask(opts Options) (*Task, error)

func (*Task) IsComplete 

func (t *Task) IsComplete() bool

func (*Task) SetCompleted 

func (t *Task) SetCompleted()

func (*Task) SetFailed 

func (t *Task) SetFailed(cause error)

func (*Task) Start 

func (t *Task) Start()

func (*Task) Update 

func (t *Task) Update(options UpdateOptions)

type TaskListOptions 

type TaskListOptions struct {
	Limit       int
	AfterTaskID uuid.UUID
	SortOrder   SortOrder

	// Optional filters (applied client-side by helper methods)
	Type     Type
	NodeName string
	Statuses []Status
}

type TaskLog 

type TaskLog struct {
	Scope       Scope      `json:"scope"`
	EntityID    string     `json:"entity_id"`
	DatabaseID  string     `json:"database_id"`
	TaskID      uuid.UUID  `json:"id"`
	LastEntryID uuid.UUID  `json:"last_entry_id"`
	Entries     []LogEntry `json:"entries"`
}

type TaskLogEntryStore 

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

func NewTaskLogEntryStore 

func NewTaskLogEntryStore(client *clientv3.Client, root string) *TaskLogEntryStore

func (*TaskLogEntryStore) DatabasePrefix 

func (s *TaskLogEntryStore) DatabasePrefix(databaseID string) string

DatabasePrefix returns the prefix for all task log entries for a given database. Deprecated: Use EntityPrefix(ScopeDatabase, databaseID) instead.

func (*TaskLogEntryStore) DeleteByDatabaseID 

func (s *TaskLogEntryStore) DeleteByDatabaseID(databaseID string) storage.DeleteOp

DeleteByDatabaseID deletes all log entries for a database. Deprecated: Use DeleteByEntity(ScopeDatabase, databaseID) instead.

func (*TaskLogEntryStore) DeleteByEntity added in v0.7.0 

func (s *TaskLogEntryStore) DeleteByEntity(scope Scope, entityID string) storage.DeleteOp

func (*TaskLogEntryStore) DeleteByTask added in v0.7.0 

func (s *TaskLogEntryStore) DeleteByTask(scope Scope, entityID string, taskID uuid.UUID) storage.DeleteOp

func (*TaskLogEntryStore) DeleteByTaskID 

func (s *TaskLogEntryStore) DeleteByTaskID(databaseID string, taskID uuid.UUID) storage.DeleteOp

DeleteByTaskID deletes all log entries for a task. Deprecated: Use DeleteByTask(ScopeDatabase, databaseID, taskID) instead.

func (*TaskLogEntryStore) EntityPrefix added in v0.7.0 

func (s *TaskLogEntryStore) EntityPrefix(scope Scope, entityID string) string

EntityPrefix returns the prefix for all task log entries for a given scope and entity.

func (*TaskLogEntryStore) GetAllByTask added in v0.7.0 

func (s *TaskLogEntryStore) GetAllByTask(scope Scope, entityID string, taskID uuid.UUID, options TaskLogOptions) storage.GetMultipleOp[*StoredTaskLogEntry]

func (*TaskLogEntryStore) GetAllByTaskID 

func (s *TaskLogEntryStore) GetAllByTaskID(databaseID string, taskID uuid.UUID, options TaskLogOptions) storage.GetMultipleOp[*StoredTaskLogEntry]

GetAllByTaskID retrieves all log entries for a task. Deprecated: Use GetAllByTask(ScopeDatabase, databaseID, taskID, options) instead.

func (*TaskLogEntryStore) Key 

func (s *TaskLogEntryStore) Key(scope Scope, entityID string, taskID, entryID uuid.UUID) string

Key returns the storage key for a task log entry.

func (*TaskLogEntryStore) KeyDeprecated added in v0.7.0 

func (s *TaskLogEntryStore) KeyDeprecated(databaseID string, taskID, entryID uuid.UUID) string

KeyDeprecated returns the storage key for a task log entry. Deprecated: Use Key(ScopeDatabase, databaseID, taskID, entryID) instead.

func (*TaskLogEntryStore) Prefix 

func (s *TaskLogEntryStore) Prefix() string

func (*TaskLogEntryStore) Put 

func (s *TaskLogEntryStore) Put(item *StoredTaskLogEntry) storage.PutOp[*StoredTaskLogEntry]

func (*TaskLogEntryStore) TaskPrefix 

func (s *TaskLogEntryStore) TaskPrefix(scope Scope, entityID string, taskID uuid.UUID) string

TaskPrefix returns the prefix for all log entries for a specific task.

func (*TaskLogEntryStore) TaskPrefixDeprecated added in v0.7.0 

func (s *TaskLogEntryStore) TaskPrefixDeprecated(databaseID string, taskID uuid.UUID) string

TaskPrefixDeprecated returns the prefix for all log entries for a specific task. Deprecated: Use TaskPrefix(ScopeDatabase, databaseID, taskID) instead.

type TaskLogOptions 

type TaskLogOptions struct {
	Limit        int
	AfterEntryID uuid.UUID
}

type TaskLogWriter 

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

func NewTaskLogWriter 

func NewTaskLogWriter(ctx context.Context, service *Service, scope Scope, entityID string, taskID uuid.UUID) *TaskLogWriter

func (*TaskLogWriter) Close 

func (w *TaskLogWriter) Close() error

func (*TaskLogWriter) Write 

func (w *TaskLogWriter) Write(p []byte) (n int, err error)

type TaskStore 

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

func NewTaskStore 

func NewTaskStore(client *clientv3.Client, root string) *TaskStore

func (*TaskStore) Create 

func (s *TaskStore) Create(item *StoredTask) storage.PutOp[*StoredTask]

func (*TaskStore) Delete 

func (s *TaskStore) Delete(scope Scope, entityID string, taskID uuid.UUID) storage.DeleteOp

func (*TaskStore) DeleteByEntity added in v0.7.0 

func (s *TaskStore) DeleteByEntity(scope Scope, entityID string) storage.DeleteOp

func (*TaskStore) EntityPrefix added in v0.7.0 

func (s *TaskStore) EntityPrefix(scope Scope, entityID string) string

func (*TaskStore) GetAllByEntity added in v0.7.0 

func (s *TaskStore) GetAllByEntity(scope Scope, entityID string, options TaskListOptions) storage.GetMultipleOp[*StoredTask]

func (*TaskStore) GetByKey 

func (s *TaskStore) GetByKey(scope Scope, entityID string, taskID uuid.UUID) storage.GetOp[*StoredTask]

func (*TaskStore) Key 

func (s *TaskStore) Key(scope Scope, entityID string, taskID uuid.UUID) string

func (*TaskStore) Prefix 

func (s *TaskStore) Prefix() string

func (*TaskStore) Update 

func (s *TaskStore) Update(item *StoredTask) storage.PutOp[*StoredTask]

func (*TaskStore) Watch 

func (s *TaskStore) Watch(scope Scope, entityID string, taskID uuid.UUID) storage.WatchOp[*StoredTask]

type Type 

type Type string
const (
	TypeCreate          Type = "create"
	TypeUpdate          Type = "update"
	TypeDelete          Type = "delete"
	TypeNodeBackup      Type = "node_backup"
	TypeRestore         Type = "restore"
	TypeNodeRestore     Type = "node_restore"
	TypeRestartInstance Type = "restart_instance"
	TypeStopInstance    Type = "stop_instance"
	TypeStartInstance   Type = "start_instance"
	TypeSwitchover      Type = "switchover"
	TypeFailover        Type = "failover"
	TypeRemoveHost      Type = "remove_host"
	TypeUpgrade         Type = "upgrade"
)

func (Type) String 

func (t Type) String() string

type UpdateOptions 

type UpdateOptions struct {
	NodeName            *string    `json:"node_name,omitempty"`
	InstanceID          *string    `json:"instance_id,omitempty"`
	HostID              *string    `json:"host_id,omitempty"`
	WorkflowInstanceID  *string    `json:"workflow_instance_id,omitempty"`
	WorkflowExecutionID *string    `json:"workflow_execution_id,omitempty"`
	CompletedAt         *time.Time `json:"completed_at,omitempty"`
	Status              *Status    `json:"status,omitempty"`
	Error               *string    `json:"error,omitempty"`
}

func UpdateCancel 

func UpdateCancel() UpdateOptions

func UpdateComplete 

func UpdateComplete() UpdateOptions

func UpdateFail 

func UpdateFail(cause error) UpdateOptions

func UpdateStart 

func UpdateStart() UpdateOptions

type Watcher 

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

Watcher is a subscription to a task's terminal state. Multiple Watchers for the same task share a single underlying etcd watch stream.

func (*Watcher) Close 

func (w *Watcher) Close()

Close releases this subscription. When the last subscription for a task is closed, the underlying etcd watch stream is stopped.

func (*Watcher) Done 

func (w *Watcher) Done() <-chan struct{}

Done returns a channel that is closed when the task reaches a terminal state or is deleted.

func (*Watcher) Err 

func (w *Watcher) Err() error

Err returns nil if the task completed successfully, ErrTaskCanceled if it was canceled (or is canceling), or ErrTaskFailed if it failed. It is only meaningful after Done() is closed.

func (*Watcher) Error 

func (w *Watcher) Error() <-chan error

Error returns a channel that receives an error if the underlying watch stream fails. The channel carries at most one value. Callers that select on Done should also select on Error so they are not blocked when the watch stream dies before the task reaches a terminal state.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.