Help us shape the future of Go! Take the 2026 Go Developer Survey

workercontract

package
v1.52.3 Latest Latest
Warning

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

 Go to latest Published: Aug 19, 2026 License: Apache-2.0 Imports: 19 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	ContractVersion       = 1
	MaxProfileBytes       = 64 * 1024
	MaxConfiguredWorkers  = 4096
	MaxQueueCapacityItems = 1_048_576
	MaxFixedDurationMS    = 86_400_000
)
View Source 
const (
	QueueSampleInterval = 15 * time.Second
	QueueSampleTimeout  = 2 * time.Second
)
View Source 
const ProfileFileCheckInterval = 60 * time.Second

Variables

This section is empty.

Functions

func DecodeSettings 

func DecodeSettings(raw json.RawMessage, destination any) error

DecodeSettings는 service-owned settings를 unknown field와 trailing JSON 없이 해석한다.

func WritePrometheus 

func WritePrometheus(writer io.Writer, families []MetricFamily) error

WritePrometheus는 descriptor와 sample을 deterministic text exposition으로 쓴다.

Types

type AdmissionResult 

type AdmissionResult string

AdmissionResult는 canonical queue ownership 결론이다. 

const (
	AdmissionAccepted       AdmissionResult = "accepted"
	AdmissionDuplicate      AdmissionResult = "duplicate"
	AdmissionRejected       AdmissionResult = "rejected"
	AdmissionFailed         AdmissionResult = "failed"
	AdmissionOutcomeUnknown AdmissionResult = "outcome_unknown"
)

type AdmissionTotals 

type AdmissionTotals struct {
	Accepted       uint64 `json:"accepted"`
	Duplicate      uint64 `json:"duplicate"`
	Rejected       uint64 `json:"rejected"`
	Failed         uint64 `json:"failed"`
	OutcomeUnknown uint64 `json:"outcomeUnknown"`
}

AdmissionTotals는 diagnostics wire vocabulary를 고정한다.

type AttemptOutcome 

type AttemptOutcome string

AttemptOutcome은 시작된 attempt의 terminal 결론이다. 

const (
	AttemptSuccess        AttemptOutcome = "success"
	AttemptFailed         AttemptOutcome = "failed"
	AttemptTimeout        AttemptOutcome = "timeout"
	AttemptCanceled       AttemptOutcome = "canceled"
	AttemptPanic          AttemptOutcome = "panic"
	AttemptOutcomeUnknown AttemptOutcome = "outcome_unknown"
)

type AttemptTotals 

type AttemptTotals struct {
	Success        uint64 `json:"success"`
	Failed         uint64 `json:"failed"`
	Timeout        uint64 `json:"timeout"`
	Canceled       uint64 `json:"canceled"`
	Panic          uint64 `json:"panic"`
	OutcomeUnknown uint64 `json:"outcomeUnknown"`
}

AttemptTotals는 diagnostics wire vocabulary를 고정한다.

type CapacityMode 

type CapacityMode string

CapacityMode는 queue가 bounded인지 명시한다. 

const (
	CapacityModeBounded   CapacityMode = "bounded"
	CapacityModeUnbounded CapacityMode = "unbounded"
)

type CapacityPolicy 

type CapacityPolicy struct {
	Mode  CapacityMode `json:"mode"`
	Items *int64       `json:"items"`
}

CapacityPolicy는 bounded/unbounded queue 용량을 표현한다.

type Counters 

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

Counters는 worker별 bounded outcome 누계를 instance 안에 보관한다.

func (*Counters) RecordAdmission 

func (c *Counters) RecordAdmission(result AdmissionResult) bool

RecordAdmission은 증명된 admission 결론만 기록한다.

func (*Counters) RecordAttempt 

func (c *Counters) RecordAttempt(outcome AttemptOutcome) bool

RecordAttempt은 시작된 attempt의 terminal 결론만 기록한다.

func (*Counters) RecordDiscard 

func (c *Counters) RecordDiscard(reason DiscardReason) bool

RecordDiscard는 attempt 시작 전 discard를 기록한다.

func (*Counters) Snapshot 

func (c *Counters) Snapshot() WorkerTotals

Snapshot은 현재 bounded outcome 누계를 반환한다.

type DiagnosticsEnvelope 

type DiagnosticsEnvelope struct {
	ContractVersion   int                          `json:"contractVersion"`
	Service           string                       `json:"service"`
	Role              string                       `json:"role"`
	ObservedAtEpochMS int64                        `json:"observedAtEpochMs"`
	Complete          bool                         `json:"complete"`
	Profile           ProfileDiagnostics           `json:"profile"`
	Workers           map[string]WorkerDiagnostics `json:"workers"`
}

DiagnosticsEnvelope은 `/diagnostics/workers`의 v1 body다.

type DiscardReason 

type DiscardReason string

DiscardReason은 attempt 시작 전 owned work를 폐기한 이유다. 

const (
	DiscardStale    DiscardReason = "stale"
	DiscardShutdown DiscardReason = "shutdown"
)

type DiscardTotals 

type DiscardTotals struct {
	Stale    uint64 `json:"stale"`
	Shutdown uint64 `json:"shutdown"`
}

DiscardTotals는 diagnostics wire vocabulary를 고정한다.

type DurationMode 

type DurationMode string

DurationMode는 실행 시간 또는 queue age 정책의 표현 방식을 고정한다. 

const (
	DurationModeFixed  DurationMode = "fixed"
	DurationModePerJob DurationMode = "per_job"
	DurationModeNone   DurationMode = "none"
)

type DurationPolicy 

type DurationPolicy struct {
	Mode         DurationMode `json:"mode"`
	Milliseconds *int64       `json:"milliseconds"`
}

DurationPolicy는 fixed 또는 per-job duration을 표현한다.

type ExecutorDiagnostics 

type ExecutorDiagnostics struct {
	Enabled             bool  `json:"enabled"`
	ConfiguredWorkers   int   `json:"configuredWorkers"`
	RunningWorkers      int64 `json:"runningWorkers"`
	InFlight            int64 `json:"inFlight"`
	OldestInFlightAgeMS int64 `json:"oldestInFlightAgeMs"`
}

ExecutorDiagnostics는 process-local executor 상태다.

type ExecutorProfile 

type ExecutorProfile struct {
	Enabled           bool           `json:"enabled"`
	ConfiguredWorkers int            `json:"configured_workers"`
	AttemptTimeout    DurationPolicy `json:"attempt_timeout"`
}

ExecutorProfile은 profile이 소유하는 executor 설정이다.

type ExecutorSnapshot 

type ExecutorSnapshot struct {
	RunningWorkers      int64
	InFlight            int64
	OldestInFlightAgeMS int64
}

ExecutorSnapshot은 process-local executor의 non-blocking snapshot이다.

type ExecutorTracker 

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

func NewExecutorTracker 

func NewExecutorTracker() *ExecutorTracker

func (*ExecutorTracker) BeginAttempt 

func (t *ExecutorTracker) BeginAttempt(now time.Time) uint64

func (*ExecutorTracker) EndAttempt 

func (t *ExecutorTracker) EndAttempt(id uint64) bool

func (*ExecutorTracker) Snapshot 

func (t *ExecutorTracker) Snapshot(now time.Time) ExecutorSnapshot

func (*ExecutorTracker) StartWorkers 

func (t *ExecutorTracker) StartWorkers(count int) bool

func (*ExecutorTracker) StopWorkers 

func (t *ExecutorTracker) StopWorkers(count int) bool

type Identity 

type Identity struct {
	Service   string
	Role      string
	WorkerIDs []string
}

Identity는 binary가 컴파일 시점에 소유하는 service, role, worker set이다.

func KnownIdentity 

func KnownIdentity(service, role string) (Identity, error)

KnownIdentity는 v1 registry에 고정된 binary identity를 반환한다.

type LoadedProfile 

type LoadedProfile struct {
	Profile Profile
	Hash    string
	// contains filtered or unexported fields
}

LoadedProfile은 검증된 profile과 exact-byte identity다.

func LoadProfileFile 

func LoadProfileFile(path string, identity Identity) (LoadedProfile, error)

LoadProfileFile은 regular file의 exact bytes를 strict profile로 해석한다.

func (LoadedProfile) Path 

func (p LoadedProfile) Path() string

Path는 file checker가 같은 source를 다시 확인할 때만 사용한다.

type MetricFamily 

type MetricFamily struct {
	Name    string
	Help    string
	Type    MetricType
	Samples []MetricSample
}

MetricFamily는 collector와 text writer가 공유하는 descriptor다.

type MetricSample 

type MetricSample struct {
	Labels map[string]string
	Value  float64
}

MetricSample은 deterministic label set과 값을 담는다.

type MetricType 

type MetricType string

MetricType은 Prometheus family type을 고정한다. 

const (
	MetricGauge   MetricType = "gauge"
	MetricCounter MetricType = "counter"
)

type Profile 

type Profile struct {
	ContractVersion int                      `json:"contract_version"`
	Service         string                   `json:"service"`
	Role            string                   `json:"role"`
	ProfileID       string                   `json:"profile_id"`
	Workers         map[string]WorkerProfile `json:"workers"`
}

Profile은 Stack Worker Profile v1의 공통 표현이다.

type ProfileDiagnostics 

type ProfileDiagnostics struct {
	ID                   string                `json:"id"`
	Hash                 string                `json:"hash"`
	FileMatch            bool                  `json:"fileMatch"`
	FileCheckedAtEpochMS int64                 `json:"fileCheckedAtEpochMs"`
	FileErrorCode        *ProfileFileErrorCode `json:"fileErrorCode"`
}

ProfileDiagnostics는 loaded identity와 file drift만 공개한다.

type ProfileFileChecker 

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

ProfileFileChecker는 hot reload 없이 같은 profile path의 drift만 관측한다.

func NewProfileFileChecker 

func NewProfileFileChecker(profile LoadedProfile, loadedAt time.Time) *ProfileFileChecker

NewProfileFileChecker는 이미 load된 exact hash를 기준으로 checker를 만든다.

func (*ProfileFileChecker) Check 

func (c *ProfileFileChecker) Check(now time.Time) ProfileFileStatus

Check는 현재 bytes를 비교하고 replacement hash를 보존하지 않는다.

func (*ProfileFileChecker) Run 

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

Run은 고정 cadence로 drift를 확인하며 context 종료 시 반환한다.

func (*ProfileFileChecker) Status 

func (c *ProfileFileChecker) Status() ProfileFileStatus

Status는 마지막 완료된 비교 결과를 반환한다.

type ProfileFileErrorCode 

type ProfileFileErrorCode string

ProfileFileErrorCode는 diagnostics에 공개 가능한 bounded code다. 

const (
	ProfileFileMissing     ProfileFileErrorCode = "profile_file_missing"
	ProfileFileUnreadable  ProfileFileErrorCode = "profile_file_unreadable"
	ProfileFileTypeInvalid ProfileFileErrorCode = "profile_file_type_invalid"
	ProfileFileChanged     ProfileFileErrorCode = "profile_file_changed"
)

type ProfileFileStatus 

type ProfileFileStatus struct {
	Match            bool
	CheckedAtEpochMS int64
	ErrorCode        *ProfileFileErrorCode
}

ProfileFileStatus는 loaded bytes와 현재 source의 match 상태만 공개한다.

type QueueBackend 

type QueueBackend string

QueueBackend는 canonical queue backend label이다. 

const (
	QueueMemory   QueueBackend = "memory"
	QueueSQLite   QueueBackend = "sqlite"
	QueuePostgres QueueBackend = "postgres"
	QueueValkey   QueueBackend = "valkey"
)

type QueueDiagnostics 

type QueueDiagnostics struct {
	Backend              QueueBackend            `json:"backend"`
	Scope                QueueScope              `json:"scope"`
	Bounded              bool                    `json:"bounded"`
	Capacity             *int64                  `json:"capacity"`
	Depth                *int64                  `json:"depth"`
	OldestQueuedAgeMS    *int64                  `json:"oldestQueuedAgeMs"`
	SnapshotStatus       QueueSnapshotStatus     `json:"snapshotStatus"`
	LastSuccessAtEpochMS *int64                  `json:"lastSuccessAtEpochMs"`
	ErrorCode            *QueueSnapshotErrorCode `json:"errorCode"`
}

QueueDiagnostics는 latest-attempt snapshot 의미를 보존한다.

type QueueProfile 

type QueueProfile struct {
	Capacity CapacityPolicy `json:"capacity"`
	MaxAge   DurationPolicy `json:"max_age"`
}

QueueProfile은 canonical queue의 공통 정책이다.

type QueueSampler 

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

QueueSampler는 HTTP 요청과 분리된 fixed-cadence queue sampler다.

func NewQueueSampler 

func NewQueueSampler(source QueueSnapshotSource) *QueueSampler

NewQueueSampler는 아직 관측되지 않은 sampler를 만든다.

func (*QueueSampler) Latest 

func (s *QueueSampler) Latest() QueueSnapshot

Latest는 queue query 없이 마지막 attempt 결과를 반환한다.

func (*QueueSampler) Run 

func (s *QueueSampler) Run(ctx context.Context)

Run은 immediate first attempt 뒤 fixed cadence로 sampling한다.

func (*QueueSampler) Sample 

func (s *QueueSampler) Sample(ctx context.Context, now time.Time) QueueSnapshot

Sample은 한 번만 조회하고 실패 시 current value를 제거한다.

type QueueScope 

type QueueScope string

QueueScope는 central aggregation semantics를 고정한다. 

const (
	QueueScopeProcess QueueScope = "process"
	QueueScopeShared  QueueScope = "shared"
)

type QueueSnapshot 

type QueueSnapshot struct {
	Status               QueueSnapshotStatus
	Depth                *int64
	OldestQueuedAgeMS    *int64
	LastSuccessAtEpochMS *int64
	ErrorCode            *QueueSnapshotErrorCode
}

QueueSnapshot은 latest attempt와 last-success metadata를 분리한다.

func CurrentQueueSnapshot 

func CurrentQueueSnapshot(depth int64, oldestQueuedAge time.Duration, observedAt time.Time) QueueSnapshot

CurrentQueueSnapshot은 memory/SQLite owner가 이미 보유한 non-blocking 값을 만든다.

type QueueSnapshotErrorCode 

type QueueSnapshotErrorCode string

QueueSnapshotErrorCode는 diagnostics에 공개 가능한 bounded code다. 

const (
	QueueNotSampled     QueueSnapshotErrorCode = "not_sampled"
	QueueSnapshotFailed QueueSnapshotErrorCode = "queue_snapshot_failed"
	QueueSnapshotStale  QueueSnapshotErrorCode = "queue_snapshot_stale"
)

type QueueSnapshotSource 

type QueueSnapshotSource func(context.Context) (QueueValues, error)

QueueSnapshotSource는 하나의 canonical queue를 bounded context로 조회한다.

type QueueSnapshotStatus 

type QueueSnapshotStatus string

QueueSnapshotStatus는 latest observation의 가용성을 나타낸다. 

const (
	QueueSnapshotCurrent     QueueSnapshotStatus = "current"
	QueueSnapshotUnavailable QueueSnapshotStatus = "unavailable"
)

type QueueValues 

type QueueValues struct {
	Depth           int64
	OldestQueuedAge time.Duration
}

QueueValues는 source가 한 번에 관측한 ready backlog다.

type Registration 

type Registration struct {
	WorkerID                string
	Runtime                 RuntimeKind
	QueueBackend            QueueBackend
	QueueScope              QueueScope
	SettingsValidated       bool
	PerJobDeadlineValidated bool
	ExecutorSnapshot        func() ExecutorSnapshot
	QueueSnapshot           func() QueueSnapshot
	Counters                *Counters
	TotalsSnapshot          func() WorkerTotals
}

Registration은 profile worker 하나와 실제 implementation adapter를 연결한다.

type Registry 

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

Registry는 process-local worker registration과 diagnostics를 소유한다.

func NewRegistry 

func NewRegistry(loaded LoadedProfile, checker *ProfileFileChecker) *Registry

NewRegistry는 검증된 profile을 기반으로 빈 instance-owned registry를 만든다.

func (*Registry) Counters 

func (r *Registry) Counters(workerID string) (*Counters, bool)

Counters는 등록된 worker의 instrumentation owner를 반환한다.

func (*Registry) Diagnostics 

func (r *Registry) Diagnostics(observedAt time.Time) (DiagnosticsEnvelope, error)

Diagnostics는 queue I/O 없이 current registry envelope를 만든다.

func (*Registry) Handler 

func (r *Registry) Handler() http.Handler

Handler는 sealed registry의 access-control-neutral HTTP handler를 반환한다.

func (*Registry) Metrics 

func (r *Registry) Metrics(observedAt time.Time) ([]MetricFamily, error)

Metrics는 현재 diagnostics를 exact metric vocabulary로 변환한다.

func (*Registry) Register 

func (r *Registry) Register(registration Registration) error

Register는 worker adapter 하나를 중복 없이 등록한다.

func (*Registry) Seal 

func (r *Registry) Seal() error

Seal은 exact worker set과 single-runtime identity를 검증한다.

func (*Registry) WorkerProfile 

func (r *Registry) WorkerProfile(workerID string) (WorkerProfile, bool)

WorkerProfile은 service validator가 strict settings를 해석할 worker entry를 반환한다.

type RuntimeKind 

type RuntimeKind string

RuntimeKind는 worker implementation runtime label이다. 

const (
	RuntimeGo   RuntimeKind = "go"
	RuntimeRust RuntimeKind = "rust"
)

type WorkerDiagnostics 

type WorkerDiagnostics struct {
	Runtime  RuntimeKind         `json:"runtime"`
	Executor ExecutorDiagnostics `json:"executor"`
	Queue    QueueDiagnostics    `json:"queue"`
	Totals   WorkerTotals        `json:"totals"`
}

WorkerDiagnostics는 executor, canonical queue, bounded totals를 묶는다.

type WorkerProfile 

type WorkerProfile struct {
	Executor ExecutorProfile `json:"executor"`
	Queue    QueueProfile    `json:"queue"`
	Settings json.RawMessage `json:"settings"`
}

WorkerProfile은 공통 정책과 service-owned settings 원문을 보존한다.

type WorkerTotals 

type WorkerTotals struct {
	Admissions AdmissionTotals `json:"admissions"`
	Attempts   AttemptTotals   `json:"attempts"`
	Discarded  DiscardTotals   `json:"discarded"`
}

WorkerTotals는 한 worker의 bounded 누계다.

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.