Silo Metadata and Placement Filtering

src/Orleans.Core.Abstractions/Manifest/GrainProperties.cs

@@ -68,6 +68,11 @@ public static class WellKnownGrainTypeProperties
         /// </summary>
         public const string PlacementStrategy = "placement-strategy";
 
+        /// <summary>
+        /// The name of the placement strategy for grains of this type.
+        /// </summary>
+        public const string PlacementFilter = "placement-filter";
+
         /// <summary>
         /// The directory policy for grains of this type.
         /// </summary>

src/Orleans.Core.Abstractions/Placement/PlacementFilterAttribute.cs

@@ -0,0 +1,25 @@
+using System;
+using System.Collections.Generic;
+using Orleans.Metadata;
+using Orleans.Runtime;
+
+namespace Orleans.Placement;
+
+/// <summary>
+/// Base for all placement filter marker attributes.
+/// </summary>
+[AttributeUsage(AttributeTargets.Class, AllowMultiple = true)]
+public abstract class PlacementFilterAttribute : Attribute, IGrainPropertiesProviderAttribute
+{
+    public PlacementFilterStrategy PlacementFilterStrategy { get; private set; }
+
+    protected PlacementFilterAttribute(PlacementFilterStrategy placement)
+    {
+        ArgumentNullException.ThrowIfNull(placement);
+        PlacementFilterStrategy = placement;
+    }
+
+    /// <inheritdoc />
+    public virtual void Populate(IServiceProvider services, Type grainClass, GrainType grainType, Dictionary<string, string> properties)
+        => PlacementFilterStrategy?.PopulateGrainProperties(services, grainClass, grainType, properties);
+}

src/Orleans.Core.Abstractions/Placement/PlacementFilterStrategy.cs

@@ -0,0 +1,76 @@
+using System;
+using System.Collections.Generic;
+using Orleans.Metadata;
+using Orleans.Runtime;
+
+namespace Orleans.Placement;
+
+public abstract class PlacementFilterStrategy
+{
+    public int Order { get; private set; }
+
+    protected PlacementFilterStrategy(int order)
+    {
+        Order = order;
+    }
+
+    /// <summary>
+    /// Initializes an instance of this type using the provided grain properties.
+    /// </summary>
+    /// <param name="properties">
+    /// The grain properties.
+    /// </param>
+    public void Initialize(GrainProperties properties)
+    {
+        var orderProperty = GetPlacementFilterGrainProperty("order", properties);
+        if (!int.TryParse(orderProperty, out var parsedOrder))
+        {
+            throw new ArgumentException("Invalid order property value.");
+        }
+
+        Order = parsedOrder;
+
+        AdditionalInitialize(properties);
+    }
+
+    public virtual void AdditionalInitialize(GrainProperties properties)
+    {
+
+    }
+
+    /// <summary>
+    /// Populates grain properties to specify the preferred placement strategy.
+    /// </summary>
+    /// <param name="services">The service provider.</param>
+    /// <param name="grainClass">The grain class.</param>
+    /// <param name="grainType">The grain type.</param>
+    /// <param name="properties">The grain properties which will be populated by this method call.</param>
+    public void PopulateGrainProperties(IServiceProvider services, Type grainClass, GrainType grainType, Dictionary<string, string> properties)
+    {
+        var typeName = GetType().Name;
+        if (properties.TryGetValue(WellKnownGrainTypeProperties.PlacementFilter, out var existingValue))
+        {
+            properties[WellKnownGrainTypeProperties.PlacementFilter] = $"{existingValue},{typeName}";
+        }
+        else
+        {
+            properties[WellKnownGrainTypeProperties.PlacementFilter] = typeName;
+        }
+
+        properties[$"{WellKnownGrainTypeProperties.PlacementFilter}.{typeName}.order"] = Order.ToString();
+
+        foreach (var additionalGrainProperty in GetAdditionalGrainProperties(services, grainClass, grainType, properties))
+        {
+            properties[$"{WellKnownGrainTypeProperties.PlacementFilter}.{typeName}.{additionalGrainProperty.Key}"] = additionalGrainProperty.Value;
+        }
+    }
+
+    protected string GetPlacementFilterGrainProperty(string key, GrainProperties properties)
+    {
+        var typeName = GetType().Name;
+        return properties.Properties.TryGetValue($"{WellKnownGrainTypeProperties.PlacementFilter}.{typeName}.{key}", out var value) ? value : null;
+    }
+
+    protected virtual IEnumerable<KeyValuePair<string, string>> GetAdditionalGrainProperties(IServiceProvider services, Type grainClass, GrainType grainType, IReadOnlyDictionary<string, string> existingProperties)
+        => Array.Empty<KeyValuePair<string, string>>();
+}

src/Orleans.Core/Placement/IPlacementFilterDirector.cs

@@ -0,0 +1,11 @@
+using System.Collections.Generic;
+using Orleans.Runtime;
+using Orleans.Runtime.Placement;
+
+namespace Orleans.Placement;
+
+public interface IPlacementFilterDirector
+{
+    IEnumerable<SiloAddress> Filter(PlacementFilterStrategy filterStrategy, PlacementTarget target,
+        IEnumerable<SiloAddress> silos);
+}

src/Orleans.Core/Placement/PlacementFilterExtensions.cs

@@ -0,0 +1,23 @@
+using Microsoft.Extensions.DependencyInjection;
+
+namespace Orleans.Placement;
+
+public static class PlacementFilterExtensions
+{
+    /// <summary>
+    /// Configures a <typeparamref name="TFilter"/> for filtering candidate grain placements.
+    /// </summary>
+    /// <typeparam name="TFilter">The placement filter.</typeparam>
+    /// <typeparam name="TDirector">The placement filter director.</typeparam>
+    /// <param name="services">The service collection.</param>
+    /// <param name="strategyLifetime">The lifetime of the placement strategy.</param>
+    /// <returns>The service collection.</returns>
+    public static void AddPlacementFilter<TFilter, TDirector>(this IServiceCollection services, ServiceLifetime strategyLifetime)
+        where TFilter : PlacementFilterStrategy
+        where TDirector : class, IPlacementFilterDirector
+    {
+        services.Add(ServiceDescriptor.DescribeKeyed(typeof(PlacementFilterStrategy), typeof(TFilter).Name, typeof(TFilter), strategyLifetime));
+        services.AddKeyedSingleton<IPlacementFilterDirector, TDirector>(typeof(TFilter));
+    }
+
+}

src/Orleans.Runtime/Hosting/DefaultSiloServices.cs

@@ -43,8 +43,7 @@
 using Orleans.Serialization.Internal;
 using Orleans.Core;
 using Orleans.Placement.Repartitioning;
-using Orleans.GrainDirectory;
-using Orleans.Runtime.Hosting;
+using Orleans.Runtime.Placement.Filtering;
 
 namespace Orleans.Hosting
 {
@@ -206,6 +205,10 @@ internal static void AddDefaultServices(ISiloBuilder builder)
             // Configure the default placement strategy.
             services.TryAddSingleton<PlacementStrategy, RandomPlacement>();
 
+            // Placement filters
+            services.AddSingleton<PlacementFilterStrategyResolver>();
+            services.AddSingleton<PlacementFilterDirectorResolver>();
+
             // Placement directors
             services.AddPlacementDirector<RandomPlacement, RandomPlacementDirector>();
             services.AddPlacementDirector<PreferLocalPlacement, PreferLocalPlacementDirector>();

src/Orleans.Runtime/MembershipService/SiloMetadata/ISiloMetadataCache.cs

@@ -0,0 +1,6 @@
+namespace Orleans.Runtime.MembershipService.SiloMetadata;
+#nullable enable
+public interface ISiloMetadataCache
+{
+    SiloMetadata GetMetadata(SiloAddress siloAddress);
+}

src/Orleans.Runtime/MembershipService/SiloMetadata/ISiloMetadataClient.cs

@@ -0,0 +1,9 @@
+using System.Threading.Tasks;
+using Orleans.Services;
+
+namespace Orleans.Runtime.MembershipService.SiloMetadata;
+
+public interface ISiloMetadataClient : IGrainServiceClient<ISiloMetadataGrainService>
+{
+    Task<SiloMetadata> GetSiloMetadata(SiloAddress siloAddress);
+}

src/Orleans.Runtime/MembershipService/SiloMetadata/ISiloMetadataGrainService.cs

@@ -0,0 +1,11 @@
+using System.Threading.Tasks;
+using Orleans.Services;
+
+namespace Orleans.Runtime.MembershipService.SiloMetadata;
+
+[Alias("Orleans.Runtime.MembershipService.SiloMetadata.ISiloMetadataGrainService")]
+public interface ISiloMetadataGrainService : IGrainService
+{
+    [Alias("GetSiloMetadata")]
+    Task<SiloMetadata> GetSiloMetadata();
+}

src/Orleans.Runtime/MembershipService/SiloMetadata/SiloMetadaCache.cs

@@ -0,0 +1,102 @@
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+using Orleans.Configuration;
+using Orleans.Internal;
+
+namespace Orleans.Runtime.MembershipService.SiloMetadata;
+#nullable enable
+internal class SiloMetadataCache(
+    ISiloMetadataClient siloMetadataClient,
+    MembershipTableManager membershipTableManager,
+    ILogger<SiloMetadataCache> logger)
+    : ISiloMetadataCache, ILifecycleParticipant<ISiloLifecycle>, IDisposable
+{
+    private readonly ConcurrentDictionary<SiloAddress, SiloMetadata> _metadata = new();
+    private readonly CancellationTokenSource _cts = new();
+
+    void ILifecycleParticipant<ISiloLifecycle>.Participate(ISiloLifecycle lifecycle)
+    {
+        var tasks = new List<Task>(1);
+        var cancellation = new CancellationTokenSource();
+        Task OnRuntimeInitializeStart(CancellationToken _)
+        {
+            tasks.Add(Task.Run(() => this.ProcessMembershipUpdates(cancellation.Token)));
+            return Task.CompletedTask;
+        }
+
+        async Task OnRuntimeInitializeStop(CancellationToken ct)
+        {
+            cancellation.Cancel(throwOnFirstException: false);
+            var shutdownGracePeriod = Task.WhenAll(Task.Delay(ClusterMembershipOptions.ClusteringShutdownGracePeriod), ct.WhenCancelled());
+            await Task.WhenAny(shutdownGracePeriod, Task.WhenAll(tasks));
+        }
+
+        lifecycle.Subscribe(
+            nameof(ClusterMembershipService),
+            ServiceLifecycleStage.RuntimeInitialize,
+            OnRuntimeInitializeStart,
+            OnRuntimeInitializeStop);
+    }
+
+
+    private async Task ProcessMembershipUpdates(CancellationToken ct)
+    {
+        try
+        {
+            if (logger.IsEnabled(LogLevel.Debug)) logger.LogDebug("Starting to process membership updates");
+            await foreach (var update in membershipTableManager.MembershipTableUpdates.WithCancellation(ct))
+            {
+                // Add entries for members that aren't already in the cache
+                foreach (var membershipEntry in update.Entries.Where(e => e.Value.Status != SiloStatus.Dead))
+                {
+                    if (!_metadata.ContainsKey(membershipEntry.Key))
+                    {
+                        try
+                        {
+                            var metadata = await siloMetadataClient.GetSiloMetadata(membershipEntry.Key);
+                            _metadata.TryAdd(membershipEntry.Key, metadata);
+                        }
+                        catch(Exception exception)
+                        {
+                            logger.LogError(exception, "Error fetching metadata for silo {Silo}", membershipEntry.Key);
+                        }
+                    }
+                }
+
+                // Add entries for members that aren't already in the cache
+                foreach (var membershipEntry in update.Entries.Where(e => e.Value.Status == SiloStatus.Dead))
+                {
+                    _metadata.TryRemove(membershipEntry.Key, out _);
+                }
+
+                // Remove entries for members that are no longer in the table
+                foreach (var silo in _metadata.Keys.ToList())
+                {
+                    if (!update.Entries.ContainsKey(silo))
+                    {
+                        _metadata.TryRemove(silo, out _);
+                    }
+                }
+            }
+        }
+        catch (Exception exception)
+        {
+            logger.LogError(exception, "Error processing membership updates");
+        }
+        finally
+        {
+            if (logger.IsEnabled(LogLevel.Debug)) logger.LogDebug("Stopping membership update processor");
+        }
+    }
+
+    public SiloMetadata GetMetadata(SiloAddress siloAddress) => _metadata.GetValueOrDefault(siloAddress) ?? SiloMetadata.Empty;
+
+    public void SetMetadata(SiloAddress siloAddress, SiloMetadata metadata) => _metadata.TryAdd(siloAddress, metadata);
+
+    public void Dispose() => _cts.Cancel();
+}

src/Orleans.Runtime/MembershipService/SiloMetadata/SiloMetadata.cs

@@ -0,0 +1,17 @@
+using System.Collections.Generic;
+using System.Collections.Immutable;
+
+namespace Orleans.Runtime.MembershipService.SiloMetadata;
+
+[GenerateSerializer]
+[Alias("Orleans.Runtime.MembershipService.SiloMetadata.SiloMetadata")]
+public record SiloMetadata
+{
+    public static SiloMetadata Empty { get; } = new SiloMetadata();
+
+    [Id(0)]
+    public ImmutableDictionary<string, string> Metadata { get; private set; } = ImmutableDictionary<string, string>.Empty;
+
+    internal void AddMetadata(IEnumerable<KeyValuePair<string, string>> metadata) => Metadata = Metadata.SetItems(metadata);
+    internal void AddMetadata(string key, string value) => Metadata = Metadata.SetItem(key, value);
+}

src/Orleans.Runtime/MembershipService/SiloMetadata/SiloMetadataClient.cs

@@ -0,0 +1,16 @@
+using System;
+using System.Threading.Tasks;
+using Orleans.Runtime.Services;
+
+namespace Orleans.Runtime.MembershipService.SiloMetadata;
+
+public class SiloMetadataClient(IServiceProvider serviceProvider)
+    : GrainServiceClient<ISiloMetadataGrainService>(serviceProvider), ISiloMetadataClient
+{
+    public async Task<SiloMetadata> GetSiloMetadata(SiloAddress siloAddress)
+    {
+        var grainService = GetGrainService(siloAddress);
+        var metadata = await grainService.GetSiloMetadata();
+        return metadata;
+    }
+}

src/Orleans.Runtime/MembershipService/SiloMetadata/SiloMetadataGrainService.cs

@@ -0,0 +1,22 @@
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+
+namespace Orleans.Runtime.MembershipService.SiloMetadata;
+
+public class SiloMetadataGrainService : GrainService, ISiloMetadataGrainService
+{
+    private readonly SiloMetadata _siloMetadata;
+
+    public SiloMetadataGrainService(IOptions<SiloMetadata> siloMetadata) : base()
+    {
+        _siloMetadata = siloMetadata.Value;
+    }
+
+    public SiloMetadataGrainService(IOptions<SiloMetadata> siloMetadata, GrainId grainId, Silo silo, ILoggerFactory loggerFactory) : base(grainId, silo, loggerFactory)
+    {
+        _siloMetadata = siloMetadata.Value;
+    }
+
+    public Task<SiloMetadata> GetSiloMetadata() => Task.FromResult(_siloMetadata);
+}