From ecd5f5d13936dd097b12766cb95a568ad5b073a9 Mon Sep 17 00:00:00 2001 From: Melodi <62381002+Melodi17@users.noreply.github.com> Date: Tue, 27 Oct 2026 23:04:59 +1000 Subject: feat: lots of stuff --- src/marlin.agents.shell/CommandRunner.cs | 38 ++++++ src/marlin.agents.shell/JobUnitOfWork.cs | 130 +++++++++++++++++++++ src/marlin.agents.shell/MarlinApiClient.cs | 33 +++++- src/marlin.agents.shell/Program.cs | 27 ++++- src/marlin.agents.shell/marlin.agents.shell.csproj | 2 + 5 files changed, 226 insertions(+), 4 deletions(-) create mode 100644 src/marlin.agents.shell/CommandRunner.cs create mode 100644 src/marlin.agents.shell/JobUnitOfWork.cs (limited to 'src/marlin.agents.shell') diff --git a/src/marlin.agents.shell/CommandRunner.cs b/src/marlin.agents.shell/CommandRunner.cs new file mode 100644 index 0000000..d4fcbb3 --- /dev/null +++ b/src/marlin.agents.shell/CommandRunner.cs @@ -0,0 +1,38 @@ +using System.Diagnostics; + +public static class CommandRunner +{ + public static async Task RunAsync(string command, Action onStdout, Action onStderr) + { + var psi = new ProcessStartInfo + { + FileName = GetShell(), + RedirectStandardInput = true, + RedirectStandardOutput = true, + RedirectStandardError = true, + UseShellExecute = false, + CreateNoWindow = true + }; + + var process = new Process { StartInfo = psi }; + process.OutputDataReceived += (_, e) => { if (e.Data != null) onStdout(e.Data); }; + process.ErrorDataReceived += (_, e) => { if (e.Data != null) onStderr(e.Data); }; + + process.Start(); + process.BeginOutputReadLine(); + process.BeginErrorReadLine(); + + // Write the command to stdin and close it + await process.StandardInput.WriteLineAsync(command); + await process.StandardInput.WriteLineAsync("exit $?"); // Exit with the command's exit code + process.StandardInput.Close(); + + await process.WaitForExitAsync(); + return process.ExitCode; + } + + private static string GetShell() + { + return OperatingSystem.IsWindows() ? "cmd.exe" : "/bin/bash"; + } +} \ No newline at end of file diff --git a/src/marlin.agents.shell/JobUnitOfWork.cs b/src/marlin.agents.shell/JobUnitOfWork.cs new file mode 100644 index 0000000..85df806 --- /dev/null +++ b/src/marlin.agents.shell/JobUnitOfWork.cs @@ -0,0 +1,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 _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(); + this._logQueue = Channel.CreateUnbounded(); + + 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 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 ExecutePlan(ExecutionPlan plan) + { + Dictionary 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 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 }); + } +} \ No newline at end of file diff --git a/src/marlin.agents.shell/MarlinApiClient.cs b/src/marlin.agents.shell/MarlinApiClient.cs index aa4f013..900101c 100644 --- a/src/marlin.agents.shell/MarlinApiClient.cs +++ b/src/marlin.agents.shell/MarlinApiClient.cs @@ -1,16 +1,28 @@ namespace marlin.agents.shell; using System.Net.Http.Json; +using core.contracts; using core.models; +using Grpc.Core; +using Grpc.Net.Client; +using ProtoBuf.Grpc.Client; public class MarlinApiClient { + private readonly string _baseUrl; + private readonly Guid _agentId; + private readonly string _agentToken; + private readonly HttpClient _http; public MarlinApiClient(string baseUrl, Guid agentId, string agentToken) { - _http = new HttpClient { BaseAddress = new Uri(baseUrl) }; - _http.DefaultRequestHeaders.Add("Agent", $"{agentId} {agentToken}"); + this._baseUrl = baseUrl; + this._agentId = agentId; + this._agentToken = agentToken; + + _http = new HttpClient { BaseAddress = new Uri(this._baseUrl) }; + _http.DefaultRequestHeaders.Add("Agent", $"{this._agentId} {this._agentToken}"); } public static MarlinApiClient FromEnvironment() @@ -35,4 +47,21 @@ public class MarlinApiClient public async Task QueueNextJob(CancellationToken ct = default) => await _http.GetFromJsonAsync("api/dispatch/queue", ct); + + public GrpcChannel CreateJobScopedRpc(DispatchJob job) + { + var credentials = CallCredentials.FromInterceptor( + async (_, metadata) => + { + metadata.Add("Agent", $"{this._agentId} {this._agentToken}"); + metadata.Add("Job", $"{job.Id} {job.Token}"); + }); + + var channel = GrpcChannel.ForAddress(this._baseUrl, new GrpcChannelOptions + { + Credentials = ChannelCredentials.Create(new SslCredentials(), credentials) + }); + + return channel; + } } diff --git a/src/marlin.agents.shell/Program.cs b/src/marlin.agents.shell/Program.cs index 02a0bc9..75ea8ad 100644 --- a/src/marlin.agents.shell/Program.cs +++ b/src/marlin.agents.shell/Program.cs @@ -1,9 +1,20 @@ using System.Runtime.InteropServices; +using System.Threading.Channels; +using Grpc.Net.Client; using marlin.agents.shell; +using marlin.core.contracts; +using marlin.core.functional; using marlin.core.models; +using marlin.core.models.meta; +using marlin.core.models.plan; +using marlin.core.util; +using ProtoBuf.Grpc.Client; DotNetEnv.Env.NoClobber().TraversePath().Load(); +var jobInstanceDir = Path.Combine(CommonPaths.AgentRoot, "instance"); +Directory.CreateDirectory(jobInstanceDir); + var client = MarlinApiClient.FromEnvironment(); var capabilities = new CapabilitySet(); @@ -16,5 +27,17 @@ AgentConfiguration config = await client.Handshake(status); Console.WriteLine($"IAM: {config.Name}"); Console.WriteLine($"MCJ: {config.MaxConcurrentJobs}"); -var job = await client.QueueNextJob(); -Console.WriteLine(job == null ? "No pending jobs." : $"Claimed job {job.Id}: {job.Recipe}"); +while (true) +{ + var job = await client.QueueNextJob(); + Console.WriteLine(job == null ? "No pending jobs." : $"Claimed job {job.Id}: {job.Recipe}"); + + if (job != null) + { + GitBinaryClient gitClient = new(); + var uow = new JobUnitOfWork(job, client, gitClient); + await uow.Run(); + } + + await Task.Delay(TimeSpan.FromSeconds(5)); +} \ No newline at end of file diff --git a/src/marlin.agents.shell/marlin.agents.shell.csproj b/src/marlin.agents.shell/marlin.agents.shell.csproj index ff86d63..0278fa0 100644 --- a/src/marlin.agents.shell/marlin.agents.shell.csproj +++ b/src/marlin.agents.shell/marlin.agents.shell.csproj @@ -9,6 +9,8 @@ + + -- cgit v1.3.1