worker

type DeploymentProvisioner ¶

type DeploymentProvisioner interface {
	ProvisionApplication(ctx context.Context, appID, name, url string) error
	ProvisionGoLink(ctx context.Context, appID, alias, url string) error
	DeleteApplication(ctx context.Context, appID string) error
}

DeploymentProvisioner is an optional external directory integration. Each operation must be idempotent: deployments and teardowns may be retried after ambiguous external responses without duplicating or orphaning applications.

type Detector ¶

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

Detector performs the advisory repository scans RequestDetection queues: Dockerfile discovery, compose preview, and a database guess.

It runs independently of Dispatcher, deliberately. A detection touches no application, deployment or provider state, so it needs none of Dispatcher's application-lock fencing, heartbeat, or interrupted-recovery machinery -- see github.com/conductorone/apphub/internal/controlplane.DetectionState's doc comment for why a worker that dies mid-scan can simply leave the record Queued for the next poll to retry, safely.

Every scan uses the same source acquisition a real deploy uses, including GitHub App credentials for a private repository. That is why detection cannot run in the authenticated API process: it never holds them (serverconfig.Load skips loading source credentials in "serve" mode). A repository's Dockerfile is discovered by the same isolated worker that will eventually build it.

func NewDetector(cfg serverconfig.Config, repo cp.Repository, keyReader GitHubAppKeyReader) (*Detector, error)

func NewDetectorWithFactory(repo cp.Repository, factory func() (Discoverer, error)) (*Detector, error)

func (d *Detector) Run(ctx context.Context) error

type DirectoryEntitlement ¶

type DirectoryEntitlement struct {
	ID, DisplayName, Description, AppID string
}

DirectoryEntitlement is one entry from an operator-configured identity directory, narrowed to what DirectorySyncer needs.

Declared here rather than imported from credentials/c1directory so this package carries no ConductorOne idiom -- internal/boundary's c1-idiom-fenced rule confines that to credentials/c1 and credentials/c1directory, and internal/worker is not one of those two. The composition root bridges credentials/c1directory.Entitlement into this shape, the same pattern cmd/apphub's logsAdapter uses to keep internal/controlplane free of the AWS SDK.

type DirectorySyncer ¶

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

DirectorySyncer periodically syncs ConductorOne groups into durable records the Workspace admin UI reads. The full entitlement catalog is not synced: role mappings and group-gated flags are group membership, and a tenant's unfiltered entitlement list is too large to be a picker.

It runs independently of Dispatcher and Detector for the same reason GitHubSyncer does: this touches no application, deployment, or provider state. Most deployments never configure a directory source at all, so a nil client (checked at the composition root, mirroring GitHubSyncer's optional construction) means this loop is never started rather than started to do nothing.

func NewDirectorySyncer(repo cp.Repository, client directoryAPI, audit cp.AuditWriter) (*DirectorySyncer, error)

func (d *DirectorySyncer) Run(ctx context.Context) error

type Discoverer ¶

type Discoverer interface {
	Discover(ctx context.Context, src deploy.Source) (commit string, tree detect.Tree, err error)
	Close() error
}

Discoverer is the source-acquisition port Detector needs: fetch one approved repository revision's file tree for advisory introspection, without exposing a build directory. *source.Checkout satisfies it through Discover; Close releases whatever Discover did not already clean up itself.

type Dispatcher ¶

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

Dispatcher owns concurrency and operation lifetimes. Its production execution path always uses the deploy module and a real, per-operation Git checkout.

func New(cfg serverconfig.Config, repo cp.Repository, eligibility cp.Eligibility, providers map[string]compute.Provider, targets map[string]cp.TargetPolicy, keyReader GitHubAppKeyReader, opts ...Option) (*Dispatcher, error)

func NewWithSourceFactory(cfg serverconfig.Config, repo cp.Repository, eligibility cp.Eligibility, providers map[string]compute.Provider, targets map[string]cp.TargetPolicy, factory func() (PreparedSource, error), opts ...Option) (*Dispatcher, error)

func (d *Dispatcher) Run(ctx context.Context) error

type GitHubAppKeyReader ¶

type GitHubAppKeyReader interface {
	Get(ctx context.Context) (credentials.Secret, error)
}

GitHubAppKeyReader resolves the admin-managed GitHub App's private key. Declared here rather than imported from internal/ghappkey so this package does not reach the AWS SDK: internal/boundary's aws-sdk-confined rule denies it to every package that is not a named provider boundary, and worker is not one -- only internal/ghappkey and the composition root ever construct the SSM client behind this interface, the same split Dispatcher's PreparedSource keeps around Git.

type GitHubSyncer ¶

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

GitHubSyncer periodically syncs the admin-managed GitHub App's installations from GitHub into durable records the Workspace admin UI reads.

It runs independently of Dispatcher and Detector for the same reason Detector does: this touches no application, deployment, or provider state, so it needs none of Dispatcher's fencing or heartbeat machinery.

A tick that finds no App configured yet, or fails to resolve or use its key, is silently skipped rather than treated as an error: this syncer always runs (cmd/apphub derives its key store automatically, with no operator toggle), but most administrators will never get around to creating an App at all, and "not yet configured" must read the same as "transiently unavailable" from a worker that has no way to distinguish an administrator who has not gotten to the Workspace yet from one whose key is temporarily unreachable.

func NewGitHubSyncer(repo cp.Repository, keyReader GitHubAppKeyReader) (*GitHubSyncer, error)

func (g *GitHubSyncer) Run(ctx context.Context) error

type Option ¶

type Option func(*Dispatcher)

Option configures optional dispatcher capabilities.

func WithDeploymentProvisioner(provisioner DeploymentProvisioner) Option

func WithSecretOpener(opener SecretOpener) Option

type OwnershipIndexer ¶

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

OwnershipIndexer files the ApplicationOwnerKind index entries an application's owner set names but the store lacks.

It exists for applications written before owner sets: those carry one ownerUserId, read as a one-user owner set (see ApplicationRecord's UnmarshalJSON), and have no index entry, so "my applications" would miss them. It only ever creates index entries, never touches an application record, so it cannot fence a running deployment.

func NewOwnershipIndexer(repo cp.Repository) (*OwnershipIndexer, error)

func (o *OwnershipIndexer) Run(ctx context.Context) error

type PreparedSource ¶

type PreparedSource interface {
	deploy.SourceFetcher
	Prepare(context.Context, deploy.Source) (string, error)
	Close() error
}

PreparedSource is the operation-scoped source port. Injection is available to trusted Go assembly (including hermetic integration), never API/config input.

type SecretOpener ¶

type SecretOpener interface {
	Open(ctx context.Context, subject cp.SecretSubject, ciphertext []byte) (compute.SecretValue, error)
}

SecretOpener decrypts a value the control plane sealed for one deployment. It must refuse a ciphertext sealed for any other subject.

type ServiceHour ¶

type ServiceHour struct {
	Service  string
	Hour     time.Time
	Class    int
	Requests int64
	First    time.Time
	Last     time.Time
}

ServiceHour is one ingress service's requests of one status class (2 for 2xx, and so on) in one UTC hour.

type SlotLease ¶

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

SlotLease is one held slot. Its methods are safe for concurrent use.

func (s *SlotLease) Confirm(ctx context.Context) error

func (s *SlotLease) Lost() <-chan struct{}

func (s *SlotLease) Release(ctx context.Context) error

func (s *SlotLease) Slot() int

type SlotLeases ¶

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

SlotLeases grants exclusive use of build slots across every worker, through one BuildSlotLease record per slot in the control-plane store.

A slot is taken by a compare-and-swap write only if its record is absent, then kept by renewing it. Expired records are quarantined until an operator verifies the old task stopped and clears the record. Every write names a fresh lease ID, so a worker that lost its lease cannot mistake a future holder's record for its own. See the compute/aws SlotLeaser port for why this cannot be a semaphore inside one process.

func NewSlotLeases(repo cp.Repository, holder string) (*SlotLeases, error)

func (l *SlotLeases) Acquire(ctx context.Context, slots []string) (*SlotLease, error)

type TrafficSource ¶

type TrafficSource interface {
	Hourly(ctx context.Context, start, end time.Time) ([]ServiceHour, error)
}

TrafficSource counts ingress requests per service over whole UTC hours. The composition root adapts internal/logs.TrafficReader to it, which keeps this package clear of the AWS SDK.

type TrafficSyncer ¶

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

TrafficSyncer periodically counts each application's ingress requests from the access log and stores them as hourly buckets (see internal/controlplane/usage.go).

It runs alongside Dispatcher for the same reason DirectorySyncer does: it touches no deployment or provider state. Every sync recomputes whole hours and replaces them, so running it twice, or on two workers at once, stores the same numbers rather than twice the numbers.

func NewTrafficSyncer(repo cp.Repository, traffic TrafficSource) (*TrafficSyncer, error)

func (t *TrafficSyncer) Run(ctx context.Context) error