worker

package
v1.6.5 Latest Latest
Warning

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

 Go to latest Published: Jul 23, 2026 License: AGPL-3.0 Imports: 39 Imported by: 0

Documentation

Overview

Package worker holds the asynq producer, task types, and handlers for background work (deploys, backups, metric scrapes, ...).

Index

Constants

View Source 
const (
	QueueDeploy  = "deploy"
	QueueDefault = "default"
	QueueLow     = "low"
	// QueueNode carries tasks that must run on a remote node (deploy/job/provision
	// against a non-local server). Agent tunnels live only in the control-plane
	// server process, so only its embedded worker consumes this queue; a
	// standalone `miabi worker` does not, and therefore never receives a
	// remote-node task it couldn't reach.
	QueueNode = "node"
)

Queue names, ordered by priority weight in the worker config.

View Source 
const (
	TypeDeploy         = "deploy:run"
	TypeProvisionDB    = "database:provision"
	TypeUpgradeDB      = "database:upgrade"
	TypeCanaryStep     = "deploy:canary-step"
	TypeNotifyFanout   = "notify:fanout"
	TypeWebhookDeliver = "webhook:deliver"
	TypeNotifyChannel  = "notify:channel-send"
	TypeRunJob         = "job:run"
	TypeVolumeBackup   = "volume:backup"
	TypeRunPipeline    = "pipeline:run"
	TypePlatformBackup = "platform:backup"
)

Task types.

Variables

View Source 
var (
	ErrGPUWithCluster           = errors.New("GPU apps must run as a single container, not a clustered (replicated) service; set the runtime to container")
	ErrGPUWithRestrictedProfile = errors.New("GPU device passthrough is incompatible with the restricted security profile; use the default profile for GPU apps")
)

ErrGPUWithCluster and ErrGPUWithRestrictedProfile are the deploy-time refusals for GPU requests that conflict with an incompatible runtime or security posture. Both fail the deploy clearly rather than silently dropping the GPU or the conflicting setting.

View Source 
var ErrRegistryRequired = errors.New("git-source deploys require the internal registry (set MIABI_REGISTRY_ENABLED): the runner builds the image and pushes it there for the node to pull")

ErrRegistryRequired is returned when a git-source app deploys but no image distributor is wired at all: the runner builds the image and pushes it to the internal registry for the target node to pull, so the registry is a hard dependency for git builds. When a distributor IS wired but misconfigured, the deploy surfaces the specific reason from DistributionUnavailableReason instead.

Functions

func DeployTopic 

func DeployTopic(deploymentID uint) string

DeployTopic is the eventbus topic carrying a deployment's live events.

func NewMux 

func NewMux(deploy *DeployHandler, provision *ProvisionDBHandler, upgrade *UpgradeDBHandler, fanout *FanoutHandler, webhook *WebhookDeliverHandler, channel *ChannelSendHandler, job *JobHandler, volumeBackup *VolumeBackupHandler, pipeline *PipelineHandler, platformBackup *PlatformBackupHandler) *asynq.ServeMux

NewMux registers task handlers and returns the asynq mux.

func NewServer 

func NewServer(redisAddr, redisPassword string, redisDB, concurrency int, consumeNodeQueue bool) *asynq.Server

NewServer builds an asynq server with Miabi's queue priorities. consumeNodeQueue must be true only for the control-plane server's embedded worker (which holds the agent tunnels); a standalone worker passes false so it never picks up a remote-node task it cannot reach.

func PipelineTopic 

func PipelineTopic(runID uint) string

PipelineTopic carries a pipeline run's live step logs and status.

Types

type AnalyticsConsumer added in v1.6.0 

type AnalyticsConsumer struct {
	// contains filtered or unexported fields
}

AnalyticsConsumer reads Goma Gateway's per-request event stream, rolls events into minute buckets (models.AnalyticsRollup) and persists closed buckets on an interval. It resolves each event's workspace from the route name and its app from a cached route→app map. Holds no per-request rows and no PII.

func NewAnalyticsConsumer added in v1.6.0 

func NewAnalyticsConsumer(rdb *redis.Client, routes *repositories.RouteRepository, store *repositories.AnalyticsRepository, stream, consumer string, flushEvery time.Duration, retentionDays func() int) *AnalyticsConsumer

NewAnalyticsConsumer wires the consumer. consumer is this worker's unique name within the group (so pending-message ownership is per-worker). retentionDays is evaluated on each prune; nil disables pruning (keep forever).

func (*AnalyticsConsumer) Run added in v1.6.0 

func (c *AnalyticsConsumer) Run(ctx context.Context)

Run consumes until ctx is cancelled. It reads batches in one goroutine and flushes closed buckets on a ticker in another, sharing the aggregator (which is concurrency-safe). Returns when ctx is done.

type BuildDispatcher 

type BuildDispatcher interface {
	RunBuild(ctx context.Context, in runners.BuildInputs, subjectUserID uint, onLog func(string)) (string, error)
	// RunnerWaitReason explains why no runner is available for a build in the
	// workspace, surfaced in the deploy's "waiting for a runner" log.
	RunnerWaitReason(workspaceID uint) string
}

BuildDispatcher dispatches a git-source app's image build to a runner and returns the pushed digest. Satisfied by *runners.Dispatcher.

type BuildResult 

type BuildResult struct {
	Digest string
	Size   int64
	Runner string
}

BuildResult is what a runner build reports for provenance: the pushed image digest, its size when known, and which runner built it.

type BuilderPolicy 

type BuilderPolicy interface {
	CustomBuilderAllowed(workspaceID uint) bool
}

BuilderPolicy reports whether a workspace may use a custom buildpack builder image. Satisfied by the quota service; nil means unrestricted.

type CanaryStepPayload 

type CanaryStepPayload struct {
	DeploymentID uint `json:"deployment_id"`
}

CanaryStepPayload identifies the canary deployment whose rollout to advance.

type ChannelSendHandler 

type ChannelSendHandler struct {
	// contains filtered or unexported fields
}

ChannelSendHandler delivers one event to one notification channel.

func NewChannelSendHandler 

func NewChannelSendHandler(channels *repositories.NotificationChannelRepository, events *repositories.AppEventRepository, apps *repositories.ApplicationRepository, registry *notify.Registry) *ChannelSendHandler

func (*ChannelSendHandler) ProcessTask 

func (h *ChannelSendHandler) ProcessTask(ctx context.Context, task *asynq.Task) error

type ClusterCap added in v1.3.0 

type ClusterCap interface {
	CapCluster() bool
}

ClusterCap reports whether the manager engine is a reachable swarm manager. Implemented by services/cluster.

type DeployHandler 

type DeployHandler struct {
	// contains filtered or unexported fields
}

DeployHandler runs the deployment pipeline: pull -> run -> health-gate -> swap -> release -> route sync. It is registered with the asynq worker.

func NewDeployHandler 

func NewDeployHandler(
	apps *repositories.ApplicationRepository,
	deployments *repositories.DeploymentRepository,
	releases *repositories.ReleaseRepository,
	registries *repositories.RegistryRepository,
	gitRepos *repositories.GitRepoRepository,
	portBindings *repositories.PortBindingRepository,
	volumes *repositories.VolumeRepository,
	stackEnv *repositories.StackEnvVarRepository,
	routeRepo *repositories.RouteRepository,
	clients NodeDocker,
	bus *eventbus.Bus,
	routes RouteSyncer,
	evts events.Recorder,
	producer *Producer,
	secrets SecretResolver,
) *DeployHandler

func (*DeployHandler) ProcessCanaryStep 

func (h *DeployHandler) ProcessCanaryStep(ctx context.Context, task *asynq.Task) error

ProcessCanaryStep advances an in-progress canary one step: it health-checks the canary container (auto-aborting on failure), then either bumps the traffic weight and reschedules, or auto-promotes once the weight reaches 100%.

func (*DeployHandler) ProcessTask 

func (h *DeployHandler) ProcessTask(ctx context.Context, task *asynq.Task) error

ProcessTask implements asynq.Handler for the deploy task.

func (DeployHandler) SetAllocator 

func (b DeployHandler) SetAllocator(a *netalloc.Service)

SetAllocator wires the subnet allocator used when recreating networks on a node (nil-safe). Shared by the deploy and job handlers.

func (*DeployHandler) SetBuildDispatch 

func (h *DeployHandler) SetBuildDispatch(d BuildDispatcher, registryHost string, runnerWaitTimeout time.Duration)

SetBuildDispatch wires runner build dispatch for git-source app deploys: the registry host runners push to and the max time a deploy waits for a runner.

func (*DeployHandler) SetBuildProvenance 

func (h *DeployHandler) SetBuildProvenance(resolver ImageRefResolver, images *imagesvc.Service)

SetBuildProvenance wires the builder-image resolver (admin builder policy) and the image catalog (build provenance). Called after construction.

func (*DeployHandler) SetBuilderPolicy 

func (h *DeployHandler) SetBuilderPolicy(p BuilderPolicy)

SetBuilderPolicy wires the custom-builder capability check (defense-in-depth: a builder set while granted stops being honored if the capability is later revoked, e.g. a plan downgrade). Optional; nil honors the app's builder.

func (DeployHandler) SetCluster added in v1.3.0 

func (b DeployHandler) SetCluster(c ClusterCap)

SetCluster wires swarm detection (nil-safe). Shared by the deploy and job handlers (both embed *runtimeBuilder).

func (*DeployHandler) SetDeployLock 

func (h *DeployHandler) SetDeployLock(l DeployLock)

SetDeployLock wires the per-app deploy serialization lock (optional; nil runs deploys without cross-worker serialization).

func (*DeployHandler) SetDistributor 

func (h *DeployHandler) SetDistributor(d Distributor)

SetDistributor wires the internal-registry image distributor (optional).

func (*DeployHandler) SetGPU added in v1.2.0 

func (h *DeployHandler) SetGPU(g GPUScheduler)

SetGPU wires the GPU scheduler (optional). Without it, a GPU request on an app is ignored at deploy (the request-time capability gate still applies).

func (*DeployHandler) SetLogStore 

func (h *DeployHandler) SetLogStore(s *logstore.Store)

SetLogStore wires the shared execution-log store. When set, a deployment's full build/deploy log is externalized to the store on terminal state and the DB row keeps only a bounded tail + a reference. nil keeps DB-tail-only.

func (DeployHandler) SetSecurity 

func (b DeployHandler) SetSecurity(r SecurityResolver, initImage string)

SetSecurity wires the container security resolver and the volume-chown init image. Shared by the deploy and job handlers (both embed *runtimeBuilder).

type DeployLock 

type DeployLock interface {
	// Acquire tries to take the per-app deploy lock without blocking. ok=true means
	// the caller holds it and must call release when done; ok=false means another
	// deploy holds it and the caller should defer and retry later.
	Acquire(ctx context.Context, appID uint) (ok bool, release func(), err error)
}

DeployLock serializes deploys per application so two concurrent deploys of the same app can't race on release-version assignment or the active-container swap (which would duplicate versions and strand containers). Optional on the handler — a nil lock disables serialization (single-worker dev, tests).

func NewRedisDeployLock 

func NewRedisDeployLock(rdb *redis.Client) DeployLock

NewRedisDeployLock builds a per-app deploy lock over the given Redis client.

type DeployPayload 

type DeployPayload struct {
	DeploymentID uint `json:"deployment_id"`
}

DeployPayload identifies the deployment to process.

type Distributor 

type Distributor interface {
	// DistributionEnabled reports whether the registry is enabled + configured to
	// receive build pushes.
	DistributionEnabled() bool
	// DistributionUnavailableReason returns "" when distribution is ready, else a
	// user-facing sentence naming the specific missing config.
	DistributionUnavailableReason() string
	// BuildRef is the registry ref a build is distributed under, e.g.
	// registry.<domain>/<workspace-name>/<app-name>:<deploymentID>.
	BuildRef(workspaceID uint, appName string, deploymentID uint) string
	// TagReleaseVersion adds a v<version> tag to the pushed build image (by digest)
	// so the registry mirrors the release number. Best-effort.
	TagReleaseVersion(ctx context.Context, workspaceID uint, appName, digest string, version int) error
	// IsBuildRef reports whether ref is one of this registry's distributed refs.
	IsBuildRef(ref string) bool
	// PushAuth is the credential to push/pull distributed images.
	PushAuth() *docker.RegistryAuth
}

Distributor distributes built images via the internal registry so any node can pull them (multi-node deploys/rollbacks of Git-built apps). Satisfied by the registry service; nil/disabled leaves builds node-local (single-node default).

type FanoutHandler 

type FanoutHandler struct {
	// contains filtered or unexported fields
}

FanoutHandler resolves a recorded event into per-endpoint delivery tasks for the event's workspace.

func NewFanoutHandler 

func NewFanoutHandler(webhooks *repositories.WebhookRepository, channels *repositories.NotificationChannelRepository, events *repositories.AppEventRepository, producer *Producer, rdb *redis.Client) *FanoutHandler

func (*FanoutHandler) ProcessTask 

func (h *FanoutHandler) ProcessTask(ctx context.Context, task *asynq.Task) error

type GPUScheduler added in v1.2.0 

type GPUScheduler interface {
	// Preflight enforces the plan capability + workspace GPU quota (no device
	// binding). Returns nil when the app requests no GPU.
	Preflight(app *models.Application) error
	// ResolveDevices binds the app's GPU request to concrete devices on its node,
	// given the node's advertised container runtimes. Returns nil when the app
	// requests no GPU.
	ResolveDevices(ctx context.Context, app *models.Application, runtimes []string) ([]docker.GPURequest, error)
}

GPUScheduler validates and resolves an app's GPU request at deploy time. Satisfied by *gpu.Service; nil means GPU handling is disabled.

type ImageRefResolver 

type ImageRefResolver interface {
	Ref(key string) string
}

ImageRefResolver resolves a platform-image catalog key (e.g. the default buildpack builder) to its effective ref (admin override -> config default -> built-in). Satisfied by platformimage.Resolver; kept as an interface so the worker doesn't import the settings stack. The control plane resolves the admin-controlled builder image with it and passes it to the runner in the build config — builds themselves run on the runner, never here.

type JobHandler 

type JobHandler struct {
	// contains filtered or unexported fields
}

JobHandler runs one-off Jobs: a command executed once in an application's runtime context (same image/env/networks/volumes/node/limits as the app), capturing the exit code and logs. It reuses the deploy pipeline's runtime substrate so a Job's environment can't drift from the real deploy.

func NewJobHandler 

func NewJobHandler(jobs *repositories.JobRepository, apps *repositories.ApplicationRepository, stackEnv *repositories.StackEnvVarRepository, routeRepo *repositories.RouteRepository, registries *repositories.RegistryRepository, clients NodeDocker, secrets SecretResolver) *JobHandler

func (*JobHandler) ProcessTask 

func (h *JobHandler) ProcessTask(ctx context.Context, task *asynq.Task) error

ProcessTask implements asynq.Handler for the run-job task.

func (JobHandler) SetAllocator 

func (b JobHandler) SetAllocator(a *netalloc.Service)

SetAllocator wires the subnet allocator used when recreating networks on a node (nil-safe). Shared by the deploy and job handlers.

func (JobHandler) SetCluster added in v1.3.0 

func (b JobHandler) SetCluster(c ClusterCap)

SetCluster wires swarm detection (nil-safe). Shared by the deploy and job handlers (both embed *runtimeBuilder).

func (*JobHandler) SetLogStore 

func (h *JobHandler) SetLogStore(s *logstore.Store)

SetLogStore wires the shared execution-log store. When set, a job's full output is externalized to the store on terminal state and the DB row keeps only a bounded tail + a reference. nil keeps DB-tail-only.

func (JobHandler) SetSecurity 

func (b JobHandler) SetSecurity(r SecurityResolver, initImage string)

SetSecurity wires the container security resolver and the volume-chown init image. Shared by the deploy and job handlers (both embed *runtimeBuilder).

type NodeDocker 

type NodeDocker interface {
	For(serverID uint) (docker.Client, error)
	LocalID() uint
}

NodeDocker resolves the Docker client for a node id (0 = local).

type NotifyChannelPayload 

type NotifyChannelPayload struct {
	ChannelID uint `json:"channel_id"`
	EventID   uint `json:"event_id"`
}

NotifyChannelPayload identifies a single notification-channel delivery.

type NotifyFanoutPayload 

type NotifyFanoutPayload struct {
	AppEventID uint `json:"app_event_id"`
}

NotifyFanoutPayload identifies a persisted event to fan out to a workspace's webhooks and notification channels.

type PipelineDeployer 

type PipelineDeployer struct {
	// contains filtered or unexported fields
}

PipelineDeployer performs the deploy-by-digest a runner build triggers: it creates a deployment of the pushed image (by its registry ref) for the app and enqueues it to the app's node, which pulls and runs it. Implements runners.Deployer.

func NewPipelineDeployer 

func NewPipelineDeployer(apps *repositories.ApplicationRepository, deployments *repositories.DeploymentRepository, producer *Producer) *PipelineDeployer

func (*PipelineDeployer) DeployByDigest 

func (d *PipelineDeployer) DeployByDigest(run *models.PipelineRun, appID uint, imageRef string) error

DeployByDigest creates and enqueues a deploy of imageRef (repo@digest) for the run's app. The deploy worker recognizes an internal-registry ref and pulls it (no rebuild), pinned to the run's commit and image for provenance.

type PipelineHandler 

type PipelineHandler struct {
	// contains filtered or unexported fields
}

PipelineHandler is the internal runner: it clones the source once into a per-run workspace, then executes the run's steps in order over that shared filesystem — container steps in one-shot containers, plus the built-in `build` (workspace -> image + digest) and `deploy` (deploy-by-digest) steps. Remote runners lease the same steps over the machine API.

func NewPipelineHandler 

func NewPipelineHandler(
	pipelines *repositories.PipelineRepository,
	apps *repositories.ApplicationRepository,
	deployments *repositories.DeploymentRepository,
	gitRepos *repositories.GitRepoRepository,
	images *imagesvc.Service,
	clients NodeDocker,
	bus *eventbus.Bus,
	producer *Producer,
) *PipelineHandler

NewPipelineHandler builds the internal runner handler.

func (*PipelineHandler) ProcessTask 

func (h *PipelineHandler) ProcessTask(ctx context.Context, task *asynq.Task) error

ProcessTask runs a pipeline end to end.

func (*PipelineHandler) SetLogStore 

func (h *PipelineHandler) SetLogStore(s *logstore.Store)

SetLogStore wires the shared execution-log store. When set, a pipeline step's full log is externalized to the store on terminal state and the DB row keeps only a bounded tail + a reference. nil keeps DB-tail-only.

func (*PipelineHandler) SetRunnerDispatch 

func (h *PipelineHandler) SetRunnerDispatch(d RunnerDispatcher, workspaces *repositories.WorkspaceRepository, registry string, hosts RegistryHostResolver, runnerWaitTimeout time.Duration)

SetRunnerDispatch wires runner dispatch: every pipeline build runs on a registered runner, and a run with no available runner waits (up to runnerWaitTimeout) rather than building on a node.

type PlatformBackupHandler 

type PlatformBackupHandler struct {
	// contains filtered or unexported fields
}

PlatformBackupHandler runs platform (control-plane) backup tasks.

func NewPlatformBackupHandler 

func NewPlatformBackupHandler(svc *platformbackup.Service) *PlatformBackupHandler

func (*PlatformBackupHandler) ProcessTask 

func (h *PlatformBackupHandler) ProcessTask(ctx context.Context, task *asynq.Task) error

ProcessTask implements asynq.Handler for the platform-backup task.

type PlatformBackupPayload 

type PlatformBackupPayload struct {
	PlatformBackupID uint `json:"platform_backup_id"`
}

PlatformBackupPayload identifies the platform backup record to execute.

type Producer 

type Producer struct {
	// contains filtered or unexported fields
}

Producer enqueues tasks into Redis via asynq.

func NewProducer 

func NewProducer(redisAddr, redisPassword string, redisDB, maxRetries int) *Producer

NewProducer creates a task producer backed by Redis on the given DB index.

func (*Producer) Close 

func (p *Producer) Close() error

Close releases the underlying asynq client.

func (*Producer) EnqueueCanaryStep 

func (p *Producer) EnqueueCanaryStep(deploymentID uint, delaySeconds int, serverID uint) error

EnqueueCanaryStep schedules the next canary progression step after delaySeconds.

func (*Producer) EnqueueDeploy 

func (p *Producer) EnqueueDeploy(deploymentID, serverID uint) error

EnqueueDeploy schedules a deployment to be processed by the worker. serverID is the app's node, used to route remote-node deploys to the server's worker.

func (*Producer) EnqueueDeployIn 

func (p *Producer) EnqueueDeployIn(deploymentID, serverID uint, delay time.Duration) error

EnqueueDeployIn re-schedules a deployment after a delay — used to defer a deploy when another for the same app is already in progress (per-app lock).

func (*Producer) EnqueueDeployToBuilder 

func (p *Producer) EnqueueDeployToBuilder(deploymentID uint) error

EnqueueDeployToBuilder routes a deployment to QueueNode, which only the control-plane server's embedded worker consumes — the process that holds the runner tunnels. A git-source deploy must build on a runner, so a standalone worker (which can't reach the tunnels) hands it off here instead of failing.

func (*Producer) EnqueueNotifyChannel 

func (p *Producer) EnqueueNotifyChannel(channelID, eventID uint) error

EnqueueNotifyChannel schedules delivery of an event to one notification channel.

func (*Producer) EnqueueNotifyFanout 

func (p *Producer) EnqueueNotifyFanout(appEventID uint) error

EnqueueNotifyFanout schedules fan-out of an event to a workspace's webhooks and channels. Cheap and safe to call from request or worker goroutines.

func (*Producer) EnqueuePipelineRun 

func (p *Producer) EnqueuePipelineRun(runID, serverID uint) error

EnqueuePipelineRun schedules a pipeline run on the internal runner. Like jobs it is not auto-retried — a failed run is recorded; the user re-runs it. The run is deploy-priority so CI feedback is prompt. serverID routes remote-node runs.

func (*Producer) EnqueuePipelineRunIn 

func (p *Producer) EnqueuePipelineRunIn(runID uint, delay time.Duration) error

EnqueuePipelineRunIn re-enqueues a pipeline run after delay — used to poll for a free runner when a run is waiting for one (no eligible runner right now).

func (*Producer) EnqueuePlatformBackup 

func (p *Producer) EnqueuePlatformBackup(backupID uint) error

EnqueuePlatformBackup schedules a platform (control-plane) backup to run in the background. Like volume backups it is not auto-retried — a failed run is recorded on the backup row and the admin re-runs it explicitly. It always runs on the manager node, so it uses the default (local) queue.

func (*Producer) EnqueueProvisionDB 

func (p *Producer) EnqueueProvisionDB(databaseID, serverID uint) error

EnqueueProvisionDB schedules a database to be provisioned by the worker.

func (*Producer) EnqueueRunJob 

func (p *Producer) EnqueueRunJob(jobID, serverID uint) error

EnqueueRunJob schedules a one-off Job to be run by the worker. Jobs are not auto-retried — a failed run is recorded; the user re-runs it explicitly.

func (*Producer) EnqueueUpgradeDB 

func (p *Producer) EnqueueUpgradeDB(databaseID, serverID uint, target, path string, stopApps bool) error

EnqueueUpgradeDB schedules a database version upgrade. MaxRetry is 0: a half-applied upgrade must not be blindly re-run (the handler records failures on the instance instead).

func (*Producer) EnqueueVolumeBackup 

func (p *Producer) EnqueueVolumeBackup(backupID, serverID uint) error

EnqueueVolumeBackup schedules a volume backup to run in the background. Like jobs, it is not auto-retried — a failed run is recorded on the backup row and the user re-runs it explicitly. serverID routes remote-node backups.

func (*Producer) EnqueueWebhookDeliver 

func (p *Producer) EnqueueWebhookDeliver(webhookID, eventID uint) error

EnqueueWebhookDeliver schedules delivery of an event to one webhook. Each endpoint retries independently with asynq's exponential backoff.

func (*Producer) SetLocalID 

func (p *Producer) SetLocalID(id uint)

SetLocalID records the local node's server id so node-affine tasks can be routed to the right queue (remote nodes -> QueueNode, local -> def).

type ProvisionDBHandler 

type ProvisionDBHandler struct {
	// contains filtered or unexported fields
}

ProvisionDBHandler runs database provisioning tasks.

func NewProvisionDBHandler 

func NewProvisionDBHandler(dbs *database.Service) *ProvisionDBHandler

func (*ProvisionDBHandler) ProcessTask 

func (h *ProvisionDBHandler) ProcessTask(ctx context.Context, task *asynq.Task) error

ProcessTask implements asynq.Handler for the database-provision task.

type ProvisionDBPayload 

type ProvisionDBPayload struct {
	DatabaseID uint `json:"database_id"`
}

ProvisionDBPayload identifies the database to provision.

type ProxyNetworkReconciler 

type ProxyNetworkReconciler struct {
	// contains filtered or unexported fields
}

ProxyNetworkReconciler attaches or detaches an application's running container(s) from the shared reverse-proxy network so only route-exposed apps stay on it. Runs on the Docker engine of the node the app lives on, so it works for local and remote (tunneled) nodes alike. Satisfies the route service's ProxyAttacher.

func NewProxyNetworkReconciler 

func NewProxyNetworkReconciler(apps *repositories.ApplicationRepository, releases *repositories.ReleaseRepository, clients NodeDocker) *ProxyNetworkReconciler

func (*ProxyNetworkReconciler) ReconcileIngressGateway 

func (r *ProxyNetworkReconciler) ReconcileIngressGateway(ctx context.Context) error

ReconcileIngressGateway joins the manager's central Goma gateway to the shared cluster ingress overlay (creating it if needed), so public traffic can reach every clustered app's service VIP. It re-asserts the attachment that a compose recreate of the gateway (docker compose up -d) silently drops, so it is meant to run at worker startup and on cluster refresh as well as on each service-app route change or deploy. No-op when the manager is not a swarm manager or the gateway container is not up yet; idempotent.

func (*ProxyNetworkReconciler) ReconcileProxyAttachment 

func (r *ProxyNetworkReconciler) ReconcileProxyAttachment(ctx context.Context, appID uint, attached bool) error

ReconcileProxyAttachment connects (attached) or disconnects the app's active and canary release containers from the proxy network, with the app's stable DNS alias. Idempotent and best-effort: a missing container or offline node is a no-op, so this never blocks a route change.

type RegistryHostResolver 

type RegistryHostResolver interface{ RegistryHost() string }

RegistryHostResolver resolves the live registry host a runner logs into and pushes to. It tracks a UI-set or domain-derived host rather than a static env value, so the login host and push host stay identical (a mismatch → "denied").

type RouteSyncer 

type RouteSyncer interface {
	SyncRoute(ctx context.Context, appID uint) error
}

RouteSyncer reconciles an application's reverse-proxy route. The domain service implements it; injected to avoid a worker->domain import.

type RunJobPayload 

type RunJobPayload struct {
	JobID uint `json:"job_id"`
}

RunJobPayload identifies the one-off Job to run.

type RunPipelinePayload 

type RunPipelinePayload struct {
	PipelineRunID uint `json:"pipeline_run_id"`
}

RunPipelinePayload identifies the pipeline run to execute.

type RunnerDispatcher 

type RunnerDispatcher interface {
	// Dispatch runs the pipeline on a runner and drives it to a terminal status,
	// or returns runners.ErrNoRunner / ErrRunnerOffline when none can take it now.
	Dispatch(ctx context.Context, in runners.JobInputs, requiredLabels []string, subjectUserID uint) error
}

RunnerDispatcher runs a pipeline on a dedicated runner over the machine API. Implemented by runners.Dispatcher. Kept as an interface so the handler stays testable.

type RuntimeContext 

type RuntimeContext struct {
	Env         []string
	Networks    []string
	Mounts      map[string]string
	Binds       []docker.BindMount
	MemoryBytes int64
	NanoCPUs    int64
}

RuntimeContext is the substrate common to any container run in an app's environment: resolved env, the networks to join (already ensured to exist on the node), volume mounts, and resource limits. Callers layer run-specific concerns (image, command, ports, aliases, healthcheck, labels) on top.

type SecretResolver 

type SecretResolver interface {
	ResolveAll(workspaceID uint, env []string) ([]string, error)
}

SecretResolver substitutes `${{ secrets.NAME }}` references in env values with the workspace's stored secret values. Implemented by the secret service; optional (nil = references are left untouched).

type Security 

type Security struct {
	User            string // "uid:gid" the container runs as; "" = image default
	NoNewPrivileges bool
	CapDrop         []string
}

Security is the resolved container hardening for a workspace's application and job containers under the "restricted" security profile. The zero value (empty User) means no restriction — run as the image's default user.

func (Security) Restricted 

func (s Security) Restricted() bool

Restricted reports whether any hardening applies.

type SecurityFunc 

type SecurityFunc func(workspaceID uint, officialTemplate bool) Security

SecurityFunc adapts a plain function to SecurityResolver.

func (SecurityFunc) ContainerSecurity 

func (f SecurityFunc) ContainerSecurity(id uint, officialTemplate bool) Security

ContainerSecurity implements SecurityResolver.

type SecurityResolver 

type SecurityResolver interface {
	ContainerSecurity(workspaceID uint, officialTemplate bool) Security
}

SecurityResolver resolves the security profile for a workspace's app/job containers. Implemented by an adapter over the quota service + platform config; optional (nil = no restriction, today's behavior). officialTemplate marks an app created from an official marketplace template, which a plan may exempt from the restricted UID so it can keep the image's own default user.

type UpgradeDBHandler 

type UpgradeDBHandler struct {
	// contains filtered or unexported fields
}

UpgradeDBHandler runs database version-upgrade tasks.

func NewUpgradeDBHandler 

func NewUpgradeDBHandler(dbs *database.Service) *UpgradeDBHandler

func (*UpgradeDBHandler) ProcessTask 

func (h *UpgradeDBHandler) ProcessTask(ctx context.Context, task *asynq.Task) error

ProcessTask implements asynq.Handler for the database-upgrade task. The job records its own success/failure on the instance, so it returns nil (no retry) except on an undecodable payload.

type UpgradeDBPayload 

type UpgradeDBPayload struct {
	DatabaseID uint   `json:"database_id"`
	Target     string `json:"target"`    // target engine version
	Path       string `json:"path"`      // in-place | dump-restore
	StopApps   bool   `json:"stop_apps"` // quiesce apps using the database during the upgrade
}

UpgradeDBPayload describes a queued database version upgrade.

type VolumeBackupHandler 

type VolumeBackupHandler struct {
	// contains filtered or unexported fields
}

VolumeBackupHandler runs volume backup tasks.

func NewVolumeBackupHandler 

func NewVolumeBackupHandler(svc *volumebackup.Service) *VolumeBackupHandler

func (*VolumeBackupHandler) ProcessTask 

func (h *VolumeBackupHandler) ProcessTask(ctx context.Context, task *asynq.Task) error

ProcessTask implements asynq.Handler for the volume-backup task.

type VolumeBackupPayload 

type VolumeBackupPayload struct {
	VolumeBackupID uint `json:"volume_backup_id"`
}

VolumeBackupPayload identifies the volume backup record to execute.

type WebhookDeliverHandler 

type WebhookDeliverHandler struct {
	// contains filtered or unexported fields
}

WebhookDeliverHandler delivers one event to one webhook endpoint, recording the outcome of each attempt. Returning an error lets asynq retry with exponential backoff.

func NewWebhookDeliverHandler 

func NewWebhookDeliverHandler(webhooks *repositories.WebhookRepository, deliveries *repositories.WebhookDeliveryRepository, events *repositories.AppEventRepository, apps *repositories.ApplicationRepository) *WebhookDeliverHandler

func (*WebhookDeliverHandler) ProcessTask 

func (h *WebhookDeliverHandler) ProcessTask(ctx context.Context, task *asynq.Task) error

type WebhookDeliverPayload 

type WebhookDeliverPayload struct {
	WebhookID uint `json:"webhook_id"`
	EventID   uint `json:"event_id"`
}

WebhookDeliverPayload identifies a single webhook delivery. EventID == 0 signals a synthetic test delivery.

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.