1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
|
namespace marlin.agents.shell;
using System.Threading.Channels;
using core.contracts;
using core.functional;
using core.models;
using core.models.meta;
using core.models.plan;
using core.util;
using Grpc.Net.Client;
using ProtoBuf.Grpc.Client;
public class JobUnitOfWork
{
private readonly IInstanceService _instanceService;
private readonly Channel<LogEventRequest> _logQueue;
protected readonly string _workingDir;
protected readonly string _recipe;
public JobUnitOfWork(IInstanceService instanceService, string workingDir, string recipe)
{
this._instanceService = instanceService;
this._logQueue = Channel.CreateUnbounded<LogEventRequest>();
this._workingDir = workingDir;
this._recipe = recipe;
}
public async Task Run()
{
var drainTask = Task.Run(this.DrainLogEvents);
await StatusUpdate(JobStatus.Active);
var plan = await this.BuildPlan();
var success = await this.ExecutePlan(plan);
await StatusUpdate(success ? JobStatus.Success : JobStatus.Failed);
this._logQueue.Writer.Complete();
await drainTask;
}
private async Task<ExecutionPlan> BuildPlan()
{
WorkflowBuilder builder = new();
builder.LoadFromDirectory(Path.Join(this._workingDir, ".marlin"));
Workflow workflow = builder.Build();
Recipe recipe = workflow.Recipes[this._recipe];
ExecutionPlan plan = ExecutionPlan.Create(workflow, recipe);
return plan;
}
private async Task<bool> ExecutePlan(ExecutionPlan plan)
{
Dictionary<string, CommandResult> results = new();
foreach (var component in plan.Components)
{
if (component.ResultSelector != null)
{
if (!results.ContainsKey(component.ResultSelector.Command.Id))
// prerequisite step did not run
continue;
var storedResult = results[component.ResultSelector.Command.Id];
if (!storedResult.Matches(component.ResultSelector.Result))
// prerequisite step returned incorrect error code
continue;
}
Console.WriteLine(component);
Task<int> run = CommandRunner.RunAsync(
component.Command,
this._workingDir,
o => Log(o, false, component),
o => Log(o, true, component));
var result = await run;
results[component.Id] = result == 0 ? CommandResult.Success : CommandResult.Failure;
}
return results.All(x => x.Value == CommandResult.Success);
}
private async Task DrainLogEvents()
{
await foreach (var evt in this._logQueue.Reader.ReadAllAsync())
await this._instanceService.SendLogEvent(evt);
}
private void Log(string message, bool error, ExecutionCommand component)
{
this._logQueue.Writer.TryWrite(
new LogEventRequest
{
At = DateTime.UtcNow,
Error = error,
Message = message,
StepId = component.Id
});
}
private async Task StatusUpdate(JobStatus status)
{
await this._instanceService.SendStatusUpdate(new StatusUpdateRequest { Status = status });
}
}
|
