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
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
|
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 static string _jobInstanceDir = Path.Combine(CommonPaths.AgentRoot, "instance");
private readonly MarlinApiClient _apiClient;
private readonly IGitClient _gitClient;
private readonly DispatchJob _job;
private readonly string _checkoutPath;
private GrpcChannel _channel;
private IInstanceService _instanceService;
private Channel<LogEventRequest> _logQueue;
public JobUnitOfWork(DispatchJob job, MarlinApiClient apiClient, IGitClient gitClient)
{
this._job = job;
this._apiClient = apiClient;
this._gitClient = gitClient;
this._checkoutPath = Path.Combine(_jobInstanceDir, job.Id.ToString("N"));
}
public async Task Run()
{
this._channel = this._apiClient.CreateJobScopedRpc(this._job);
this._instanceService = this._channel.CreateGrpcService<IInstanceService>();
this._logQueue = Channel.CreateUnbounded<LogEventRequest>();
var drainTask = Task.Run(this.DrainLogEvents);
await StatusUpdate(JobStatus.Active);
await this.CheckoutSource();
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 CheckoutSource()
{
Directory.CreateDirectory(_jobInstanceDir);
if (Path.Exists(this._checkoutPath))
Directory.Delete(this._checkoutPath, true);
await this._gitClient.Checkout(this._checkoutPath, this._job.Vcs, this._job.Ref, this._job.Commit);
}
private async Task DrainLogEvents()
{
await foreach (var evt in this._logQueue.Reader.ReadAllAsync())
await this._instanceService.SendLogEvent(evt);
}
private async Task<ExecutionPlan> BuildPlan()
{
WorkflowBuilder builder = new();
builder.LoadFromDirectory(Path.Join(this._checkoutPath, ".marlin"));
Workflow workflow = builder.Build();
Recipe recipe = workflow.Recipes[this._job.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,
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 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 });
}
}
|
