dispatcher

package
v0.0.0-...-abab66f Latest Latest
Warning

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

 Go to latest Published: Sep 8, 2026 License: Apache-2.0 Imports: 46 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	// SnapshotFormatVersion is the persisted snapshot schema version.
	SnapshotFormatVersion = 1
	// SnapshotCompilerVersion identifies placement semantics included in the input digest.
	SnapshotCompilerVersion = "dispatcher-future-v1"
)

Variables

This section is empty.

Functions

func CompileSnapshot 

func CompileSnapshot(input CompileInput) (*PolicySnapshot, CompileSummary, error)

CompileSnapshot produces a complete immutable snapshot and an override impact summary.

func DetermineCloud 

func DetermineCloud(jobBase prowconfig.JobBase) string

DetermineCloud determines which cloud this job should run. It returns the value of ci-operator.openshift.io/cloud if it is none empty. The label is set by prow-gen for multistage tests. For template tests and hand-crafted tests, it returns the value of env. var. CLUSTER_TYPE from the job's spec.

func DetermineTargetCluster 

func DetermineTargetCluster(cluster, determinedCluster, defaultCluster string, canBeRelocated bool, blocked sets.Set[string]) string

func FindMostUsedCluster 

func FindMostUsedCluster(jc *prowconfig.JobConfig) string

func FormatDurationSeconds 

func FormatDurationSeconds(seconds int64) string

FormatDurationSeconds converts an API duration field to a human-readable value.

func GetJobVolumesFromPrometheus 

func GetJobVolumesFromPrometheus(ctx context.Context, prometheusAPI PrometheusAPI, ts time.Time) (map[string]float64, error)

GetJobVolumesFromPrometheus gets job volumes from a Prometheus server for the given time

func GetWeightedJobDemandFromPrometheus 

func GetWeightedJobDemandFromPrometheus(ctx context.Context, prometheusAPI PrometheusAPI, ts time.Time, options PrometheusOptions) (map[string]float64, error)

GetWeightedJobDemandFromPrometheus combines run count with optional aggregate runtime and resource-consumption queries. Optional queries must return a vector labelled by job_name, with duration/CPU in seconds and memory in byte-seconds.

func HasCapacityOrCapabilitiesChanged 

func HasCapacityOrCapabilitiesChanged(prev, next ClusterMap) bool

func IsGobWriteCommitted 

func IsGobWriteCommitted(err error) bool

IsGobWriteCommitted reports whether err indicates that the Gob destination was replaced before a later failure.

func NewEphemeralClusterDispatcher 

func NewEphemeralClusterDispatcher(clusters []string) *ephemeralClusterScheduler

func NewOverride 

func NewOverride(spec dispatcherv1.DispatchOverrideSpec) dispatcherv1.DispatchOverride

NewOverride constructs an override with Kubernetes timestamps for callers that do not otherwise need to import metav1.

func NewPrometheusVolumes 

func NewPrometheusVolumes(promOptions PrometheusOptions, prometheusDaysBefore int) (prometheusVolumes, error)

func OverrideIsActive 

func OverrideIsActive(override *dispatcherv1.DispatchOverride, now time.Time) bool

OverrideIsActive reports whether an override is approved and effective at now.

func ParseCapacity 

func ParseCapacity(value string) (*int32, error)

ParseCapacity parses a 1-100 capacity value.

func ParsePositiveDuration 

func ParsePositiveDuration(value string) (int64, error)

ParsePositiveDuration parses a bounded positive duration for command adapters.

func ReadGob 

func ReadGob(filename string, data interface{}) error

func SaveConfig 

func SaveConfig(config *Config, configPath string) error

SaveConfig saves config to a file

func SortOverrides 

func SortOverrides(overrides []dispatcherv1.DispatchOverride)

SortOverrides sorts overrides by stable identifier for presentation.

func ValidateOverrideSet 

func ValidateOverrideSet(overrides []dispatcherv1.DispatchOverride, now time.Time) error

ValidateOverrideSet rejects ambiguous overlapping runtime policy.

func WriteGob 

func WriteGob(filename string, data interface{}) error

Types

type ApplyRequest 

type ApplyRequest struct {
	UserID         string `json:"userID"`
	ChannelID      string `json:"channelID"`
	IdempotencyKey string `json:"idempotencyKey"`
	SlackThreadTS  string `json:"slackThreadTS,omitempty"`
}

ApplyRequest applies or approves a plan.

type BindThreadRequest 

type BindThreadRequest struct {
	UserID    string `json:"userID"`
	ChannelID string `json:"channelID"`
	ThreadTS  string `json:"threadTS"`
}

BindThreadRequest records the Slack thread used for lifecycle notifications.

type BuildFarmConfig 

type BuildFarmConfig struct {
	FilenamesRaw []string         `json:"filenames,omitempty"`
	Filenames    sets.Set[string] `json:"-"`
}

type CancelRequest 

type CancelRequest struct {
	UserID         string `json:"userID"`
	ChannelID      string `json:"channelID"`
	IdempotencyKey string `json:"idempotencyKey"`
}

CancelRequest revokes an override idempotently.

type Client 

type Client interface {
	ClusterForJob(jobName string) (string, error)
}

func NewClient 

func NewClient(address string) Client

type ClusterInfo 

type ClusterInfo struct {
	Provider     string
	Capacity     int
	Capabilities []string
}

ClusterInfo holds the provider, capacity, and capabilities.

type ClusterMap 

type ClusterMap map[string]ClusterInfo

ClusterMap maps a cluster name to its corresponding ClusterInfo.

func LoadClusterConfig 

func LoadClusterConfig(filePath string) (ClusterMap, sets.Set[string], error)

LoadClusterConfig loads cluster configuration from a YAML file, returning a ClusterMap and a set of blocked clusters.

type CompileInput 

type CompileInput struct {
	Baseline   map[string]ProwJobData
	Inventory  ClusterMap
	Blocked    sets.Set[string]
	Overrides  []dispatcherv1.DispatchOverride
	Generation uint64
	Now        time.Time
}

CompileInput contains every policy input used to build a snapshot.

type CompileSummary 

type CompileSummary struct {
	AffectedJobs   int                       `json:"affectedJobs"`
	AffectedGroups int                       `json:"affectedGroups"`
	AffectedDemand float64                   `json:"affectedDemand"`
	MovableJobs    int                       `json:"movableJobs"`
	MovableDemand  float64                   `json:"movableDemand"`
	MovedJobs      int                       `json:"movedJobs"`
	MovedDemand    float64                   `json:"movedDemand"`
	ImmovableJobs  []string                  `json:"immovableJobs,omitempty"`
	Destinations   map[string]float64        `json:"destinations,omitempty"`
	ByOverride     map[string]CompileSummary `json:"-"`
}

CompileSummary describes the impact of active overrides.

type Config 

type Config struct {
	// the job will be run on the same cloud as the one for the e2e test
	DetermineE2EByJob bool `json:"determineE2EByJob,omitempty"`
	// the job will be run on the target cloud if it otherwise runs on the source cloud.
	// The field has effect only when DetermineE2EByJob is true.
	CloudMapping map[api.Cloud]api.Cloud `json:"cloudMapping,omitempty"`
	// the cluster cluster name if no other condition matches
	Default api.Cluster `json:"default"`
	// the cluster name for ssh bastion jobs
	SSHBastion api.Cluster `json:"sshBastion"`
	// the cluster names for kvm jobs
	KVM []api.Cluster `json:"kvm"`
	// the cluster names for no-builds jobs
	NoBuilds []api.Cluster `json:"noBuilds,omitempty"`
	// ManualClusters contains valid assignment targets that must never participate in automatic build-farm scheduling.
	ManualClusters []api.Cluster `json:"manualClusters,omitempty"`
	// Groups maps a group of jobs to a cluster
	Groups JobGroups `json:"groups"`
	// BuildFarm maps groups of jobs to a cloud provider, like GCP
	BuildFarm map[api.Cloud]map[api.Cluster]*BuildFarmConfig `json:"buildFarm,omitempty"`
	// BuildFarmCloud maps sets of clusters to a cloud provider, like GCP
	BuildFarmCloud map[api.Cloud][]string `json:"-"`
}

Config is the configuration file of this tools, which defines the cluster parameter for each Prow job, i.e., where it runs

func LoadConfig 

func LoadConfig(configPath string) (*Config, error)

LoadConfig loads config from a file

func (*Config) DetermineCloudMapping 

func (config *Config) DetermineCloudMapping(jobBase prowconfig.JobBase) string

DetermineCloudMapping determines if for a given cloud there is a replacement to map, eg for cost saving reasons

func (*Config) DetermineClusterForJob 

func (config *Config) DetermineClusterForJob(jobBase prowconfig.JobBase, path string, cm ClusterMap) (clusterName api.Cluster, mayBeRelocated bool, _ error)

DetermineClusterForJob return the cluster for a prow job and if it can be relocated to a cluster in build farm

func (*Config) GetBuildFarmSize 

func (config *Config) GetBuildFarmSize() int

GetBuildFarmSize returns build farm size

func (*Config) GetClusterForJob 

func (config *Config) GetClusterForJob(jobBase prowconfig.JobBase, path string, cm ClusterMap) (api.Cluster, error)

GetClusterForJob returns a cluster for a prow job

func (*Config) IsInBuildFarm 

func (config *Config) IsInBuildFarm(clusterName api.Cluster) api.Cloud

IsInBuildFarm returns the cloudProvider if the cluster is in the build farm; empty string otherwise.

func (*Config) IsManualCluster 

func (config *Config) IsManualCluster(clusterName api.Cluster) bool

IsManualCluster reports whether clusterName is an allowed manual-only assignment target.

func (*Config) MatchingPathRegEx 

func (config *Config) MatchingPathRegEx(path string) bool

MatchingPathRegEx returns true if the given path matches a path regular expression defined in a config's group

func (*Config) SynchronizeBuildFarm 

func (config *Config) SynchronizeBuildFarm(clusterMap ClusterMap) (bool, error)

SynchronizeBuildFarm makes the cluster inventory authoritative for which build-farm clusters exist and which provider owns each cluster. Existing generated filename assignments are preserved when a cluster moves between providers. The returned boolean reports whether synchronization changed the loaded dispatcher configuration.

func (*Config) Validate 

func (config *Config) Validate() error

Validate checks if the config is valid

type ControlClient 

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

ControlClient is used by the DPTP bot to call the dispatcher control API.

func NewControlClient 

func NewControlClient(baseURL string, token func() []byte) *ControlClient

NewControlClient creates a dispatcher control API client with bounded requests.

func (*ControlClient) Apply 

func (c *ControlClient) Apply(ctx context.Context, planID string, request ApplyRequest) (*dispatcherv1.DispatchOverride, error)

Apply applies or approves a plan.

func (*ControlClient) BindThread 

func (c *ControlClient) BindThread(ctx context.Context, overrideID string, request BindThreadRequest) (*dispatcherv1.DispatchOverride, error)

BindThread records the Slack notification thread for an override.

func (*ControlClient) Cancel 

func (c *ControlClient) Cancel(ctx context.Context, overrideID string, request CancelRequest) (*dispatcherv1.DispatchOverride, error)

Cancel revokes an override.

func (*ControlClient) Explain 

func (c *ControlClient) Explain(ctx context.Context, job string) (Decision, error)

Explain returns the current scheduling decision for a job.

func (*ControlClient) GetPlan 

func (c *ControlClient) GetPlan(ctx context.Context, id string) (DispatchPlan, error)

GetPlan returns a previously created plan.

func (*ControlClient) Overrides 

func (c *ControlClient) Overrides(ctx context.Context) ([]dispatcherv1.DispatchOverride, error)

Overrides returns durable runtime overrides.

func (*ControlClient) Plan 

func (c *ControlClient) Plan(ctx context.Context, request PlanRequest) (DispatchPlan, error)

Plan previews a runtime override.

func (*ControlClient) Status 

func (c *ControlClient) Status(ctx context.Context, cluster string) (ControlStatus, error)

Status returns dispatcher control status.

type ControlOptions 

type ControlOptions struct {
	AllowedChannelID          string
	MaxTTL                    time.Duration
	MaxDrainTTL               time.Duration
	PlanTTL                   time.Duration
	EnableCapacity            bool
	EnableDrain               bool
	EnableCapabilityScope     bool
	ReconcileInterval         time.Duration
	SchedulerPropagationBound time.Duration
	WriteSafetyCheck          func() error
}

ControlOptions defines safety policy and staged feature gates for runtime controls.

func (*ControlOptions) Validate 

func (o *ControlOptions) Validate() error

Validate checks control-plane safety configuration.

type ControlPlane 

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

ControlPlane compiles durable overrides into serving snapshots and implements plan/apply/cancel.

func NewControlPlane 

func NewControlPlane(manager *SnapshotManager, store OverrideStore, options ControlOptions) (*ControlPlane, error)

NewControlPlane creates a single-replica control plane.

func (*ControlPlane) Apply 

func (c *ControlPlane) Apply(ctx context.Context, planID string, request ApplyRequest) (*dispatcherv1.DispatchOverride, error)

Apply creates or adds a distinct approval to the override represented by planID.

func (*ControlPlane) BindThread 

func (c *ControlPlane) BindThread(ctx context.Context, overrideID string, request BindThreadRequest) (*dispatcherv1.DispatchOverride, error)

BindThread idempotently associates an override with a notification thread in the allowed channel.

func (*ControlPlane) Cancel 

func (c *ControlPlane) Cancel(ctx context.Context, overrideID string, request CancelRequest) (*dispatcherv1.DispatchOverride, error)

Cancel durably revokes an override. Repeated cancellation is successful.

func (*ControlPlane) Explain 

func (c *ControlPlane) Explain(job string) (Decision, error)

Explain returns the current scheduling decision for a job.

func (*ControlPlane) GetPlan 

func (c *ControlPlane) GetPlan(id string) (DispatchPlan, error)

GetPlan returns a live plan or an error matching apply for missing or expired IDs.

func (*ControlPlane) Overrides 

func (c *ControlPlane) Overrides(ctx context.Context) ([]dispatcherv1.DispatchOverride, error)

Overrides returns every durable override.

func (*ControlPlane) Plan 

func (c *ControlPlane) Plan(ctx context.Context, request PlanRequest) (DispatchPlan, error)

Plan validates and previews an override without mutating durable state.

func (*ControlPlane) Reconcile 

func (c *ControlPlane) Reconcile(ctx context.Context) error

Reconcile compiles and publishes one complete generation, preserving the last good snapshot on failure.

func (*ControlPlane) Run 

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

Run reconciles baseline and override changes until ctx is cancelled.

func (*ControlPlane) Status 

func (c *ControlPlane) Status(ctx context.Context, cluster string) (ControlStatus, error)

Status returns current generation and optional cluster details.

func (*ControlPlane) UpdateBaseline 

func (c *ControlPlane) UpdateBaseline(baseline map[string]ProwJobData, inventory ClusterMap, blocked sets.Set[string])

UpdateBaseline installs the latest durable Git baseline inputs and triggers compilation.

type ControlServer 

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

ControlServer exposes the authenticated dispatcher operator API.

func NewControlServer 

func NewControlServer(control *ControlPlane, token func() []byte) *ControlServer

NewControlServer creates an authenticated control API handler.

func (*ControlServer) ServeHTTP 

func (s *ControlServer) ServeHTTP(writer http.ResponseWriter, request *http.Request)

ServeHTTP routes control API requests after bearer-token authentication.

type ControlStatus 

type ControlStatus struct {
	Ready             bool                            `json:"ready"`
	Generation        uint64                          `json:"generation,omitempty"`
	PolicyInputDigest string                          `json:"policyInputDigest,omitempty"`
	SnapshotChecksum  string                          `json:"snapshotChecksum,omitempty"`
	GeneratedAt       time.Time                       `json:"generatedAt,omitempty"`
	Cluster           string                          `json:"cluster,omitempty"`
	ClusterInfo       *ClusterInfo                    `json:"clusterInfo,omitempty"`
	EffectiveCapacity *int                            `json:"effectiveCapacity,omitempty"`
	Overrides         []dispatcherv1.DispatchOverride `json:"overrides,omitempty"`
}

ControlStatus describes the serving policy and optional cluster state.

type Decision 

type Decision struct {
	Cluster          string    `json:"cluster"`
	Source           string    `json:"source"`
	PolicyGeneration uint64    `json:"policyGeneration"`
	PolicyDigest     string    `json:"policyDigest"`
	OverrideID       string    `json:"overrideID,omitempty"`
	ValidUntil       time.Time `json:"validUntil,omitempty"`
	Explanation      string    `json:"explanation"`
}

Decision explains the cluster selected from a snapshot.

type DispatchPlan 

type DispatchPlan struct {
	ID                         string         `json:"id"`
	CreatedAt                  time.Time      `json:"createdAt"`
	ExpiresAt                  time.Time      `json:"expiresAt"`
	SourceGeneration           uint64         `json:"sourceGeneration"`
	PolicyInputDigest          string         `json:"policyInputDigest"`
	Request                    PlanRequest    `json:"request"`
	Impact                     CompileSummary `json:"impact"`
	RequiredApprovals          int32          `json:"requiredApprovals"`
	PropagationBound           time.Duration  `json:"propagationBound"`
	CurrentEffectiveCapacity   int            `json:"currentEffectiveCapacity"`
	RequestedEffectiveCapacity int            `json:"requestedEffectiveCapacity"`
}

DispatchPlan is an immutable impact preview tied to one policy generation.

type GobWriteCommittedError 

type GobWriteCommittedError struct {
	// Err is the underlying error encountered after the Gob file was replaced.
	Err error
}

GobWriteCommittedError reports a failure that happened after the destination file was atomically replaced. Callers must publish the new in-memory state, but should retry so the directory entry can be durably synced.

func (*GobWriteCommittedError) Error 

func (e *GobWriteCommittedError) Error() string

Error returns the underlying committed-write error message.

func (*GobWriteCommittedError) Unwrap 

func (e *GobWriteCommittedError) Unwrap() error

Unwrap returns the underlying committed-write error for errors.Is and errors.As.

type Group 

type Group struct {
	// a list of job names
	Jobs []string `json:"jobs,omitempty"`
	// a list of regexes of the file paths
	Paths []string `json:"paths,omitempty"`

	PathREs []*regexp.Regexp `json:"-"`
}

Group is a group of jobs

type JobGroups 

type JobGroups = map[api.Cluster]Group

JobGroups maps a group of jobs to a cluster

type KubernetesOverrideStore 

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

KubernetesOverrideStore persists DispatchOverride custom resources.

func NewKubernetesOverrideStore 

func NewKubernetesOverrideStore(client ctrlruntimeclient.Client, namespace string) *KubernetesOverrideStore

NewKubernetesOverrideStore creates a namespace-scoped Kubernetes override store.

func (*KubernetesOverrideStore) Create 

func (s *KubernetesOverrideStore) Create(ctx context.Context, override *dispatcherv1.DispatchOverride) error

Create persists a new override.

func (*KubernetesOverrideStore) Get 

func (s *KubernetesOverrideStore) Get(ctx context.Context, name string) (*dispatcherv1.DispatchOverride, error)

Get returns a named override.

func (*KubernetesOverrideStore) List 

func (s *KubernetesOverrideStore) List(ctx context.Context) ([]dispatcherv1.DispatchOverride, error)

List returns every override in deterministic name order.

func (*KubernetesOverrideStore) Update 

func (s *KubernetesOverrideStore) Update(ctx context.Context, override *dispatcherv1.DispatchOverride) error

Update persists spec changes such as approval or cancellation.

func (*KubernetesOverrideStore) UpdateStatus 

func (s *KubernetesOverrideStore) UpdateStatus(ctx context.Context, override *dispatcherv1.DispatchOverride) error

UpdateStatus persists controller-observed state.

type MemoryOverrideStore 

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

MemoryOverrideStore is an in-process store intended for tests and shadow-mode development.

func NewMemoryOverrideStore 

func NewMemoryOverrideStore() *MemoryOverrideStore

NewMemoryOverrideStore creates an empty memory store.

func (*MemoryOverrideStore) Create 

func (s *MemoryOverrideStore) Create(_ context.Context, override *dispatcherv1.DispatchOverride) error

Create stores a new override and rejects duplicate names.

func (*MemoryOverrideStore) Get 

func (s *MemoryOverrideStore) Get(_ context.Context, name string) (*dispatcherv1.DispatchOverride, error)

Get returns a named override.

func (*MemoryOverrideStore) List 

func (s *MemoryOverrideStore) List(_ context.Context) ([]dispatcherv1.DispatchOverride, error)

List returns every stored override in deterministic order.

func (*MemoryOverrideStore) Update 

func (s *MemoryOverrideStore) Update(_ context.Context, override *dispatcherv1.DispatchOverride) error

Update applies spec changes while preserving status as the Kubernetes status subresource does.

func (*MemoryOverrideStore) UpdateStatus 

func (s *MemoryOverrideStore) UpdateStatus(_ context.Context, override *dispatcherv1.DispatchOverride) error

UpdateStatus replaces status on an existing override.

type OverrideStore 

type OverrideStore interface {
	List(context.Context) ([]dispatcherv1.DispatchOverride, error)
	Get(context.Context, string) (*dispatcherv1.DispatchOverride, error)
	Create(context.Context, *dispatcherv1.DispatchOverride) error
	Update(context.Context, *dispatcherv1.DispatchOverride) error
	UpdateStatus(context.Context, *dispatcherv1.DispatchOverride) error
}

OverrideStore is the durable source of dispatcher runtime overrides.

type PlanRequest 

type PlanRequest struct {
	Kind            dispatcherv1.OverrideKind `json:"kind"`
	Cluster         string                    `json:"cluster"`
	Capability      string                    `json:"capability,omitempty"`
	Capacity        *int32                    `json:"capacity,omitempty"`
	DurationSeconds int64                     `json:"durationSeconds"`
	Reason          string                    `json:"reason"`
	IncidentURL     string                    `json:"incidentURL,omitempty"`
	UserID          string                    `json:"userID"`
	ChannelID       string                    `json:"channelID"`
	IdempotencyKey  string                    `json:"idempotencyKey"`
}

PlanRequest is a read-only request to preview a temporary override.

type PolicySnapshot 

type PolicySnapshot struct {
	FormatVersion   int                           `json:"formatVersion"`
	CompilerVersion string                        `json:"compilerVersion"`
	Generation      uint64                        `json:"generation"`
	GeneratedAt     time.Time                     `json:"generatedAt"`
	InputDigest     string                        `json:"inputDigest"`
	BaselineDigest  string                        `json:"baselineDigest"`
	InventoryDigest string                        `json:"inventoryDigest"`
	OverridesDigest string                        `json:"overridesDigest"`
	Checksum        string                        `json:"checksum"`
	Assignments     map[string]SnapshotAssignment `json:"assignments"`
	Baseline        map[string]ProwJobData        `json:"baseline"`
	Inventory       ClusterMap                    `json:"inventory"`
	Blocked         []string                      `json:"blocked,omitempty"`
	OverrideIDs     []string                      `json:"overrideIDs,omitempty"`
}

PolicySnapshot is a complete, immutable dispatcher policy generation.

type PrometheusAPI 

type PrometheusAPI interface {
	// Query performs a query for the given time.
	Query(ctx context.Context, query string, ts time.Time, opts ...prometheusapi.Option) (model.Value, prometheusapi.Warnings, error)
}

PrometheusAPI defines what we expect Prometheus to do in the package

type PrometheusOptions 

type PrometheusOptions struct {
	PrometheusURL             string
	PrometheusUsername        string
	PrometheusPasswordPath    string
	PrometheusBearerTokenPath string
	JobDurationQuery          string
	JobCPUQuery               string
	JobMemoryQuery            string
	RunWeight                 float64
	DurationHourWeight        float64
	CPUHourWeight             float64
	MemoryGBHourWeight        float64
	MinimumJobDemand          float64
}

PrometheusOptions exposes options used in contacting a Prometheus instance

func (*PrometheusOptions) AddFlags 

func (o *PrometheusOptions) AddFlags(fs *flag.FlagSet)

AddFlags sets up the flags for PrometheusOptions

func (*PrometheusOptions) NewPrometheusClient 

func (o *PrometheusOptions) NewPrometheusClient(secretGetter func(string) []byte) (api.Client, error)

NewPrometheusClient return a Prometheus client

func (*PrometheusOptions) Validate 

func (o *PrometheusOptions) Validate() error

Validate validates the values in the options

type ProwJobData 

type ProwJobData struct {
	Cluster      string
	Capabilities []string
	// Demand is the estimated scheduling load for this job. A non-positive value
	// is treated as one so jobs with no historical runs still participate in plans.
	Demand float64
	// Group is the repository-relative job configuration path used for impact summaries.
	Group string
}

type Prowjobs 

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

func NewProwjobs 

func NewProwjobs(jobsStoragePath string) *Prowjobs

func (*Prowjobs) GetCluster 

func (pjs *Prowjobs) GetCluster(pj string) string

func (*Prowjobs) GetDataCopy 

func (pjs *Prowjobs) GetDataCopy() map[string]ProwJobData

func (*Prowjobs) HasAnyOfClusters 

func (pjs *Prowjobs) HasAnyOfClusters(clusters sets.Set[string]) bool

func (*Prowjobs) Regenerate 

func (pjs *Prowjobs) Regenerate(prowjobs map[string]ProwJobData)

type SchedulingRequest 

type SchedulingRequest struct {
	Job string `json:"job"`
}

SchedulingRequest represents the incoming request structure

type SchedulingResponse 

type SchedulingResponse struct {
	Cluster          string     `json:"cluster"`
	Source           string     `json:"source,omitempty"`
	PolicyGeneration uint64     `json:"policyGeneration,omitempty"`
	PolicyDigest     string     `json:"policyDigest,omitempty"`
	OverrideID       string     `json:"overrideID,omitempty"`
	ValidUntil       *time.Time `json:"validUntil,omitempty"`
	Explanation      string     `json:"explanation,omitempty"`
}

Response represents the response structure

type Server 

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

func NewServer 

func NewServer(jobs *Prowjobs, ecd *ephemeralClusterScheduler, dispatch func(bool)) *Server

func NewSnapshotServer 

func NewSnapshotServer(jobs *Prowjobs, ecd *ephemeralClusterScheduler, dispatch func(bool), snapshots *SnapshotManager) *Server

NewSnapshotServer creates a server that performs normal lookups from immutable snapshots.

func (*Server) EventHandler 

func (s *Server) EventHandler(w http.ResponseWriter, r *http.Request)

EventHandler handles the /event route with dispatch logic

func (*Server) HealthHandler 

func (s *Server) HealthHandler(w http.ResponseWriter, _ *http.Request)

HealthHandler reports process liveness without claiming policy readiness.

func (*Server) ReadyHandler 

func (s *Server) ReadyHandler(w http.ResponseWriter, _ *http.Request)

ReadyHandler reports ready only after a complete valid snapshot is loaded.

func (*Server) RequestHandler 

func (s *Server) RequestHandler(w http.ResponseWriter, r *http.Request)

RequestHandler handles scheduling requests for jobs

type SnapshotAssignment 

type SnapshotAssignment struct {
	Cluster         string    `json:"cluster"`
	BaselineCluster string    `json:"baselineCluster"`
	Capabilities    []string  `json:"capabilities,omitempty"`
	Demand          float64   `json:"demand"`
	OverrideID      string    `json:"overrideID,omitempty"`
	ValidUntil      time.Time `json:"validUntil,omitempty"`
	Explanation     string    `json:"explanation,omitempty"`
}

SnapshotAssignment is one immutable scheduling decision in a PolicySnapshot.

type SnapshotManager 

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

SnapshotManager atomically publishes and serves immutable policy snapshots.

func NewSnapshotManager 

func NewSnapshotManager(path string) *SnapshotManager

NewSnapshotManager creates a snapshot manager using path as an optional restart cache.

func (*SnapshotManager) Current 

func (m *SnapshotManager) Current() *PolicySnapshot

Current returns a defensive copy of the current snapshot.

func (*SnapshotManager) Load 

func (m *SnapshotManager) Load() error

Load restores and validates the local snapshot restart cache.

func (*SnapshotManager) Lookup 

func (m *SnapshotManager) Lookup(job string, now time.Time) (Decision, bool)

Lookup returns the current decision for job. Expired runtime policy fails back to the Git-materialized baseline even if a cleanup controller is unavailable.

func (*SnapshotManager) Publish 

func (m *SnapshotManager) Publish(snapshot *PolicySnapshot) error

Publish validates, durably caches, and atomically installs a snapshot.

func (*SnapshotManager) Ready 

func (m *SnapshotManager) Ready() bool

Ready reports whether a valid snapshot is currently loaded.

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.