dal

package
v0.130.1 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Feb 22, 2024 License: Apache-2.0 Imports: 27 Imported by: 0

Documentation

Overview

Package dal provides a data abstraction layer for the Controller

Index

Constants

View Source
const (
	RunnerStateIdle     = RunnerState(sql.RunnerStateIdle)
	RunnerStateReserved = RunnerState(sql.RunnerStateReserved)
	RunnerStateAssigned = RunnerState(sql.RunnerStateAssigned)
	RunnerStateDead     = RunnerState(sql.RunnerStateDead)
)

Runner states.

View Source
const (
	ControllerStateLive = ControllerState(sql.ControllerStateLive)
	ControllerStateDead = ControllerState(sql.ControllerStateDead)
)

Controller states.

View Source
const (
	RequestOriginIngress = RequestOrigin(sql.OriginIngress)
	RequestOriginCron    = RequestOrigin(sql.OriginCron)
	RequestOriginPubsub  = RequestOrigin(sql.OriginPubsub)
)
View Source
const (
	EventTypeLog               = sql.EventTypeLog
	EventTypeCall              = sql.EventTypeCall
	EventTypeDeploymentCreated = sql.EventTypeDeploymentCreated
	EventTypeDeploymentUpdated = sql.EventTypeDeploymentUpdated
)

Supported event types.

Variables

View Source
var (
	// ErrConflict is returned by select methods in the DAL when a resource already exists.
	//
	// Its use will be documented in the corresponding methods.
	ErrConflict = errors.New("conflict")
	// ErrNotFound is returned by select methods in the DAL when no results are found.
	ErrNotFound = errors.New("not found")
)

Functions

func WithReservation

func WithReservation(ctx context.Context, reservation Reservation, fn func() error) error

Types

type CallEvent

type CallEvent struct {
	ID             int64
	DeploymentName model.DeploymentName
	RequestName    optional.Option[model.RequestName]
	Time           time.Time
	SourceVerb     optional.Option[schema.VerbRef]
	DestVerb       schema.VerbRef
	Duration       time.Duration
	Request        []byte
	Response       []byte
	Error          optional.Option[string]
	Stack          optional.Option[string]
}

func (*CallEvent) GetID

func (e *CallEvent) GetID() int64

type Controller

type Controller struct {
	Key      model.ControllerKey
	Endpoint string
	State    ControllerState
}

type ControllerState

type ControllerState string

func ControllerStateFromProto

func ControllerStateFromProto(state ftlv1.ControllerState) ControllerState

func (ControllerState) ToProto

type DAL

type DAL struct {

	// DeploymentChanges is a Topic that receives changes to the deployments table.
	DeploymentChanges *pubsub.Topic[DeploymentNotification]
	// contains filtered or unexported fields
}

func New

func New(ctx context.Context, pool *pgxpool.Pool) (*DAL, error)

func (*DAL) CreateArtefact

func (d *DAL) CreateArtefact(ctx context.Context, content []byte) (digest sha256.SHA256, err error)

CreateArtefact inserts a new artefact into the database and returns its ID.

func (*DAL) CreateDeployment

func (d *DAL) CreateDeployment(ctx context.Context, language string, moduleSchema *schema.Module, artefacts []DeploymentArtefact, ingressRoutes []IngressRoutingEntry) (key model.DeploymentName, err error)

CreateDeployment (possibly) creates a new deployment and associates previously created artefacts with it.

If an existing deployment with identical artefacts exists, it is returned.

func (*DAL) CreateIngressRequest

func (d *DAL) CreateIngressRequest(ctx context.Context, route, addr string) (model.RequestName, error)

func (*DAL) DeregisterRunner

func (d *DAL) DeregisterRunner(ctx context.Context, key model.RunnerKey) error

DeregisterRunner deregisters the given runner.

func (*DAL) ExpireRunnerClaims

func (d *DAL) ExpireRunnerClaims(ctx context.Context) (int64, error)

func (*DAL) GetActiveDeploymentSchemas added in v0.79.3

func (d *DAL) GetActiveDeploymentSchemas(ctx context.Context) ([]*schema.Module, error)

func (*DAL) GetActiveDeployments

func (d *DAL) GetActiveDeployments(ctx context.Context) ([]Deployment, error)

GetActiveDeployments returns all active deployments.

func (*DAL) GetActiveRunners

func (d *DAL) GetActiveRunners(ctx context.Context) ([]Runner, error)

func (*DAL) GetControllers added in v0.94.1

func (d *DAL) GetControllers(ctx context.Context, allControllers bool) ([]Controller, error)

func (*DAL) GetDeployment

func (d *DAL) GetDeployment(ctx context.Context, name model.DeploymentName) (*model.Deployment, error)

func (*DAL) GetDeploymentsNeedingReconciliation

func (d *DAL) GetDeploymentsNeedingReconciliation(ctx context.Context) ([]Reconciliation, error)

GetDeploymentsNeedingReconciliation returns deployments that have a mismatch between the number of assigned and required replicas.

func (*DAL) GetIdleRunners

func (d *DAL) GetIdleRunners(ctx context.Context, limit int, labels model.Labels) ([]Runner, error)

GetIdleRunners returns up to limit idle runners matching the given labels.

"labels" is a single level map of key-value pairs. Values may be scalar or lists of scalars. If a value is a list, it will match the labels if all the values in the list match.

e.g. {"languages": ["kotlin"], "arch": "arm64"}' will match a runner with the labels '{"languages": ["go", "kotlin"], "os": "linux", "arch": "amd64", "pid": 1234}'

If no runners are available, it will return an empty slice.

func (*DAL) GetIngressRoutes

func (d *DAL) GetIngressRoutes(ctx context.Context, method string) ([]IngressRoute, error)

func (*DAL) GetMissingArtefacts

func (d *DAL) GetMissingArtefacts(ctx context.Context, digests []sha256.SHA256) ([]sha256.SHA256, error)

GetMissingArtefacts returns the digests of the artefacts that are missing from the database.

func (*DAL) GetProcessList

func (d *DAL) GetProcessList(ctx context.Context) ([]Process, error)

GetProcessList returns a list of all "processes".

func (*DAL) GetRoutingTable

func (d *DAL) GetRoutingTable(ctx context.Context, modules []string) (map[string][]Route, error)

GetRoutingTable returns the endpoints for all runners for the given modules, or all routes if modules is empty.

Returns route map keyed by module.

func (*DAL) GetRunner

func (d *DAL) GetRunner(ctx context.Context, runnerKey model.RunnerKey) (Runner, error)

func (*DAL) GetRunnerState

func (d *DAL) GetRunnerState(ctx context.Context, runnerKey model.RunnerKey) (RunnerState, error)

func (*DAL) GetRunnersForDeployment

func (d *DAL) GetRunnersForDeployment(ctx context.Context, deployment model.DeploymentName) ([]Runner, error)

func (*DAL) GetStatus

func (d *DAL) GetStatus(
	ctx context.Context,
	allControllers, allRunners, allDeployments, allIngressRoutes bool,
) (Status, error)

func (*DAL) InsertCallEvent

func (d *DAL) InsertCallEvent(ctx context.Context, call *CallEvent) error

func (*DAL) InsertLogEvent

func (d *DAL) InsertLogEvent(ctx context.Context, log *LogEvent) error

func (*DAL) KillStaleControllers

func (d *DAL) KillStaleControllers(ctx context.Context, age time.Duration) (int64, error)

KillStaleControllers deletes controllers that have not had heartbeats for the given duration.

func (*DAL) KillStaleRunners

func (d *DAL) KillStaleRunners(ctx context.Context, age time.Duration) (int64, error)

KillStaleRunners deletes runners that have not had heartbeats for the given duration.

func (*DAL) QueryEvents

func (d *DAL) QueryEvents(ctx context.Context, limit int, filters ...EventFilter) ([]Event, error)

func (*DAL) ReplaceDeployment

func (d *DAL) ReplaceDeployment(ctx context.Context, newDeploymentName model.DeploymentName, minReplicas int) (err error)

ReplaceDeployment replaces an old deployment of a module with a new deployment.

func (*DAL) ReserveRunnerForDeployment

func (d *DAL) ReserveRunnerForDeployment(ctx context.Context, deployment model.DeploymentName, reservationTimeout time.Duration, labels model.Labels) (Reservation, error)

ReserveRunnerForDeployment reserves a runner for the given deployment.

It returns a Reservation that must be committed or rolled back.

func (*DAL) SetDeploymentReplicas

func (d *DAL) SetDeploymentReplicas(ctx context.Context, key model.DeploymentName, minReplicas int) error

SetDeploymentReplicas activates the given deployment.

func (*DAL) UpsertController

func (d *DAL) UpsertController(ctx context.Context, key model.ControllerKey, addr string) (int64, error)

func (*DAL) UpsertModule

func (d *DAL) UpsertModule(ctx context.Context, language, name string) (err error)

func (*DAL) UpsertRunner

func (d *DAL) UpsertRunner(ctx context.Context, runner Runner) error

UpsertRunner registers or updates a new runner.

ErrConflict will be returned if a runner with the same endpoint and a different key already exists.

type Deployment

type Deployment struct {
	Name        model.DeploymentName
	Language    string
	Module      string
	MinReplicas int
	Schema      *schema.Module
	CreatedAt   time.Time
	Labels      model.Labels
}

func (Deployment) String

func (d Deployment) String() string

type DeploymentArtefact

type DeploymentArtefact struct {
	Digest     sha256.SHA256
	Executable bool
	Path       string
}

func DeploymentArtefactFromProto

func DeploymentArtefactFromProto(in *ftlv1.DeploymentArtefact) (DeploymentArtefact, error)

func (*DeploymentArtefact) ToProto

type DeploymentCreatedEvent

type DeploymentCreatedEvent struct {
	ID                 int64
	DeploymentName     model.DeploymentName
	Time               time.Time
	Language           string
	ModuleName         string
	MinReplicas        int
	ReplacedDeployment optional.Option[model.DeploymentName]
}

func (*DeploymentCreatedEvent) GetID

func (e *DeploymentCreatedEvent) GetID() int64

type DeploymentNotification

type DeploymentNotification = Notification[Deployment, model.DeploymentName, *model.DeploymentName]

DeploymentNotification is a notification from the database when a deployment changes.

type DeploymentUpdatedEvent

type DeploymentUpdatedEvent struct {
	ID              int64
	DeploymentName  model.DeploymentName
	Time            time.Time
	MinReplicas     int
	PrevMinReplicas int
}

func (*DeploymentUpdatedEvent) GetID

func (e *DeploymentUpdatedEvent) GetID() int64

type Event

type Event interface {
	GetID() int64
	// contains filtered or unexported methods
}

Event types.

type EventFilter

type EventFilter func(query *eventFilter)

func FilterCall

func FilterCall(sourceModule optional.Option[string], destModule string, destVerb optional.Option[string]) EventFilter

FilterCall filters call events between the given modules.

May be called multiple times.

func FilterDeployments

func FilterDeployments(deploymentNames ...model.DeploymentName) EventFilter

func FilterDescending

func FilterDescending() EventFilter

FilterDescending returns events in descending order.

func FilterIDRange

func FilterIDRange(higherThan, lowerThan int64) EventFilter

FilterIDRange filters events between the given IDs, inclusive.

func FilterLogLevel

func FilterLogLevel(level log.Level) EventFilter

func FilterRequests

func FilterRequests(requestNames ...model.RequestName) EventFilter

func FilterTimeRange

func FilterTimeRange(olderThan, newerThan time.Time) EventFilter

FilterTimeRange filters events between the given times, inclusive.

Either maybe be zero to indicate no upper or lower bound.

func FilterTypes

func FilterTypes(types ...sql.EventType) EventFilter

type EventType

type EventType = sql.EventType

type IngressRoute

type IngressRoute struct {
	Runner     model.RunnerKey
	Deployment model.DeploymentName
	Endpoint   string
	Path       string
	Module     string
	Verb       string
}

type IngressRouteEntry

type IngressRouteEntry struct {
	Deployment model.DeploymentName
	Module     string
	Verb       string
	Method     string
	Path       string
}

type IngressRoutingEntry

type IngressRoutingEntry struct {
	Verb   string
	Method string
	Path   string
}

type LogEvent

type LogEvent struct {
	ID             int64
	DeploymentName model.DeploymentName
	RequestName    optional.Option[model.RequestName]
	Time           time.Time
	Level          int32
	Attributes     map[string]string
	Message        string
	Error          optional.Option[string]
	Stack          optional.Option[string]
}

func (*LogEvent) GetID

func (e *LogEvent) GetID() int64

type Notification

type Notification[T NotificationPayload, Key any, KeyP interface {
	*Key
	encoding.TextUnmarshaler
}] struct {
	Deleted optional.Option[Key] // If present the object was deleted.
	Message optional.Option[T]
}

A Notification from the database.

func (Notification[T, Key, KeyP]) String

func (n Notification[T, Key, KeyP]) String() string

type NotificationPayload

type NotificationPayload interface {
	// contains filtered or unexported methods
}

NotificationPayload is a row from the database.

type Process

type Process struct {
	Deployment  model.DeploymentName
	MinReplicas int
	Labels      model.Labels
	Runner      optional.Option[ProcessRunner]
}

type ProcessRunner

type ProcessRunner struct {
	Key      model.RunnerKey
	Endpoint string
	Labels   model.Labels
}

type Reconciliation

type Reconciliation struct {
	Deployment model.DeploymentName
	Module     string
	Language   string

	AssignedReplicas int
	RequiredReplicas int
}

type RequestOrigin

type RequestOrigin string

type Reservation

type Reservation interface {
	Runner() Runner
	Commit(ctx context.Context) error
	Rollback(ctx context.Context) error
}

A Reservation of a Runner.

type Route

type Route struct {
	Module     string
	Runner     model.RunnerKey
	Deployment model.DeploymentName
	Endpoint   string
}

func (Route) String

func (r Route) String() string

type Runner

type Runner struct {
	Key                model.RunnerKey
	Endpoint           string
	State              RunnerState
	ReservationTimeout optional.Option[time.Duration]
	Module             optional.Option[string]
	// Assigned deployment key, if any.
	Deployment optional.Option[model.DeploymentName]
	Labels     model.Labels
}

type RunnerState

type RunnerState string

func RunnerStateFromProto

func RunnerStateFromProto(state ftlv1.RunnerState) RunnerState

func (RunnerState) ToProto

func (s RunnerState) ToProto() ftlv1.RunnerState

type Status

type Status struct {
	Controllers   []Controller
	Runners       []Runner
	Deployments   []Deployment
	IngressRoutes []IngressRouteEntry
	Routes        []Route
}

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL