feat: Key prefix redis options

Author: jaironalves

src/Redis/Orleans.DurableJobs.Redis/Hosting/RedisJobShardOptions.cs

@@ -21,6 +21,15 @@ public class RedisJobShardOptions
     /// </remarks>
     public Func<RedisJobShardOptions, Task<IConnectionMultiplexer>> CreateMultiplexer { get; set; } = DefaultCreateMultiplexer;
 
+    /// <summary>
+    /// Gets or sets the prefix for Redis keys used by the durable jobs provider.
+    /// </summary>
+    /// <remarks>
+    /// This prefix is combined with the service ID to create the final key prefix.
+    /// If not set, defaults to null and the standard pattern is used: "{ServiceId}/durablejobs".
+    /// </remarks>
+    public string? KeyPrefix { get; set; }
+
     /// <summary>
     /// Gets or sets the prefix for shard identifiers.
     /// </summary>

src/Redis/Orleans.DurableJobs.Redis/README.md

@@ -24,12 +24,28 @@ siloBuilder.UseRedisDurableJobs(options =>
 });
 ```
 
+### Advanced Configuration - Custom Key Prefix
+
+By default, Redis keys are prefixed with `{ServiceId}/durablejobs` (e.g., `my-service/durablejobs:shards:shard`). You can customize this prefix:
+
+```csharp
+siloBuilder.UseRedisDurableJobs(options =>
+{
+    options.ConfigurationOptions = ConfigurationOptions.Parse("localhost:6379");
+    options.KeyPrefix = "custom-prefix"; // Custom Redis key prefix
+    options.ShardPrefix = "my-app";
+});
+```
+
+This will result in Redis keys like `custom-prefix:shards:my-app` instead of the default pattern.
+
 ### Configuration Options
 
 | Option | Description | Default |
 |--------|-------------|---------|
 | `ConfigurationOptions` | Redis client configuration options (from StackExchange.Redis) | Required |
 | `CreateMultiplexer` | Optional delegate for custom connection logic (advanced scenarios) | Uses `ConfigurationOptions` |
+| `KeyPrefix` | Custom prefix for all Redis keys. If not set, defaults to `{ServiceId}/durablejobs` | `null` |
 | `ShardPrefix` | Prefix for shard identifiers in Redis | `"shard"` |
 | `MaxShardCreationRetries` | Maximum retries when creating a shard | `5` |
 | `MaxBatchSize` | Maximum operations per batch write | `128` |

src/Redis/Orleans.DurableJobs.Redis/RedisJobShard.cs

@@ -1,6 +1,8 @@
 using System.Diagnostics;
 using System.Threading.Channels;
 using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using Orleans.Configuration;
 using StackExchange.Redis;
 
 namespace Orleans.DurableJobs.Redis;
@@ -28,13 +30,15 @@ internal sealed partial class RedisJobShard : JobShard
     /// <param name="endTime">The end time of the shard's time range.</param>
     /// <param name="redis">The Redis connection multiplexer.</param>
     /// <param name="metadata">The shard metadata.</param>
+    /// <param name="keyPrefix">The Redis key prefix to use.</param>
     /// <param name="options">The Redis job shard options.</param>
     /// <param name="logger">The logger.</param>
     public RedisJobShard(string shardId,
             DateTimeOffset startTime,
             DateTimeOffset endTime,
             IConnectionMultiplexer redis,
             IDictionary<string, string> metadata,
+            string keyPrefix,
             RedisJobShardOptions options,
             ILogger<RedisJobShard> logger)
             : base(shardId, startTime, endTime)
@@ -52,8 +56,8 @@ public RedisJobShard(string shardId,
             MetadataVersion = version;
         }
 
-        _streamKey = $"durablejobs:shard:{Id}:stream";
-        _metaKey = $"durablejobs:shard:{Id}:meta";
+        _streamKey = $"{keyPrefix}:shard:{Id}:stream";
+        _metaKey = $"{keyPrefix}:shard:{Id}:meta";
 
         _storageOperationChannel = Channel.CreateUnbounded<StorageOperation>(new UnboundedChannelOptions
         {

src/Redis/Orleans.DurableJobs.Redis/RedisJobShardManager.cs

@@ -1,7 +1,9 @@
 using System.Collections.Concurrent;
 using System.Globalization;
+using System.Text;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
+using Orleans.Configuration;
 using StackExchange.Redis;
 
 namespace Orleans.DurableJobs.Redis;
@@ -14,8 +16,10 @@ public sealed partial class RedisJobShardManager : JobShardManager
     private readonly ILocalSiloDetails _localSiloDetails;
     private readonly IClusterMembershipService _clusterMembership;
     private readonly RedisJobShardOptions _options;
+    private readonly ClusterOptions _clusterOptions;
     private readonly ILogger<RedisJobShardManager> _logger;
     private readonly ILoggerFactory _loggerFactory;
+    private readonly RedisKey _keyPrefix;
 
     private IConnectionMultiplexer? _multiplexer;
     private RedisOperationsManager? _redisOps;
@@ -30,20 +34,26 @@ public sealed partial class RedisJobShardManager : JobShardManager
     /// </summary>
     /// <param name="localSiloDetails">The local silo details.</param>
     /// <param name="options">The Redis job shard options.</param>
+    /// <param name="clusterOptions">The cluster options.</param>
     /// <param name="clusterMembership">The cluster membership service.</param>
     /// <param name="loggerFactory">The logger factory.</param>
     public RedisJobShardManager(
         ILocalSiloDetails localSiloDetails,
         IOptions<RedisJobShardOptions> options,
+        IOptions<ClusterOptions> clusterOptions,
         IClusterMembershipService clusterMembership,
         ILoggerFactory loggerFactory)
         : base(localSiloDetails.SiloAddress)
     {
         _localSiloDetails = localSiloDetails ?? throw new ArgumentNullException(nameof(localSiloDetails));
         _options = options?.Value ?? throw new ArgumentNullException(nameof(options));
+        _clusterOptions = clusterOptions?.Value ?? throw new ArgumentNullException(nameof(clusterOptions));
         _clusterMembership = clusterMembership ?? throw new ArgumentNullException(nameof(clusterMembership));
         _loggerFactory = loggerFactory ?? throw new ArgumentNullException(nameof(loggerFactory));
         _logger = loggerFactory.CreateLogger<RedisJobShardManager>();
+
+        _keyPrefix = Encoding.UTF8.GetBytes(
+            _options.KeyPrefix ?? $"{_clusterOptions.ServiceId}/durablejobs");
     }
 
     private async ValueTask InitializeIfNeeded()
@@ -60,8 +70,8 @@ private async ValueTask InitializeIfNeeded()
         LogInitialized(_logger);
     }
 
-    private string ShardSetKey => $"durablejobs:shards:{_options.ShardPrefix}";
-    private static string MetaKeyForShard(string shardId) => $"durablejobs:shard:{shardId}:meta";
+    private string ShardSetKey => $"{_keyPrefix}:shards:{_options.ShardPrefix}";
+    private string MetaKeyForShard(string shardId) => $"{_keyPrefix}:shard:{shardId}:meta";
 
     public override async Task<List<IJobShard>> AssignJobShardsAsync(DateTimeOffset maxDueTime, CancellationToken cancellationToken)
     {
@@ -165,7 +175,7 @@ public override async Task<List<IJobShard>> AssignJobShardsAsync(DateTimeOffset
             var minDue = ParseDateTimeOffset(metadata, "MinDueTime", DateTimeOffset.MinValue);
             var maxDue = ParseDateTimeOffset(metadata, "MaxDueTime", DateTimeOffset.MaxValue);
 
-            var shard = new RedisJobShard(shardId, minDue, maxDue, _multiplexer!, metadata, _options, _loggerFactory.CreateLogger<RedisJobShard>());
+            var shard = new RedisJobShard(shardId, minDue, maxDue, _multiplexer!, metadata, _keyPrefix.ToString(), _options, _loggerFactory.CreateLogger<RedisJobShard>());
             try
             {
                 await shard.InitializeAsync(cancellationToken).ConfigureAwait(false);
@@ -233,7 +243,7 @@ public override async Task<IJobShard> CreateShardAsync(DateTimeOffset minDueTime
                     continue;
                 }
 
-                var shard = new RedisJobShard(shardId, minDueTime, maxDueTime, _multiplexer!, metadataInfo, _options, _loggerFactory.CreateLogger<RedisJobShard>());
+                var shard = new RedisJobShard(shardId, minDueTime, maxDueTime, _multiplexer!, metadataInfo, _keyPrefix.ToString(), _options, _loggerFactory.CreateLogger<RedisJobShard>());
                 await shard.InitializeAsync(cancellationToken).ConfigureAwait(false);
                 _jobShardCache[shardId] = shard;
                 LogShardRegistered(_logger, shardId);
@@ -313,8 +323,8 @@ private async Task DeleteShardAsync(string shardId)
     {
         await InitializeIfNeeded().ConfigureAwait(false);
 
-        var streamKey = $"durablejobs:shard:{shardId}:stream";
-        var metaKey = $"durablejobs:shard:{shardId}:meta";
+        var streamKey = $"{_keyPrefix}:shard:{shardId}:stream";
+        var metaKey = $"{_keyPrefix}:shard:{shardId}:meta";
 
         // Delete all shard-related keys
         await _redisOps!.DeleteKeysAsync([streamKey, metaKey]).ConfigureAwait(false);

test/Extensions/Tester.Redis/DurableJobs/RedisJobShardManagerTestFixture.cs

@@ -1,5 +1,6 @@
 using Microsoft.Extensions.Logging.Abstractions;
 using Microsoft.Extensions.Options;
+using Orleans.Configuration;
 using Orleans.DurableJobs;
 using Orleans.DurableJobs.Redis;
 using StackExchange.Redis;
@@ -15,6 +16,7 @@ namespace Tester.Redis.DurableJobs;
 internal sealed class RedisJobShardManagerTestFixture : IJobShardManagerTestFixture
 {
     private readonly IOptions<RedisJobShardOptions> _options;
+    private readonly IOptions<ClusterOptions> _clusterOptions;
     private ConnectionMultiplexer _multiplexer;
     private readonly string _shardPrefix;
 
@@ -33,6 +35,12 @@ public RedisJobShardManagerTestFixture()
             MinBatchSize = 1,
             BatchFlushInterval = TimeSpan.FromMilliseconds(100)
         });
+
+        _clusterOptions = Options.Create(new ClusterOptions
+        {
+            ServiceId = "test-service",
+            ClusterId = "test-cluster"
+        });
     }
 
     private async Task<IConnectionMultiplexer> CreateMultiplexerAsync(RedisJobShardOptions options)
@@ -46,6 +54,7 @@ public JobShardManager CreateManager(ILocalSiloDetails localSiloDetails, ICluste
         return new RedisJobShardManager(
             localSiloDetails,
             _options,
+            _clusterOptions,
             membershipService,
             NullLoggerFactory.Instance);
     }
@@ -58,14 +67,15 @@ public async ValueTask DisposeAsync()
             var db = _multiplexer.GetDatabase();
             var server = _multiplexer.GetServer(_multiplexer.GetEndPoints()[0]);
 
-            // Delete all keys with our test prefix
-            await foreach (var key in server.KeysAsync(pattern: $"durablejobs:shard:{_shardPrefix}*"))
+            // Delete all keys with our test prefix (using the default key prefix pattern)
+            var keyPrefix = _options.Value.KeyPrefix ?? $"{_clusterOptions.Value.ServiceId}/durablejobs";
+            await foreach (var key in server.KeysAsync(pattern: $"{keyPrefix}:shard:{_shardPrefix}*"))
             {
                 await db.KeyDeleteAsync(key);
             }
 
             // Delete the shard set key
-            await db.KeyDeleteAsync($"durablejobs:shards:{_shardPrefix}");
+            await db.KeyDeleteAsync($"{keyPrefix}:shards:{_shardPrefix}");
 
             await _multiplexer.CloseAsync();
             _multiplexer.Dispose();