Skip to content
Merged
Show file tree
Hide file tree
Changes from 3 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.
Resolved

public Task SubscribeAsync(
IDistributedApplicationEventing eventing,
DistributedApplicationExecutionContext executionContext,
Expand All @@ -30,6 +37,12 @@ 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>();
// Ports configured anywhere in the app model are bound later, so probing cannot see them.
var reservedPorts = @event.Model.Resources
.SelectMany(resource => resource.Annotations.OfType<EndpointAnnotation>())
.Select(endpoint => endpoint.Port)

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This probably needs to filter to endpoint.IsAllocated. endpoint allocation can occur asyncronously, so not everythign is guaranteed to have a port allocated at this point, in which case this method may throw or stall.

Either

  1. Guard by only reading port if endpoint.IsAllocated
  2. await endpoint.Property(EndpointProperty.Port).GetValueAsync() to guarantee you get hte port

Probably the former as otherwise you may hang forever.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Neither read blocks here. EndpointAnnotation.Port is the configured value (a plain property), and AllocatedEndpoint returns null until the snapshot is set, without waiting. So an endpoint that is still being allocated never throws or stalls this handler.

I didn't filter to IsAllocated, though. At BeforeStartEvent almost nothing is allocated yet, so that filter would drop the configured ports. Those are the ones this set must hold, because their sockets don't exist yet and probing can't see them.

ef2af1d adds already-allocated ports to the set as well, so it is now configured ports plus allocated ports. I moved the logic into GetReservedPorts and added GetReservedPortsIncludesConfiguredAndAllocatedPortsOnly: a configured port and an allocated port are both reserved, and a pending dynamic endpoint is skipped.

.OfType<int>()
.ToHashSet();
var allocatedResources = new List<FlociAzureServiceBusResource>();
try
{
Expand All @@ -52,7 +65,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 +99,60 @@ await eventing.PublishAsync(
}
}

private static int SelectPort(List<TcpListener> listeners)
internal static int SelectPort(List<TcpListener> listeners, HashSet<int> reservedPorts)
Comment thread
afscrome marked this conversation as resolved.
Resolved
{
// 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;
// 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.
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)
{
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,74 @@ await builder.Eventing.PublishAsync(
.GetValueAsync(TestContext.Current.CancellationToken));
}

[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.
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