Documentation
¶
Index ¶
- Constants
- func CheckPassword(plaintext, hashed string) bool
- func FetchWorkersForWatchdog(ctx context.Context, db *sql.DB) ([]watchdog.WorkerInfo, error)
- func HttpServer(cfg config.Config) (tls.Certificate, string, http.Handler, error)
- func IsAdmin(ctx context.Context) bool
- func LoadEmbeddedCertificates() (tls.Certificate, string, error)
- func LogoutHandler(w http.ResponseWriter, r *http.Request)
- func NewLogin(s *taskQueueServer) http.HandlerFunc
- func Serve(cfg config.Config, ctx context.Context, cancel context.CancelFunc) error
- type Accumulator
- type AuthenticatedUser
- type Job
- type ObjectiveDefinition
- type ParamSpec
- type QualityDefinition
- type QualityResult
- type StepAgg
- type StepStatsAgg
- type TaskUpdateBroadcast
Constants ¶
const DefaultAssignTrigger int32 = 500 // 5 sec
Variables ¶
This section is empty.
Functions ¶
func CheckPassword ¶
func FetchWorkersForWatchdog ¶
FetchWorkersForWatchdog reads workers and builds WorkerInfo list
func HttpServer ¶
func LoadEmbeddedCertificates ¶
func LoadEmbeddedCertificates() (tls.Certificate, string, error)
func LogoutHandler ¶
func LogoutHandler(w http.ResponseWriter, r *http.Request)
LogoutHandler clears the session_token cookie to log out the user.
func NewLogin ¶
func NewLogin(s *taskQueueServer) http.HandlerFunc
NewLogin returns an HTTP handler function to process login requests.
Types ¶
type Accumulator ¶
In-memory aggregation types for step/workflow stats Accumulator holds aggregation stats for durations/counts.
type AuthenticatedUser ¶
func GetUserFromContext ¶
func GetUserFromContext(ctx context.Context) *AuthenticatedUser
type Job ¶
type Job struct {
JobID int32
WorkerID int32
WorkerName string
ProviderID int32
ProviderName string
Region string
Flavor string
Action rune // "C", "D", "R"
Retry int
Timeout time.Duration
FromRecruiter bool
}
Job represents a task to be executed.
type ObjectiveDefinition ¶ added in v0.7.5
type ObjectiveDefinition struct {
Formula string `json:"formula"`
Direction string `json:"direction"` // "maximize" or "minimize"
}
ObjectiveDefinition defines one objective in a multi-objective quality definition.
type QualityDefinition ¶ added in v0.7.2
type QualityDefinition struct {
Variables map[string]string `json:"variables"` // name -> regex pattern (one capture group)
Formula string `json:"formula,omitempty"` // single-objective formula (backward compat)
Objectives []ObjectiveDefinition `json:"objectives,omitempty"` // multi-objective: list of {formula, direction}
}
QualityDefinition is the JSON structure stored in step.quality_definition.
func ParseQualityDefinition ¶ added in v0.7.2
func ParseQualityDefinition(jsonStr string) (*QualityDefinition, error)
ParseQualityDefinition parses a JSON quality definition string.
type QualityResult ¶ added in v0.7.2
type QualityResult struct {
Vars map[string]float64 `json:"vars"`
Score float64 `json:"score"` // primary score (first objective, or single formula)
Scores []float64 `json:"scores,omitempty"` // all objective scores (multi-objective only)
}
QualityResult holds extracted variables and computed scores.
func ExtractQuality ¶ added in v0.7.2
func ExtractQuality(def *QualityDefinition, stdout, stderr string) (*QualityResult, error)
ExtractQuality runs quality variable regexes against log text and evaluates the formula. Each regex captures the LAST match (for iterative programs that output metrics repeatedly).
type StepAgg ¶
type StepAgg struct {
// Task status counters
Waiting int32
Pending int32
Accepted int32
Running int32
Uploading int32
Succeeded int32
Failed int32
ReallyFailed int32 // tasks that have exhausted all retries and are failed
Retrying int32 // tasks currently being retried
RetryingSet map[int32]bool // set of task IDs currently retrying
Total int32
// Accumulators for durations
Download Accumulator
Upload Accumulator
SuccessRun Accumulator
FailRun Accumulator
// RunningTasks maps taskID to start time
RunningTasks map[int32]time.Time
// Time bounds (epoch seconds)
StartTime *int64
EndTime *int64
}
StepAgg aggregates statistics for a step within a workflow.
func NewStepAgg ¶
func NewStepAgg() *StepAgg
NewStepAgg creates a new StepAgg with sensible defaults.
type StepStatsAgg ¶
type StepStatsAgg struct {
// contains filtered or unexported fields
}
StepStatsAgg holds in-memory aggregated statistics for steps in workflows.
func NewStepStatsAgg ¶
func NewStepStatsAgg(db *sql.DB) (*StepStatsAgg, error)
NewStepStatsAgg creates a new StepStatsAgg with initialized internal maps.
func (*StepStatsAgg) Adjust ¶ added in v0.7.5
func (a *StepStatsAgg) Adjust(workflowID, stepID int32, fn func(*StepAgg))
Adjust runs fn with the StepAgg for (workflowID, stepID) while holding the aggregator mutex. It lazily creates any missing map entries so call sites don't need to care about initialisation order. This is the canonical way to mutate counters from anywhere outside UpdateTaskStatus; using it eliminates the two classes of drift caused by unlocked direct map access:
- races with concurrent writers (int32 increments are not atomic)
- updates to a StepAgg that Reconcile has just orphaned by swapping data
func (*StepStatsAgg) EnsureStep ¶
func (a *StepStatsAgg) EnsureStep(workflowID, stepID int32)
EnsureStep guarantees an entry exists for (workflowID, stepID).
func (*StepStatsAgg) Reconcile ¶ added in v0.7.2
func (a *StepStatsAgg) Reconcile(db *sql.DB)
Reconcile rebuilds the in-memory stats from the database, fixing any drift between cached counters and actual task states.
func (*StepStatsAgg) RemoveStep ¶
func (a *StepStatsAgg) RemoveStep(stepID int32) int32
RemoveStep deletes a step entry by its stepID, scanning all workflows to locate it.
func (*StepStatsAgg) RemoveWorkflow ¶
func (a *StepStatsAgg) RemoveWorkflow(workflowID int32)