policy

Documentation

Overview

Package policy contains implementation of triggering policy functions.

Index

• func UnmarshalDefinition(b []byte) (*messages.TriggeringPolicy, error)
• func ValidateDefinition(ctx *validation.Context, p *messages.TriggeringPolicy)
• type Environment
• type Func
  • func Default() Func
  • func GreedyBatchingPolicy(maxConcurrentInvs, maxBatchSize int) (Func, error)
  • func LogarithmicBatchingPolicy(maxConcurrentInvs, maxBatchSize int, logBase float64) (Func, error)
  • func New(p *messages.TriggeringPolicy) (Func, error)
  • func NewestFirstPolicy(maxConcurrentInvs int, pendingTimeout time.Duration) (Func, error)
• type In
• type Out
• type RequestBuilder
  • func (r *RequestBuilder) DebugLog(format string, args ...any)
  • func (r *RequestBuilder) FromBuildbucketTrigger(t *scheduler.BuildbucketTrigger)
  • func (r *RequestBuilder) FromCronTrigger(t *scheduler.CronTrigger)
  • func (r *RequestBuilder) FromGitilesTrigger(t *scheduler.GitilesTrigger)
  • func (r *RequestBuilder) FromNoopTrigger(t *scheduler.NoopTrigger)
  • func (r *RequestBuilder) FromTrigger(t *internal.Trigger)
  • func (r *RequestBuilder) FromWebUITrigger(t *scheduler.WebUITrigger)
• type SimulatedEnvironment
  • func (s *SimulatedEnvironment) DebugLog(format string, args ...any)
• type SimulatedInvocation
• type Simulator
  • func (s *Simulator) AddTrigger(delay time.Duration, t ...internal.Trigger)
  • func (s *Simulator) AdvanceTime(d time.Duration)
  • func (s *Simulator) Last() *SimulatedInvocation

Functions

func UnmarshalDefinition

func UnmarshalDefinition(b []byte) (*messages.TriggeringPolicy, error)

UnmarshalDefinition deserializes TriggeringPolicy, filling in defaults.

func ValidateDefinition

func ValidateDefinition(ctx *validation.Context, p *messages.TriggeringPolicy)

ValidateDefinition validates the triggering policy message.

Emits errors into the given context.

Types

type Environment

type Environment interface {
	// DebugLog appends a line to the triage text log.
	DebugLog(format string, args ...any)
}

Environment is used by the triggering policy for getting transient information about the environment and for logging.

TODO(vadimsh): This is intentionally mostly empty for now. Will be extended on as-needed basis.

type Func

type Func func(Environment, In) Out

Func is the concrete implementation of a triggering policy.

It looks at the current state of the job and its pending triggers list and (optionally) emits a bunch of requests to start new invocations.

It is a pure function without any side effects (except, perhaps, logging into the given environment).

func Default

func Default() Func

Default instantiates default triggering policy function.

Is is used if jobs do not define a triggering policy or it can't be understood.

func GreedyBatchingPolicy

func GreedyBatchingPolicy(maxConcurrentInvs, maxBatchSize int) (Func, error)

GreedyBatchingPolicy instantiates new GREEDY_BATCHING policy function.

It takes all pending triggers and collapses them into one new invocation, deriving its properties from the most recent trigger alone.

func LogarithmicBatchingPolicy

func LogarithmicBatchingPolicy(maxConcurrentInvs, maxBatchSize int, logBase float64) (Func, error)

LogarithmicBatchingPolicy instantiates new LOGARITHMIC_BATCHING policy function.

It takes all pending triggers and collapses floor(log(base,N)) of them into one new invocation, deriving its properties from the most recent trigger alone.

func New

func New(p *messages.TriggeringPolicy) (Func, error)

New is a factory that takes TriggeringPolicy proto and returns a concrete function that implements this policy.

The returned function will be used only during one triage round and then discarded.

The caller is responsible for filling in all default values in the proto if necessary. Use UnmarshalDefinition to deserialize TriggeringPolicy filling in the defaults.

Returns an error if the TriggeringPolicy message can't be understood (for example, it references an undefined policy kind).

func NewestFirstPolicy

func NewestFirstPolicy(maxConcurrentInvs int, pendingTimeout time.Duration) (Func, error)

NewestFirstPolicy instantiates new NEWEST_FIRST policy function.

It takes the most newly added pending triggers and creates invocations for them. It also discards triggers that have been pending for a long time.

type In

type In struct {
	// Now is the time when the policy function was called.
	Now time.Time
	// ActiveInvocations is a set of currently running invocations of the job.
	ActiveInvocations []int64
	// Triggers is a list of pending triggers sorted by time, more recent last.
	Triggers []*internal.Trigger
}

In contains parameters for a triggering policy function.

type Out

type Out struct {
	// Requests is a list of requests to start new invocations (if any).
	//
	// Each request contains parameters that will be passed to a new invocation by
	// the engine. The policy is responsible for filling them in.
	//
	// Triggers specified in the each request will be removed from the set of
	// pending triggers (they are consumed).
	Requests []task.Request

	// Discard is a list of triggers that the policy decided aren't needed anymore
	// and should be discarded by the engine.
	//
	// A trigger associated with a request must not be also slated for discard.
	Discard []*internal.Trigger
}

Out contains the decision of a triggering policy function.

type RequestBuilder

type RequestBuilder struct {
	task.Request
	// contains filtered or unexported fields
}

RequestBuilder is a task.Request in a process of being prepared.

func (*RequestBuilder) DebugLog

func (r *RequestBuilder) DebugLog(format string, args ...any)

DebugLog adds a line to the request log and the triage log.

func (*RequestBuilder) FromBuildbucketTrigger

func (r *RequestBuilder) FromBuildbucketTrigger(t *scheduler.BuildbucketTrigger)

FromBuildbucketTrigger derives the request properties from the given buildbucket trigger.

func (*RequestBuilder) FromCronTrigger

func (r *RequestBuilder) FromCronTrigger(t *scheduler.CronTrigger)

FromCronTrigger derives the request properties from the given cron trigger.

func (*RequestBuilder) FromGitilesTrigger

func (r *RequestBuilder) FromGitilesTrigger(t *scheduler.GitilesTrigger)

FromGitilesTrigger derives the request properties from the given gitiles trigger.

TODO(crbug.com/1182002): Remove properties/tags modifications here once there are no in-flight triggers that might hit Buildbucket v1 code path.

func (*RequestBuilder) FromNoopTrigger

func (r *RequestBuilder) FromNoopTrigger(t *scheduler.NoopTrigger)

FromNoopTrigger derives the request properties from the given noop trigger.

func (*RequestBuilder) FromTrigger

func (r *RequestBuilder) FromTrigger(t *internal.Trigger)

FromTrigger derives the request properties from the given trigger.

func (*RequestBuilder) FromWebUITrigger

func (r *RequestBuilder) FromWebUITrigger(t *scheduler.WebUITrigger)

FromWebUITrigger derives the request properties from the given web UI trigger.

type SimulatedEnvironment

type SimulatedEnvironment struct {
	OnDebugLog func(format string, args ...any)
}

SimulatedEnvironment implements Environment interface for use by Simulator.

func (*SimulatedEnvironment) DebugLog

func (s *SimulatedEnvironment) DebugLog(format string, args ...any)

DebugLog is part of Environment interface.

type SimulatedInvocation

type SimulatedInvocation struct {
	// Request is the original invocation request as emitted by the policy.
	Request task.Request
	// Created is when the invocation was created, relative to the epoch.
	Created time.Duration
	// Duration of the invocation, as returned by OnRequest.
	Duration time.Duration
	// Running is true if the invocation is still running.
	Running bool
}

SimulatedInvocation contains details of an invocation.

type Simulator

type Simulator struct {
	// Policy is the policy function under test.
	//
	// Must be set by the caller.
	Policy Func

	// OnRequest is called whenever a new invocation request is emitted by the
	// policy.
	//
	// It decides for how long the invocation will run.
	//
	// Must be set by the caller.
	OnRequest func(s *Simulator, r task.Request) time.Duration

	// OnDebugLog is called whenever the triggering policy logs something.
	//
	// May be set by the caller to collect the policy logs.
	OnDebugLog func(format string, args ...any)

	// Epoch is the timestamp of when the simulation started.
	//
	// Used to calculate SimulatedInvocation.Created. It is fine to leave it
	// default if you aren't looking at absolute times (which will be weird with
	// zero epoch time).
	Epoch time.Time

	// Now is the current time inside the simulation.
	//
	// It is advanced on various events (like new triggers or finishing
	// invocations). Use AdvanceTime to move it manually.
	Now time.Time

	// PendingTriggers is a set of currently pending triggers, sorted by time
	// (most recent last).
	//
	// Do not modify this list directly, use AddTrigger instead.
	PendingTriggers []*internal.Trigger

	// Invocations is a log of all produced invocations.
	//
	// They are ordered by the creation time. Contains invocations that are still
	// running (based on Now). Use Last() as a shortcut to get the last item of
	// this list.
	Invocations []*SimulatedInvocation

	// DiscardedTriggers is a log of all triggers that were discarded, sorted by time
	// (most recent last).
	DiscardedTriggers []*internal.Trigger
	// contains filtered or unexported fields
}

Simulator is used to test policies.

It simulates the scheduler engine logic and passage of time. It takes a stream of triggers as input, passes them through the policy under the test, and collects the resulting invocation requests.

func (*Simulator) AddTrigger

func (s *Simulator) AddTrigger(delay time.Duration, t ...internal.Trigger)

AddTrigger submits a trigger (one or many) to the pending trigger set.

This causes the execution of the policy function to decide what to do with the new triggers.

'delay' is time interval from the previously submitted trigger. It is used to advance time. The current simulation time will be used to populate trigger's Created field.

func (*Simulator) AdvanceTime

func (s *Simulator) AdvanceTime(d time.Duration)

AdvanceTime moves the simulated time, executing all events that happen.

func (*Simulator) Last

func (s *Simulator) Last() *SimulatedInvocation

Last returns the last invocation in Invocations list or nil if its empty.