dal

package
v0.43.2 Latest Latest
Warning

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

 Go to latest Published: Oct 13, 2023 License: Apache-2.0 Imports: 29 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    types.Option[model.RequestName]
	Time           time.Time
	SourceVerb     types.Option[schema.VerbRef]
	DestVerb       schema.VerbRef
	Duration       time.Duration
	Request        []byte
	Response       []byte
	Error          types.Option[string]
}

func (*CallEvent) GetID added in v0.6.0 

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 

func (s ControllerState) ToProto() ftlv1.ControllerState

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, schema *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) 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) 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, path 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 added in v0.14.0 

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 

func (d *DeploymentArtefact) ToProto() *ftlv1.DeploymentArtefact

type DeploymentCreatedEvent added in v0.22.0 

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

func (*DeploymentCreatedEvent) GetID added in v0.22.0 

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 added in v0.22.0 

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

func (*DeploymentUpdatedEvent) GetID added in v0.22.0 

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 types.Option[string], destModule string, destVerb types.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 added in v0.6.0 

func FilterDescending() EventFilter

FilterDescending returns events in descending order.

func FilterIDRange added in v0.5.0 

func FilterIDRange(higherThan, lowerThan int64) EventFilter

FilterIDRange filters events between the given IDs, inclusive.

func FilterLogLevel added in v0.5.0 

func FilterLogLevel(level log.Level) EventFilter

func FilterRequests 

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

func FilterTimeRange added in v0.5.0 

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
	Endpoint 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    types.Option[model.RequestName]
	Time           time.Time
	Level          int32
	Attributes     map[string]string
	Message        string
	Error          types.Option[string]
}

func (*LogEvent) GetID added in v0.6.0 

func (e *LogEvent) GetID() int64

type Notification 

type Notification[T NotificationPayload, Key any, KeyP interface {
	*Key
	encoding.TextUnmarshaler
}] struct {
	Deleted types.Option[Key] // If present the object was deleted.
	Message types.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 added in v0.14.0 

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

type ProcessRunner added in v0.14.0 

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 added in v0.12.0 

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 types.Option[time.Duration]
	Module             types.Option[string]
	// Assigned deployment key, if any.
	Deployment types.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
}

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.