Documentation
¶
Overview ¶
Package enginelink is the in-engine Link: a registration is an engine Worker row created and activated in process, an operator-backed dispatcher session that forwards assigned actions, and direct calls into the dispatcher for step events and durable invocations. It replays the steps grpcoperator.Listen performs server-side without a stream, so the serverless operator core runs inside the engine's dispatcher process with the same behaviour it has out of process over grpclink.
Index ¶
Constants ¶
const DefaultOperatorName = "serverless"
DefaultOperatorName matches grpclink's default so both links share one operator row per tenant.
const LinkName = "engine"
LinkName labels the in-engine mode's metrics.
Variables ¶
var ErrClosed = errors.New("serverless registration is closed")
ErrClosed is returned once the registration's action stream has ended.
var ErrFlowControl = errors.New("serverless registration action buffer is full, flow control is active")
ErrFlowControl is returned by HandleAction when the registration's action buffer is full, which only happens when the core has stopped reading. The dispatcher treats it like a stream in flow control: the assignment fails and the task is requeued.
Functions ¶
func ConfigFromServer ¶
func ConfigFromServer(cf server.ServerlessOperatorConfigFile) serverlessoperator.Config
ConfigFromServer maps the engine's SERVER_SERVERLESS_OPERATOR_* settings onto the core's Config. The health server is disabled: the engine serves its own health and metrics endpoints, and the core's metrics are registered on the default Prometheus registry either way. Zero values fall through to the core's defaults.
Types ¶
type Deps ¶
type Deps struct {
Dispatcher Dispatcher
AdminV1 adminv1.AdminService
Repo repository.Repository
Validator validator.Validator
Logger *zerolog.Logger
// DispatcherId pins every worker the link creates to the local dispatcher, since the
// operator session that delivers its actions lives there.
DispatcherId uuid.UUID
// OperatorName is the v1_operator row name registrations attach their workers to; it is
// the same name grpclink connects as. Defaults to "serverless".
OperatorName string
}
Deps is what New needs from the engine.
type Dispatcher ¶
type Dispatcher interface {
// AddOperatorSession registers op as the operator-backed session for workerId; release
// removes it.
AddOperatorSession(workerId uuid.UUID, op operator.Operator) (release func())
// NotifyNewWorker tells the tenant's scheduler that the worker (or its action set) changed.
NotifyNewWorker(ctx context.Context, tenant *sqlcv1.Tenant, workerId uuid.UUID)
// SendStepActionEvent reports task progress; ctx must carry the tenant.
SendStepActionEvent(ctx context.Context, req *contracts.StepActionEvent) (*contracts.ActionEventResponse, error)
// RegisterDurableTask opens the in-process durable task session; ctx must carry the tenant.
RegisterDurableTask(ctx context.Context, externalId uuid.UUID) (chan<- *v1.DurableTaskRequest, <-chan *v1.DurableTaskResponse, error)
}
Dispatcher is the seam to the engine dispatcher, satisfied by *dispatcher.DispatcherImpl. It is an interface so the link can be tested without a running dispatcher.
type Link ¶
type Link struct {
// contains filtered or unexported fields
}
Link opens in-engine registrations.
func New ¶
New builds a Link. It panics on a nil dispatcher, admin service or repository, since the engine wires it once at startup and a nil there is a programming error.
func (*Link) Open ¶
func (e *Link) Open(ctx context.Context, tenantId uuid.UUID, opts link.OpenOpts) (link.Registration, error)
Open implements link.Link. It runs the registration steps of grpcoperator.Register and Listen in process: upsert the SERVERLESS operator row, create the worker with its slot config, link the initial action set in bulk chunks (the same path a streamed delta takes, not one upsert per action), write labels, activate the worker under a fresh listener session id, register the operator session with the dispatcher and notify the scheduler. The worker is activated only once its actions are linked. Heartbeats start with the registration and stop on Close.
func (*Link) ReleaseTenant ¶
ReleaseTenant implements link.TenantReleaser. The link holds no per-tenant state.
Source Files