Support 'rzfs:host:port/template-id/subpath' engine spec (#106)

UET/Redpoint.Uet.BuildPipeline/Executors/BuildEngineSpecification.cs

@@ -13,6 +13,7 @@ public class BuildEngineSpecification
         internal string? _gitSharedWindowsCachePath { get; private set; }
         internal string? _gitSharedMacCachePath { get; private set; }
         internal string? _sesNetworkShare { get; private set; }
+        internal string? _remoteZfs { get; private set; }
         public bool IsEngineBuild { get; private set; } = false;
 
         public static BuildEngineSpecification ForVersionWithPath(string version, string localPath)
@@ -48,6 +49,14 @@ public static BuildEngineSpecification ForSESNetworkShare(string sesNetworkShare
             };
         }
 
+        public static BuildEngineSpecification ForRemoteZfs(string remoteZfs)
+        {
+            return new BuildEngineSpecification
+            {
+                _remoteZfs = remoteZfs,
+            };
+        }
+
         [SuppressMessage("Design", "CA1054:URI-like parameters should not be strings", Justification = "Git URLs are not compatible with the Uri object.")]
         public static BuildEngineSpecification ForGitCommitWithZips(
             string uefsGitUrl,
@@ -81,6 +90,10 @@ public string ToReparsableString()
             {
                 return $"ses:{_sesNetworkShare}";
             }
+            else if (!string.IsNullOrWhiteSpace(_remoteZfs))
+            {
+                return $"rzfs:{_remoteZfs}";
+            }
             else if (!string.IsNullOrWhiteSpace(_engineVersion))
             {
                 return _engineVersion;

UET/Redpoint.Uet.BuildPipeline/Executors/Engine/DefaultEngineWorkspaceProvider.cs

@@ -64,6 +64,12 @@ public async Task<IWorkspace> GetEngineWorkspace(
                     },
                     cancellationToken).ConfigureAwait(false);
             }
+            else if (buildEngineSpecification._remoteZfs != null)
+            {
+                return await _workspaceProvider.GetWorkspaceAsync(
+                    RemoteZfsWorkspaceDescriptor.Parse(buildEngineSpecification._remoteZfs),
+                    cancellationToken).ConfigureAwait(false);
+            }
             else if (buildEngineSpecification._gitCommit != null)
             {
                 return await _workspaceProvider.GetWorkspaceAsync(

UET/Redpoint.Uet.Workspace/Descriptors/RemoteZfsWorkspaceDescriptor.cs

@@ -0,0 +1,41 @@
+namespace Redpoint.Uet.Workspace.Descriptors
+{
+    using System.Globalization;
+
+    public record class RemoteZfsWorkspaceDescriptor : IWorkspaceDescriptor
+    {
+        public required string Hostname { get; set; }
+
+        public required int Port { get; set; }
+
+        public required string TemplateId { get; set; }
+
+        public required string Subpath { get; set; }
+
+        public static RemoteZfsWorkspaceDescriptor Parse(string spec)
+        {
+            ArgumentNullException.ThrowIfNull(spec);
+
+            var components = spec.Replace('\\', '/').Split('/', 3);
+            if (components.Length < 2)
+            {
+                throw new ArgumentException("Invalid remote ZFS spec; expected 'host:port/template-id/optional-subpath'.", nameof(spec));
+            }
+
+            var hostPort = components[0].Split(':');
+            var templateId = components[1];
+            var subpath = components.Length > 2 ? components[2] : string.Empty;
+
+            var host = hostPort[0];
+            var port = hostPort.Length > 1 ? int.Parse(hostPort[1], CultureInfo.InvariantCulture) : 9000;
+
+            return new RemoteZfsWorkspaceDescriptor
+            {
+                Hostname = host,
+                Port = port,
+                TemplateId = templateId,
+                Subpath = subpath,
+            };
+        }
+    }
+}

UET/Redpoint.Uet.Workspace/Instance/RemoteZfsWorkspace.cs

@@ -0,0 +1,36 @@
+namespace Redpoint.Uet.Workspace.Instance
+{
+    using Grpc.Core;
+    using Redpoint.Reservation;
+    using Redpoint.Uet.Workspace.RemoteZfs;
+    using System.Threading.Tasks;
+
+    internal class RemoteZfsWorkspace : IWorkspace
+    {
+        private readonly AsyncServerStreamingCall<AcquireWorkspaceResponse> _connection;
+        private readonly IReservation _physicalReservation;
+
+        public RemoteZfsWorkspace(
+            AsyncServerStreamingCall<AcquireWorkspaceResponse> connection,
+            IReservation physicalReservation)
+        {
+            _connection = connection;
+            _physicalReservation = physicalReservation;
+        }
+
+        public string Path => System.IO.Path.Combine(_physicalReservation.ReservedPath, "S");
+
+        public async ValueTask DisposeAsync()
+        {
+            if (Directory.Exists(System.IO.Path.Combine(_physicalReservation.ReservedPath, "S")))
+            {
+                Directory.Delete(System.IO.Path.Combine(_physicalReservation.ReservedPath, "S"));
+            }
+
+            await _physicalReservation.DisposeAsync().ConfigureAwait(false);
+
+            _connection.Dispose();
+        }
+    }
+
+}

UET/Redpoint.Uet.Workspace/PhysicalWorkspaceProvider.cs

@@ -2,6 +2,7 @@
 {
     using Microsoft.Extensions.Logging;
     using Microsoft.Win32.SafeHandles;
+    using Redpoint.GrpcPipes;
     using Redpoint.IO;
     using Redpoint.ProcessExecution;
     using Redpoint.Reservation;
@@ -10,12 +11,15 @@
     using Redpoint.Uet.Workspace.Instance;
     using Redpoint.Uet.Workspace.ParallelCopy;
     using Redpoint.Uet.Workspace.PhysicalGit;
+    using Redpoint.Uet.Workspace.RemoteZfs;
     using Redpoint.Uet.Workspace.Reservation;
     using System;
     using System.Collections.Generic;
     using System.Linq;
+    using System.Net;
     using System.Threading;
     using System.Threading.Tasks;
+    using static Redpoint.Uet.Workspace.RemoteZfs.RemoteZfs;
 
     internal class PhysicalWorkspaceProvider : IPhysicalWorkspaceProvider
     {
@@ -26,6 +30,7 @@ internal class PhysicalWorkspaceProvider : IPhysicalWorkspaceProvider
         private readonly IPhysicalGitCheckout _physicalGitCheckout;
         private readonly IWorkspaceReservationParameterGenerator _parameterGenerator;
         private readonly IProcessExecutor _processExecutor;
+        private readonly IGrpcPipeFactory _grpcPipeFactory;
 
         public PhysicalWorkspaceProvider(
             ILogger<PhysicalWorkspaceProvider> logger,
@@ -34,7 +39,8 @@ public PhysicalWorkspaceProvider(
             IVirtualWorkspaceProvider virtualWorkspaceProvider,
             IPhysicalGitCheckout physicalGitCheckout,
             IWorkspaceReservationParameterGenerator parameterGenerator,
-            IProcessExecutor processExecutor)
+            IProcessExecutor processExecutor,
+            IGrpcPipeFactory grpcPipeFactory)
         {
             _logger = logger;
             _reservationManager = reservationManager;
@@ -43,6 +49,7 @@ public PhysicalWorkspaceProvider(
             _physicalGitCheckout = physicalGitCheckout;
             _parameterGenerator = parameterGenerator;
             _processExecutor = processExecutor;
+            _grpcPipeFactory = grpcPipeFactory;
         }
 
         public bool ProvidesFastCopyOnWrite => false;
@@ -63,6 +70,8 @@ public async Task<IWorkspace> GetWorkspaceAsync(IWorkspaceDescriptor workspaceDe
                     return await _virtualWorkspaceProvider.GetWorkspaceAsync(descriptor, cancellationToken).ConfigureAwait(false);
                 case SharedEngineSourceWorkspaceDescriptor descriptor:
                     return await AllocateSharedEngineSourceAsync(descriptor, cancellationToken).ConfigureAwait(false);
+                case RemoteZfsWorkspaceDescriptor descriptor:
+                    return await AllocateRemoteZfsAsync(descriptor, cancellationToken).ConfigureAwait(false);
                 default:
                     throw new NotSupportedException();
             }
@@ -210,5 +219,90 @@ await _processExecutor.ExecuteAsync(
 
             throw new OperationCanceledException(cancellationToken);
         }
+
+        private async Task<IWorkspace> AllocateRemoteZfsAsync(RemoteZfsWorkspaceDescriptor descriptor, CancellationToken cancellationToken)
+        {
+            if (!OperatingSystem.IsWindowsVersionAtLeast(5, 1, 2600))
+            {
+                throw new PlatformNotSupportedException();
+            }
+
+            if (!IPAddress.TryParse(descriptor.Hostname, out var address))
+            {
+                var addresses = await Dns.GetHostAddressesAsync(descriptor.Hostname, cancellationToken)
+                    .ConfigureAwait(false);
+                if (addresses.Length != 0)
+                {
+                    address = addresses[0];
+                }
+            }
+            if (address == null)
+            {
+                throw new InvalidOperationException($"Unable to resolve '{descriptor.Hostname}' to an IP address.");
+            }
+
+            _logger.LogInformation($"Resolved hostname '{descriptor.Hostname}' to address '{address}' for ZFS client.");
+
+            _logger.LogInformation($"Connecting to ZFS snapshot server...");
+            var client = _grpcPipeFactory.CreateNetworkClient(
+                new IPEndPoint(address!, descriptor.Port),
+                invoker => new RemoteZfsClient(invoker));
+
+            _logger.LogInformation($"Acquiring workspace from ZFS snapshot server...");
+            var response = client.AcquireWorkspace(new AcquireWorkspaceRequest
+            {
+                TemplateId = descriptor.TemplateId,
+            }, cancellationToken: cancellationToken);
+
+            var usingOuterReservation = false;
+            try
+            {
+                _logger.LogInformation($"Waiting for workspace to be acquired on ZFS snapshot server...");
+                var acquired = await response.ResponseStream.MoveNext(cancellationToken).ConfigureAwait(false);
+                if (!acquired)
+                {
+                    throw new InvalidOperationException($"Unable to acquire workspace on ZFS server!");
+                }
+
+                // Allocate a workspace to put our symbolic link in.
+                _logger.LogInformation($"Allocating physical workspace for symbolic link...");
+                var usingInnerReservation = false;
+                var reservation = await _reservationManager.ReserveAsync(
+                    "RemoteZfs",
+                    _parameterGenerator.ConstructReservationParameters(descriptor.TemplateId)).ConfigureAwait(false);
+                try
+                {
+                    if (Directory.Exists(Path.Combine(reservation.ReservedPath, "S")))
+                    {
+                        Directory.Delete(Path.Combine(reservation.ReservedPath, "S"));
+                    }
+
+                    var targetPath = string.IsNullOrWhiteSpace(descriptor.Subpath)
+                        ? response.ResponseStream.Current.WindowsShareRemotePath
+                        : Path.Combine(response.ResponseStream.Current.WindowsShareRemotePath, descriptor.Subpath);
+                    Directory.CreateSymbolicLink(
+                        Path.Combine(reservation.ReservedPath, "S"),
+                        targetPath);
+
+                    _logger.LogInformation($"Remote ZFS workspace '{targetPath}' is now available at '{Path.Combine(reservation.ReservedPath, "S")}'.");
+
+                    return new RemoteZfsWorkspace(response, reservation);
+                }
+                finally
+                {
+                    if (!usingInnerReservation)
+                    {
+                        await reservation.DisposeAsync().ConfigureAwait(false);
+                    }
+                }
+            }
+            finally
+            {
+                if (!usingOuterReservation)
+                {
+                    response.Dispose();
+                }
+            }
+        }
     }
 }

UET/uet/Commands/Build/BuildCommand.cs

@@ -401,6 +401,9 @@ public async Task<int> ExecuteAsync(InvocationContext context)
                     case EngineSpecType.SESNetworkShare:
                         engineSpec = BuildEngineSpecification.ForSESNetworkShare(engine.SESNetworkShare!);
                         break;
+                    case EngineSpecType.RemoteZfs:
+                        engineSpec = BuildEngineSpecification.ForRemoteZfs(engine.RemoteZfs!);
+                        break;
                     case EngineSpecType.Version:
                         engineSpec = BuildEngineSpecification.ForVersionWithPath(engine.Version!, engine.Path!);
                         break;

UET/uet/Commands/Internal/CIBuild/CIBuildCommand.cs

@@ -133,6 +133,9 @@ await _storageManagement.AutoPurgeStorageAsync(
                     case EngineSpecType.SESNetworkShare:
                         engineSpec = BuildEngineSpecification.ForSESNetworkShare(engine.SESNetworkShare!);
                         break;
+                    case EngineSpecType.RemoteZfs:
+                        engineSpec = BuildEngineSpecification.ForRemoteZfs(engine.RemoteZfs!);
+                        break;
                     case EngineSpecType.Version:
                         engineSpec = BuildEngineSpecification.ForVersionWithPath(engine.Version!, engine.Path!);
                         break;

UET/uet/Commands/Internal/RemoteZfsServer/RemoteZfsServerCommand.cs

@@ -200,7 +200,7 @@ await responseStream.WriteAsync(new AcquireWorkspaceResponse
                     new HttpRequestMessage
                     {
                         Method = HttpMethod.Delete,
-                        RequestUri = new Uri($"{_config.TrueNasUrl}/pool/dataset/id/{HttpUtility.UrlEncode(rzfsId)}"),
+                        RequestUri = new Uri($"{_config.TrueNasUrl}/pool/dataset/id/{HttpUtility.UrlEncode(rzfsId)}?force=true"),
                     },
                     cancellationToken: CancellationToken.None).ConfigureAwait(false);
                 deleteResponse.EnsureSuccessStatusCode();

UET/uet/Commands/ParameterSpec/EngineSpec.cs

@@ -193,6 +193,7 @@ private enum EngineParseFlags
             LauncherInstalled = 1 << 6,
             WindowsUserRegistry = 1 << 7,
             SESNetworkShare = 1 << 8,
+            RemoteZfs = 1 << 9,
 
             All = 0xFFFFFF,
         }
@@ -220,7 +221,6 @@ private enum EngineParseFlags
 
             if ((flags & EngineParseFlags.SESNetworkShare) != 0)
             {
-                // Detect UEFS tags.
                 if (engine.StartsWith("ses:", StringComparison.Ordinal))
                 {
                     return new EngineSpec
@@ -233,6 +233,19 @@ private enum EngineParseFlags
                 }
             }
 
+            if ((flags & EngineParseFlags.RemoteZfs) != 0)
+            {
+                if (engine.StartsWith("rzfs:", StringComparison.Ordinal))
+                {
+                    return new EngineSpec
+                    {
+                        Type = EngineSpecType.RemoteZfs,
+                        OriginalSpec = engine,
+                        RemoteZfs = engine["rzfs:".Length..],
+                    };
+                }
+            }
+
             if ((flags & EngineParseFlags.Git) != 0)
             {
                 // Detect commits.
@@ -486,6 +499,8 @@ public static EngineSpec ParseEngineSpecWithoutPath(ArgumentResult result)
 
         public string? SESNetworkShare { get; private init; }
 
+        public string? RemoteZfs { get; private init; }
+
         public string? GitUrl { get; private init; }
 
         public string? GitCommit { get; private init; }

UET/uet/Commands/ParameterSpec/EngineSpecType.cs

@@ -19,5 +19,7 @@ internal enum EngineSpecType
         SelfEngineByBuildConfig,
 
         SESNetworkShare,
+
+        RemoteZfs,
     }
 }

UET/uet/Commands/Test/TestCommand.cs

@@ -152,6 +152,9 @@ public async Task<int> ExecuteAsync(InvocationContext context)
                     case EngineSpecType.SESNetworkShare:
                         engineSpec = BuildEngineSpecification.ForSESNetworkShare(engine.SESNetworkShare!);
                         break;
+                    case EngineSpecType.RemoteZfs:
+                        engineSpec = BuildEngineSpecification.ForRemoteZfs(engine.RemoteZfs!);
+                        break;
                     case EngineSpecType.Version:
                         engineSpec = BuildEngineSpecification.ForVersionWithPath(engine.Version!, engine.Path!);
                         break;