engine.go
Rama main · réplica de lectura del repositorio original.
// Copyright (C) 2026 Yota Hamada
// SPDX-License-Identifier: GPL-3.0-or-later
package dagu
import (
"context"
"fmt"
"log/slog"
"maps"
"os"
"path/filepath"
"slices"
"time"
"github.com/dagucloud/dagu/v2/internal/cmn/config"
iengine "github.com/dagucloud/dagu/v2/internal/engine"
"github.com/dagucloud/dagu/v2/internal/persis"
"github.com/dagucloud/dagu/v2/internal/persis/file"
filematerialization "github.com/dagucloud/dagu/v2/internal/persis/file/materialization"
"github.com/dagucloud/dagu/v2/internal/persis/store"
_ "github.com/dagucloud/dagu/v2/internal/runtime/builtin" // Register built-in executors for embedded use.
)
// ExecutionMode controls how a DAG run is dispatched.
type ExecutionMode string
const (
// ExecutionModeLocal runs the DAG in the current process.
ExecutionModeLocal ExecutionMode = "local"
// ExecutionModeDistributed dispatches the DAG to configured coordinators.
ExecutionModeDistributed ExecutionMode = "distributed"
)
// Options configures an embedded Dagu engine.
type Options struct {
// HomeDir is the Dagu application home used for default config and data paths.
HomeDir string
// ConfigFile loads Dagu configuration from an explicit config file.
ConfigFile string
// DAGsDir overrides the directory used to resolve named DAGs and sub-DAGs.
DAGsDir string
// DataDir overrides the file-backed state directory.
DataDir string
// LogDir overrides the run log directory.
LogDir string
// ArtifactDir overrides the artifact directory.
ArtifactDir string
// BaseConfig points at a base configuration file applied during DAG loading.
BaseConfig string
// Logger receives embedded engine logs. A quiet logger is used when nil.
Logger *slog.Logger
// DefaultMode is used when a run does not set WithMode.
DefaultMode ExecutionMode
// Distributed configures dispatch and worker clients.
Distributed *DistributedOptions
}
// DistributedOptions configures distributed execution.
type DistributedOptions struct {
// Coordinators are coordinator gRPC addresses.
Coordinators []string
// TLS configures coordinator client TLS.
TLS TLSOptions
// WorkerSelector constrains distributed runs to matching workers.
WorkerSelector map[string]string
// PollInterval controls distributed run status polling.
PollInterval time.Duration
// MaxStatusErrors is the number of consecutive status failures before Wait fails.
MaxStatusErrors int
}
// TLSOptions configures TLS for coordinator and worker peer clients.
type TLSOptions struct {
// Insecure explicitly allows plaintext coordinator connections.
Insecure bool
// CertFile is the client certificate file for TLS connections.
CertFile string
// KeyFile is the client private key file for TLS connections.
KeyFile string
// ClientCAFile is the CA file used to verify coordinator certificates.
ClientCAFile string
// SkipTLSVerify skips coordinator certificate verification.
SkipTLSVerify bool
}
// RunRef identifies a DAG run.
type RunRef struct {
Name string
ID string
}
// Status is a stable snapshot of a DAG run.
type Status struct {
Name string
RunID string
AttemptID string
Status string
StartedAt time.Time
FinishedAt time.Time
Error string
LogFile string
ArchiveDir string
WorkerID string
TriggerType string
}
// WorkerOptions configures an embedded distributed worker.
type WorkerOptions struct {
// ID is the worker identifier. A host and process based ID is generated when empty.
ID string
// MaxActiveRuns limits concurrent DAG runs. A default is used when zero or negative.
MaxActiveRuns int
// Labels are advertised to coordinators and matched by worker selectors.
Labels map[string]string
// Coordinators overrides DistributedOptions.Coordinators when non-empty.
// If empty, the worker falls back to the engine-level DistributedOptions.
// The resolved coordinator list must contain at least one non-empty address.
Coordinators []string
// TLS overrides DistributedOptions.TLS when non-zero. If zero, the worker
// falls back to the engine-level DistributedOptions TLS settings.
TLS TLSOptions
// HealthPort starts the worker health endpoint on the given port. Zero disables it.
HealthPort int
}
// Engine is an embedded Dagu engine backed by the configured file stores.
type Engine struct {
inner *iengine.Engine
}
// Run is a handle for an asynchronous DAG run.
type Run struct {
inner *iengine.Run
}
// Worker is a distributed worker connected to configured coordinators.
type Worker struct {
inner *iengine.Worker
}
// RunOption customizes a single DAG run.
type RunOption func(*runOptions)
type runOptions struct {
runID string
name string
params map[string]string
paramsList []string
defaultWorkingDir string
mode ExecutionMode
workerSelector map[string]string
labels []string
dryRun bool
noReuse bool
}
// New creates an embedded Dagu engine.
func New(ctx context.Context, opts Options) (*Engine, error) {
inner, err := iengine.New(ctx, internalOptions(opts))
if err != nil {
return nil, err
}
return &Engine{inner: inner}, nil
}
// Close releases engine resources.
func (e *Engine) Close(ctx context.Context) error {
if e == nil || e.inner == nil {
return nil
}
return e.inner.Close(ctx)
}
// RunFile loads a DAG definition from a file and starts it asynchronously.
func (e *Engine) RunFile(ctx context.Context, path string, opts ...RunOption) (*Run, error) {
if e == nil || e.inner == nil {
return nil, fmt.Errorf("engine is not initialized")
}
runOpts := applyRunOptions(opts)
inner, err := e.inner.RunFile(ctx, path, internalRunOptions(runOpts))
if err != nil {
return nil, err
}
return &Run{inner: inner}, nil
}
// RunYAML loads a DAG definition from YAML bytes and starts it asynchronously.
func (e *Engine) RunYAML(ctx context.Context, yaml []byte, opts ...RunOption) (*Run, error) {
if e == nil || e.inner == nil {
return nil, fmt.Errorf("engine is not initialized")
}
runOpts := applyRunOptions(opts)
inner, err := e.inner.RunYAML(ctx, yaml, internalRunOptions(runOpts))
if err != nil {
return nil, err
}
return &Run{inner: inner}, nil
}
// Status reads the latest status for a local DAG run.
func (e *Engine) Status(ctx context.Context, ref RunRef) (*Status, error) {
if e == nil || e.inner == nil {
return nil, fmt.Errorf("engine is not initialized")
}
status, err := e.inner.Status(ctx, internalRunRef(ref))
if err != nil {
return nil, err
}
return publicStatus(status), nil
}
// Outputs reads the collected step outputs for a local DAG run.
func (e *Engine) Outputs(ctx context.Context, ref RunRef) (map[string]string, error) {
if e == nil || e.inner == nil {
return nil, fmt.Errorf("engine is not initialized")
}
return e.inner.Outputs(ctx, internalRunRef(ref))
}
// Stop requests cancellation for a local DAG run.
func (e *Engine) Stop(ctx context.Context, ref RunRef) error {
if e == nil || e.inner == nil {
return fmt.Errorf("engine is not initialized")
}
return e.inner.Stop(ctx, internalRunRef(ref))
}
// NewWorker creates an embedded distributed worker.
func (e *Engine) NewWorker(opts WorkerOptions) (*Worker, error) {
if e == nil || e.inner == nil {
return nil, fmt.Errorf("engine is not initialized")
}
inner, err := e.inner.NewWorker(internalWorkerOptions(opts))
if err != nil {
return nil, err
}
return &Worker{inner: inner}, nil
}
// Ref returns the run reference.
func (r *Run) Ref() RunRef {
if r == nil || r.inner == nil {
return RunRef{}
}
return publicRunRef(r.inner.Ref())
}
// ID returns the DAG run ID.
func (r *Run) ID() string {
if r == nil || r.inner == nil {
return ""
}
return r.inner.ID()
}
// Name returns the DAG name.
func (r *Run) Name() string {
if r == nil || r.inner == nil {
return ""
}
return r.inner.Name()
}
// Wait blocks until the DAG run reaches a terminal state or ctx is canceled.
func (r *Run) Wait(ctx context.Context) (*Status, error) {
if r == nil || r.inner == nil {
return nil, fmt.Errorf("run is not initialized")
}
status, err := r.inner.Wait(ctx)
return publicStatus(status), err
}
// Status returns the current run status.
func (r *Run) Status(ctx context.Context) (*Status, error) {
if r == nil || r.inner == nil {
return nil, fmt.Errorf("run is not initialized")
}
status, err := r.inner.Status(ctx)
return publicStatus(status), err
}
// Outputs reads the collected step outputs for this run.
func (r *Run) Outputs(ctx context.Context) (map[string]string, error) {
if r == nil || r.inner == nil {
return nil, fmt.Errorf("run is not initialized")
}
return r.inner.Outputs(ctx)
}
// Stop requests cancellation for this run.
func (r *Run) Stop(ctx context.Context) error {
if r == nil || r.inner == nil {
return fmt.Errorf("run is not initialized")
}
return r.inner.Stop(ctx)
}
// Start registers and starts the worker. It blocks until ctx is canceled or the
// worker exits with an error.
func (w *Worker) Start(ctx context.Context) error {
if w == nil || w.inner == nil {
return fmt.Errorf("worker is not initialized")
}
return w.inner.Start(ctx)
}
// Stop stops polling for tasks and drains the running DAG runs. Each run gets
// SIGTERM and Stop blocks until it finishes its cleanup window
// (max_clean_up_time_sec) and lifecycle handlers, so Stop can take that long
// plus handler time. The ctx deadline bounds the wait: when it is reached, Stop
// cancels the remaining runs, allows them up to 10 seconds to exit and report
// their final status, then up to 5 seconds to release its resources. Without a
// deadline, Stop waits for every run.
func (w *Worker) Stop(ctx context.Context) error {
if w == nil || w.inner == nil {
return nil
}
return w.inner.Stop(ctx)
}
// WaitReady blocks until the worker has registered with a coordinator.
func (w *Worker) WaitReady(ctx context.Context) error {
if w == nil || w.inner == nil {
return fmt.Errorf("worker is not initialized")
}
return w.inner.WaitReady(ctx)
}
func applyRunOptions(opts []RunOption) runOptions {
var runOpts runOptions
for _, opt := range opts {
opt(&runOpts)
}
return runOpts
}
// WithRunID sets an explicit DAG run ID.
func WithRunID(id string) RunOption {
return func(o *runOptions) {
o.runID = id
}
}
// WithName overrides the loaded DAG name.
func WithName(name string) RunOption {
return func(o *runOptions) {
o.name = name
}
}
// WithParams sets DAG parameters from a key-value map.
func WithParams(params map[string]string) RunOption {
return func(o *runOptions) {
o.params = cloneMap(params)
}
}
// WithParamsList sets DAG parameters from Dagu-style KEY=VALUE entries.
func WithParamsList(params []string) RunOption {
return func(o *runOptions) {
o.paramsList = cloneSlice(params)
}
}
// WithDefaultWorkingDir sets the default working directory while loading a DAG.
func WithDefaultWorkingDir(dir string) RunOption {
return func(o *runOptions) {
o.defaultWorkingDir = dir
}
}
// WithMode overrides the engine default execution mode.
func WithMode(mode ExecutionMode) RunOption {
return func(o *runOptions) {
o.mode = mode
}
}
// WithWorkerSelector sets the distributed worker selector for one run.
func WithWorkerSelector(selector map[string]string) RunOption {
return func(o *runOptions) {
o.workerSelector = cloneMap(selector)
}
}
// WithLabels adds labels to one run.
func WithLabels(labels ...string) RunOption {
return func(o *runOptions) {
o.labels = cloneSlice(labels)
}
}
// WithTags adds labels to one run.
//
// Deprecated: use WithLabels.
func WithTags(tags ...string) RunOption {
return WithLabels(tags...)
}
// WithDryRun enables or disables dry-run mode.
func WithDryRun(enabled bool) RunOption {
return func(o *runOptions) {
o.dryRun = enabled
}
}
// WithNoReuse forces build steps to recompute for this run.
func WithNoReuse(enabled bool) RunOption {
return func(o *runOptions) {
o.noReuse = enabled
}
}
func internalOptions(opts Options) iengine.Options {
out := iengine.Options{
HomeDir: opts.HomeDir,
ConfigFile: opts.ConfigFile,
DAGsDir: opts.DAGsDir,
DataDir: opts.DataDir,
LogDir: opts.LogDir,
ArtifactDir: opts.ArtifactDir,
BaseConfig: opts.BaseConfig,
Logger: opts.Logger,
PersistenceFactory: filePersistenceFactory,
DefaultMode: iengine.ExecutionMode(opts.DefaultMode),
}
if opts.Distributed != nil {
distributed := internalDistributedOptions(*opts.Distributed)
out.Distributed = &distributed
}
return out
}
func filePersistenceFactory(ctx context.Context, cfg *config.Config) (iengine.Persistence, error) {
if err := os.MkdirAll(cfg.Paths.DataDir, 0o750); err != nil {
return iengine.Persistence{}, fmt.Errorf("create data directory: %w", err)
}
if err := os.MkdirAll(cfg.Paths.DAGStateDir, 0o750); err != nil {
return iengine.Persistence{}, fmt.Errorf("create DAG state directory: %w", err)
}
backend := file.NewBackend(cfg.Paths)
procRepository := file.NewProcRepository(cfg)
dagRepository, err := fileEngineDAGRepository(ctx, cfg, iengine.DAGRepositoryFactoryOptions{})
if err != nil {
return iengine.Persistence{}, err
}
return iengine.Persistence{
DAGRepository: dagRepository,
DAGRunRepository: file.NewDAGRunRepository(cfg, file.WithDAGRunLatestStatusToday(false)),
ProcRepository: procRepository,
StateStore: store.NewDAGStateStore(backend.Collection(persis.CollectionDAGState)),
ServiceRegistry: file.NewServiceRegistry(cfg),
DAGRepositoryFactory: fileEngineDAGRepository,
RuntimeStoresFactory: func(ctx context.Context, cfg *config.Config) iengine.RuntimeStores {
return iengine.RuntimeStores{
SecretStore: file.NewSecretStore(ctx, cfg, backend.Collection(persis.CollectionSecrets)),
ProfileStore: file.NewProfileStore(ctx, cfg, backend.Collection(persis.CollectionProfiles)),
MaterializationStore: filematerialization.New(filepath.Join(cfg.Paths.DataDir, "materializations")),
}
},
}, nil
}
func fileEngineDAGRepository(_ context.Context, cfg *config.Config, opts iengine.DAGRepositoryFactoryOptions) (*persis.DAGRepository, error) {
var fileOpts []file.DAGRepositoryOption
if len(opts.SearchPaths) > 0 {
fileOpts = append(fileOpts, file.WithDAGSearchPaths(opts.SearchPaths))
}
return file.NewDAGRepository(cfg, fileOpts...)
}
func internalDistributedOptions(opts DistributedOptions) iengine.DistributedOptions {
return iengine.DistributedOptions{
Coordinators: cloneSlice(opts.Coordinators),
TLS: internalTLSOptions(opts.TLS),
WorkerSelector: cloneMap(opts.WorkerSelector),
PollInterval: opts.PollInterval,
MaxStatusErrors: opts.MaxStatusErrors,
}
}
func internalTLSOptions(opts TLSOptions) iengine.TLSOptions {
return iengine.TLSOptions{
Insecure: opts.Insecure,
CertFile: opts.CertFile,
KeyFile: opts.KeyFile,
ClientCAFile: opts.ClientCAFile,
SkipTLSVerify: opts.SkipTLSVerify,
}
}
func internalRunOptions(opts runOptions) iengine.RunOptions {
return iengine.RunOptions{
RunID: opts.runID,
Name: opts.name,
Params: opts.params,
ParamsList: opts.paramsList,
DefaultWorkingDir: opts.defaultWorkingDir,
Mode: iengine.ExecutionMode(opts.mode),
WorkerSelector: opts.workerSelector,
Labels: opts.labels,
DryRun: opts.dryRun,
NoReuse: opts.noReuse,
}
}
func internalRunRef(ref RunRef) iengine.RunRef {
return iengine.RunRef{Name: ref.Name, ID: ref.ID}
}
func publicRunRef(ref iengine.RunRef) RunRef {
return RunRef{Name: ref.Name, ID: ref.ID}
}
func internalWorkerOptions(opts WorkerOptions) iengine.WorkerOptions {
return iengine.WorkerOptions{
ID: opts.ID,
MaxActiveRuns: opts.MaxActiveRuns,
Labels: cloneMap(opts.Labels),
Coordinators: cloneSlice(opts.Coordinators),
TLS: internalTLSOptions(opts.TLS),
HealthPort: opts.HealthPort,
}
}
func publicStatus(status *iengine.Status) *Status {
if status == nil {
return nil
}
return &Status{
Name: status.Name,
RunID: status.RunID,
AttemptID: status.AttemptID,
Status: status.Status,
StartedAt: status.StartedAt,
FinishedAt: status.FinishedAt,
Error: status.Error,
LogFile: status.LogFile,
ArchiveDir: status.ArchiveDir,
WorkerID: status.WorkerID,
TriggerType: status.TriggerType,
}
}
func cloneMap(values map[string]string) map[string]string {
if len(values) == 0 {
return nil
}
out := make(map[string]string, len(values))
maps.Copy(out, values)
return out
}
func cloneSlice(values []string) []string {
if len(values) == 0 {
return nil
}
return slices.Clone(values)
}