Documentation
¶
Index ¶
- Constants
- func CanTransition(from string, to string) bool
- func EnsureTransition(from string, to string) error
- func IsSuccessStatus(status string) bool
- func IsTerminalStatus(status string) bool
- func SortNodesByStartedAt(nodes []*types.AnalysisNode)
- type DagRuntimeEventNotifier
- type DagScheduler
- type NodeCompletionCoordinator
- type NodeContainerOperator
- type NodeDispatcher
- type NodeFailureCleanupFunc
- type RunningEntry
- type RunningRegistry
- func (r *RunningRegistry) IsRunning(analysisID int64) bool
- func (r *RunningRegistry) IsStopping(analysisID int64) bool
- func (r *RunningRegistry) MarkFinished(analysisID int64, status string)
- func (r *RunningRegistry) Register(entry *RunningEntry)
- func (r *RunningRegistry) RequestStop(analysisID int64) bool
- type RuntimeEngine
- func (e *RuntimeEngine) ClaimNextReadyNode(ctx context.Context, analysisID int64) (*types.AnalysisNode, error)
- func (e *RuntimeEngine) CompleteNode(ctx context.Context, analysisNodeID int64, status string, ...) (*types.AnalysisNode, error)
- func (e *RuntimeEngine) GetSnapshot(ctx context.Context, analysisID int64) (*RuntimeSnapshot, error)
- func (e *RuntimeEngine) MarkNodeRunning(ctx context.Context, analysisNodeID int64) (*types.AnalysisNode, error)
- func (e *RuntimeEngine) RefreshReadyStatus(ctx context.Context, analysisID int64) error
- type RuntimeEvent
- type RuntimeSnapshot
- type SchedulerConfig
- type SchedulerResult
- type WorkerPool
Constants ¶
View Source
const ( EventDagStarted = "dag.started" EventDagCompleted = "dag.completed" EventDagFailed = "dag.failed" EventNodeSubmitted = "dag.node.submitted" EventNodeRunning = "dag.node.running" EventNodeCompleted = "dag.node.completed" EventNodeFailed = "dag.node.failed" EventNodeStateChange = "dag.node.state_changed" )
View Source
const ( StatusPending = "pending" StatusReady = "ready" StatusSubmitted = "submitted" StatusRunning = "running" StatusStopping = "stopping" StatusStopped = "stopped" StatusDone = "done" StatusFailed = "failed" StatusCached = "cached" StatusSkipped = "skipped" )
Variables ¶
This section is empty.
Functions ¶
func CanTransition ¶
func EnsureTransition ¶
func IsSuccessStatus ¶
func IsTerminalStatus ¶
func SortNodesByStartedAt ¶
func SortNodesByStartedAt(nodes []*types.AnalysisNode)
Types ¶
type DagRuntimeEventNotifier ¶
type DagRuntimeEventNotifier struct {
// contains filtered or unexported fields
}
DagRuntimeEventNotifier subscribes DAG runtime events and pushes UI action messages to project users through the unified realtime hub.
func NewDagRuntimeEventNotifier ¶
func NewDagRuntimeEventNotifier(db *gorm.DB, projectRepo interfaces.ProjectRepository, analysisRepo interfaces.AnalysisRepository, hub *realtime.Hub) *DagRuntimeEventNotifier
func (*DagRuntimeEventNotifier) Handle ¶
func (n *DagRuntimeEventNotifier) Handle(evt event.Event)
type DagScheduler ¶
type DagScheduler struct {
// contains filtered or unexported fields
}
func NewDagScheduler ¶
func NewDagScheduler(analysisID int64, runtime *RuntimeEngine, dispatcher *NodeDispatcher, bus event.Bus, cfg SchedulerConfig) *DagScheduler
func (*DagScheduler) Run ¶
func (s *DagScheduler) Run(ctx context.Context) (*SchedulerResult, error)
type NodeCompletionCoordinator ¶
type NodeCompletionCoordinator struct {
// contains filtered or unexported fields
}
func NewNodeCompletionCoordinator ¶
func NewNodeCompletionCoordinator( analysisRepo interfaces.AnalysisRepository, containerRepo interfaces.ContainerRepository, containerOps NodeContainerOperator, bus event.Bus, cfg *config.Config, ) *NodeCompletionCoordinator
func (*NodeCompletionCoordinator) Handle ¶
func (c *NodeCompletionCoordinator) Handle(evt event.Event)
type NodeContainerOperator ¶
type NodeDispatcher ¶
type NodeDispatcher struct {
// contains filtered or unexported fields
}
func NewNodeDispatcher ¶
func NewNodeDispatcher( repo interfaces.AnalysisRepository, bus event.Bus, factory *executor.ExecuterFactory, workflowRepo interfaces.WorkflowRepository, projectRepo interfaces.ProjectRepository, workflowService interfaces.WorkflowService, cfg *config.Config, runScriptBuilders map[string]prepare.RunScriptBuilder, ) *NodeDispatcher
func (*NodeDispatcher) Dispatch ¶
func (d *NodeDispatcher) Dispatch(ctx context.Context, analysisNodeID int64) error
func (*NodeDispatcher) Stop ¶
func (d *NodeDispatcher) Stop(ctx context.Context, node *types.AnalysisNode) (*executor.Result, error)
type NodeFailureCleanupFunc ¶
type NodeFailureCleanupFunc func(ctx context.Context, node *types.AnalysisNode)
type RunningEntry ¶
type RunningRegistry ¶
type RunningRegistry struct {
// contains filtered or unexported fields
}
func NewRunningRegistry ¶
func NewRunningRegistry() *RunningRegistry
func (*RunningRegistry) IsRunning ¶
func (r *RunningRegistry) IsRunning(analysisID int64) bool
func (*RunningRegistry) IsStopping ¶
func (r *RunningRegistry) IsStopping(analysisID int64) bool
func (*RunningRegistry) MarkFinished ¶
func (r *RunningRegistry) MarkFinished(analysisID int64, status string)
func (*RunningRegistry) Register ¶
func (r *RunningRegistry) Register(entry *RunningEntry)
func (*RunningRegistry) RequestStop ¶
func (r *RunningRegistry) RequestStop(analysisID int64) bool
type RuntimeEngine ¶
type RuntimeEngine struct {
// contains filtered or unexported fields
}
func NewRuntimeEngine ¶
func NewRuntimeEngine(repo interfaces.AnalysisRepository) *RuntimeEngine
func (*RuntimeEngine) ClaimNextReadyNode ¶
func (e *RuntimeEngine) ClaimNextReadyNode(ctx context.Context, analysisID int64) (*types.AnalysisNode, error)
func (*RuntimeEngine) CompleteNode ¶
func (*RuntimeEngine) GetSnapshot ¶
func (e *RuntimeEngine) GetSnapshot(ctx context.Context, analysisID int64) (*RuntimeSnapshot, error)
func (*RuntimeEngine) MarkNodeRunning ¶
func (e *RuntimeEngine) MarkNodeRunning(ctx context.Context, analysisNodeID int64) (*types.AnalysisNode, error)
func (*RuntimeEngine) RefreshReadyStatus ¶
func (e *RuntimeEngine) RefreshReadyStatus(ctx context.Context, analysisID int64) error
type RuntimeEvent ¶
type RuntimeEvent struct {
Name string `json:"name"`
AnalysisID int64 `json:"analysis_id,string"`
AnalysisNodeID int64 `json:"analysis_node_id,string,omitempty"`
NodeID string `json:"node_id,omitempty"`
OccurredAt time.Time `json:"occurred_at"`
Payload map[string]interface{} `json:"payload,omitempty"`
}
type RuntimeSnapshot ¶
type RuntimeSnapshot struct {
AnalysisID int64 `json:"analysis_id,string"`
TotalNodes int `json:"total_nodes"`
StatusCount map[string]int `json:"status_count"`
CompletedCount int `json:"completed_count"`
CompletionPercent int `json:"completion_percent"`
ReadyCount int `json:"ready_count"`
RunningCount int `json:"running_count"`
IsFinished bool `json:"is_finished"`
}
type SchedulerConfig ¶
type SchedulerResult ¶
type SchedulerResult struct {
AnalysisID int64 `json:"analysis_id"`
SubmittedCount int `json:"submitted_count"`
FailedToSubmitCount int `json:"failed_to_submit_count"`
TimedOut bool `json:"timed_out"`
Snapshot *RuntimeSnapshot `json:"snapshot"`
}
type WorkerPool ¶
type WorkerPool struct {
// contains filtered or unexported fields
}
func NewWorkerPool ¶
func NewWorkerPool(dispatcher *NodeDispatcher, workers int, queueSize int) *WorkerPool
func (*WorkerPool) Enqueue ¶
func (p *WorkerPool) Enqueue(nodeID int64) bool
func (*WorkerPool) QueueLen ¶
func (p *WorkerPool) QueueLen() int
func (*WorkerPool) Start ¶
func (p *WorkerPool) Start(ctx context.Context)
func (*WorkerPool) Stop ¶
func (p *WorkerPool) Stop()
Source Files
¶
Click to show internal directories.
Click to hide internal directories.