ACA cluster provider support for multiple images + pluggable member store (#3981)
* Decouple ACA cluster provider storage * Add readme
This commit is contained in:
Sipke Schoorstra
GitHub
parent
213b0e29b9
commit
4112fbfc9a
7
Elsa.sln
7
Elsa.sln
|
|
@ -194,6 +194,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Samples.JsonWorkflowPr
|
|||
EndProject
|
||||
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.WorkflowProviders.FluentStorage", "src\modules\Elsa.WorkflowProviders.FluentStorage\Elsa.WorkflowProviders.FluentStorage.csproj", "{044C3108-FE79-460A-9C31-A03C30228836}"
|
||||
EndProject
|
||||
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Samples.AspNet.ProtoActorRuntime.AzureContainerApps", "src\samples\aspnet\Elsa.Samples.AspNet.ProtoActorRuntime.AzureContainerApps\Elsa.Samples.AspNet.ProtoActorRuntime.AzureContainerApps.csproj", "{3212A999-4AC4-4911-9AA4-92AB906BCB5E}"
|
||||
EndProject
|
||||
Global
|
||||
GlobalSection(SolutionConfigurationPlatforms) = preSolution
|
||||
Debug|Any CPU = Debug|Any CPU
|
||||
|
|
@ -492,6 +494,10 @@ Global
|
|||
{044C3108-FE79-460A-9C31-A03C30228836}.Debug|Any CPU.Build.0 = Debug|Any CPU
|
||||
{044C3108-FE79-460A-9C31-A03C30228836}.Release|Any CPU.ActiveCfg = Release|Any CPU
|
||||
{044C3108-FE79-460A-9C31-A03C30228836}.Release|Any CPU.Build.0 = Release|Any CPU
|
||||
{3212A999-4AC4-4911-9AA4-92AB906BCB5E}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
|
||||
{3212A999-4AC4-4911-9AA4-92AB906BCB5E}.Debug|Any CPU.Build.0 = Debug|Any CPU
|
||||
{3212A999-4AC4-4911-9AA4-92AB906BCB5E}.Release|Any CPU.ActiveCfg = Release|Any CPU
|
||||
{3212A999-4AC4-4911-9AA4-92AB906BCB5E}.Release|Any CPU.Build.0 = Release|Any CPU
|
||||
EndGlobalSection
|
||||
GlobalSection(NestedProjects) = preSolution
|
||||
{155227F0-A33B-40AA-A4B4-06F813EB921B} = {61017E64-6D00-49CB-9E81-5002DC8F7D5F}
|
||||
|
|
@ -578,5 +584,6 @@ Global
|
|||
{4FA6955C-6860-493F-ABD4-CE327A33EEA3} = {873BFC3E-63C2-4495-A503-5EC05DCD84E4}
|
||||
{C5043453-5FB8-4796-9A80-C4C766F2CB62} = {56C2FFB8-EA54-45B5-A095-4A78142EB4B5}
|
||||
{044C3108-FE79-460A-9C31-A03C30228836} = {5BA4A8FA-F7F4-45B3-AEC8-8886D35AAC79}
|
||||
{3212A999-4AC4-4911-9AA4-92AB906BCB5E} = {56C2FFB8-EA54-45B5-A095-4A78142EB4B5}
|
||||
EndGlobalSection
|
||||
EndGlobal
|
||||
|
|
|
|||
|
|
@ -0,0 +1,14 @@
|
|||
using Azure.ResourceManager;
|
||||
|
||||
namespace Proto.Cluster.AzureContainerApps;
|
||||
|
||||
/// <summary>
|
||||
/// Provides an <see cref="ArmClient"/> instance.
|
||||
/// </summary>
|
||||
public static class ArmClientProviders
|
||||
{
|
||||
/// <summary>
|
||||
/// A default <see cref="IArmClientProvider"/> that uses <see cref="Azure.Identity.DefaultAzureCredential"/>
|
||||
/// </summary>
|
||||
public static readonly DefaultAzureCredentialArmClientProvider DefaultAzureCredential = new();
|
||||
}
|
||||
|
|
@ -1,118 +0,0 @@
|
|||
using Azure;
|
||||
using Azure.ResourceManager;
|
||||
using Azure.ResourceManager.AppContainers;
|
||||
using Azure.ResourceManager.Resources;
|
||||
using Azure.ResourceManager.Resources.Models;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Proto;
|
||||
using Proto.Cluster;
|
||||
|
||||
namespace Elsa.ProtoActor.Cluster.AzureContainerApps;
|
||||
|
||||
public static class ArmClientUtils
|
||||
{
|
||||
private static readonly ILogger Logger = Log.CreateLogger(nameof(ArmClientUtils));
|
||||
|
||||
public static async Task<Member[]> GetClusterMembers(this ArmClient client, string resourceGroupName, string containerAppName)
|
||||
{
|
||||
var members = new List<Member>();
|
||||
|
||||
var containerApp = await (await client.GetResourceGroupByName(resourceGroupName)).Value.GetContainerAppAsync(containerAppName);
|
||||
|
||||
if (containerApp is null || !containerApp.HasValue)
|
||||
{
|
||||
Logger.LogError("Container App: {ContainerApp} in resource group: {ResourceGroup} is not found", containerApp, resourceGroupName);
|
||||
return members.ToArray();
|
||||
}
|
||||
|
||||
var containerAppRevisions = GetActiveRevisionsWithTraffic(containerApp).ToList();
|
||||
if (!containerAppRevisions.Any())
|
||||
{
|
||||
Logger.LogError("Container App: {ContainerApp} in resource group: {ResourceGroup} does not contain any active revisions with traffic", containerAppName, resourceGroupName);
|
||||
return members.ToArray();
|
||||
}
|
||||
|
||||
var replicasWithTraffic = containerAppRevisions.SelectMany(r => r.GetContainerAppReplicas());
|
||||
|
||||
var allTags = (await containerApp.Value.GetTagResource().GetAsync()).Value.Data.TagValues;
|
||||
|
||||
foreach (var replica in replicasWithTraffic)
|
||||
{
|
||||
var replicaNameTag = allTags.FirstOrDefault(kvp => kvp.Value == replica.Data.Name);
|
||||
if (replicaNameTag.Key == null)
|
||||
{
|
||||
Logger.LogWarning("Skipping Replica with name: {Name}, no Proto Tags found", replica.Data.Name);
|
||||
continue;
|
||||
}
|
||||
|
||||
var replicaNameTagPrefix = replicaNameTag.Key.Replace(ResourceTagLabels.LabelReplicaNameWithoutPrefix, string.Empty);
|
||||
var currentReplicaTags = allTags.Where(kvp => kvp.Key.StartsWith(replicaNameTagPrefix)).ToDictionary(x => x.Key, x => x.Value);
|
||||
|
||||
var memberId = currentReplicaTags.FirstOrDefault(kvp => kvp.Key.ToString().Contains(ResourceTagLabels.LabelMemberIdWithoutPrefix)).Value;
|
||||
|
||||
var kinds = currentReplicaTags
|
||||
.Where(kvp => kvp.Key.StartsWith(ResourceTagLabels.LabelKind(memberId)))
|
||||
.Select(kvp => kvp.Key[(ResourceTagLabels.LabelKind(memberId).Length + 1)..])
|
||||
.ToArray();
|
||||
|
||||
var member = new Member
|
||||
{
|
||||
Id = currentReplicaTags[ResourceTagLabels.LabelMemberId(memberId)],
|
||||
Port = int.Parse(currentReplicaTags[ResourceTagLabels.LabelPort(memberId)]),
|
||||
Host = currentReplicaTags[ResourceTagLabels.LabelHost(memberId)],
|
||||
Kinds = { kinds }
|
||||
};
|
||||
|
||||
members.Add(member);
|
||||
}
|
||||
|
||||
return members.ToArray();
|
||||
}
|
||||
|
||||
public static async Task AddMemberTags(this ArmClient client, string resourceGroupName, string containerAppName, Dictionary<string, string> newTags)
|
||||
{
|
||||
var resourceTag = new Tag();
|
||||
foreach (var tag in newTags)
|
||||
{
|
||||
resourceTag.TagValues.Add(tag);
|
||||
}
|
||||
|
||||
var resourceGroup = await client.GetResourceGroupByName(resourceGroupName);
|
||||
var containerApp = await resourceGroup.Value.GetContainerAppAsync(containerAppName);
|
||||
var tagResource = containerApp.Value.GetTagResource();
|
||||
|
||||
var existingTags = (await tagResource.GetAsync()).Value.Data.TagValues;
|
||||
foreach (var tag in existingTags)
|
||||
{
|
||||
resourceTag.TagValues.Add(tag);
|
||||
}
|
||||
|
||||
await tagResource.CreateOrUpdateAsync(WaitUntil.Completed, new TagResourceData(resourceTag));
|
||||
}
|
||||
|
||||
public static async Task ClearMemberTags(this ArmClient client, string resourceGroupName, string containerAppName, string memberId)
|
||||
{
|
||||
var resourceGroup = await client.GetResourceGroupByName(resourceGroupName);
|
||||
var containerApp = await resourceGroup.Value.GetContainerAppAsync(containerAppName);
|
||||
var tagResource = containerApp.Value.GetTagResource();
|
||||
|
||||
var resourceTag = new Tag();
|
||||
var existingTags = (await tagResource.GetAsync()).Value.Data.TagValues;
|
||||
|
||||
foreach (var tag in existingTags)
|
||||
{
|
||||
if (!tag.Key.StartsWith(ResourceTagLabels.LabelPrefix(memberId)))
|
||||
{
|
||||
resourceTag.TagValues.Add(tag);
|
||||
}
|
||||
}
|
||||
|
||||
await tagResource.CreateOrUpdateAsync(WaitUntil.Completed, new TagResourceData(resourceTag));
|
||||
}
|
||||
|
||||
public static async Task<Response<ResourceGroupResource>> GetResourceGroupByName(this ArmClient client, string resourceGroupName) =>
|
||||
await (await client.GetDefaultSubscriptionAsync()).GetResourceGroups().GetAsync(resourceGroupName);
|
||||
|
||||
private static IEnumerable<ContainerAppRevisionResource> GetActiveRevisionsWithTraffic(ContainerAppResource containerApp) =>
|
||||
containerApp.GetContainerAppRevisions().Where(r => r.HasData && (r.Data.IsActive ?? false) && r.Data.TrafficWeight > 0);
|
||||
}
|
||||
|
|
@ -1,59 +1,64 @@
|
|||
using Azure.ResourceManager;
|
||||
using System;
|
||||
using System.Linq;
|
||||
using System.Threading.Tasks;
|
||||
using Azure.ResourceManager;
|
||||
using Azure.ResourceManager.AppContainers;
|
||||
using JetBrains.Annotations;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Proto;
|
||||
using Proto.Cluster;
|
||||
using Microsoft.Extensions.Options;
|
||||
using Proto.Cluster.AzureContainerApps.Stores.ResourceTags;
|
||||
using Proto.Utils;
|
||||
|
||||
namespace Elsa.ProtoActor.Cluster.AzureContainerApps;
|
||||
namespace Proto.Cluster.AzureContainerApps;
|
||||
|
||||
/// <summary>
|
||||
/// A cluster provider that uses Azure Container Apps to host the cluster.
|
||||
/// </summary>
|
||||
[PublicAPI]
|
||||
public class AzureContainerAppsProvider : IClusterProvider
|
||||
{
|
||||
private readonly ArmClient _client;
|
||||
private readonly string _resourceGroup;
|
||||
private readonly string _containerAppName;
|
||||
private readonly IArmClientProvider _armClientProvider;
|
||||
private readonly IClusterMemberStore _clusterMemberStore;
|
||||
private readonly IOptions<AzureContainerAppsProviderOptions> _options;
|
||||
private readonly ILogger _logger;
|
||||
private readonly string? _containerAppName;
|
||||
private readonly string _revisionName;
|
||||
private readonly string _replicaName;
|
||||
private readonly string _advertisedHost;
|
||||
|
||||
|
||||
private string _memberId = null!;
|
||||
private string _address = null!;
|
||||
private global::Proto.Cluster.Cluster _cluster = null!;
|
||||
private Cluster _cluster = null!;
|
||||
private string _clusterName = null!;
|
||||
private string[] _kinds = null!;
|
||||
private int _port;
|
||||
|
||||
private static readonly ILogger Logger = Log.CreateLogger<AzureContainerAppsProvider>();
|
||||
private static readonly TimeSpan PollIntervalInSeconds = TimeSpan.FromSeconds(5);
|
||||
private ArmClient _client = null!;
|
||||
|
||||
/// <summary>
|
||||
/// Use this constructor to create a new instance.
|
||||
/// </summary>
|
||||
/// <param name="client">An existing <see cref="ArmClient"/></param> instance that you need to bring yourself.
|
||||
/// <param name="resourceGroup">The resource group name containing your Azure Container App.</param>
|
||||
/// <param name="containerAppName">The name of the container app. If not specified, the CONTAINER_APP_NAME environment variable is used.</param>
|
||||
/// <param name="revision">The revision of the container app. If not specified, the CONTAINER_APP_REVISION environment variable is used.</param>
|
||||
/// <param name="replicaName">The replica name of the container app. If not specified, the HOSTNAME environment variable is used.</param>
|
||||
/// <param name="advertisedHost">The host or IP address of the container app. If not specified, will take the smallest local IP address (e.g. 127.0.0.1).</param>
|
||||
/// <param name="armClientProvider">An <see cref="IArmClientProvider"/> to create <see cref="ArmClient"/> instances.</param>
|
||||
/// <param name="clusterMemberStore">The store to use for storing member information.</param>
|
||||
/// <param name="options">The options for this provider.</param>
|
||||
/// <param name="logger">The logger to use.</param>
|
||||
public AzureContainerAppsProvider(
|
||||
ArmClient client,
|
||||
string resourceGroup,
|
||||
string? containerAppName = default,
|
||||
string? revision = default,
|
||||
string? replicaName = default,
|
||||
string? advertisedHost = default)
|
||||
IArmClientProvider armClientProvider,
|
||||
IClusterMemberStore clusterMemberStore,
|
||||
IOptions<AzureContainerAppsProviderOptions> options,
|
||||
ILogger<AzureContainerAppsProvider> logger)
|
||||
{
|
||||
_client = client;
|
||||
_resourceGroup = resourceGroup;
|
||||
_containerAppName = containerAppName ?? Environment.GetEnvironmentVariable("CONTAINER_APP_NAME") ?? throw new Exception("No app name provided");
|
||||
_revisionName = revision ?? Environment.GetEnvironmentVariable("CONTAINER_APP_REVISION") ?? throw new Exception("No app revision provided");
|
||||
_replicaName = replicaName ?? Environment.GetEnvironmentVariable("HOSTNAME") ?? throw new Exception("No replica name provided");
|
||||
_advertisedHost = !string.IsNullOrEmpty(advertisedHost) ? advertisedHost : ConfigUtils.FindSmallestIpAddress().ToString();
|
||||
_armClientProvider = armClientProvider;
|
||||
_clusterMemberStore = clusterMemberStore;
|
||||
_options = options;
|
||||
_logger = logger;
|
||||
_containerAppName = Environment.GetEnvironmentVariable("CONTAINER_APP_NAME") ?? throw new Exception("No app name provided");
|
||||
_revisionName = Environment.GetEnvironmentVariable("CONTAINER_APP_REVISION") ?? throw new Exception("No app revision provided");
|
||||
_replicaName = Environment.GetEnvironmentVariable("HOSTNAME") ?? throw new Exception("No replica name provided");
|
||||
_advertisedHost = ConfigUtils.FindSmallestIpAddress().ToString();
|
||||
}
|
||||
|
||||
public async Task StartMemberAsync(global::Proto.Cluster.Cluster cluster)
|
||||
/// <inheritdoc />
|
||||
public async Task StartMemberAsync(Cluster cluster)
|
||||
{
|
||||
var clusterName = cluster.Config.ClusterName;
|
||||
var (host, port) = cluster.System.GetAddress();
|
||||
|
|
@ -64,12 +69,15 @@ public class AzureContainerAppsProvider : IClusterProvider
|
|||
_port = port;
|
||||
_kinds = kinds;
|
||||
_address = $"{host}:{port}";
|
||||
_client = await _armClientProvider.CreateClientAsync();
|
||||
|
||||
await RegisterMemberAsync();
|
||||
//await CleanupStoreAsync(cluster);
|
||||
await RegisterMemberAsync().ConfigureAwait(false);
|
||||
StartClusterMonitor();
|
||||
}
|
||||
|
||||
public Task StartClientAsync(global::Proto.Cluster.Cluster cluster)
|
||||
/// <inheritdoc />
|
||||
public Task StartClientAsync(Cluster cluster)
|
||||
{
|
||||
var clusterName = cluster.Config.ClusterName;
|
||||
var (_, port) = cluster.System.GetAddress();
|
||||
|
|
@ -83,103 +91,95 @@ public class AzureContainerAppsProvider : IClusterProvider
|
|||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public async Task ShutdownAsync(bool graceful) => await DeregisterMemberAsync();
|
||||
/// <inheritdoc />
|
||||
public async Task ShutdownAsync(bool graceful) => await DeregisterMemberAsync().ConfigureAwait(false);
|
||||
|
||||
private async Task CleanupStoreAsync(Cluster cluster)
|
||||
{
|
||||
await _clusterMemberStore.ClearAsync(cluster.Config.ClusterName);
|
||||
}
|
||||
|
||||
private async Task RegisterMemberAsync()
|
||||
{
|
||||
await Retry.Try(RegisterMemberInner, retryCount: Retry.Forever, onError: OnError, onFailed: OnFailed);
|
||||
await Retry.Try(RegisterMemberInternal, retryCount: Retry.Forever, onError: OnError, onFailed: OnFailed).ConfigureAwait(false);
|
||||
|
||||
static void OnError(int attempt, Exception exception) => Logger.LogWarning(exception, "Failed to register service");
|
||||
static void OnFailed(Exception exception) => Logger.LogError(exception, "Failed to register service");
|
||||
void OnError(int attempt, Exception exception) => _logger.LogWarning(exception, "Failed to register service");
|
||||
void OnFailed(Exception exception) => _logger.LogError(exception, "Failed to register service");
|
||||
}
|
||||
|
||||
private async Task RegisterMemberInner()
|
||||
private async Task RegisterMemberInternal()
|
||||
{
|
||||
var resourceGroup = await _client.GetResourceGroupByName(_resourceGroup);
|
||||
var containerApp = await resourceGroup.Value.GetContainerAppAsync(_containerAppName);
|
||||
var revision = await containerApp.Value.GetContainerAppRevisionAsync(_revisionName);
|
||||
var subscriptionId = _options.Value.SubscriptionId;
|
||||
var resourceGroupName = _options.Value.ResourceGroupName;
|
||||
var resourceGroup = await _client.GetResourceGroupByNameAsync(resourceGroupName, subscriptionId).ConfigureAwait(false);
|
||||
var containerApp = await resourceGroup.GetContainerAppAsync(_containerAppName).ConfigureAwait(false);
|
||||
var revision = await containerApp.Value.GetContainerAppRevisionAsync(_revisionName).ConfigureAwait(false);
|
||||
|
||||
if (revision.Value.Data.TrafficWeight.GetValueOrDefault(0) == 0)
|
||||
if ((revision.Value.Data.TrafficWeight ?? 0) == 0)
|
||||
return;
|
||||
|
||||
Logger.LogInformation(
|
||||
var member = new Member
|
||||
{
|
||||
Id = _memberId,
|
||||
Host = _advertisedHost,
|
||||
Port = _port,
|
||||
};
|
||||
|
||||
_logger.LogInformation(
|
||||
"[Cluster][AzureContainerAppsProvider] Registering service {ReplicaName} on {IpAddress}",
|
||||
_replicaName,
|
||||
_address);
|
||||
|
||||
var tags = new Dictionary<string, string>
|
||||
{
|
||||
[ResourceTagLabels.LabelCluster(_memberId)] = _clusterName,
|
||||
[ResourceTagLabels.LabelHost(_memberId)] = _advertisedHost,
|
||||
[ResourceTagLabels.LabelPort(_memberId)] = _port.ToString(),
|
||||
[ResourceTagLabels.LabelMemberId(_memberId)] = _memberId,
|
||||
[ResourceTagLabels.LabelReplicaName(_memberId)] = _replicaName
|
||||
};
|
||||
|
||||
foreach (var kind in _kinds)
|
||||
{
|
||||
var labelKey = $"{ResourceTagLabels.LabelKind(_memberId)}-{kind}";
|
||||
tags.TryAdd(labelKey, "true");
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
await _client.AddMemberTags(_resourceGroup, _containerAppName, tags);
|
||||
}
|
||||
catch (Exception x)
|
||||
{
|
||||
Logger.LogError(x, "Failed to update metadata");
|
||||
}
|
||||
member.Kinds.AddRange(_kinds);
|
||||
await _clusterMemberStore.RegisterAsync(_clusterName, member).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private void StartClusterMonitor() =>
|
||||
private void StartClusterMonitor()
|
||||
{
|
||||
var pollInterval = _options.Value.PollInterval;
|
||||
var storeName = _clusterMemberStore.GetType().Name;
|
||||
|
||||
_ = SafeTask.Run(async () =>
|
||||
{
|
||||
while (!_cluster.System.Shutdown.IsCancellationRequested)
|
||||
{
|
||||
Logger.LogInformation("Calling ACS API");
|
||||
_logger.LogInformation("Looking for members in {Store}", storeName);
|
||||
|
||||
try
|
||||
{
|
||||
var members = await _client.GetClusterMembers(_resourceGroup, _containerAppName);
|
||||
var members = (await _clusterMemberStore.ListAsync().ConfigureAwait(false)).ToArray();
|
||||
|
||||
if (members.Any())
|
||||
{
|
||||
Logger.LogInformation("Got members {Members}", members.Length);
|
||||
_logger.LogInformation("Got members {Members}", members.Length);
|
||||
_cluster.MemberList.UpdateClusterTopology(members);
|
||||
}
|
||||
else
|
||||
{
|
||||
Logger.LogWarning("Failed to get members from Azure Container Apps");
|
||||
_logger.LogWarning("Failed to get members from {Store}", storeName);
|
||||
}
|
||||
}
|
||||
catch (Exception x)
|
||||
{
|
||||
Logger.LogError(x, "Failed to get members from Azure Container Apps");
|
||||
_logger.LogError(x, "Failed to get members from {Store}", storeName);
|
||||
}
|
||||
|
||||
await Task.Delay(PollIntervalInSeconds);
|
||||
await Task.Delay(pollInterval).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
private async Task DeregisterMemberAsync()
|
||||
{
|
||||
await Retry.Try(DeregisterMemberInner, onError: OnError, onFailed: OnFailed);
|
||||
|
||||
static void OnError(int attempt, Exception exception) =>
|
||||
Logger.LogWarning(exception, "Failed to deregister service");
|
||||
|
||||
static void OnFailed(Exception exception) => Logger.LogError(exception, "Failed to deregister service");
|
||||
await Retry.Try(DeregisterMemberInner, onError: OnError, onFailed: OnFailed).ConfigureAwait(false);
|
||||
void OnError(int attempt, Exception exception) => _logger.LogWarning(exception, "Failed to deregister service");
|
||||
void OnFailed(Exception exception) => _logger.LogError(exception, "Failed to deregister service");
|
||||
}
|
||||
|
||||
private async Task DeregisterMemberInner()
|
||||
{
|
||||
Logger.LogInformation(
|
||||
"[Cluster][AzureContainerAppsProvider] Unregistering member {ReplicaName} on {IpAddress}",
|
||||
_replicaName,
|
||||
_address);
|
||||
|
||||
await _client.ClearMemberTags(_resourceGroup, _containerAppName, _memberId);
|
||||
_logger.LogInformation("[Cluster][AzureContainerAppsProvider] Unregistering member {ReplicaName} on {IpAddress}", _replicaName, _address);
|
||||
await _clusterMemberStore.UnregisterAsync(_memberId).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,25 @@
|
|||
using JetBrains.Annotations;
|
||||
|
||||
namespace Proto.Cluster.AzureContainerApps;
|
||||
|
||||
/// <summary>
|
||||
/// Options for <see cref="AzureContainerAppsProvider"/>
|
||||
/// </summary>
|
||||
[PublicAPI]
|
||||
public class AzureContainerAppsProviderOptions
|
||||
{
|
||||
/// <summary>
|
||||
/// The subscription ID to use. If not set, the default subscription will be used.
|
||||
/// </summary>
|
||||
public string? SubscriptionId { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// The name of the resource group to use.
|
||||
/// </summary>
|
||||
public string ResourceGroupName { get; set; } = default!;
|
||||
|
||||
/// <summary>
|
||||
/// The name of the container app to use.
|
||||
/// </summary>
|
||||
public TimeSpan PollInterval { get; set; } = TimeSpan.FromSeconds(5);
|
||||
}
|
||||
|
|
@ -1,13 +1,14 @@
|
|||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using System.Net;
|
||||
using System.Net.NetworkInformation;
|
||||
using System.Net.Sockets;
|
||||
using JetBrains.Annotations;
|
||||
|
||||
namespace Elsa.ProtoActor.Cluster.AzureContainerApps;
|
||||
namespace Proto.Cluster.AzureContainerApps;
|
||||
|
||||
public static class ConfigUtils
|
||||
{
|
||||
[PublicAPI]
|
||||
public static IPAddress FindSmallestIpAddress(AddressFamily family = AddressFamily.InterNetwork)
|
||||
{
|
||||
var addressCandidates = NetworkInterface.GetAllNetworkInterfaces()
|
||||
|
|
@ -29,7 +30,7 @@ public static class ConfigUtils
|
|||
|
||||
return result;
|
||||
|
||||
static bool CompareIpAddresses(IPAddress lhs, IPAddress? rhs)
|
||||
static bool CompareIpAddresses(IPAddress lhs, [CanBeNull] IPAddress rhs)
|
||||
{
|
||||
if (rhs == null)
|
||||
return true;
|
||||
|
|
|
|||
|
|
@ -0,0 +1,19 @@
|
|||
using Azure.Identity;
|
||||
using Azure.ResourceManager;
|
||||
using JetBrains.Annotations;
|
||||
|
||||
namespace Proto.Cluster.AzureContainerApps;
|
||||
|
||||
/// <summary>
|
||||
/// Provides an <see cref="ArmClient"/> instance using <see cref="Azure.Identity.DefaultAzureCredential"/>
|
||||
/// </summary>
|
||||
[PublicAPI]
|
||||
public class DefaultAzureCredentialArmClientProvider : IArmClientProvider
|
||||
{
|
||||
/// <inheritdoc />
|
||||
public ValueTask<ArmClient> CreateClientAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
var client = new ArmClient(new DefaultAzureCredential());
|
||||
return new(client);
|
||||
}
|
||||
}
|
||||
|
|
@ -10,9 +10,11 @@
|
|||
This is a temporary stand-in until Proto.Actor provides an updated version of Proto.Cluster.AzureContainerApps.
|
||||
</Description>
|
||||
<PackageTags>elsa module runtime protoactor cluster azure container apps</PackageTags>
|
||||
<RootNamespace>Proto.Cluster.AzureContainerApps</RootNamespace>
|
||||
</PropertyGroup>
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Azure.ResourceManager.AppContainers" Version="1.0.2" />
|
||||
<PackageReference Include="Azure.Identity" Version="1.8.2" />
|
||||
<PackageReference Include="Azure.ResourceManager.AppContainers" Version="1.0.3" />
|
||||
<PackageReference Include="Azure.ResourceManager.Resources" Version="1.4.0" />
|
||||
<PackageReference Include="Proto.Cluster" Version="1.1.0" />
|
||||
</ItemGroup>
|
||||
|
|
|
|||
|
|
@ -0,0 +1,15 @@
|
|||
using Azure.ResourceManager;
|
||||
|
||||
namespace Proto.Cluster.AzureContainerApps;
|
||||
|
||||
/// <summary>
|
||||
/// Provides an <see cref="ArmClient"/> instance.
|
||||
/// </summary>
|
||||
public interface IArmClientProvider
|
||||
{
|
||||
/// <summary>
|
||||
/// Creates an <see cref="ArmClient"/> instance.
|
||||
/// </summary>
|
||||
/// <returns>An <see cref="ArmClient"/> instance.</returns>
|
||||
ValueTask<ArmClient> CreateClientAsync(CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
|
@ -0,0 +1,40 @@
|
|||
using System.Collections.Generic;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
|
||||
namespace Proto.Cluster.AzureContainerApps;
|
||||
|
||||
/// <summary>
|
||||
/// Represents a repository of members in a cluster.
|
||||
/// </summary>
|
||||
public interface IClusterMemberStore
|
||||
{
|
||||
/// <summary>
|
||||
/// Returns a list of all members in the cluster.
|
||||
/// </summary>
|
||||
/// <param name="cancellationToken">The cancellation token.</param>
|
||||
/// <returns>A list of all members in the cluster.</returns>
|
||||
ValueTask<ICollection<Member>> ListAsync(CancellationToken cancellationToken = default);
|
||||
|
||||
/// <summary>
|
||||
/// Registers a member in the cluster.
|
||||
/// </summary>
|
||||
/// <param name="clusterName">The name of the cluster.</param>
|
||||
/// <param name="member">The member to register.</param>
|
||||
/// <param name="cancellationToken">The cancellation token.</param>
|
||||
ValueTask RegisterAsync(string clusterName, Member member, CancellationToken cancellationToken = default);
|
||||
|
||||
/// <summary>
|
||||
/// Unregisters a member from the cluster.
|
||||
/// </summary>
|
||||
/// <param name="memberId">The ID of the member to unregister.</param>
|
||||
/// <param name="cancellationToken">The cancellation token.</param>
|
||||
ValueTask UnregisterAsync(string memberId, CancellationToken cancellationToken = default);
|
||||
|
||||
/// <summary>
|
||||
/// Clears all members from the cluster.
|
||||
/// </summary>
|
||||
/// <param name="clusterName">The name of the cluster.</param>
|
||||
/// <param name="cancellationToken">The cancellation token.</param>
|
||||
ValueTask ClearAsync(string clusterName, CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
|
@ -0,0 +1,59 @@
|
|||
# Azure Container Apps cluster provider
|
||||
|
||||
Use this cluster provider when you're hosting your application in an Azure Container Apps cluster.
|
||||
|
||||
The provider stores cluster member information using Azure Resource Tags on the container application and uses these tags to discover other cluster members within the same Azure Container Application Managed Environment.
|
||||
|
||||
## Installation
|
||||
|
||||
To install the provider, add the following code to your program:
|
||||
|
||||
```csharp
|
||||
services.AddAzureContainerAppsProvider(ArmClientProviders.DefaultAzureCredential, options =>
|
||||
{
|
||||
options.ResourceGroupName = "{the resource group name containing your container application}";
|
||||
|
||||
// Optionally, you can specify the subscription ID. If not specified, the provider will use the default subscription.
|
||||
options.SubscriptionId = "{your subscription id}";
|
||||
});
|
||||
```
|
||||
|
||||
## Appsettings.json
|
||||
|
||||
Instead of hardcoding the options above, you should instead bind the options using your app's configuration.
|
||||
For example, consider the following appsettings.json:
|
||||
|
||||
```json
|
||||
{
|
||||
"AzureContainerApps": {
|
||||
"ResourceGroupName": "{the resource group name containing your container application}",
|
||||
"SubscriptionId": "{your subscription id}"
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
You can then bind the options using the following code:
|
||||
|
||||
```csharp
|
||||
services.AddAzureContainerAppsProvider(ArmClientProviders.DefaultAzureCredential, options => configuration.Bind("AzureContainerApps", options));
|
||||
```
|
||||
|
||||
## Custom member store
|
||||
|
||||
By default, the provider will store cluster member information using Azure Resource Tags on the container application.
|
||||
If you want to use a custom storage mechanism, you can do so by implementing the `IClusterMemberStore` interface and registering it with the service collection.
|
||||
|
||||
For example, consider the following implementation:
|
||||
|
||||
```csharp
|
||||
public class RedisClusterMemberStorage : IClusterMemberStore
|
||||
{
|
||||
...
|
||||
}
|
||||
```
|
||||
|
||||
You can then register the implementation with the service collection:
|
||||
|
||||
```csharp
|
||||
services.AddSingleton<IClusterMemberStore, RedisClusterMemberStore>();
|
||||
```
|
||||
|
|
@ -1,14 +0,0 @@
|
|||
namespace Elsa.ProtoActor.Cluster.AzureContainerApps;
|
||||
|
||||
public static class ResourceTagLabels
|
||||
{
|
||||
public static string LabelPrefix(string memberId) => $"proto.cluster-{memberId}|";
|
||||
public static string LabelHost(string memberId) => LabelPrefix(memberId) + "host";
|
||||
public static string LabelPort(string memberId) => LabelPrefix(memberId) + "port";
|
||||
public static string LabelKind(string memberId) => LabelPrefix(memberId) + "kind";
|
||||
public static string LabelCluster(string memberId) => LabelPrefix(memberId) + "cluster";
|
||||
public static string LabelMemberId(string memberId) => LabelPrefix(memberId) + LabelMemberIdWithoutPrefix;
|
||||
public const string LabelMemberIdWithoutPrefix = "memberId";
|
||||
public static string LabelReplicaName(string memberId) => LabelPrefix(memberId) + LabelReplicaNameWithoutPrefix;
|
||||
public const string LabelReplicaNameWithoutPrefix = "replicaName";
|
||||
}
|
||||
|
|
@ -0,0 +1,17 @@
|
|||
using Microsoft.Extensions.Options;
|
||||
using Proto.Cluster.AzureContainerApps.Stores.ResourceTags;
|
||||
|
||||
namespace Proto.Cluster.AzureContainerApps;
|
||||
|
||||
/// <summary>
|
||||
/// Validates the <see cref="ResourceTagsMemberStoreOptions"/> to ensure that the required options are provided.
|
||||
/// </summary>
|
||||
public class AzureContainerAppsProviderOptionsValidator : IPostConfigureOptions<AzureContainerAppsProviderOptions>
|
||||
{
|
||||
/// <inheritdoc />
|
||||
public void PostConfigure(string name, AzureContainerAppsProviderOptions options)
|
||||
{
|
||||
if (string.IsNullOrEmpty(options.ResourceGroupName))
|
||||
throw new Exception("No resource group provided");
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,46 @@
|
|||
using System;
|
||||
using System.Diagnostics.CodeAnalysis;
|
||||
using Azure.ResourceManager;
|
||||
using JetBrains.Annotations;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Options;
|
||||
using Proto.Cluster.AzureContainerApps.Stores.ResourceTags;
|
||||
|
||||
namespace Proto.Cluster.AzureContainerApps;
|
||||
|
||||
/// <summary>
|
||||
/// Adds extension methods to <see cref="IServiceCollection"/> for registering the Azure Container Apps provider
|
||||
/// </summary>
|
||||
[PublicAPI]
|
||||
public static class ServiceCollectionExtensions
|
||||
{
|
||||
/// <summary>
|
||||
/// Adds the Azure Container Apps provider to the service collection.
|
||||
/// </summary>
|
||||
/// <param name="services">The service collection to add the provider to.</param>
|
||||
/// <param name="armClientProvider">An <see cref="IArmClientProvider"/> to create <see cref="ArmClient"/> instances.</param>
|
||||
/// <param name="configure">An optional action to configure the provider options.</param>
|
||||
/// <returns>The service collection.</returns>
|
||||
public static IServiceCollection AddAzureContainerAppsProvider(this IServiceCollection services, IArmClientProvider? armClientProvider = default, [AllowNull] Action<AzureContainerAppsProviderOptions> configure = null)
|
||||
{
|
||||
var configureOptions = configure ?? (_ => { });
|
||||
services.Configure(configureOptions);
|
||||
services.ConfigureOptions<AzureContainerAppsProviderOptionsValidator>();
|
||||
services.AddSingleton<AzureContainerAppsProvider>();
|
||||
|
||||
if (armClientProvider != null)
|
||||
services.AddSingleton(armClientProvider);
|
||||
|
||||
// Register the default member store.
|
||||
services.AddSingleton<IClusterMemberStore, ResourceTagsClusterMemberStore>(sp =>
|
||||
{
|
||||
var clientProvider = sp.GetRequiredService<IArmClientProvider>();
|
||||
var logger = sp.GetRequiredService<ILogger<ResourceTagsClusterMemberStore>>();
|
||||
var options = sp.GetRequiredService<IOptions<AzureContainerAppsProviderOptions>>().Value;
|
||||
return new ResourceTagsClusterMemberStore(clientProvider, logger, options.ResourceGroupName, options.SubscriptionId);
|
||||
});
|
||||
|
||||
return services;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,28 @@
|
|||
using System.Threading.Tasks;
|
||||
using Azure.Core;
|
||||
using Azure.ResourceManager;
|
||||
using Azure.ResourceManager.Resources;
|
||||
|
||||
namespace Proto.Cluster.AzureContainerApps.Stores.ResourceTags;
|
||||
|
||||
/// <summary>
|
||||
/// Adds extension methods to the <see cref="ArmClient"/> class.
|
||||
/// </summary>
|
||||
public static class ArmClientUtils
|
||||
{
|
||||
/// <summary>
|
||||
/// Returns the specified resource group
|
||||
/// </summary>
|
||||
/// <param name="client">The <see cref="ArmClient"/> being extended.</param>
|
||||
/// <param name="resourceGroupName">The name of the resource group.</param>
|
||||
/// <param name="subscriptionId">The subscription ID. If not set, the default subscription will be used.</param>
|
||||
/// <param name="cancellationToken">The cancellation token.</param>
|
||||
/// <returns>The resource group.</returns>
|
||||
public static async Task<ResourceGroupResource> GetResourceGroupByNameAsync(this ArmClient client, string resourceGroupName, string? subscriptionId = default, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var resourceIdentifier = $"/subscriptions/{subscriptionId}";
|
||||
var subscription = subscriptionId != null ? client.GetSubscriptionResource(ResourceIdentifier.Parse(resourceIdentifier)) : await client.GetDefaultSubscriptionAsync(cancellationToken);
|
||||
var response = await subscription.GetResourceGroupAsync(resourceGroupName, cancellationToken).ConfigureAwait(false);
|
||||
return response.Value;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,40 @@
|
|||
namespace Proto.Cluster.AzureContainerApps.Stores.ResourceTags;
|
||||
|
||||
/// <summary>
|
||||
/// Static helpers for creating Azure resource tag names.
|
||||
/// </summary>
|
||||
public static class ResourceTagNames
|
||||
{
|
||||
/// <summary>
|
||||
/// The prefix for the tag name.
|
||||
/// </summary>
|
||||
public const string NamePrefix = "proto.cluster:member:";
|
||||
|
||||
/// <summary>
|
||||
/// The prefix for the tag name.
|
||||
/// </summary>
|
||||
public const string KindPrefix = "kind:";
|
||||
|
||||
/// <summary>
|
||||
/// Gets the prefixed name for the given member ID.
|
||||
/// </summary>
|
||||
/// <param name="memberId">The member ID.</param>
|
||||
/// <returns>The prefixed name.</returns>
|
||||
public static string Prefix(string memberId) => $"{NamePrefix}{memberId}";
|
||||
|
||||
/// <summary>
|
||||
/// Gets the prefixed name for the given member ID and name.
|
||||
/// </summary>
|
||||
/// <param name="memberId">The member ID.</param>
|
||||
/// <param name="name">The name.</param>
|
||||
/// <returns>The prefixed name.</returns>
|
||||
public static string Prefix(string memberId, string name) => $"{Prefix(memberId)}:{name}";
|
||||
|
||||
/// <summary>
|
||||
/// Gets the name for the cluster tag for the given member ID.
|
||||
/// </summary>
|
||||
/// <param name="memberId">The member ID.</param>
|
||||
/// <param name="kind">The kind.</param>
|
||||
/// <returns>The prefixed kind.</returns>
|
||||
public static string PrefixKind(string memberId, string kind) => Prefix(memberId, $"{KindPrefix}{kind}");
|
||||
}
|
||||
|
|
@ -0,0 +1,213 @@
|
|||
using System.Text.Json;
|
||||
using Azure.Core;
|
||||
using Azure.ResourceManager;
|
||||
using Azure.ResourceManager.AppContainers;
|
||||
using JetBrains.Annotations;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Options;
|
||||
|
||||
namespace Proto.Cluster.AzureContainerApps.Stores.ResourceTags;
|
||||
|
||||
/// <summary>
|
||||
/// Stores members in the form of resource tags of the Azure Container Application resource.
|
||||
/// </summary>
|
||||
[PublicAPI]
|
||||
public class ResourceTagsClusterMemberStore : IClusterMemberStore
|
||||
{
|
||||
private readonly IArmClientProvider _armClientProvider;
|
||||
private readonly ILogger _logger;
|
||||
private readonly string _containerAppName;
|
||||
private readonly string _resourceGroupName;
|
||||
private readonly string? _subscriptionId;
|
||||
|
||||
private ArmClient? _armClient;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="ResourceTagsClusterMemberStore"/> class.
|
||||
/// </summary>
|
||||
/// <param name="armArmClientProvider">The <see cref="IArmClientProvider"/> to use.</param>
|
||||
/// <param name="options">The options for this store.</param>
|
||||
/// <param name="logger">The logger to use.</param>
|
||||
public ResourceTagsClusterMemberStore(
|
||||
IArmClientProvider armArmClientProvider,
|
||||
IOptions<ResourceTagsMemberStoreOptions> options,
|
||||
ILogger<ResourceTagsClusterMemberStore> logger) : this(armArmClientProvider, logger, options.Value.ResourceGroupName, options.Value.SubscriptionId)
|
||||
{
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="ResourceTagsClusterMemberStore"/> class.
|
||||
/// </summary>
|
||||
/// <param name="armArmClientProvider">The <see cref="IArmClientProvider"/> to use.</param>
|
||||
/// <param name="logger">The logger to use.</param>
|
||||
/// <param name="resourceGroupName">The name of the resource group.</param>
|
||||
/// <param name="subscriptionId">The subscription ID.</param>
|
||||
internal ResourceTagsClusterMemberStore(
|
||||
IArmClientProvider armArmClientProvider,
|
||||
ILogger<ResourceTagsClusterMemberStore> logger,
|
||||
string resourceGroupName,
|
||||
string? subscriptionId = default)
|
||||
{
|
||||
_armClientProvider = armArmClientProvider;
|
||||
_logger = logger;
|
||||
_resourceGroupName = resourceGroupName;
|
||||
_subscriptionId = subscriptionId;
|
||||
_containerAppName = Environment.GetEnvironmentVariable("CONTAINER_APP_NAME") ?? throw new Exception("No app name provided");
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async ValueTask<ICollection<Member>> ListAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
var members = new List<Member>();
|
||||
var resourceGroupName = _resourceGroupName;
|
||||
var containerAppName = _containerAppName;
|
||||
var containerApp = await GetContainerAppAsync(cancellationToken).ConfigureAwait(false);
|
||||
|
||||
if (containerApp == null)
|
||||
{
|
||||
_logger.LogError("Resource: {ResourceName} in resource group: {ResourceGroup} is not found", containerAppName, resourceGroupName);
|
||||
return members.ToArray();
|
||||
}
|
||||
|
||||
// Get the app container managed environment in order to get the other container apps.
|
||||
var environmentId = containerApp.Data.EnvironmentId;
|
||||
var containerApps = (await GetContainerAppsAsync(environmentId, cancellationToken).ConfigureAwait(false)).ToList();
|
||||
|
||||
// Build a list of all tags from all container apps.
|
||||
var allTags = containerApps.SelectMany(x => x.Data.Tags).ToList();
|
||||
|
||||
var taggedMemberTags = allTags
|
||||
.Where(kvp => kvp.Key.StartsWith(ResourceTagNames.NamePrefix) && !kvp.Key.Contains(ResourceTagNames.KindPrefix))
|
||||
.Select(x => x);
|
||||
|
||||
foreach (var serializedTaggedMember in taggedMemberTags)
|
||||
{
|
||||
var taggedMember = Deserialize(serializedTaggedMember.Value);
|
||||
var memberId = serializedTaggedMember.Key[ResourceTagNames.NamePrefix.Length..];
|
||||
var member = new Member
|
||||
{
|
||||
Id = memberId,
|
||||
Host = taggedMember.Host,
|
||||
Port = taggedMember.Port,
|
||||
};
|
||||
|
||||
var kinds = allTags
|
||||
.Where(x => x.Key.StartsWith(ResourceTagNames.Prefix(member.Id, ResourceTagNames.KindPrefix)))
|
||||
.Select(x => x.Value);
|
||||
|
||||
member.Kinds.AddRange(kinds);
|
||||
members.Add(member);
|
||||
}
|
||||
|
||||
return members.ToArray();
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async ValueTask RegisterAsync(string clusterName, Member member, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var taggedMember = new TaggedMember(member.Host, member.Port, clusterName);
|
||||
var serializedTaggedMember = Serialize(taggedMember);
|
||||
|
||||
var tags = new Dictionary<string, string>
|
||||
{
|
||||
[ResourceTagNames.Prefix(member.Id)] = serializedTaggedMember
|
||||
};
|
||||
|
||||
foreach (var kind in member.Kinds)
|
||||
tags[ResourceTagNames.PrefixKind(member.Id, kind)] = kind;
|
||||
|
||||
try
|
||||
{
|
||||
await AddMemberTags(tags, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception x)
|
||||
{
|
||||
_logger.LogError(x, "Failed to update metadata");
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async ValueTask UnregisterAsync(string memberId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var containerApp = await GetContainerAppAsync(cancellationToken).ConfigureAwait(false);
|
||||
|
||||
if(containerApp == null)
|
||||
return;
|
||||
|
||||
var existingTags = containerApp.Data.Tags;
|
||||
var prefixedName = ResourceTagNames.Prefix(memberId);
|
||||
|
||||
foreach (var tag in existingTags)
|
||||
if (tag.Key.StartsWith(prefixedName))
|
||||
existingTags.Remove(tag.Key);
|
||||
|
||||
await containerApp.SetTagsAsync(existingTags, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async ValueTask ClearAsync(string clusterName, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var containerApp = await GetContainerAppAsync(cancellationToken).ConfigureAwait(false);
|
||||
|
||||
if(containerApp == null)
|
||||
return;
|
||||
|
||||
var existingTags = containerApp.Data.Tags;
|
||||
var prefixedName = ResourceTagNames.NamePrefix;
|
||||
|
||||
foreach (var tag in existingTags)
|
||||
if (tag.Key.StartsWith(prefixedName))
|
||||
existingTags.Remove(tag.Key);
|
||||
|
||||
await containerApp.SetTagsAsync(existingTags, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private async Task AddMemberTags(Dictionary<string, string> newTags, CancellationToken cancellationToken)
|
||||
{
|
||||
var containerApp = await GetContainerAppAsync(cancellationToken).ConfigureAwait(false);
|
||||
|
||||
if(containerApp == null)
|
||||
return;
|
||||
|
||||
var tags = containerApp.Data.Tags;
|
||||
|
||||
foreach (var tag in newTags)
|
||||
tags[tag.Key] = tag.Value;
|
||||
|
||||
await containerApp.SetTagsAsync(tags, cancellationToken);
|
||||
}
|
||||
|
||||
private async Task<ContainerAppResource?> GetContainerAppAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
var armClient = await GetArmClientAsync().ConfigureAwait(false);
|
||||
var subscriptionId = _subscriptionId;
|
||||
var resourceGroupName = _resourceGroupName;
|
||||
var resourceGroup = await armClient.GetResourceGroupByNameAsync(resourceGroupName, subscriptionId, cancellationToken).ConfigureAwait(false);
|
||||
var resource = await resourceGroup.GetContainerAppAsync(_containerAppName, cancellationToken).ConfigureAwait(false);
|
||||
return resource.HasValue ? resource.Value : default;
|
||||
}
|
||||
|
||||
private async Task<ContainerAppManagedEnvironmentResource> GetContainerAppManagedEnvironmentResourceAsync(ContainerAppResource containerApp, CancellationToken cancellationToken)
|
||||
{
|
||||
var armClient = await GetArmClientAsync().ConfigureAwait(false);
|
||||
var environmentId = containerApp.Data.EnvironmentId;
|
||||
var response = armClient.GetContainerAppManagedEnvironmentResource(environmentId);
|
||||
return response;
|
||||
}
|
||||
|
||||
private async Task<IEnumerable<ContainerAppResource>> GetContainerAppsAsync(ResourceIdentifier environmentId, CancellationToken cancellationToken)
|
||||
{
|
||||
var armClient = await GetArmClientAsync();
|
||||
var resourceGroupName = _resourceGroupName;
|
||||
var subscriptionId = _subscriptionId;
|
||||
var resourceGroup = await armClient.GetResourceGroupByNameAsync(resourceGroupName, subscriptionId, cancellationToken).ConfigureAwait(false);
|
||||
return resourceGroup.GetContainerApps().Where(x => x.Data.EnvironmentId == environmentId);
|
||||
}
|
||||
|
||||
private static IEnumerable<ContainerAppRevisionResource> GetActiveRevisionsWithTraffic(ContainerAppResource containerApp) =>
|
||||
containerApp.GetContainerAppRevisions().Where(r => r.HasData && (r.Data.IsActive ?? false) && r.Data.TrafficWeight > 0);
|
||||
|
||||
private static string Serialize(TaggedMember taggedMember) => JsonSerializer.Serialize(taggedMember);
|
||||
private static TaggedMember Deserialize(string json) => JsonSerializer.Deserialize<TaggedMember>(json)!;
|
||||
private async Task<ArmClient> GetArmClientAsync() => _armClient ??= await _armClientProvider.CreateClientAsync().ConfigureAwait(false);
|
||||
}
|
||||
|
|
@ -0,0 +1,17 @@
|
|||
namespace Proto.Cluster.AzureContainerApps.Stores.ResourceTags;
|
||||
|
||||
/// <summary>
|
||||
/// Options for the <see cref="ResourceTagsClusterMemberStore"/>.
|
||||
/// </summary>
|
||||
public class ResourceTagsMemberStoreOptions
|
||||
{
|
||||
/// <summary>
|
||||
/// The subscription ID to use. If not set, the default subscription will be used.
|
||||
/// </summary>
|
||||
public string? SubscriptionId { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// The name of the resource group to use.
|
||||
/// </summary>
|
||||
public string ResourceGroupName { get; set; } = default!;
|
||||
}
|
||||
|
|
@ -0,0 +1,17 @@
|
|||
using System;
|
||||
using Microsoft.Extensions.Options;
|
||||
|
||||
namespace Proto.Cluster.AzureContainerApps.Stores.ResourceTags;
|
||||
|
||||
/// <summary>
|
||||
/// Validates the <see cref="ResourceTagsMemberStoreOptions"/> to ensure that the required options are provided.
|
||||
/// </summary>
|
||||
public class ResourceTagsMemberStoreOptionsValidator : IPostConfigureOptions<ResourceTagsMemberStoreOptions>
|
||||
{
|
||||
/// <inheritdoc />
|
||||
public void PostConfigure(string name, ResourceTagsMemberStoreOptions options)
|
||||
{
|
||||
if (string.IsNullOrEmpty(options.ResourceGroupName))
|
||||
throw new Exception("No resource group provided");
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,28 @@
|
|||
using System;
|
||||
using System.Diagnostics.CodeAnalysis;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.DependencyInjection.Extensions;
|
||||
|
||||
namespace Proto.Cluster.AzureContainerApps.Stores.ResourceTags;
|
||||
|
||||
/// <summary>
|
||||
/// Adds extension methods to <see cref="IServiceCollection"/> for registering the Azure Container Apps provider
|
||||
/// </summary>
|
||||
public static class ServiceCollectionExtensions
|
||||
{
|
||||
/// <summary>
|
||||
/// Adds the <see cref="ResourceTagsClusterMemberStore"/> to the service collection.
|
||||
/// </summary>
|
||||
/// <param name="services">The service collection to add the provider to.</param>
|
||||
/// <param name="configure">An optional action to configure the provider options.</param>
|
||||
/// <returns>The service collection.</returns>
|
||||
public static IServiceCollection AddResourceTagsMemberStore(this IServiceCollection services, [AllowNull]Action<ResourceTagsMemberStoreOptions> configure = null)
|
||||
{
|
||||
var configureOptions = configure ?? (_ => { });
|
||||
services.Configure(configureOptions);
|
||||
services.ConfigureOptions<ResourceTagsMemberStoreOptionsValidator>();
|
||||
services.Replace(new ServiceDescriptor(typeof(IClusterMemberStore), typeof(ResourceTagsClusterMemberStore), ServiceLifetime.Singleton));
|
||||
|
||||
return services;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,6 @@
|
|||
namespace Proto.Cluster.AzureContainerApps.Stores.ResourceTags;
|
||||
|
||||
/// <summary>
|
||||
/// A member with a cluster name.
|
||||
/// </summary>
|
||||
public record TaggedMember(string Host, int Port, string Cluster);
|
||||
|
|
@ -0,0 +1,20 @@
|
|||
FROM mcr.microsoft.com/dotnet/aspnet:7.0 AS base
|
||||
WORKDIR /app
|
||||
EXPOSE 80
|
||||
EXPOSE 443
|
||||
|
||||
FROM mcr.microsoft.com/dotnet/sdk:7.0 AS build
|
||||
WORKDIR /src
|
||||
COPY ["src/samples/aspnet/Elsa.Samples.AspNet.ProtoActorRuntime.AzureContainerApps/Elsa.Samples.AspNet.ProtoActorRuntime.AzureContainerApps.csproj", "src/samples/aspnet/Elsa.Samples.AspNet.ProtoActorRuntime.AzureContainerApps/"]
|
||||
RUN dotnet restore "src/samples/aspnet/Elsa.Samples.AspNet.ProtoActorRuntime.AzureContainerApps/Elsa.Samples.AspNet.ProtoActorRuntime.AzureContainerApps.csproj"
|
||||
COPY . .
|
||||
WORKDIR "/src/src/samples/aspnet/Elsa.Samples.AspNet.ProtoActorRuntime.AzureContainerApps"
|
||||
RUN dotnet build "Elsa.Samples.AspNet.ProtoActorRuntime.AzureContainerApps.csproj" -c Release -o /app/build
|
||||
|
||||
FROM build AS publish
|
||||
RUN dotnet publish "Elsa.Samples.AspNet.ProtoActorRuntime.AzureContainerApps.csproj" -c Release -o /app/publish
|
||||
|
||||
FROM base AS final
|
||||
WORKDIR /app
|
||||
COPY --from=publish /app/publish .
|
||||
ENTRYPOINT ["dotnet", "Elsa.Samples.AspNet.ProtoActorRuntime.AzureContainerApps.dll"]
|
||||
|
|
@ -0,0 +1,34 @@
|
|||
<Project Sdk="Microsoft.NET.Sdk.Web">
|
||||
|
||||
<PropertyGroup>
|
||||
<TargetFramework>net7.0</TargetFramework>
|
||||
<Nullable>enable</Nullable>
|
||||
<ImplicitUsings>enable</ImplicitUsings>
|
||||
<DockerDefaultTargetOS>Linux</DockerDefaultTargetOS>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<Content Include="..\..\..\..\.dockerignore">
|
||||
<Link>.dockerignore</Link>
|
||||
</Content>
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\..\..\bundles\Elsa\Elsa.csproj" />
|
||||
<ProjectReference Include="..\..\..\modules\Elsa.EntityFrameworkCore.Sqlite\Elsa.EntityFrameworkCore.Sqlite.csproj" />
|
||||
<ProjectReference Include="..\..\..\modules\Elsa.EntityFrameworkCore\Elsa.EntityFrameworkCore.csproj" />
|
||||
<ProjectReference Include="..\..\..\modules\Elsa.Identity\Elsa.Identity.csproj" />
|
||||
<ProjectReference Include="..\..\..\modules\Elsa.JavaScript\Elsa.JavaScript.csproj" />
|
||||
<ProjectReference Include="..\..\..\modules\Elsa.ProtoActor.Cluster.AzureContainerApps\Elsa.ProtoActor.Cluster.AzureContainerApps.csproj" />
|
||||
<ProjectReference Include="..\..\..\modules\Elsa.ProtoActor\Elsa.ProtoActor.csproj" />
|
||||
<ProjectReference Include="..\..\..\modules\Elsa.Scheduling\Elsa.Scheduling.csproj" />
|
||||
<ProjectReference Include="..\..\..\modules\Elsa.Workflows.Api\Elsa.Workflows.Api.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Azure.Identity" Version="1.8.2" />
|
||||
<PackageReference Include="Azure.ResourceManager" Version="1.4.0" />
|
||||
<PackageReference Include="Proto.Persistence.Sqlite" Version="1.1.0" />
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
|
|
@ -0,0 +1,108 @@
|
|||
using Elsa.EntityFrameworkCore.Extensions;
|
||||
using Elsa.EntityFrameworkCore.Modules.Labels;
|
||||
using Elsa.EntityFrameworkCore.Modules.Management;
|
||||
using Elsa.EntityFrameworkCore.Modules.Runtime;
|
||||
using Elsa.Extensions;
|
||||
using Elsa.ProtoActor.Protos;
|
||||
using Google.Protobuf.WellKnownTypes;
|
||||
using Microsoft.Data.Sqlite;
|
||||
using Proto.Cluster.AzureContainerApps;
|
||||
using Proto.Persistence.Sqlite;
|
||||
using Proto.Remote;
|
||||
using Proto.Remote.GrpcNet;
|
||||
|
||||
var builder = WebApplication.CreateBuilder(args);
|
||||
var services = builder.Services;
|
||||
var configuration = builder.Configuration;
|
||||
var sqliteConnectionString = configuration.GetConnectionString("Sqlite")!;
|
||||
var identitySection = configuration.GetSection("Identity");
|
||||
var identityTokenSection = identitySection.GetSection("Tokens");
|
||||
var protoActorSection = configuration.GetSection("ProtoActor");
|
||||
var protoActorClusterSection = protoActorSection.GetSection("Cluster");
|
||||
|
||||
// Configure Proto Actor cluster provider services.
|
||||
services.AddAzureContainerAppsProvider(ArmClientProviders.DefaultAzureCredential, options => protoActorClusterSection.GetSection("AzureContainerApps").Bind(options));
|
||||
|
||||
// Add Elsa services.
|
||||
services
|
||||
.AddElsa(elsa => elsa
|
||||
.AddActivitiesFrom<Program>()
|
||||
.UseIdentity(identity =>
|
||||
{
|
||||
identity.IdentityOptions = options => identitySection.Bind(options);
|
||||
identity.TokenOptions = options => identityTokenSection.Bind(options);
|
||||
identity.UseConfigurationBasedUserProvider(options => identitySection.Bind(options));
|
||||
identity.UseConfigurationBasedApplicationProvider(options => identitySection.Bind(options));
|
||||
identity.UseConfigurationBasedRoleProvider(options => identitySection.Bind(options));
|
||||
})
|
||||
.UseDefaultAuthentication()
|
||||
.UseWorkflowManagement(management =>
|
||||
{
|
||||
// Use EF core for workflow definitions and instances.
|
||||
management.UseEntityFrameworkCore(m => m.UseSqlite(sqliteConnectionString));
|
||||
})
|
||||
.UseWorkflowRuntime(runtime =>
|
||||
{
|
||||
// Use EF core for triggers and bookmarks.
|
||||
runtime.UseEntityFrameworkCore(ef => ef.UseSqlite(sqliteConnectionString));
|
||||
|
||||
// Use EF core for execution log records.
|
||||
runtime.UseExecutionLogRecords(log => log.UseEntityFrameworkCore(ef => ef.UseSqlite(sqliteConnectionString)));
|
||||
|
||||
// Install a workflow state exporter to capture workflow states and store them in IWorkflowInstanceStore.
|
||||
runtime.UseAsyncWorkflowStateExporter();
|
||||
|
||||
// Use Proto.Actor for workflow execution.
|
||||
runtime.UseProtoActor(protoActor =>
|
||||
{
|
||||
var advertisedHost = ConfigUtils.FindSmallestIpAddress().ToString();
|
||||
|
||||
protoActor.ClusterProvider = sp => sp.GetRequiredService<AzureContainerAppsProvider>();
|
||||
|
||||
protoActor.RemoteConfig = _ => GrpcNetRemoteConfig
|
||||
.BindTo(advertisedHost)
|
||||
.WithProtoMessages(EmptyReflection.Descriptor)
|
||||
.WithProtoMessages(MessagesReflection.Descriptor)
|
||||
.WithLogLevelForDeserializationErrors(LogLevel.Critical)
|
||||
.WithRemoteDiagnostics(true); // required by proto.actor dashboard
|
||||
|
||||
protoActor.PersistenceProvider = _ => new SqliteProvider(new SqliteConnectionStringBuilder(sqliteConnectionString));
|
||||
});
|
||||
})
|
||||
.UseLabels(labels => labels.UseEntityFrameworkCore(ef => ef.UseSqlite(sqliteConnectionString)))
|
||||
.UseScheduling()
|
||||
.UseWorkflowsApi(api => api.AddFastEndpointsAssembly<Program>())
|
||||
.UseJavaScript()
|
||||
.UseLiquid()
|
||||
.UseHttp()
|
||||
);
|
||||
|
||||
services.AddHealthChecks();
|
||||
services.AddCors(cors => cors.AddDefaultPolicy(policy => policy.AllowAnyHeader().AllowAnyMethod().AllowAnyOrigin()));
|
||||
|
||||
// Configure middleware pipeline.
|
||||
var app = builder.Build();
|
||||
|
||||
if (app.Environment.IsDevelopment())
|
||||
app.UseDeveloperExceptionPage();
|
||||
|
||||
// CORS.
|
||||
app.UseCors();
|
||||
|
||||
// Health checks.
|
||||
app.MapHealthChecks("/");
|
||||
|
||||
app.UseAuthentication();
|
||||
app.UseAuthorization();
|
||||
|
||||
// Elsa API endpoints for designer.
|
||||
app.UseWorkflowsApi();
|
||||
|
||||
// Captures unhandled exceptions and returns a JSON response.
|
||||
app.UseJsonSerializationErrorHandler();
|
||||
|
||||
// Elsa HTTP Endpoint activities
|
||||
app.UseWorkflows();
|
||||
|
||||
// Run.
|
||||
app.Run();
|
||||
|
|
@ -0,0 +1,37 @@
|
|||
{
|
||||
"iisSettings": {
|
||||
"windowsAuthentication": false,
|
||||
"anonymousAuthentication": true,
|
||||
"iisExpress": {
|
||||
"applicationUrl": "http://localhost:23224",
|
||||
"sslPort": 44356
|
||||
}
|
||||
},
|
||||
"profiles": {
|
||||
"http": {
|
||||
"commandName": "Project",
|
||||
"dotnetRunMessages": true,
|
||||
"launchBrowser": true,
|
||||
"applicationUrl": "http://localhost:5244",
|
||||
"environmentVariables": {
|
||||
"ASPNETCORE_ENVIRONMENT": "Development"
|
||||
}
|
||||
},
|
||||
"https": {
|
||||
"commandName": "Project",
|
||||
"dotnetRunMessages": true,
|
||||
"launchBrowser": true,
|
||||
"applicationUrl": "https://localhost:7187;http://localhost:5244",
|
||||
"environmentVariables": {
|
||||
"ASPNETCORE_ENVIRONMENT": "Development"
|
||||
}
|
||||
},
|
||||
"IIS Express": {
|
||||
"commandName": "IISExpress",
|
||||
"launchBrowser": true,
|
||||
"environmentVariables": {
|
||||
"ASPNETCORE_ENVIRONMENT": "Development"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,10 @@
|
|||
# Server
|
||||
|
||||
This project represents an Elsa application that hosts workflows and exposes API endpoints to manage & execute workflows using the Proto Actor runtime and Azure Container Apps.
|
||||
|
||||
## Secrets
|
||||
The following are the secrets stored in hashed form in appsettings.json:
|
||||
|
||||
**API key**: `4E753976726458745954355043687772-e54d5a2c-33a3-4c05-a216-b09569062aed`
|
||||
**Admin user**: `admin`
|
||||
**Admin password**: `password`
|
||||
|
|
@ -0,0 +1,18 @@
|
|||
{
|
||||
"Logging": {
|
||||
"LogLevel": {
|
||||
"Default": "Debug",
|
||||
"Microsoft.AspNetCore": "Warning"
|
||||
}
|
||||
},
|
||||
"AllowedHosts": "*",
|
||||
"ProtoActor": {
|
||||
"Cluster": {
|
||||
"AzureContainerApps": {
|
||||
"SubscriptionId": "8e23037a-420f-4ad0-9594-9d194de29e84",
|
||||
"ResourceGroupName": "elsa-workflows-test",
|
||||
"PollInterval": "00:00:05"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,54 @@
|
|||
{
|
||||
"Logging": {
|
||||
"LogLevel": {
|
||||
"Default": "Information",
|
||||
"Microsoft.AspNetCore": "Warning"
|
||||
}
|
||||
},
|
||||
"AllowedHosts": "*",
|
||||
"ConnectionStrings": {
|
||||
"Sqlite": "Data Source=elsa.sqlite.db;Cache=Shared;"
|
||||
},
|
||||
"Identity": {
|
||||
"Tokens": {
|
||||
"SigningKey": "secret-signing-key",
|
||||
"AccessTokenLifetime": "1:00:00:00",
|
||||
"RefreshTokenLifetime": "1:00:10:00"
|
||||
},
|
||||
"Roles": [{
|
||||
"Id": "admin",
|
||||
"Name": "Administrator",
|
||||
"Permissions": ["*"]
|
||||
}],
|
||||
"Users": [
|
||||
{
|
||||
"Id": "a2323f46-42db-4e15-af8b-94238717d817",
|
||||
"Name": "admin",
|
||||
"HashedPassword": "TfKzh9RLix6FPcCNeHLkGrysFu3bYxqzGqduNdi8v1U=",
|
||||
"HashedPasswordSalt": "JEy9kBlhHCNsencitRHlGxmErmSgY+FVyMJulCH27Ds=",
|
||||
"Roles": ["admin"]
|
||||
}
|
||||
],
|
||||
"Applications": [{
|
||||
"id": "529572c2df854b13807b8bf23f1784cd",
|
||||
"name": "Postman",
|
||||
"roles": [
|
||||
"admin"
|
||||
],
|
||||
"clientId": "Nu9vrdXtYT5PChwr",
|
||||
"clientSecret": "011pp2C$|j01-qrMZpC9VC0F00XCJq(5",
|
||||
"hashedApiKey": "d0rDld3A+ugKmdctGtMzOLTYjQFkOlUWN+kt0VyW9D0=",
|
||||
"hashedApiKeySalt": "EnutGOyy5MuJWV0fF5jCQiciK7a8PU/DRF+fr6nekSY=",
|
||||
"hashedClientSecret": "ERia2zBcCSWb/9dvB0grQ9yf7fWgFrClNeR8A5RMTzk=",
|
||||
"hashedClientSecretSalt": "z3z8KmzHt+xkAj/zYTXcB8I7y0xAkLm95v4Er/oNqiY="
|
||||
}]
|
||||
},
|
||||
"ProtoActor": {
|
||||
"ClusterProvider": {
|
||||
"AzureContainerApps": {
|
||||
"ResourceGroupName": "ElsaWorkflows",
|
||||
"PollInterval": "00:00:05"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue