Merge pull request #390 from serverlessworkflow/fix-runner-container-support

Fix the Runner to inject and configure an IContainerPlatform

Author: cdavernas

deployments/docker-compose/docker-compose.build.yml

@@ -29,7 +29,8 @@ services:
       CONNECTIONSTRINGS__REDIS: ${GARNET_URI}
       SYNAPSE_OPERATOR_NAMESPACE: default
       SYNAPSE_OPERATOR_NAME: operator-1
-      SYNAPSE_OPERATOR_RUNNER_API: http://api:8080
+      SYNAPSE_RUNNER_API: http://api:8080
+      SYNAPSE_RUNNER_LIFECYCLE_EVENTS: true
       DOCKER_HOST: unix:///var/run/docker.sock
     extra_hosts:
       - "host.docker.internal:host-gateway"

deployments/docker-compose/docker-compose.yml

@@ -25,7 +25,8 @@ services:
       CONNECTIONSTRINGS__REDIS: ${GARNET_URI}
       SYNAPSE_OPERATOR_NAMESPACE: default
       SYNAPSE_OPERATOR_NAME: operator-1
-      SYNAPSE_OPERATOR_RUNNER_API: http://api:8080
+      SYNAPSE_RUNNER_API: http://api:8080
+      SYNAPSE_RUNNER_LIFECYCLE_EVENTS: true
       DOCKER_HOST: unix:///var/run/docker.sock
     extra_hosts:
       - "host.docker.internal:host-gateway"

src/core/Synapse.Core.Infrastructure.Containers.Docker/Configuration/DockerContainerPlatformOptions.cs

@@ -11,6 +11,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+using Synapse.Resources;
+using System.Runtime.Serialization;
+using System.Text.Json.Serialization;
+using YamlDotNet.Serialization;
+
 namespace Synapse.Core.Infrastructure.Containers.Configuration;
 
 /// <summary>
@@ -20,8 +25,20 @@ public class DockerContainerPlatformOptions
 {
 
     /// <summary>
-    /// Gets/sets the Docker runtime host's network name
+    /// Gets the default network to connect containers to
+    /// </summary>
+    public const string DefaultNetwork = "synapse";
+
+    /// <summary>
+    /// Gets/sets the Docker API to use
+    /// </summary>
+    [DataMember(Order = 1, Name = "api"), JsonPropertyOrder(1), JsonPropertyName("api"), YamlMember(Order = 1, Alias = "api")]
+    public virtual DockerApiConfiguration Api { get; set; } = new();
+
+    /// <summary>
+    /// Gets/sets the network to connect containers to, if any
     /// </summary>
-    public virtual string Network { get; set; } = "synapse";
+    [DataMember(Order = 2, Name = "network"), JsonPropertyOrder(2), JsonPropertyName("network"), YamlMember(Order = 2, Alias = "network")]
+    public virtual string? Network { get; set; } = DefaultNetwork;
 
 }

src/core/Synapse.Core.Infrastructure.Containers.Docker/DockerContainerPlatform.cs

@@ -24,12 +24,13 @@ namespace Synapse.Core.Infrastructure.Containers;
 /// </summary>
 /// <param name="logger">The service used to perform logging</param>
 /// <param name="hostEnvironment">The current <see cref="IHostEnvironment"/></param>
-/// <param name="dockerClient">The service used to interact with the Docker API</param>
 /// <param name="options">The current <see cref="DockerContainerPlatformOptions"/></param>
-public class DockerContainerPlatform(ILogger<DockerContainerPlatform> logger, IHostEnvironment hostEnvironment, IDockerClient dockerClient, IOptions<DockerContainerPlatformOptions> options)
-    : IHostedService, IContainerPlatform
+public class DockerContainerPlatform(ILogger<DockerContainerPlatform> logger, IHostEnvironment hostEnvironment, IOptions<DockerContainerPlatformOptions> options)
+    : IHostedService, IContainerPlatform, IDisposable, IAsyncDisposable
 {
 
+    bool _disposed;
+
     /// <summary>
     /// Gets the service used to perform logging
     /// </summary>
@@ -43,7 +44,7 @@ public class DockerContainerPlatform(ILogger<DockerContainerPlatform> logger, IH
     /// <summary>
     /// Gets the service used to interact with the Docker API
     /// </summary>
-    protected IDockerClient DockerClient { get; } = dockerClient;
+    protected IDockerClient? Docker { get; set; }
 
     /// <summary>
     /// Gets the current <see cref="DockerContainerPlatformOptions"/>
@@ -53,40 +54,42 @@ public class DockerContainerPlatform(ILogger<DockerContainerPlatform> logger, IH
     /// <inheritdoc/>
     public virtual async Task StartAsync(CancellationToken cancellationToken)
     {
+        var dockerConfiguration = new DockerClientConfiguration(this.Options.Api.Endpoint);
+        this.Docker = dockerConfiguration.CreateClient(string.IsNullOrWhiteSpace(this.Options.Api.Version) ? null : System.Version.Parse(this.Options.Api.Version!));
         if (!this.Environment.RunsInDocker()) return;
         var containerShortId = System.Environment.MachineName;
-        var containerId = (await this.DockerClient.Containers.InspectContainerAsync(containerShortId, cancellationToken)).ID;
+        var containerId = (await this.Docker.Containers.InspectContainerAsync(containerShortId, cancellationToken)).ID;
         var response = null as NetworkResponse;
         try
         {
-            response = await this.DockerClient.Networks.InspectNetworkAsync(this.Options.Network, cancellationToken);
+            response = await this.Docker.Networks.InspectNetworkAsync(this.Options.Network, cancellationToken);
         }
         catch (DockerNetworkNotFoundException)
         {
-            await this.DockerClient.Networks.CreateNetworkAsync(new() { Name = this.Options.Network }, cancellationToken);
+            await this.Docker.Networks.CreateNetworkAsync(new() { Name = this.Options.Network }, cancellationToken);
         }
         finally
         {
-            if (response == null || !response!.Containers.ContainsKey(containerId))
-                await this.DockerClient.Networks.ConnectNetworkAsync(this.Options.Network, new NetworkConnectParameters() { Container = containerId }, cancellationToken);
+            if (response == null || !response!.Containers.ContainsKey(containerId)) await this.Docker.Networks.ConnectNetworkAsync(this.Options.Network, new NetworkConnectParameters() { Container = containerId }, cancellationToken);
         }
     }
 
     /// <inheritdoc/>
     public virtual async Task<IContainer> CreateAsync(ContainerProcessDefinition definition, CancellationToken cancellationToken = default)
     {
         ArgumentNullException.ThrowIfNull(definition);
+        if (this.Docker == null) throw new NullReferenceException("The DockerContainerPlatform has not been properly initialized");
         try
         {
-            await this.DockerClient.Images.InspectImageAsync(definition.Image, cancellationToken).ConfigureAwait(false);
+            await this.Docker.Images.InspectImageAsync(definition.Image, cancellationToken).ConfigureAwait(false);
         }
         catch (DockerApiException ex) when (ex.StatusCode == HttpStatusCode.NotFound)
         {
             var downloadProgress = new Progress<JSONMessage>();
             var imageComponents = definition.Image.Split(':');
             var imageName = imageComponents[0];
             var imageTag = imageComponents.Length > 1 ? imageComponents[1] : null;
-            await this.DockerClient.Images.CreateImageAsync(new() { FromImage = imageName, Tag = imageTag }, new(), downloadProgress, cancellationToken).ConfigureAwait(false);
+            await this.Docker.Images.CreateImageAsync(new() { FromImage = imageName, Tag = imageTag }, new(), downloadProgress, cancellationToken).ConfigureAwait(false);
         }
         var parameters = new CreateContainerParameters()
         {
@@ -99,16 +102,60 @@ public virtual async Task<IContainer> CreateAsync(ContainerProcessDefinition def
                 Binds = definition.Volumes?.Select(e => $"{e.Key}={e.Value}")?.ToList() ?? []
             }
         };
-        var response = await this.DockerClient.Containers.CreateContainerAsync(parameters, cancellationToken).ConfigureAwait(false);
-        if (this.Environment.RunsInDocker()) await this.DockerClient.Networks.ConnectNetworkAsync(this.Options.Network, new NetworkConnectParameters() { Container = response.ID }, cancellationToken);
+        var response = await this.Docker.Containers.CreateContainerAsync(parameters, cancellationToken).ConfigureAwait(false);
+        if (this.Environment.RunsInDocker()) await this.Docker.Networks.ConnectNetworkAsync(this.Options.Network, new NetworkConnectParameters() { Container = response.ID }, cancellationToken);
         foreach (var warning in response.Warnings)
         {
             this.Logger.LogWarning(warning);
         }
-        return new DockerContainer(response.ID, this.DockerClient);
+        return new DockerContainer(response.ID, this.Docker);
     }
 
     /// <inheritdoc/>
     public virtual Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;
 
+    /// <summary>
+    /// Disposes of the <see cref="DockerContainerPlatform"/>
+    /// </summary>
+    /// <param name="disposing">A boolean indicating whether or not the <see cref="DockerContainerPlatform"/> is being disposed of</param>
+    /// <returns>A new <see cref="ValueTask"/></returns>
+    protected virtual async ValueTask DisposeAsync(bool disposing)
+    {
+        if (this._disposed) return;
+        if (disposing)
+        {
+            this.Docker?.Dispose();
+        }
+        this._disposed = true;
+        await Task.CompletedTask.ConfigureAwait(false);
+    }
+
+    /// <inheritdoc/>
+    public async ValueTask DisposeAsync()
+    {
+        await this.DisposeAsync(disposing: true).ConfigureAwait(false);
+        GC.SuppressFinalize(this);
+    }
+
+    /// <summary>
+    /// Disposes of the <see cref="DockerContainerPlatform"/>
+    /// </summary>
+    /// <param name="disposing">A boolean indicating whether or not the <see cref="DockerContainerPlatform"/> is being disposed of</param>
+    protected virtual void Dispose(bool disposing)
+    {
+        if (this._disposed) return;
+        if (disposing)
+        {
+            this.Docker?.Dispose();
+        }
+        this._disposed = true;
+    }
+
+    /// <inheritdoc/>
+    public void Dispose()
+    {
+        this.Dispose(disposing: true);
+        GC.SuppressFinalize(this);
+    }
+
 }

src/core/Synapse.Core.Infrastructure.Containers.Docker/Extensions/DockerContainerServiceCollectionExtensions.cs

@@ -28,12 +28,9 @@ public static class DockerContainerServiceCollectionExtensions
     /// Adds and configures a new <see cref="DockerContainerPlatform"/>
     /// </summary>
     /// <param name="services">The <see cref="IServiceCollection"/> to configure</param>
-    /// <param name="dockerClientConfiguration">The <see cref="DockerClientConfiguration"/> used to configure the <see cref="IDockerClient"/> to use, if any</param>
     /// <returns>The configured <see cref="IServiceCollection"/></returns>
-    public static IServiceCollection AddDockerContainerPlatform(this IServiceCollection services, DockerClientConfiguration? dockerClientConfiguration = null)
+    public static IServiceCollection AddDockerContainerPlatform(this IServiceCollection services)
     {
-        dockerClientConfiguration ??= new DockerClientConfiguration();
-        services.TryAddSingleton<IDockerClient>(dockerClientConfiguration.CreateClient());
         services.TryAddSingleton<DockerContainerPlatform>();
         services.AddSingleton<IContainerPlatform>(provider => provider.GetRequiredService<DockerContainerPlatform>());
         services.AddSingleton<IHostedService>(provider => provider.GetRequiredService<DockerContainerPlatform>());

src/core/Synapse.Core/ContainerPlatform.cs

@@ -0,0 +1,41 @@
+// Copyright © 2024-Present The Synapse Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License"),
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+namespace Synapse;
+
+/// <summary>
+/// Enumerates all supported container platforms
+/// </summary>
+public static class ContainerPlatform
+{
+
+    /// <summary>
+    /// Gets the Docker container platform
+    /// </summary>
+    public const string Docker = "docker";
+    /// <summary>
+    /// Gets the Kubernetes container platform
+    /// </summary>
+    public const string Kubernetes = "kubernetes";
+
+    /// <summary>
+    /// Gets a new <see cref="IEnumerable{T}"/> containing all default container platforms
+    /// </summary>
+    /// <returns>A new <see cref="IEnumerable{T}"/> containing all default container platforms</returns>
+    public static IEnumerable<string> AsEnumerable()
+    {
+        yield return Docker;
+        yield return Kubernetes;
+    }
+
+}

src/core/Synapse.Core/OperatorRuntimeMode.cs

@@ -38,4 +38,4 @@ public static IEnumerable<string> AsEnumerable()
         yield return Containerized;
     }
 
-}
+}

src/core/Synapse.Core/Resources/CorrelationLifetime.cs

@@ -39,4 +39,4 @@ public static IEnumerable<string> AsEnumerable()
         yield return Ephemeral;
     }
 
-}
+}

src/core/Synapse.Core/Resources/DockerApiConfiguration.cs

@@ -0,0 +1,37 @@
+// Copyright © 2024-Present The Synapse Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License"),
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System.Runtime.InteropServices;
+
+namespace Synapse.Resources;
+
+/// <summary>
+/// Represents an object used to configure the Docker API to use
+/// </summary>
+[DataContract]
+public record DockerApiConfiguration
+{
+
+    /// <summary>
+    /// Gets/sets the endpoint of the Docker API to use
+    /// </summary>
+    [DataMember(Order = 1, Name = "endpoint"), JsonPropertyOrder(1), JsonPropertyName("endpoint"), YamlMember(Order = 1, Alias = "endpoint")]
+    public virtual Uri Endpoint { get; set; } = RuntimeInformation.IsOSPlatform(OSPlatform.Windows) ? new Uri("npipe://./pipe/docker_engine") : new Uri("unix:/var/run/docker.sock");
+
+    /// <summary>
+    /// Gets/sets the version of the Docker API to use
+    /// </summary>
+    [DataMember(Order = 2, Name = "version"), JsonPropertyOrder(2), JsonPropertyName("version"), YamlMember(Order = 2, Alias = "version")]
+    public virtual string? Version { get; set; }
+
+}

src/core/Synapse.Core/Resources/RunnerDefinition.cs

@@ -40,10 +40,22 @@ public record RunnerDefinition
         }
     };
 
+    /// <summary>
+    /// Gets/sets the container platform used by runners to spawn containers
+    /// </summary>
+    [Required]
+    [DataMember(Order = 3, Name = "containerPlatform"), JsonPropertyOrder(3), JsonPropertyName("containerPlatform"), YamlMember(Order = 3, Alias = "containerPlatform")]
+    public virtual string ContainerPlatform { get; set; } = Synapse.ContainerPlatform.Docker;
+
     /// <summary>
     /// Gets/sets the endpoint that references the base address and authentication policy for the Synapse API used by runners
     /// </summary>
-    [DataMember(Order = 3, Name = "certificates"), JsonPropertyOrder(3), JsonPropertyName("certificates"), YamlMember(Order = 3, Alias = "certificates")]
+    [DataMember(Order = 4, Name = "certificates"), JsonPropertyOrder(4), JsonPropertyName("certificates"), YamlMember(Order = 4, Alias = "certificates")]
     public virtual CertificateValidationStrategyDefinition? Certificates { get; set; }
-    
+
+    /// <summary>
+    /// Gets/sets a boolean indicating whether or not runners spawned by the configured Synapse Operators should publish lifecycle events
+    /// </summary>
+    [DataMember(Order = 5, Name = "publishLifecycleEvents"), JsonPropertyOrder(5), JsonPropertyName("publishLifecycleEvents"), YamlMember(Order = 5, Alias = "publishLifecycleEvents")]
+    public virtual bool? PublishLifecycleEvents { get; set; } = true;
 }