scheduler

Documentation

Index

• Constants
• Variables
• func CleanSpec(spec *schedpb.ScheduleSpec)
• func GetListInfoFromStartArgs(args *schedspb.StartScheduleArgs, now time.Time, specBuilder *SpecBuilder) *schedpb.ScheduleListInfo
• func NewResult(dc *dynamicconfig.Collection, specBuilder *SpecBuilder, params activityDeps) fxResult
• func SchedulerWorkflow(ctx workflow.Context, args *schedspb.StartScheduleArgs) error
• func ValidateVisibilityQuery(namespaceName namespace.Name, saNameType searchattribute.NameTypeMap, ...) error
• type CompiledSpec
  • func (cs *CompiledSpec) CanonicalForm() *schedpb.ScheduleSpec
  • func (cs *CompiledSpec) GetNextTime(jitterSeed string, after time.Time) GetNextTimeResult
• type GetNextTimeResult
• type Overlappable
• type ProcessBufferResult
  • func ProcessBuffer[T Overlappable](buffer []T, isRunning bool, ...) ProcessBufferResult[T]
• type SchedulerWorkflowVersion
• type SpecBuilder
  • func NewSpecBuilder() *SpecBuilder
  • func (b *SpecBuilder) NewCompiledSpec(spec *schedpb.ScheduleSpec) (*CompiledSpec, error)
• type TweakablePolicies

Constants

const (
	WorkflowType      = "temporal-sys-scheduler-workflow"
	NamespaceDivision = "TemporalScheduler"
)

const (

	// represents the state before Version is introduced
	InitialVersion SchedulerWorkflowVersion = 0
	// skip over entire time range if paused and batch and cache GetNextTime queries
	BatchAndCacheTimeQueries = 1
	// use cache v2, and include ids in jitter
	NewCacheAndJitter = 2
	// Don't put possibly-overlapping runs (from SCHEDULE_OVERLAP_POLICY_ALLOW_ALL) in
	// RunningWorkflows.
	DontTrackOverlapping = 3
	// start time in backfill is inclusive rather than exclusive
	InclusiveBackfillStartTime = 4
	// do backfill incrementally
	IncrementalBackfill = 5
	// update from previous action instead of current time
	UpdateFromPrevious = 6
	// do continue-as-new after pending signals
	CANAfterSignals = 7
	// set LastProcessedTime to last action instead of now
	UseLastAction = 8
	// getFutureActionTimes accounts for UpdateTime and RemainingActions
	AccurateFutureActionTimes = 9
)

const (
	// Schedules are implemented by a workflow whose ID is this string plus the schedule ID.
	WorkflowIDPrefix = "temporal-sys-scheduler:"

	// This is an example of a timestamp that's appended to the workflow
	// id, used for validation in the frontend.
	AppendedTimestampForValidation = "-2009-11-10T23:00:00Z"

	SignalNameUpdate   = "update"
	SignalNamePatch    = "patch"
	SignalNameRefresh  = "refresh"
	SignalNameForceCAN = "force-continue-as-new"

	QueryNameDescribe          = "describe"
	QueryNameListMatchingTimes = "listMatchingTimes"

	MemoFieldInfo = "ScheduleInfo"

	InitialConflictToken = 1
)

Variables

var (

	// CurrentTweakablePolicies is  a handful of options in a static value and use it as a MutableSideEffect within
	// the workflow so that we can change them without breaking existing executions or having
	// to use versioning.
	CurrentTweakablePolicies = TweakablePolicies{
		DefaultCatchupWindow:              365 * 24 * time.Hour,
		MinCatchupWindow:                  10 * time.Second,
		RetentionTime:                     7 * 24 * time.Hour,
		CanceledTerminatedCountAsFailures: false,
		AlwaysAppendTimestamp:             true,
		FutureActionCount:                 10,
		RecentActionCount:                 10,
		FutureActionCountForList:          5,
		RecentActionCountForList:          5,
		IterationsBeforeContinueAsNew:     0,
		SleepWhilePaused:                  true,
		MaxBufferSize:                     1000,
		BackfillsPerIteration:             10,
		AllowZeroSleep:                    true,
		ReuseTimer:                        true,
		NextTimeCacheV2Size:               14,
		Version:                           UseLastAction,
	}
)

var Module = fx.Options(
	fx.Provide(NewResult),
	fx.Provide(NewSpecBuilder),
)

var (
	VisibilityBaseListQuery = fmt.Sprintf(
		"%s = '%s' AND %s = '%s' AND %s = '%s'",
		searchattribute.WorkflowType,
		WorkflowType,
		searchattribute.TemporalNamespaceDivision,
		NamespaceDivision,
		searchattribute.ExecutionStatus,
		enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING.String(),
	)
)

Functions

func CleanSpec

func CleanSpec(spec *schedpb.ScheduleSpec)

CleanSpec sets default values in ranges.

func GetListInfoFromStartArgs

func GetListInfoFromStartArgs(args *schedspb.StartScheduleArgs, now time.Time, specBuilder *SpecBuilder) *schedpb.ScheduleListInfo

func NewResult

func NewResult(
	dc *dynamicconfig.Collection,
	specBuilder *SpecBuilder,
	params activityDeps,
) fxResult

func SchedulerWorkflow

func SchedulerWorkflow(ctx workflow.Context, args *schedspb.StartScheduleArgs) error

func ValidateVisibilityQuery

func ValidateVisibilityQuery(
	namespaceName namespace.Name,
	saNameType searchattribute.NameTypeMap,
	saMapperProvider searchattribute.MapperProvider,
	queryString string,
) error

Types

type CompiledSpec

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

func (*CompiledSpec) CanonicalForm

func (cs *CompiledSpec) CanonicalForm() *schedpb.ScheduleSpec

func (*CompiledSpec) GetNextTime

func (cs *CompiledSpec) GetNextTime(jitterSeed string, after time.Time) GetNextTimeResult

Returns the earliest time that matches the schedule spec that is after the given time. Returns: Nominal is the time that matches, pre-jitter. Next is the nominal time with jitter applied. If there is no matching time, Nominal and Next will be the zero time.

type GetNextTimeResult

type GetNextTimeResult struct {
	Nominal time.Time // scheduled time before adding jitter
	Next    time.Time // scheduled time after adding jitter
}

type Overlappable

type Overlappable interface {
	comparable
	GetOverlapPolicy() enumspb.ScheduleOverlapPolicy
}

type ProcessBufferResult

type ProcessBufferResult[T Overlappable] struct {
	// We can start allow-all all at once
	OverlappingStarts []T
	// Ignoring allow-all, we can start either zero or one now.
	// This is the one that we want to start, or nil.
	NonOverlappingStart T
	// The remaining buffer
	NewBuffer []T
	// Whether to cancel/terminate the currently-running one
	NeedCancel    bool
	NeedTerminate bool
	// Stats
	OverlapSkipped int64
}

func ProcessBuffer

func ProcessBuffer[T Overlappable](
	buffer []T,
	isRunning bool,
	resolve func(enumspb.ScheduleOverlapPolicy) enumspb.ScheduleOverlapPolicy,
) ProcessBufferResult[T]

type SchedulerWorkflowVersion

type SchedulerWorkflowVersion int64

type SpecBuilder

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

func NewSpecBuilder

func NewSpecBuilder() *SpecBuilder

func (*SpecBuilder) NewCompiledSpec

func (b *SpecBuilder) NewCompiledSpec(spec *schedpb.ScheduleSpec) (*CompiledSpec, error)

type TweakablePolicies

type TweakablePolicies struct {
	DefaultCatchupWindow              time.Duration            // Default for catchup window
	MinCatchupWindow                  time.Duration            // Minimum for catchup window
	RetentionTime                     time.Duration            // How long to keep schedules after they're done
	CanceledTerminatedCountAsFailures bool                     // Whether cancelled+terminated count for pause-on-failure
	AlwaysAppendTimestamp             bool                     // Whether to append timestamp for non-overlapping workflows too
	FutureActionCount                 int                      // The number of future action times to include in Describe.
	RecentActionCount                 int                      // The number of recent actual action results to include in Describe.
	FutureActionCountForList          int                      // The number of future action times to include in List (search attr).
	RecentActionCountForList          int                      // The number of recent actual action results to include in List (search attr).
	IterationsBeforeContinueAsNew     int                      // Number of iterations per run, or 0 to use server-suggested
	SleepWhilePaused                  bool                     // If true, don't set timers while paused/out of actions
	MaxBufferSize                     int                      // MaxBufferSize limits the number of buffered starts and backfills
	BackfillsPerIteration             int                      // How many backfilled actions to take per iteration (implies rate limit since min sleep is 1s)
	AllowZeroSleep                    bool                     // Whether to allow a zero-length timer. Used for workflow compatibility.
	ReuseTimer                        bool                     // Whether to reuse timer. Used for workflow compatibility.
	NextTimeCacheV2Size               int                      // Size of next time cache (v2)
	Version                           SchedulerWorkflowVersion // Used to keep track of schedules version to release new features and for backward compatibility

}