Documentation
¶
Index ¶
- type AcquireAsyncCallRow
- type AssociateArtefactWithDeploymentParams
- type AsyncCall
- type AsyncCallState
- type CreateAsyncCallParams
- type DBTX
- type Deployment
- type FailAsyncCallWithRetryParams
- type GetActiveDeploymentSchemasRow
- type GetActiveDeploymentsRow
- type GetActiveRunnersRow
- type GetDeploymentArtefactsRow
- type GetDeploymentRow
- type GetDeploymentsWithArtefactsRow
- type GetDeploymentsWithMinReplicasRow
- type GetExistingDeploymentForModuleRow
- type GetNextEventForSubscriptionRow
- type GetProcessListRow
- type GetRandomSubscriberRow
- type GetRunnerRow
- type GetRunnersForDeploymentRow
- type GetSubscriptionsNeedingUpdateRow
- type InsertSubscriberParams
- type Module
- type NullAsyncCallState
- type NullTopicSubscriptionState
- type PublishEventForTopicParams
- type Querier
- type Queries
- func (q *Queries) AcquireAsyncCall(ctx context.Context) (AcquireAsyncCallRow, error)
- func (q *Queries) AssociateArtefactWithDeployment(ctx context.Context, arg AssociateArtefactWithDeploymentParams) error
- func (q *Queries) AsyncCallQueueDepth(ctx context.Context) (int64, error)
- func (q *Queries) BeginConsumingTopicEvent(ctx context.Context, subscription model.SubscriptionKey, ...) error
- func (q *Queries) CompleteEventForSubscription(ctx context.Context, name string, module string) error
- func (q *Queries) CreateAsyncCall(ctx context.Context, arg CreateAsyncCallParams) (int64, error)
- func (q *Queries) CreateDeployment(ctx context.Context, moduleName string, schema *schema.Module, ...) error
- func (q *Queries) DeleteSubscribers(ctx context.Context, deployment model.DeploymentKey) ([]model.SubscriberKey, error)
- func (q *Queries) DeleteSubscriptions(ctx context.Context, deployment model.DeploymentKey) ([]model.SubscriptionKey, error)
- func (q *Queries) DeregisterRunner(ctx context.Context, key model.RunnerKey) (int64, error)
- func (q *Queries) FailAsyncCall(ctx context.Context, error string, iD int64) (bool, error)
- func (q *Queries) FailAsyncCallWithRetry(ctx context.Context, arg FailAsyncCallWithRetryParams) (bool, error)
- func (q *Queries) GetActiveDeploymentSchemas(ctx context.Context) ([]GetActiveDeploymentSchemasRow, error)
- func (q *Queries) GetActiveDeployments(ctx context.Context) ([]GetActiveDeploymentsRow, error)
- func (q *Queries) GetActiveRunners(ctx context.Context) ([]GetActiveRunnersRow, error)
- func (q *Queries) GetArtefactDigests(ctx context.Context, digests [][]byte) ([][]byte, error)
- func (q *Queries) GetDeployment(ctx context.Context, key model.DeploymentKey) (GetDeploymentRow, error)
- func (q *Queries) GetDeploymentArtefacts(ctx context.Context, deploymentID int64) ([]GetDeploymentArtefactsRow, error)
- func (q *Queries) GetDeploymentsByID(ctx context.Context, ids []int64) ([]Deployment, error)
- func (q *Queries) GetDeploymentsWithArtefacts(ctx context.Context, digests [][]byte, schema []byte, count int64) ([]GetDeploymentsWithArtefactsRow, error)
- func (q *Queries) GetDeploymentsWithMinReplicas(ctx context.Context) ([]GetDeploymentsWithMinReplicasRow, error)
- func (q *Queries) GetExistingDeploymentForModule(ctx context.Context, name string) (GetExistingDeploymentForModuleRow, error)
- func (q *Queries) GetModulesByID(ctx context.Context, ids []int64) ([]Module, error)
- func (q *Queries) GetNextEventForSubscription(ctx context.Context, consumptionDelay sqltypes.Duration, topic model.TopicKey, ...) (GetNextEventForSubscriptionRow, error)
- func (q *Queries) GetProcessList(ctx context.Context) ([]GetProcessListRow, error)
- func (q *Queries) GetRandomSubscriber(ctx context.Context, key model.SubscriptionKey) (GetRandomSubscriberRow, error)
- func (q *Queries) GetRunner(ctx context.Context, key model.RunnerKey) (GetRunnerRow, error)
- func (q *Queries) GetRunnersForDeployment(ctx context.Context, key model.DeploymentKey) ([]GetRunnersForDeploymentRow, error)
- func (q *Queries) GetSchemaForDeployment(ctx context.Context, key model.DeploymentKey) (*schema.Module, error)
- func (q *Queries) GetSubscription(ctx context.Context, column1 string, column2 string) (TopicSubscription, error)
- func (q *Queries) GetSubscriptionsNeedingUpdate(ctx context.Context) ([]GetSubscriptionsNeedingUpdateRow, error)
- func (q *Queries) GetTopic(ctx context.Context, dollar_1 int64) (Topic, error)
- func (q *Queries) GetTopicEvent(ctx context.Context, dollar_1 int64) (TopicEvent, error)
- func (q *Queries) GetZombieAsyncCalls(ctx context.Context, limit int32) ([]AsyncCall, error)
- func (q *Queries) InsertSubscriber(ctx context.Context, arg InsertSubscriberParams) error
- func (q *Queries) KillStaleRunners(ctx context.Context, timeout sqltypes.Duration) (int64, error)
- func (q *Queries) LoadAsyncCall(ctx context.Context, id int64) (AsyncCall, error)
- func (q *Queries) PublishEventForTopic(ctx context.Context, arg PublishEventForTopicParams) error
- func (q *Queries) SetDeploymentDesiredReplicas(ctx context.Context, key model.DeploymentKey, minReplicas int32) error
- func (q *Queries) SetSubscriptionCursor(ctx context.Context, column1 model.SubscriptionKey, ...) error
- func (q *Queries) SucceedAsyncCall(ctx context.Context, response interface{}, iD int64) (bool, error)
- func (q *Queries) UpdateDeploymentSchema(ctx context.Context, schema *schema.Module, key model.DeploymentKey) error
- func (q *Queries) UpsertModule(ctx context.Context, language string, name string) (int64, error)
- func (q *Queries) UpsertRunner(ctx context.Context, arg UpsertRunnerParams) (int64, error)
- func (q *Queries) UpsertSubscription(ctx context.Context, arg UpsertSubscriptionParams) (UpsertSubscriptionRow, error)
- func (q *Queries) UpsertTopic(ctx context.Context, arg UpsertTopicParams) error
- func (q *Queries) WithTx(tx *sql.Tx) *Queries
- type Topic
- type TopicEvent
- type TopicSubscription
- type TopicSubscriptionState
- type UpsertRunnerParams
- type UpsertSubscriptionParams
- type UpsertSubscriptionRow
- type UpsertTopicParams
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type AcquireAsyncCallRow ¶
type AcquireAsyncCallRow struct { AsyncCallID int64 QueueDepth int64 Origin string Verb schema.RefKey CatchVerb optional.Option[schema.RefKey] Request json.RawMessage ScheduledAt time.Time RemainingAttempts int32 Error optional.Option[string] Backoff sqltypes.Duration MaxBackoff sqltypes.Duration ParentRequestKey optional.Option[string] TraceContext pqtype.NullRawMessage Catching bool }
type AssociateArtefactWithDeploymentParams ¶
type AssociateArtefactWithDeploymentParams struct { Key model.DeploymentKey Digest []byte Executable bool Path string }
type AsyncCall ¶
type AsyncCall struct { ID int64 CreatedAt time.Time Verb schema.RefKey State AsyncCallState Origin string ScheduledAt time.Time Response interface{} Error optional.Option[string] RemainingAttempts int32 Backoff sqltypes.Duration MaxBackoff sqltypes.Duration CatchVerb optional.Option[schema.RefKey] Catching bool ParentRequestKey optional.Option[string] TraceContext pqtype.NullRawMessage Request json.RawMessage }
type AsyncCallState ¶
type AsyncCallState string
const ( AsyncCallStatePending AsyncCallState = "pending" AsyncCallStateExecuting AsyncCallState = "executing" AsyncCallStateSuccess AsyncCallState = "success" AsyncCallStateError AsyncCallState = "error" )
func (*AsyncCallState) Scan ¶
func (e *AsyncCallState) Scan(src interface{}) error
type CreateAsyncCallParams ¶
type CreateAsyncCallParams struct { ScheduledAt time.Time Verb schema.RefKey Origin string Request json.RawMessage RemainingAttempts int32 Backoff sqltypes.Duration MaxBackoff sqltypes.Duration CatchVerb optional.Option[schema.RefKey] ParentRequestKey optional.Option[string] TraceContext json.RawMessage }
type Deployment ¶
type GetActiveDeploymentSchemasRow ¶
type GetActiveDeploymentSchemasRow struct { Key model.DeploymentKey Schema *schema.Module }
type GetActiveDeploymentsRow ¶
type GetActiveDeploymentsRow struct { Deployment Deployment ModuleName string Language string Replicas int64 }
type GetActiveRunnersRow ¶
type GetDeploymentRow ¶
type GetDeploymentRow struct { Deployment Deployment Language string ModuleName string MinReplicas int32 }
type GetDeploymentsWithMinReplicasRow ¶
type GetDeploymentsWithMinReplicasRow struct { Deployment Deployment ModuleName string Language string }
type GetNextEventForSubscriptionRow ¶
type GetNextEventForSubscriptionRow struct { Event optional.Option[model.TopicEventKey] Payload pqtype.NullRawMessage CreatedAt sqltypes.OptionalTime Caller optional.Option[string] RequestKey optional.Option[string] TraceContext pqtype.NullRawMessage Ready bool }
type GetProcessListRow ¶
type GetProcessListRow struct { MinReplicas int32 DeploymentKey model.DeploymentKey DeploymentLabels json.RawMessage RunnerKey optional.Option[model.RunnerKey] Endpoint optional.Option[string] RunnerLabels pqtype.NullRawMessage }
type GetRandomSubscriberRow ¶
type GetRunnerRow ¶
type GetRunnersForDeploymentRow ¶
type GetRunnersForDeploymentRow struct { ID int64 Key model.RunnerKey Created time.Time LastSeen time.Time Endpoint string ModuleName optional.Option[string] DeploymentID int64 Labels json.RawMessage ID_2 int64 CreatedAt time.Time ModuleID int64 Key_2 model.DeploymentKey Schema *schema.Module Labels_2 json.RawMessage MinReplicas int32 LastActivatedAt time.Time }
type GetSubscriptionsNeedingUpdateRow ¶
type GetSubscriptionsNeedingUpdateRow struct { Key model.SubscriptionKey Cursor optional.Option[model.TopicEventKey] Topic model.TopicKey Name string DeploymentKey model.DeploymentKey RequestKey optional.Option[string] }
type InsertSubscriberParams ¶
type NullAsyncCallState ¶
type NullAsyncCallState struct { AsyncCallState AsyncCallState Valid bool // Valid is true if AsyncCallState is not NULL }
func (*NullAsyncCallState) Scan ¶
func (ns *NullAsyncCallState) Scan(value interface{}) error
Scan implements the Scanner interface.
type NullTopicSubscriptionState ¶
type NullTopicSubscriptionState struct { TopicSubscriptionState TopicSubscriptionState Valid bool // Valid is true if TopicSubscriptionState is not NULL }
func (*NullTopicSubscriptionState) Scan ¶
func (ns *NullTopicSubscriptionState) Scan(value interface{}) error
Scan implements the Scanner interface.
type PublishEventForTopicParams ¶
type PublishEventForTopicParams struct { Key model.TopicEventKey Module string Topic string Caller string Payload json.RawMessage RequestKey string TraceContext json.RawMessage }
type Querier ¶
type Querier interface { // Reserve a pending async call for execution, returning the associated lease // reservation key and accompanying metadata. AcquireAsyncCall(ctx context.Context) (AcquireAsyncCallRow, error) AssociateArtefactWithDeployment(ctx context.Context, arg AssociateArtefactWithDeploymentParams) error AsyncCallQueueDepth(ctx context.Context) (int64, error) BeginConsumingTopicEvent(ctx context.Context, subscription model.SubscriptionKey, event model.TopicEventKey) error CompleteEventForSubscription(ctx context.Context, name string, module string) error CreateAsyncCall(ctx context.Context, arg CreateAsyncCallParams) (int64, error) CreateDeployment(ctx context.Context, moduleName string, schema *schema.Module, key model.DeploymentKey) error DeleteSubscribers(ctx context.Context, deployment model.DeploymentKey) ([]model.SubscriberKey, error) DeleteSubscriptions(ctx context.Context, deployment model.DeploymentKey) ([]model.SubscriptionKey, error) DeregisterRunner(ctx context.Context, key model.RunnerKey) (int64, error) FailAsyncCall(ctx context.Context, error string, iD int64) (bool, error) FailAsyncCallWithRetry(ctx context.Context, arg FailAsyncCallWithRetryParams) (bool, error) GetActiveDeploymentSchemas(ctx context.Context) ([]GetActiveDeploymentSchemasRow, error) GetActiveDeployments(ctx context.Context) ([]GetActiveDeploymentsRow, error) GetActiveRunners(ctx context.Context) ([]GetActiveRunnersRow, error) // Return the digests that exist in the database. GetArtefactDigests(ctx context.Context, digests [][]byte) ([][]byte, error) GetDeployment(ctx context.Context, key model.DeploymentKey) (GetDeploymentRow, error) // Get all artefacts matching the given digests. GetDeploymentArtefacts(ctx context.Context, deploymentID int64) ([]GetDeploymentArtefactsRow, error) GetDeploymentsByID(ctx context.Context, ids []int64) ([]Deployment, error) // Get all deployments that have artefacts matching the given digests. GetDeploymentsWithArtefacts(ctx context.Context, digests [][]byte, schema []byte, count int64) ([]GetDeploymentsWithArtefactsRow, error) GetDeploymentsWithMinReplicas(ctx context.Context) ([]GetDeploymentsWithMinReplicasRow, error) GetExistingDeploymentForModule(ctx context.Context, name string) (GetExistingDeploymentForModuleRow, error) GetModulesByID(ctx context.Context, ids []int64) ([]Module, error) GetNextEventForSubscription(ctx context.Context, consumptionDelay sqltypes.Duration, topic model.TopicKey, cursor optional.Option[model.TopicEventKey]) (GetNextEventForSubscriptionRow, error) GetProcessList(ctx context.Context) ([]GetProcessListRow, error) GetRandomSubscriber(ctx context.Context, key model.SubscriptionKey) (GetRandomSubscriberRow, error) GetRunner(ctx context.Context, key model.RunnerKey) (GetRunnerRow, error) GetRunnersForDeployment(ctx context.Context, key model.DeploymentKey) ([]GetRunnersForDeploymentRow, error) GetSchemaForDeployment(ctx context.Context, key model.DeploymentKey) (*schema.Module, error) GetSubscription(ctx context.Context, column1 string, column2 string) (TopicSubscription, error) // Results may not be ready to be scheduled yet due to event consumption delay // Sorting ensures that brand new events (that may not be ready for consumption) // don't prevent older events from being consumed // We also make sure that the subscription belongs to a deployment that has at least one runner GetSubscriptionsNeedingUpdate(ctx context.Context) ([]GetSubscriptionsNeedingUpdateRow, error) GetTopic(ctx context.Context, dollar_1 int64) (Topic, error) GetTopicEvent(ctx context.Context, dollar_1 int64) (TopicEvent, error) GetZombieAsyncCalls(ctx context.Context, limit int32) ([]AsyncCall, error) InsertSubscriber(ctx context.Context, arg InsertSubscriberParams) error KillStaleRunners(ctx context.Context, timeout sqltypes.Duration) (int64, error) LoadAsyncCall(ctx context.Context, id int64) (AsyncCall, error) PublishEventForTopic(ctx context.Context, arg PublishEventForTopicParams) error SetDeploymentDesiredReplicas(ctx context.Context, key model.DeploymentKey, minReplicas int32) error SetSubscriptionCursor(ctx context.Context, column1 model.SubscriptionKey, column2 model.TopicEventKey) error SucceedAsyncCall(ctx context.Context, response interface{}, iD int64) (bool, error) // Note that this can result in a race condition if the deployment is being updated by another process. This will go // away once we ditch the DB. // UpdateDeploymentSchema(ctx context.Context, schema *schema.Module, key model.DeploymentKey) error UpsertModule(ctx context.Context, language string, name string) (int64, error) // Upsert a runner and return the deployment ID that it is assigned to, if any. UpsertRunner(ctx context.Context, arg UpsertRunnerParams) (int64, error) UpsertSubscription(ctx context.Context, arg UpsertSubscriptionParams) (UpsertSubscriptionRow, error) UpsertTopic(ctx context.Context, arg UpsertTopicParams) error }
type Queries ¶
type Queries struct {
// contains filtered or unexported fields
}
func (*Queries) AcquireAsyncCall ¶
func (q *Queries) AcquireAsyncCall(ctx context.Context) (AcquireAsyncCallRow, error)
Reserve a pending async call for execution, returning the associated lease reservation key and accompanying metadata.
func (*Queries) AssociateArtefactWithDeployment ¶
func (q *Queries) AssociateArtefactWithDeployment(ctx context.Context, arg AssociateArtefactWithDeploymentParams) error
func (*Queries) AsyncCallQueueDepth ¶
func (*Queries) BeginConsumingTopicEvent ¶
func (q *Queries) BeginConsumingTopicEvent(ctx context.Context, subscription model.SubscriptionKey, event model.TopicEventKey) error
func (*Queries) CompleteEventForSubscription ¶
func (*Queries) CreateAsyncCall ¶
func (*Queries) CreateDeployment ¶
func (*Queries) DeleteSubscribers ¶
func (q *Queries) DeleteSubscribers(ctx context.Context, deployment model.DeploymentKey) ([]model.SubscriberKey, error)
func (*Queries) DeleteSubscriptions ¶
func (q *Queries) DeleteSubscriptions(ctx context.Context, deployment model.DeploymentKey) ([]model.SubscriptionKey, error)
func (*Queries) DeregisterRunner ¶
func (*Queries) FailAsyncCall ¶
func (*Queries) FailAsyncCallWithRetry ¶
func (*Queries) GetActiveDeploymentSchemas ¶
func (q *Queries) GetActiveDeploymentSchemas(ctx context.Context) ([]GetActiveDeploymentSchemasRow, error)
func (*Queries) GetActiveDeployments ¶
func (q *Queries) GetActiveDeployments(ctx context.Context) ([]GetActiveDeploymentsRow, error)
func (*Queries) GetActiveRunners ¶
func (q *Queries) GetActiveRunners(ctx context.Context) ([]GetActiveRunnersRow, error)
func (*Queries) GetArtefactDigests ¶
Return the digests that exist in the database.
func (*Queries) GetDeployment ¶
func (q *Queries) GetDeployment(ctx context.Context, key model.DeploymentKey) (GetDeploymentRow, error)
func (*Queries) GetDeploymentArtefacts ¶
func (q *Queries) GetDeploymentArtefacts(ctx context.Context, deploymentID int64) ([]GetDeploymentArtefactsRow, error)
Get all artefacts matching the given digests.
func (*Queries) GetDeploymentsByID ¶
func (*Queries) GetDeploymentsWithArtefacts ¶
func (q *Queries) GetDeploymentsWithArtefacts(ctx context.Context, digests [][]byte, schema []byte, count int64) ([]GetDeploymentsWithArtefactsRow, error)
Get all deployments that have artefacts matching the given digests.
func (*Queries) GetDeploymentsWithMinReplicas ¶
func (q *Queries) GetDeploymentsWithMinReplicas(ctx context.Context) ([]GetDeploymentsWithMinReplicasRow, error)
func (*Queries) GetExistingDeploymentForModule ¶
func (*Queries) GetModulesByID ¶
func (*Queries) GetNextEventForSubscription ¶
func (*Queries) GetProcessList ¶
func (q *Queries) GetProcessList(ctx context.Context) ([]GetProcessListRow, error)
func (*Queries) GetRandomSubscriber ¶
func (q *Queries) GetRandomSubscriber(ctx context.Context, key model.SubscriptionKey) (GetRandomSubscriberRow, error)
func (*Queries) GetRunnersForDeployment ¶
func (q *Queries) GetRunnersForDeployment(ctx context.Context, key model.DeploymentKey) ([]GetRunnersForDeploymentRow, error)
func (*Queries) GetSchemaForDeployment ¶
func (*Queries) GetSubscription ¶
func (*Queries) GetSubscriptionsNeedingUpdate ¶
func (q *Queries) GetSubscriptionsNeedingUpdate(ctx context.Context) ([]GetSubscriptionsNeedingUpdateRow, error)
Results may not be ready to be scheduled yet due to event consumption delay Sorting ensures that brand new events (that may not be ready for consumption) don't prevent older events from being consumed We also make sure that the subscription belongs to a deployment that has at least one runner
func (*Queries) GetTopicEvent ¶
func (*Queries) GetZombieAsyncCalls ¶
func (*Queries) InsertSubscriber ¶
func (q *Queries) InsertSubscriber(ctx context.Context, arg InsertSubscriberParams) error
func (*Queries) KillStaleRunners ¶
func (*Queries) LoadAsyncCall ¶
func (*Queries) PublishEventForTopic ¶
func (q *Queries) PublishEventForTopic(ctx context.Context, arg PublishEventForTopicParams) error
func (*Queries) SetDeploymentDesiredReplicas ¶
func (*Queries) SetSubscriptionCursor ¶
func (q *Queries) SetSubscriptionCursor(ctx context.Context, column1 model.SubscriptionKey, column2 model.TopicEventKey) error
func (*Queries) SucceedAsyncCall ¶
func (*Queries) UpdateDeploymentSchema ¶ added in v0.409.0
func (q *Queries) UpdateDeploymentSchema(ctx context.Context, schema *schema.Module, key model.DeploymentKey) error
Note that this can result in a race condition if the deployment is being updated by another process. This will go away once we ditch the DB.
func (*Queries) UpsertModule ¶
func (*Queries) UpsertRunner ¶
Upsert a runner and return the deployment ID that it is assigned to, if any.
func (*Queries) UpsertSubscription ¶
func (q *Queries) UpsertSubscription(ctx context.Context, arg UpsertSubscriptionParams) (UpsertSubscriptionRow, error)
func (*Queries) UpsertTopic ¶
func (q *Queries) UpsertTopic(ctx context.Context, arg UpsertTopicParams) error
type TopicEvent ¶
type TopicSubscription ¶
type TopicSubscriptionState ¶
type TopicSubscriptionState string
const ( TopicSubscriptionStateIdle TopicSubscriptionState = "idle" TopicSubscriptionStateExecuting TopicSubscriptionState = "executing" )
func (*TopicSubscriptionState) Scan ¶
func (e *TopicSubscriptionState) Scan(src interface{}) error
type UpsertRunnerParams ¶
type UpsertRunnerParams struct { Key model.RunnerKey Endpoint string Labels json.RawMessage DeploymentKey model.DeploymentKey }
type UpsertSubscriptionParams ¶
type UpsertSubscriptionParams struct { Key model.SubscriptionKey TopicModule string TopicName string Module string Deployment model.DeploymentKey Name string }