direktiv

Documentation

Index

• Constants
• Variables
• func InstanceMemory(rec *ent.WorkflowInstance) ([]byte, error)
• func NewInternalError(err error) error
• func SyncSubscribeTo(dbConnString string, topic int, fn func(interface{})) error
• func WrapCatchableError(msg string, err error) error
• type CatchableError
  • func NewCatchableError(code string, msg string, a ...interface{}) *CatchableError
  • func (err *CatchableError) Error() string
• type CmdErrorResponse
• type Config
  • func ReadConfig(file string) (*Config, error)
• type ErrorType
  • func GetErrorType(err error) ErrorType
• type InternalError
  • func NewInternalErrorWithDepth(err error, depth int) *InternalError
  • func (err *InternalError) Error() string
  • func (err *InternalError) Unwrap() error
• type ServiceResponse
• type SyncRequest
• type UncatchableError
  • func NewUncatchableError(code, msg string, a ...interface{}) *UncatchableError
  • func (err *UncatchableError) Error() string
• type WorkflowServer
  • func NewWorkflowServer(config *Config) (*WorkflowServer, error)
  • func (s *WorkflowServer) GetConfig() Config
  • func (s *WorkflowServer) Kill()
  • func (s *WorkflowServer) Lifeline() chan bool
  • func (s *WorkflowServer) Run() error
  • func (s *WorkflowServer) SetVariableStorage(vs varstore.VarStorage)
  • func (s *WorkflowServer) Stop()

Constants

const (
	DirektivActionIDHeader    = "Direktiv-ActionID"
	DirektivInstanceIDHeader  = "Direktiv-InstanceID"
	DirektivExchangeKeyHeader = "Direktiv-ExchangeKey"
	DirektivPingAddrHeader    = "Direktiv-PingAddr"
	DirektivDeadlineHeader    = "Direktiv-Deadline"
	DirektivTimeoutHeader     = "Direktiv-Timeout"
	DirektivStepHeader        = "Direktiv-Step"
	DirektivResponseHeader    = "Direktiv-Response"
	DirektivNamespaceHeader   = "Direktiv-Namespace"
	DirektivSourceHeader      = "Direktiv-Source"
	DirektivFileHeader        = "Direktiv-Files"

	DirektivErrorCodeHeader    = "Direktiv-ErrorCode"
	DirektivErrorMessageHeader = "Direktiv-ErrorMessage"
)

headers for flow->container communication

const (
	ServiceResponseNoError = ""
	ServiceErrorInternal   = "au.com.direktiv.error.internal"
	ServiceErrorImage      = "au.com.direktiv.error.image"
	ServiceErrorNetwork    = "au.com.direktiv.error.network"
	ServiceErrorIO         = "au.com.direktiv.error.io"
)

internal error codes for knative services

const (
	CancelFunction = iota
	CancelSubflow
	CancelTimer
	CancelInstanceTimers
	AddCron
	UpdateEventDelays
)

direktiv pub/sub items

const ApiSync = "apisync"

const FlowSync = "flowsync"

FlowSync is the name of postgres pubsub channel

const (
	// WorkflowStateSubscription is the channel that runs workflow states.
	WorkflowStateSubscription = "workflow-state"
)

Variables

var (
	ErrCodeJQBadQuery        = "direktiv.jq.badCommand"
	ErrCodeJQNotObject       = "direktiv.jq.notObject"
	ErrCodeMultipleErrors    = "direktiv.workflow.multipleErrors"
	ErrCodeAllBranchesFailed = "direktiv.parallel.allFailed"
)

internal errors

Functions

func InstanceMemory

func InstanceMemory(rec *ent.WorkflowInstance) ([]byte, error)

func NewInternalError

func NewInternalError(err error) error

func SyncSubscribeTo

func SyncSubscribeTo(dbConnString string, topic int,
	fn func(interface{})) error

SyncSubscribeTo subscribes to direktiv interna postgres pub/sub

func WrapCatchableError

func WrapCatchableError(msg string, err error) error

Types

type CatchableError

type CatchableError struct {
	Code    string `json:"code"`
	Message string `json:"msg"`
}

func NewCatchableError

func NewCatchableError(code string, msg string, a ...interface{}) *CatchableError

func (*CatchableError) Error

func (err *CatchableError) Error() string

type CmdErrorResponse

type CmdErrorResponse struct {
	Error string    `json:"error"`
	Type  ErrorType `json:"type"`
}

CmdErrorResponse struct for responding when command has an error

type Config

type Config struct {
	FunctionsProtocol string `yaml:"functions-protocol"`

	Database struct {
		DB string
	}

	InstanceLogging struct {
		Driver string
	}

	VariablesStorage struct {
		Driver string
	}
}

Config is the configuration for workflow and runner server

func ReadConfig

func ReadConfig(file string) (*Config, error)

ReadConfig reads the configuration file and overwrites with environment variables if set

type ErrorType

type ErrorType int

ErrorType types of errors direktiv commands can return

const (
	DirektivError ErrorType = iota
	// Ent Errors
	ValidationError
	NotFoundError
	NotSingularError
	NotLoadedError
	ConstraintError
)

func GetErrorType

func GetErrorType(err error) ErrorType

GetErrorType get Error Type from passed error

type InternalError

type InternalError struct {
	Err      error
	Function string
	File     string
	Line     int
}

func NewInternalErrorWithDepth

func NewInternalErrorWithDepth(err error, depth int) *InternalError

func (*InternalError) Error

func (err *InternalError) Error() string

func (*InternalError) Unwrap

func (err *InternalError) Unwrap() error

type ServiceResponse

type ServiceResponse struct {
	ErrorCode    string      `json:"errorCode"`
	ErrorMessage string      `json:"errorMessage"`
	Data         interface{} `json:"data"`
}

ServiceResponse is the response structure for internal knative services

type SyncRequest

type SyncRequest struct {
	Cmd    int
	Sender uuid.UUID
	ID     interface{}
}

SyncRequest sync maintenance requests between instances subscribed to FlowSync

type UncatchableError

type UncatchableError struct {
	Code    string
	Message string
}

func NewUncatchableError

func NewUncatchableError(code, msg string, a ...interface{}) *UncatchableError

func (*UncatchableError) Error

func (err *UncatchableError) Error() string

type WorkflowServer

type WorkflowServer struct {
	LifeLine chan bool
	// contains filtered or unexported fields
}

WorkflowServer is a direktiv server

func NewWorkflowServer

func NewWorkflowServer(config *Config) (*WorkflowServer, error)

NewWorkflowServer creates a new workflow server

func (*WorkflowServer) GetConfig

func (s *WorkflowServer) GetConfig() Config

func (*WorkflowServer) Kill

func (s *WorkflowServer) Kill()

Kill kills the server

func (*WorkflowServer) Lifeline

func (s *WorkflowServer) Lifeline() chan bool

Lifeline interface impl

func (*WorkflowServer) Run

func (s *WorkflowServer) Run() error

Run starts all components of direktiv

func (*WorkflowServer) SetVariableStorage

func (s *WorkflowServer) SetVariableStorage(vs varstore.VarStorage)

func (*WorkflowServer) Stop

func (s *WorkflowServer) Stop()

Stop stops the server gracefully