(job params.WorkflowJob)
| 404 | } |
| 405 | |
| 406 | func (r *basePoolManager) HandleWorkflowJob(job params.WorkflowJob) error { |
| 407 | ctx := garmUtil.WithSlogContext( |
| 408 | r.ctx, |
| 409 | slog.Any("action", "handle_workflow"), |
| 410 | ) |
| 411 | // Validate job ownership |
| 412 | if err := r.ValidateOwner(job); err != nil { |
| 413 | slog.ErrorContext(ctx, "failed to validate owner", "error", err) |
| 414 | return fmt.Errorf("error validating owner: %w", err) |
| 415 | } |
| 416 | |
| 417 | slog.DebugContext(ctx, "handling job", "workflow_job", job.WorkflowJob.Name, "workflow_job_id", job.WorkflowJob.ID) |
| 418 | |
| 419 | // Jobs without labels cannot be processed |
| 420 | if len(job.WorkflowJob.Labels) == 0 { |
| 421 | slog.WarnContext(ctx, "job has no labels", "workflow_job", job.WorkflowJob.Name) |
| 422 | return nil |
| 423 | } |
| 424 | |
| 425 | // Convert webhook payload to internal job format |
| 426 | jobParams, err := r.paramsWorkflowJobToParamsJob(job) |
| 427 | if err != nil { |
| 428 | slog.ErrorContext(ctx, "failed to convert job to params", "error", err) |
| 429 | return fmt.Errorf("error converting job to params: %w", err) |
| 430 | } |
| 431 | |
| 432 | // For in_progress/completed jobs, check if the runner belongs to a scale set. |
| 433 | // Scale set jobs are handled by the scale set listener, not by webhooks. |
| 434 | if job.Action == "in_progress" || job.Action == "completed" { |
| 435 | if jobParams.RunnerName != "" { |
| 436 | instance, err := r.store.GetInstance(ctx, jobParams.RunnerName) |
| 437 | if err == nil && instance.ScaleSetID != 0 { |
| 438 | slog.DebugContext(ctx, "job belongs to a scale set instance, skipping webhook processing", |
| 439 | "runner_name", util.SanitizeLogEntry(jobParams.RunnerName), |
| 440 | "scale_set_id", instance.ScaleSetID) |
| 441 | // Clean up any orphaned queued job that may have been recorded |
| 442 | // via webhook before we knew it belonged to a scale set. |
| 443 | if err := r.store.DeleteJob(ctx, jobParams.WorkflowJobID); err != nil && !errors.Is(err, runnerErrors.ErrNotFound) { |
| 444 | slog.With(slog.Any("error", err)).ErrorContext( |
| 445 | ctx, "failed to delete orphaned webhook job for scale set instance", |
| 446 | "job_id", jobParams.WorkflowJobID) |
| 447 | } |
| 448 | return nil |
| 449 | } |
| 450 | } |
| 451 | } |
| 452 | |
| 453 | // Validate and persist job to database first, then act on it. |
| 454 | if err := r.persistJobToDB(ctx, jobParams); err != nil { |
| 455 | return err |
| 456 | } |
| 457 | |
| 458 | // Process job based on action type |
| 459 | var actionErr error |
| 460 | |
| 461 | switch job.Action { |
| 462 | case "queued": |
| 463 | // Queued jobs are just recorded; they'll be picked up by consumeQueuedJobs() |
nothing calls this directly
no test coverage detected