flow

README

Flow Distributed Workflow System

Flow Server

The Flow server manages job submissions. It provides a GRPC API that is used by clients to submit and manage workflows.

Flow CLI

The Flow CLI manages workflows from the command line. You can queue, view, and delete workflows.

NAME:
   fctl workflows queue - queue workflow

USAGE:
   fctl workflows queue [command options] [arguments...]

OPTIONS:
   --name value, -n value               workflow instance name
   --type value, -t value               workflow type
   --input-file value, -f value         file path to use as workflow input
   --input-workflow-id value, -w value  output from another workflow to use as input (can be multiple)
   --priority value                     workflow priority (low, normal, urgent) (default: "normal")
   --parameter value, -p value          specify workflow parameters (KEY=VAL)
   --help, -h                           show help (default: false)

PostgreSQL

PostgreSQL is used for database storage for workflow metadata, user accounts, and queueing.

MinIO

MinIO is used for workflow input and output storage. Minio is lightweight and works very well with Flow but any compatible S3 system should work.

Flow Processors

Flow processors provide the processing logic for queued workflows. Processors use GRPC to access the Flow server which streams workflow events to the corresponding processor type.

Documentation

Index

• Constants
• Variables
• func GenerateHash(v ...string) string
• func GenerateToken(v ...string) string
• func GetContentType(f *os.File) (string, error)
• func GetMinioClient(cfg *Config) (*minio.Client, error)
• func GetNamespaceFromContext(ctx context.Context) (string, error)
• func GetTokenFromContext(ctx context.Context) (string, error)
• type AuthenticatorConfig
• type Config
  • func DefaultConfig() *Config
  • func LoadConfig(configPath string) (*Config, error)

Constants

const (
	// S3WorkflowPath is the path in the s3 bucket for workflow content
	S3WorkflowPath = "workflows"

	// CtxTokenKey is the key stored in the context for the token
	CtxTokenKey = "token"
	// CtxAPITokenKey is the user api token key stored in the context
	CtxAPITokenKey = "api-token"
	// CtxServiceTokenKey is the service key stored in the context
	CtxServiceTokenKey = "service-token"
	// CtxTokenKey is the key stored in the context for the username
	CtxUsernameKey = "username"
	// CtxTokenKey is the key stored in the context if the user is an admin
	CtxAdminKey = "isAdmin"
	// CtxNamespaceKey is the key stored in the context for the namespace
	CtxNamespaceKey = "namespace"
	// CtxDefaultNamespace is the default key used when unauthenticated and no auth
	CtxDefaultNamespace = "default"

	// CtxDatastoreLabels is the key stored in the context for datastore queries with labels
	CtxDatastoreLabels = "labels"

	// GRPCMaxMessagSize is the max message size when sending over GRPC
	GRPCMaxMessageSize = 32 * 1024 * 1024
)

Variables

var (
	// ErrAccountExists is returned if an account already exists
	ErrAccountExists = errors.New("an account with that username already exists")
	// ErrAccountDoesNotExist is returned when an account cannot be found
	ErrAccountDoesNotExist = errors.New("account does not exist")
	// ErrNamespaceDoesNotExist is returned when an namespace cannot be found
	ErrNamespaceDoesNotExist = errors.New("namespace does not exist")
	// ErrWorkflowDoesNotExist is returned when a workflow cannot be found
	ErrWorkflowDoesNotExist = errors.New("workflow does not exist")
	// ErrServiceTokenDoesNotExist is returned when a service token cannot be found
	ErrServiceTokenDoesNotExist = errors.New("service token does not exist")
	// ErrAPITokenDoesNotExist is returned when an api token cannot be found
	ErrAPITokenDoesNotExist = errors.New("api token does not exist")
)

Functions

func GenerateHash

func GenerateHash(v ...string) string

GenerateHash generates a sha256 hash of the string

func GenerateToken

func GenerateToken(v ...string) string

GenerateToken returns a sha256 of the specified data

func GetContentType

func GetContentType(f *os.File) (string, error)

GetContentType returns the content type of the specified file

func GetMinioClient

func GetMinioClient(cfg *Config) (*minio.Client, error)

GetMinioClient returns a MinIO client using the specified flow.Config

func GetNamespaceFromContext

func GetNamespaceFromContext(ctx context.Context) (string, error)

GetNamespaceFromContext returns the namespace from the specified context

func GetTokenFromContext

func GetTokenFromContext(ctx context.Context) (string, error)

GetTokenFromContext returns the first auth token from the context

Types

type AuthenticatorConfig

type AuthenticatorConfig struct {
	// Name is the name of the authenticator
	Name string
	// Options are passed to the authenticator
	Options map[string]string
}

type Config

type Config struct {
	// GRPCAddress is the address for the grpc server
	GRPCAddress string
	// TLSCertificate is the certificate used for grpc communication
	TLSServerCertificate string
	// TLSKey is the key used for grpc communication
	TLSServerKey string
	// TLSClientCertificate is the client certificate used for client communication
	TLSClientCertificate string
	// TLSClientKey is the client key used for client communication
	TLSClientKey string
	// TLSInsecureSkipVerify disables certificate verification
	TLSInsecureSkipVerify bool
	// EnableTLS enables TLS security for connections and loads the system certificates
	EnableTLS bool
	// S3Endpoint is the endpoint for the S3 compatible service
	S3Endpoint string
	// S3AccessID is the S3 access id
	S3AccessID string
	// S3AccessKey is the S3 key
	S3AccessKey string
	// S3Bucket is the S3 bucket
	S3Bucket string
	// S3UseSSL enables SSL for the S3 service
	S3UseSSL bool
	// DatastoreAddress is the address of the datastore
	DatastoreAddress string
	// ProfilerAddress enables the performance profiler on the specified address
	ProfilerAddress string
	// MetricsAddress enables builtin Prometheus metrics
	MetricsAddress string
	// TraceEndpoint is the endpoint of the telemetry tracer
	TraceEndpoint string
	// Environment is the environment the app is running in
	Environment string
	// InitialAdminPassword is the password used when creating the initial admin account. If empty, a random one is generated.
	InitialAdminPassword string
	// Authenticator is the auth configuration
	Authenticator *AuthenticatorConfig
}

Config is the configuration used for the server

func DefaultConfig

func DefaultConfig() *Config

func LoadConfig

func LoadConfig(configPath string) (*Config, error)

LoadConfig returns a Flow config from the specified file path