pingregistrywf

Documentation

Overview

Package `pingregistrywf` implements the ping-registry ephemeral workflow, which gathers pings from the daemons that watch a registry. The workflow illustrates how to implement a workflow that requires coordination between multiple servers and an admin.

Workflow Events

An admin uses `nogfsoctl ping-registry begin` to initialized the workflow with `WorkflowEvent_EV_FSO_PING_REGISTRY_STARTED` on the workflow and a corresponding `WorkflowEvent_EV_FSO_PING_REGISTRY_STARTED` on the ephemeral workflow index.

The Nogfsostad servers that watch the registry observe the workflow, and each of them posts a specified number of `WorkflowEvent_EV_FSO_SERVER_PINGED`.

Nogfsoregd observes the workflow and posts a specified number of `WorkflowEvent_EV_FSO_SERVER_PINGED`. It then waits until the workflow deadline and summarizes the pings in a `WorkflowEvent_EV_FSO_SERVER_PINGS_GATHERED`.

An admin uses `nogfsoctl ping-registry commit` to complete the workflow with `WorkflowEvent_EV_FSO_PING_REGISTRY_COMPLETED`, followed by `WorkflowEvent_EV_FSO_PING_REGISTRY_COMPLETED` on the workflow index, followed by a final `WorkflowEvent_EV_FSO_PING_REGISTRY_COMMITTED` on the workflow.

The final workflow event has no observable side effect. Its only purpose is to explicitly confirm termination of the workflow history. The final event may be missing if a multi-step command to complete the workflow gets interrupted.

Workflows are eventually deleted from the index with `WorkflowEvent_EV_FSO_PING_REGISTRY_DELETED` on the index. Workflows may be deleted with or without the final `WorkflowEvent_EV_FSO_PING_REGISTRY_COMMITTED` on the workflow.

Index

• Variables
• func IsPackageError(err error) bool
• type Advancer
  • func (a *Advancer) Advance(s events.State, ev events.Event) events.State
• type ArgumentNotIdempotentError
  • func (err *ArgumentNotIdempotentError) Error() string
• type Behavior
  • func (Behavior) NewAdvancer() events.Advancer
  • func (Behavior) NewEvent() events.Event
  • func (Behavior) NewState(id uuid.I) events.State
  • func (Behavior) Tell(s events.State, c events.Command) ([]events.Event, error)
• type CmdAbortExpired
  • func (*CmdAbortExpired) AggregateCommand()
• type CmdAppendPing
  • func (*CmdAppendPing) AggregateCommand()
• type CmdCommit
  • func (*CmdCommit) AggregateCommand()
• type CmdDelete
  • func (*CmdDelete) AggregateCommand()
• type CmdEnd
  • func (*CmdEnd) AggregateCommand()
• type CmdInit
  • func (*CmdInit) AggregateCommand()
• type CmdPostSummary
  • func (*CmdPostSummary) AggregateCommand()
• type Event
  • func (ev *Event) UnmarshalProto(data []byte) error
• type EventTypeError
  • func (err *EventTypeError) Error() string
• type InvalidCommandError
  • func (err *InvalidCommandError) Error() string
• type JournalError
  • func (err *JournalError) Error() string
  • func (err *JournalError) Unwrap() error
• type NewEventsError
  • func (err *NewEventsError) Error() string
  • func (err *NewEventsError) Unwrap() error
• type State
  • func (*State) AggregateState()
  • func (s *State) Id() uuid.I
  • func (st *State) NumPings() int
  • func (st *State) Pings() []Status
  • func (st *State) RegistryId() uuid.I
  • func (s *State) SetVid(vid ulid.I)
  • func (st *State) StateCode() StateCode
  • func (st *State) SummaryStatus() Status
  • func (s *State) Vid() ulid.I
• type StateCode
• type StateConflictError
  • func (err *StateConflictError) Error() string
• type Status
• type UninitializedError
  • func (err *UninitializedError) Error() string
• type Workflows
  • func New(journal *events.Journal) *Workflows
  • func (r *Workflows) AbortExpired(id uuid.I, vid ulid.I) (ulid.I, error)
  • func (r *Workflows) AppendPing(id uuid.I, vid ulid.I, code int32, message string) (ulid.I, error)
  • func (r *Workflows) Commit(id uuid.I, vid ulid.I) (ulid.I, error)
  • func (w *Workflows) Delete(id uuid.I, vid ulid.I) error
  • func (r *Workflows) End(id uuid.I, vid ulid.I) (ulid.I, error)
  • func (r *Workflows) FindId(id uuid.I) (*State, error)
  • func (r *Workflows) Init(id uuid.I, cmd *CmdInit) (ulid.I, error)
  • func (r *Workflows) PostSummary(id uuid.I, vid ulid.I, code int32, message string) (ulid.I, error)

Variables

var NoVC = events.NoVC

var RetryNoVC = events.RetryNoVC

Functions

func IsPackageError

func IsPackageError(err error) bool

Types

type Advancer

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

The bools indicate which part of the state has been duplicated.

func (*Advancer) Advance

func (a *Advancer) Advance(s events.State, ev events.Event) events.State

type ArgumentNotIdempotentError

type ArgumentNotIdempotentError struct {
	Arg string
}

func (*ArgumentNotIdempotentError) Error

func (err *ArgumentNotIdempotentError) Error() string

type Behavior

type Behavior struct{}

func (Behavior) NewAdvancer

func (Behavior) NewAdvancer() events.Advancer

func (Behavior) NewEvent

func (Behavior) NewEvent() events.Event

func (Behavior) NewState

func (Behavior) NewState(id uuid.I) events.State

func (Behavior) Tell

func (Behavior) Tell(
	s events.State, c events.Command,
) ([]events.Event, error)

type CmdAbortExpired

type CmdAbortExpired struct{}

func (*CmdAbortExpired) AggregateCommand

func (*CmdAbortExpired) AggregateCommand()

type CmdAppendPing

type CmdAppendPing struct {
	Code    int32
	Message string
}

func (*CmdAppendPing) AggregateCommand

func (*CmdAppendPing) AggregateCommand()

type CmdCommit

type CmdCommit struct{}

func (*CmdCommit) AggregateCommand

func (*CmdCommit) AggregateCommand()

type CmdDelete

type CmdDelete struct{}

func (*CmdDelete) AggregateCommand

func (*CmdDelete) AggregateCommand()

type CmdEnd

type CmdEnd struct{}

func (*CmdEnd) AggregateCommand

func (*CmdEnd) AggregateCommand()

type CmdInit

type CmdInit struct {
	RegistryId uuid.I
}

func (*CmdInit) AggregateCommand

func (*CmdInit) AggregateCommand()

type CmdPostSummary

type CmdPostSummary struct {
	Code    int32
	Message string
}

func (*CmdPostSummary) AggregateCommand

func (*CmdPostSummary) AggregateCommand()

type Event

type Event struct{ wfev.Event }

func (*Event) UnmarshalProto

func (ev *Event) UnmarshalProto(data []byte) error

type EventTypeError

type EventTypeError struct{}

func (*EventTypeError) Error

func (err *EventTypeError) Error() string

type InvalidCommandError

type InvalidCommandError struct{}

func (*InvalidCommandError) Error

func (err *InvalidCommandError) Error() string

type JournalError

type JournalError struct {
	Err error
}

func (*JournalError) Error

func (err *JournalError) Error() string

func (*JournalError) Unwrap

func (err *JournalError) Unwrap() error

type NewEventsError

type NewEventsError struct {
	Err error
}

func (*NewEventsError) Error

func (err *NewEventsError) Error() string

func (*NewEventsError) Unwrap

func (err *NewEventsError) Unwrap() error

type State

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

func (*State) AggregateState

func (*State) AggregateState()

func (*State) Id

func (s *State) Id() uuid.I

func (*State) NumPings

func (st *State) NumPings() int

func (*State) Pings

func (st *State) Pings() []Status

func (*State) RegistryId

func (st *State) RegistryId() uuid.I

func (*State) SetVid

func (s *State) SetVid(vid ulid.I)

func (*State) StateCode

func (st *State) StateCode() StateCode

func (*State) SummaryStatus

func (st *State) SummaryStatus() Status

func (*State) Vid

func (s *State) Vid() ulid.I

type StateCode

type StateCode int

const (
	StateUninitialized StateCode = iota
	StateInitialized
	StateAppending
	StateSummarized
	StateCompleted
	StateTerminated
)

type StateConflictError

type StateConflictError struct{}

func (*StateConflictError) Error

func (err *StateConflictError) Error() string

type Status

type Status struct {
	Code    int32
	Message string
	EventId ulid.I
}

type UninitializedError

type UninitializedError struct{}

func (*UninitializedError) Error

func (err *UninitializedError) Error() string

type Workflows

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

func New

func New(journal *events.Journal) *Workflows

func (*Workflows) AbortExpired

func (r *Workflows) AbortExpired(id uuid.I, vid ulid.I) (ulid.I, error)

func (*Workflows) AppendPing

func (r *Workflows) AppendPing(
	id uuid.I, vid ulid.I, code int32, message string,
) (ulid.I, error)

func (*Workflows) Commit

func (r *Workflows) Commit(id uuid.I, vid ulid.I) (ulid.I, error)

func (*Workflows) Delete

func (w *Workflows) Delete(id uuid.I, vid ulid.I) error

func (*Workflows) End

func (r *Workflows) End(id uuid.I, vid ulid.I) (ulid.I, error)

func (*Workflows) FindId

func (r *Workflows) FindId(id uuid.I) (*State, error)

func (*Workflows) Init

func (r *Workflows) Init(id uuid.I, cmd *CmdInit) (ulid.I, error)

func (*Workflows) PostSummary

func (r *Workflows) PostSummary(
	id uuid.I, vid ulid.I, code int32, message string,
) (ulid.I, error)