flow

Documentation

Index

• Constants
• Variables
• func GetInodePath(path string) string
• func IsNotFound(err error) bool
• func NewInternalError(err error) error
• func PublishKnativeEvent(ce *cloudevents.Event)
• func Run(ctx context.Context, logger *zap.SugaredLogger, conf *util.Config) error
• func SyncSubscribeTo(log *zap.Logger, addr, topic string, fn func(interface{})) error
• func WrapCatchableError(msg string, err error) error
• type Carrier
  • func (c *Carrier) Get(key string) string
  • func (c *Carrier) Keys() []string
  • func (c *Carrier) Set(key, val string)
• type CatchableError
  • func NewCatchableError(code string, msg string, a ...interface{}) *CatchableError
  • func (err *CatchableError) Error() string
• type Config
• type ConfigBroadcast
• type EventingCtxKey
• type InternalError
  • func NewInternalErrorWithDepth(err error, depth int) *InternalError
  • func (err *InternalError) Error() string
  • func (err *InternalError) Unwrap() error
• type NotFoundError
  • func (err *NotFoundError) Error() string
• type PubsubUpdate
• type ServiceResponse
• type UncatchableError
  • func NewUncatchableError(code, msg string, a ...interface{}) *UncatchableError
  • func (err *UncatchableError) Error() string

Constants

const (
	BroadcastEventTypeCreate string = "create"
	BroadcastEventTypeUpdate string = "update"
	BroadcastEventTypeDelete string = "delete"

	BroadcastEventTypeInstanceStarted string = "started"
	BroadcastEventTypeInstanceFailed  string = "failed"
	BroadcastEventTypeInstanceSuccess string = "success"
)

const (
	BroadcastEventScopeWorkflow  string = "workflow"
	BroadcastEventScopeNamespace string = "namespace"
	BroadcastEventScopeInstance  string = "instance"
)

const (
	BroadcastEventPrefixWorkflow  string = "workflow"
	BroadcastEventPrefixDirectory string = "directory"
	BroadcastEventPrefixVariable  string = "variable"
	BroadcastEventPrefixInstance  string = "instance"
)

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 (
	StatusPending  = "pending"
	StatusFailed   = "failed"
	StatusCrashed  = "crashed"
	StatusComplete = "complete"
)

const CancelActionMessage = "cancelAction"

Variables

var (
	ErrCodeInternal               = "direktiv.internal.error"
	ErrCodeWorkflowUnparsable     = "direktiv.workflow.unparsable"
	ErrCodeMultipleErrors         = "direktiv.workflow.multipleErrors"
	ErrCodeCancelledByParent      = "direktiv.cancels.parent"
	ErrCodeSoftTimeout            = "direktiv.cancels.timeout.soft"
	ErrCodeHardTimeout            = "direktiv.cancels.timeout.hard"
	ErrCodeJQBadQuery             = "direktiv.jq.badCommand"
	ErrCodeJQNotObject            = "direktiv.jq.notObject"
	ErrCodeAllBranchesFailed      = "direktiv.parallel.allFailed"
	ErrCodeFailedSchemaValidation = "direktiv.schema.failed"
)

var (
	ErrNotDir      = errors.New("not a directory")
	ErrNotWorkflow = errors.New("not a workflow")
)

Functions

func GetInodePath

func GetInodePath(path string) string

func IsNotFound

func IsNotFound(err error) bool

func NewInternalError

func NewInternalError(err error) error

func PublishKnativeEvent

func PublishKnativeEvent(ce *cloudevents.Event)

func Run

func Run(ctx context.Context, logger *zap.SugaredLogger, conf *util.Config) error

func SyncSubscribeTo

func SyncSubscribeTo(log *zap.Logger, addr, topic string, fn func(interface{})) error

SyncSubscribeTo subscribes to direktiv interna postgres pub/sub

func WrapCatchableError

func WrapCatchableError(msg string, err error) error

Types

type Carrier

type Carrier struct {
	Trace map[string]string
}

func (*Carrier) Get

func (c *Carrier) Get(key string) string

func (*Carrier) Keys

func (c *Carrier) Keys() []string

func (*Carrier) Set

func (c *Carrier) Set(key, val string)

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 Config

type Config struct {
	Broadcast *ConfigBroadcast `json:"broadcast"`
}

type ConfigBroadcast

type ConfigBroadcast struct {
	WorkflowCreate          *bool `json:"workflow.create"`
	WorkflowUpdate          *bool `json:"workflow.update"`
	WorkflowDelete          *bool `json:"workflow.delete"`
	DirectoryCreate         *bool `json:"directory.create"`
	DirectoryDelete         *bool `json:"directory.delete"`
	WorkflowVariableCreate  *bool `json:"workflow.variable.create"`
	WorkflowVariableUpdate  *bool `json:"workflow.variable.update"`
	WorkflowVariableDelete  *bool `json:"workflow.variable.delete"`
	NamespaceVariableCreate *bool `json:"namespace.variable.create"`
	NamespaceVariableUpdate *bool `json:"namespace.variable.update"`
	NamespaceVariableDelete *bool `json:"namespace.variable.delete"`
	InstanceVariableCreate  *bool `json:"instance.variable.create"`
	InstanceVariableUpdate  *bool `json:"instance.variable.update"`
	InstanceVariableDelete  *bool `json:"instance.variable.delete"`
	InstanceStarted         *bool `json:"instance.started"`
	InstanceSuccess         *bool `json:"instance.success"`
	InstanceFailed          *bool `json:"instance.failed"`
}

type EventingCtxKey

type EventingCtxKey string

const EventingCtxKeySource EventingCtxKey = "source"

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 NotFoundError

type NotFoundError struct {
	Label string
}

func (*NotFoundError) Error

func (err *NotFoundError) Error() string

type PubsubUpdate

type PubsubUpdate struct {
	Handler  string
	Sender   string
	Key      string
	Hostname string
}

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 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