Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,13 @@ namespace CommunityToolkit.Aspire.Hosting.Floci;

internal sealed class FlociServiceBusEndpointAllocator : IDistributedApplicationEventingSubscriber
{
// Random host ports are taken from below every OS ephemeral range (Linux starts at 32768,
// Windows and macOS at 49152). An OS-assigned port is released before Docker binds it, and DCP
// can hand it to a project proxy in the meantime: on Windows the proxy's loopback listener then
// shadows Docker's wildcard bind, and on Linux Docker's bind fails.
private const int MinSelectedPort = 20000;
private const int MaxSelectedPortExclusive = 32768;
Comment thread
afscrome marked this conversation as resolved.

public Task SubscribeAsync(
IDistributedApplicationEventing eventing,
DistributedApplicationExecutionContext executionContext,
Expand All @@ -30,6 +37,7 @@ private static async Task AllocateEndpointsAsync(
// DCP does not allocate endpoints on these custom resources. Allocate before the
// parent's environment is evaluated, otherwise resolving its child's ports blocks startup.
var listeners = new List<TcpListener>();
var reservedPorts = GetReservedPorts(@event.Model);
var allocatedResources = new List<FlociAzureServiceBusResource>();
try
{
Expand All @@ -52,7 +60,7 @@ private static async Task AllocateEndpointsAsync(
continue;
}

endpoint.Port ??= SelectPort(listeners);
endpoint.Port ??= SelectPort(listeners, reservedPorts);
endpoint.TargetPort = endpoint.Port;
endpoint.AllocatedEndpoint = new AllocatedEndpoint(endpoint, KnownHostNames.Localhost, endpoint.Port.Value,
EndpointBindingMode.SingleAddress, null, KnownNetworkIdentifiers.LocalhostNetwork);
Expand Down Expand Up @@ -86,13 +94,72 @@ await eventing.PublishAsync(
}
}

private static int SelectPort(List<TcpListener> listeners)
internal static HashSet<int> GetReservedPorts(DistributedApplicationModel model)
{
// Ports configured anywhere in the app model are bound later, so probing cannot see them.
// Both reads are synchronous: Port is the configured value and AllocatedEndpoint is null until
// allocation completes, so endpoints that are still being allocated never block startup.
return model.Resources
.SelectMany(resource => resource.Annotations.OfType<EndpointAnnotation>())
.SelectMany(endpoint => new[] { endpoint.Port, endpoint.AllocatedEndpoint?.Port })
.OfType<int>()
.ToHashSet();
}

internal static int SelectPort(List<TcpListener> listeners, HashSet<int> reservedPorts)
Comment thread
afscrome marked this conversation as resolved.
{
// Start at a random offset so concurrent AppHosts spread out, then scan the whole range so a
// free port is always found when one exists. Reserved ports are skipped without probing.
int rangeSize = MaxSelectedPortExclusive - MinSelectedPort;
int offset = Random.Shared.Next(rangeSize);
for (int i = 0; i < rangeSize; i++)
{
int port = MinSelectedPort + (offset + i) % rangeSize;
Comment thread
thomhurst marked this conversation as resolved.
if (reservedPorts.Contains(port))
{
continue;
}

// Match Docker's bind address, as in the k3s allocator, and also probe IPv4 and IPv6 loopback, where
// another process's listener would shadow Docker's wildcard bind. This probes availability;
// it cannot reserve the port through Docker startup, which needs the socket released.
if (!IsAvailable(IPAddress.Loopback, port)
|| (Socket.OSSupportsIPv6 && !IsAvailable(IPAddress.IPv6Loopback, port)))
{
continue;
}

var listener = new TcpListener(IPAddress.Any, port);
try
{
listener.Start();
}
catch (SocketException)
{
listener.Dispose();
continue;
}

listeners.Add(listener);
reservedPorts.Add(port);
return port;
}

throw new DistributedApplicationException(
$"No free Service Bus host port was found in {MinSelectedPort}-{MaxSelectedPortExclusive - 1}.");
}

private static bool IsAvailable(IPAddress address, int port)
{
// Match Docker's bind address, as in the k3s allocator. This probes availability;
// it cannot reserve the port through Docker startup, which needs the socket released.
var listener = new TcpListener(IPAddress.Any, 0);
listeners.Add(listener);
listener.Start();
return ((IPEndPoint)listener.LocalEndpoint).Port;
using var listener = new TcpListener(address, port);
try
{
listener.Start();
return true;
}
catch (SocketException)
{
return false;
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -95,8 +95,6 @@ await builder.Eventing.PublishAsync(
.GetValueAsync(TestContext.Current.CancellationToken);
string? amqpTlsPort = await serviceBus.Resource.AmqpTlsEndpoint.Property(EndpointProperty.Port)
.GetValueAsync(TestContext.Current.CancellationToken);
Assert.InRange(int.Parse(amqpPort!), 1, 65535);
Assert.InRange(int.Parse(amqpTlsPort!), 1, 65535);
Assert.NotEqual(amqpPort, amqpTlsPort);
foreach (var endpoint in new[] { serviceBus.Resource.AmqpEndpoint, serviceBus.Resource.AmqpTlsEndpoint })
{
Expand All @@ -110,10 +108,19 @@ await builder.Eventing.PublishAsync(
{
Assert.Equal(configuredAmqpPort.Value.ToString(), amqpPort);
}
else
{
// Selected ports stay below every OS ephemeral range, where DCP allocates proxy ports.
Assert.InRange(int.Parse(amqpPort!), 20000, 32767);
}
if (configuredAmqpTlsPort.HasValue)
{
Assert.Equal(configuredAmqpTlsPort.Value.ToString(), amqpTlsPort);
}
else
{
Assert.InRange(int.Parse(amqpTlsPort!), 20000, 32767);
}

var resource = Assert.Single(appModel.Resources.OfType<FlociAzureContainerResource>());
Assert.True(resource.TryGetAnnotationsOfType(out IEnumerable<EnvironmentCallbackAnnotation>? envAnnotations));
Expand All @@ -137,6 +144,90 @@ await builder.Eventing.PublishAsync(
.GetValueAsync(TestContext.Current.CancellationToken));
}

[Fact]
public void GetReservedPortsIncludesConfiguredAndAllocatedPortsOnly()
{
IDistributedApplicationBuilder builder = DistributedApplication.CreateBuilder();

builder.AddContainer("configured", "image").WithEndpoint("tcp", endpoint => endpoint.Port = 21000);
builder.AddContainer("allocated", "image").WithEndpoint("tcp", endpoint =>
endpoint.AllocatedEndpoint = new AllocatedEndpoint(endpoint, "localhost", 22000));
builder.AddContainer("pending", "image").WithEndpoint(targetPort: 8080, name: "tcp");

using var app = builder.Build();
var appModel = app.Services.GetRequiredService<DistributedApplicationModel>();

Assert.Equal([21000, 22000], FlociServiceBusEndpointAllocator.GetReservedPorts(appModel).Order());
}

[Fact]
public void SelectPortSkipsReservedPorts()
{
// Leave exactly one free, unreserved port in the selection range.
int freePort = Enumerable.Range(20000, 12768).First(port =>
{
using var probe = new System.Net.Sockets.TcpListener(System.Net.IPAddress.Any, port);
try
{
probe.Start();
return true;
}
catch (System.Net.Sockets.SocketException)
{
return false;
}
});
var reserved = Enumerable.Range(20000, 12768).Where(port => port != freePort).ToHashSet();
var listeners = new List<System.Net.Sockets.TcpListener>();
try
{
Assert.Equal(freePort, FlociServiceBusEndpointAllocator.SelectPort(listeners, reserved));
Assert.Contains(freePort, reserved);
}
finally
{
listeners.ForEach(listener => listener.Stop());
}
}

[Fact]
public void SelectPortSkipsPortsHeldOnIPv6Loopback()
{
Assert.SkipUnless(System.Net.Sockets.Socket.OSSupportsIPv6, "IPv6 is not available.");

// Hold one port on ::1 only, and leave it plus one other free port unreserved.
var free = Enumerable.Range(20000, 12768).Where(IsFreeEverywhere).Take(2).ToArray();
using var ipv6Holder = new System.Net.Sockets.TcpListener(System.Net.IPAddress.IPv6Loopback, free[0]);
ipv6Holder.Start();
var reserved = Enumerable.Range(20000, 12768).Where(port => !free.Contains(port)).ToHashSet();
var listeners = new List<System.Net.Sockets.TcpListener>();
try
{
Assert.Equal(free[1], FlociServiceBusEndpointAllocator.SelectPort(listeners, reserved));
}
finally
{
listeners.ForEach(listener => listener.Stop());
}
}

private static bool IsFreeEverywhere(int port)
{
foreach (var address in new[] { System.Net.IPAddress.Any, System.Net.IPAddress.IPv6Loopback })
Comment thread
afscrome marked this conversation as resolved.
{
using var probe = new System.Net.Sockets.TcpListener(address, port);
try
{
probe.Start();
}
catch (System.Net.Sockets.SocketException)
{
return false;
}
}
return true;
}

[Fact]
public async Task WithServiceBusDoesNotConfigureTheDataPlaneInPublishMode()
{
Expand Down
Loading