coordinator

Documentation

Index

• type Coordinator
  • func New(cfg *CoordinatorConfig) *Coordinator
  • func (c *Coordinator) CancelRun(id ulid.ULID, namespace string) error
  • func (c *Coordinator) CreateTrigger(trigger *state.Trigger) error
  • func (c *Coordinator) Getlogs(namespace, runID, stepID, logType string) ([]string, error)
  • func (c *Coordinator) HandleWebhook(w http.ResponseWriter, r *http.Request, trigger *state.Trigger)
  • func (c *Coordinator) RunFlow(id, namespace, trigger string, vars map[string]any) (ulid.ULID, error)
  • func (c *Coordinator) Start() error
  • func (c *Coordinator) Stop()
  • func (c *Coordinator) StreamLogs(namespace, runID, stepID, logType string) *LogStream
  • func (c *Coordinator) TriggerDelete(trigger *state.Trigger) error
  • func (c *Coordinator) WriteLogsBatch(namespace, runID, stepID, logType string, lines []string) error
• type CoordinatorConfig
• type LogStream
  • func NewLogStream(path string) *LogStream
  • func (s *LogStream) ErrorCh() <-chan error
  • func (s *LogStream) Run(ctx context.Context)
  • func (s *LogStream) StreamCh() <-chan string

Types

type Coordinator

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

func New

func New(cfg *CoordinatorConfig) *Coordinator

func (*Coordinator) CancelRun

func (c *Coordinator) CancelRun(id ulid.ULID, namespace string) error

func (*Coordinator) CreateTrigger

func (c *Coordinator) CreateTrigger(trigger *state.Trigger) error

func (*Coordinator) Getlogs

func (c *Coordinator) Getlogs(namespace, runID, stepID, logType string) ([]string, error)

func (*Coordinator) HandleWebhook

func (c *Coordinator) HandleWebhook(w http.ResponseWriter, r *http.Request, trigger *state.Trigger)

func (*Coordinator) RunFlow

func (c *Coordinator) RunFlow(
	id, namespace, trigger string,
	vars map[string]any,
) (ulid.ULID, error)

func (*Coordinator) Start

func (c *Coordinator) Start() error

Start starts the coordinator and its trigger coordinator

func (*Coordinator) Stop

func (c *Coordinator) Stop()

Stop gracefully stops the coordinator

func (*Coordinator) StreamLogs

func (c *Coordinator) StreamLogs(namespace, runID, stepID, logType string) *LogStream

func (*Coordinator) TriggerDelete

func (c *Coordinator) TriggerDelete(trigger *state.Trigger) error

func (*Coordinator) WriteLogsBatch

func (c *Coordinator) WriteLogsBatch(namespace, runID, stepID, logType string, lines []string) error

type CoordinatorConfig

type CoordinatorConfig struct {
	Logger      *zap.Logger
	NomadClient *api.Client
	State       serverstate.State
	DataDir     string
	RPCAddr     string
}

type LogStream

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

func NewLogStream

func NewLogStream(path string) *LogStream

func (*LogStream) ErrorCh

func (s *LogStream) ErrorCh() <-chan error

func (*LogStream) Run

func (s *LogStream) Run(ctx context.Context)

func (*LogStream) StreamCh

func (s *LogStream) StreamCh() <-chan string