Skip to content

Commit 92fe2d1

Browse files
ReubenBondCopilot
ReubenBond
and
Copilot
authored
feat(streaming): generalize Azure Table checkpointer (#10355)
* feat(streaming): generalize Azure Table checkpointer Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> * fix(streaming): configure Azure checkpoint key prefix Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 31b5b49f-6c0c-4fab-baee-7833c2e56363 * test(streaming): lock Event Hubs checkpoint compatibility Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 31b5b49f-6c0c-4fab-baee-7833c2e56363 --------- Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 31b5b49f-6c0c-4fab-baee-7833c2e56363
1 parent 7a639b7 commit 92fe2d1

12 files changed

Lines changed: 625 additions & 174 deletions

‎src/Azure/Orleans.Streaming.EventHubs/Providers/Streams/EventHub/AzureTableStreamConfiguratorExtensions.cs‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
1+
using System;
2+
using Microsoft.Extensions.DependencyInjection;
3+
using Microsoft.Extensions.Options;
4+
using Orleans.Configuration;
5+
using Orleans.Runtime;
6+
using Orleans.Streams;
7+
8+
namespace Orleans.Hosting
9+
{
10+
/// <summary>
11+
/// Extension methods for configuring Azure Table stream checkpointers.
12+
/// </summary>
13+
public static class AzureTableStreamConfiguratorExtensions
14+
{
15+
/// <summary>
16+
/// Configures the stream provider to persist checkpoints using Azure Table Storage.
17+
/// </summary>
18+
/// <param name="configurator">The configuration builder.</param>
19+
/// <param name="configureOptions">The Azure Table checkpointer configuration.</param>
20+
public static void UseAzureTableCheckpointer(
21+
this ISiloPersistentStreamConfigurator configurator,
22+
Action<OptionsBuilder<AzureTableStreamCheckpointerOptions>> configureOptions)
23+
{
24+
configurator.ConfigureDelegate(services =>
25+
services.AddTransient<IConfigurationValidator>(sp =>
26+
new AzureTableStreamCheckpointerOptionsValidator(
27+
sp.GetRequiredService<IOptionsMonitor<AzureTableStreamCheckpointerOptions>>().Get(configurator.Name),
28+
configurator.Name)));
29+
configurator.ConfigureComponent<AzureTableStreamCheckpointerOptions, IStreamQueueCheckpointerFactory>(
30+
AzureTableStreamQueueCheckpointerFactory.CreateFactory,
31+
options =>
32+
{
33+
options.Validate(
34+
static value => value.PersistInterval > TimeSpan.Zero,
35+
$"{nameof(AzureTableStreamCheckpointerOptions.PersistInterval)} must be greater than zero.");
36+
configureOptions(options);
37+
});
38+
}
39+
}
40+
}

‎src/Azure/Orleans.Streaming.EventHubs/Providers/Streams/EventHub/AzureTableStreamQueueCheckpointer.cs‎

Lines changed: 225 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,225 @@
1+
using System;
2+
using System.Collections.Generic;
3+
using System.Threading;
4+
using System.Threading.Tasks;
5+
using Microsoft.Extensions.Logging;
6+
using Orleans.Configuration;
7+
using Orleans.Streaming.EventHubs;
8+
9+
namespace Orleans.Streams
10+
{
11+
/// <summary>
12+
/// Persists stream queue checkpoints using Azure Table Storage.
13+
/// </summary>
14+
public partial class AzureTableStreamQueueCheckpointer : IStreamQueueCheckpointer<string>
15+
{
16+
private readonly AzureTableDataManager<StreamQueueCheckpointEntity> _dataManager;
17+
private readonly TimeSpan _persistInterval;
18+
private readonly IComparer<string>? _checkpointComparer;
19+
private readonly object _lock = new();
20+
21+
private StreamQueueCheckpointEntity _entity;
22+
private Task _inProgressSave = Task.CompletedTask;
23+
private DateTime? _throttleSavesUntilUtc;
24+
private string _latestCheckpoint = string.Empty;
25+
private string _persistedCheckpoint = string.Empty;
26+
27+
private AzureTableStreamQueueCheckpointer(
28+
AzureTableStreamCheckpointerOptions options,
29+
string streamProviderName,
30+
string partition,
31+
string serviceId,
32+
ILoggerFactory loggerFactory,
33+
IComparer<string>? defaultComparer = null,
34+
string? partitionKeyPrefix = null)
35+
{
36+
ArgumentNullException.ThrowIfNull(options);
37+
ArgumentException.ThrowIfNullOrWhiteSpace(streamProviderName);
38+
ArgumentException.ThrowIfNullOrWhiteSpace(partition);
39+
ArgumentNullException.ThrowIfNull(loggerFactory);
40+
if (options.PersistInterval <= TimeSpan.Zero)
41+
{
42+
throw new ArgumentOutOfRangeException(
43+
nameof(options),
44+
options.PersistInterval,
45+
$"{nameof(AzureTableStreamCheckpointerOptions.PersistInterval)} must be greater than zero.");
46+
}
47+
48+
_persistInterval = options.PersistInterval;
49+
_checkpointComparer = options.CheckpointComparer ?? defaultComparer;
50+
_dataManager = new AzureTableDataManager<StreamQueueCheckpointEntity>(
51+
options,
52+
loggerFactory.CreateLogger<StreamQueueCheckpointEntity>());
53+
_entity = StreamQueueCheckpointEntity.Create(
54+
partitionKeyPrefix ?? options.PartitionKeyPrefix,
55+
streamProviderName,
56+
serviceId,
57+
partition);
58+
LogCreatingCheckpointer(
59+
loggerFactory.CreateLogger<AzureTableStreamQueueCheckpointer>(),
60+
partition,
61+
streamProviderName,
62+
serviceId);
63+
}
64+
65+
/// <inheritdoc />
66+
public bool CheckpointExists
67+
{
68+
get
69+
{
70+
lock (_lock)
71+
{
72+
return !string.IsNullOrEmpty(_latestCheckpoint);
73+
}
74+
}
75+
}
76+
77+
/// <summary>
78+
/// Creates and initializes an Azure Table stream queue checkpointer.
79+
/// </summary>
80+
public static Task<IStreamQueueCheckpointer<string>> Create(
81+
AzureTableStreamCheckpointerOptions options,
82+
string streamProviderName,
83+
string partition,
84+
string serviceId,
85+
ILoggerFactory loggerFactory)
86+
{
87+
return Create(options, streamProviderName, partition, serviceId, loggerFactory, defaultComparer: null);
88+
}
89+
90+
internal static async Task<IStreamQueueCheckpointer<string>> Create(
91+
AzureTableStreamCheckpointerOptions options,
92+
string streamProviderName,
93+
string partition,
94+
string serviceId,
95+
ILoggerFactory loggerFactory,
96+
IComparer<string>? defaultComparer,
97+
string? partitionKeyPrefix = null)
98+
{
99+
var checkpointer = new AzureTableStreamQueueCheckpointer(
100+
options,
101+
streamProviderName,
102+
partition,
103+
serviceId,
104+
loggerFactory,
105+
defaultComparer,
106+
partitionKeyPrefix);
107+
await checkpointer._dataManager.InitTableAsync();
108+
return checkpointer;
109+
}
110+
111+
/// <inheritdoc />
112+
public async Task<string> Load()
113+
{
114+
var result = await _dataManager.ReadSingleTableEntryAsync(_entity.PartitionKey, _entity.RowKey);
115+
var checkpoint = result.Entity?.Offset ?? string.Empty;
116+
lock (_lock)
117+
{
118+
if (result.Entity is not null)
119+
{
120+
_entity = result.Entity;
121+
}
122+
123+
_latestCheckpoint = checkpoint;
124+
_persistedCheckpoint = checkpoint;
125+
}
126+
127+
return checkpoint;
128+
}
129+
130+
/// <inheritdoc />
131+
public void Update(string offset, DateTime utcNow)
132+
{
133+
ArgumentNullException.ThrowIfNull(offset);
134+
135+
lock (_lock)
136+
{
137+
if (string.Equals(_latestCheckpoint, offset, StringComparison.Ordinal)
138+
|| (_checkpointComparer is { } comparer
139+
&& !string.IsNullOrEmpty(_latestCheckpoint)
140+
&& comparer.Compare(offset, _latestCheckpoint) <= 0))
141+
{
142+
return;
143+
}
144+
145+
_latestCheckpoint = offset;
146+
if (_throttleSavesUntilUtc.HasValue
147+
&& (_throttleSavesUntilUtc.Value > utcNow || !_inProgressSave.IsCompleted))
148+
{
149+
return;
150+
}
151+
152+
_throttleSavesUntilUtc = utcNow + _persistInterval;
153+
_inProgressSave = Save(offset);
154+
_inProgressSave.Ignore();
155+
}
156+
}
157+
158+
/// <inheritdoc />
159+
public async Task FlushAsync(CancellationToken cancellationToken)
160+
{
161+
var retryingSave = false;
162+
while (true)
163+
{
164+
Task inProgressSave;
165+
lock (_lock)
166+
{
167+
inProgressSave = _inProgressSave;
168+
}
169+
170+
if (retryingSave)
171+
{
172+
await inProgressSave.WaitAsync(cancellationToken);
173+
}
174+
else
175+
{
176+
try
177+
{
178+
await inProgressSave.WaitAsync(cancellationToken);
179+
}
180+
catch (Exception) when (!cancellationToken.IsCancellationRequested)
181+
{
182+
}
183+
184+
cancellationToken.ThrowIfCancellationRequested();
185+
}
186+
187+
lock (_lock)
188+
{
189+
if (!ReferenceEquals(inProgressSave, _inProgressSave))
190+
{
191+
retryingSave = false;
192+
continue;
193+
}
194+
195+
if (string.Equals(_persistedCheckpoint, _latestCheckpoint, StringComparison.Ordinal))
196+
{
197+
return;
198+
}
199+
200+
_inProgressSave = Save(_latestCheckpoint);
201+
retryingSave = true;
202+
}
203+
}
204+
}
205+
206+
private async Task Save(string checkpoint)
207+
{
208+
_entity.Offset = checkpoint;
209+
await _dataManager.UpsertTableEntryAsync(_entity);
210+
lock (_lock)
211+
{
212+
_persistedCheckpoint = checkpoint;
213+
}
214+
}
215+
216+
[LoggerMessage(
217+
Level = LogLevel.Information,
218+
Message = "Creating Azure Table checkpointer for partition {Partition} of stream provider {StreamProviderName} with service ID {ServiceId}.")]
219+
private static partial void LogCreatingCheckpointer(
220+
ILogger logger,
221+
string partition,
222+
string streamProviderName,
223+
string serviceId);
224+
}
225+
}

‎src/Azure/Orleans.Streaming.EventHubs/Providers/Streams/EventHub/AzureTableStreamQueueCheckpointerFactory.cs‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
1+
using System;
2+
using System.Threading.Tasks;
3+
using Microsoft.Extensions.DependencyInjection;
4+
using Microsoft.Extensions.Logging;
5+
using Microsoft.Extensions.Options;
6+
using Orleans.Configuration;
7+
using Orleans.Configuration.Overrides;
8+
9+
namespace Orleans.Streams
10+
{
11+
/// <summary>
12+
/// Creates Azure Table stream queue checkpointers.
13+
/// </summary>
14+
public class AzureTableStreamQueueCheckpointerFactory : IStreamQueueCheckpointerFactory
15+
{
16+
private readonly string _providerName;
17+
private readonly AzureTableStreamCheckpointerOptions _options;
18+
private readonly ClusterOptions _clusterOptions;
19+
private readonly ILoggerFactory _loggerFactory;
20+
21+
/// <summary>
22+
/// Initializes a new instance.
23+
/// </summary>
24+
public AzureTableStreamQueueCheckpointerFactory(
25+
string providerName,
26+
AzureTableStreamCheckpointerOptions options,
27+
IOptions<ClusterOptions> clusterOptions,
28+
ILoggerFactory loggerFactory)
29+
{
30+
_providerName = providerName;
31+
_options = options;
32+
_clusterOptions = clusterOptions.Value;
33+
_loggerFactory = loggerFactory;
34+
}
35+
36+
/// <summary>
37+
/// Creates a factory from a service provider.
38+
/// </summary>
39+
public static IStreamQueueCheckpointerFactory CreateFactory(IServiceProvider services, string providerName)
40+
{
41+
var options = services.GetOptionsByName<AzureTableStreamCheckpointerOptions>(providerName);
42+
var clusterOptions = services.GetProviderClusterOptions(providerName);
43+
return ActivatorUtilities.CreateInstance<AzureTableStreamQueueCheckpointerFactory>(
44+
services,
45+
providerName,
46+
options,
47+
clusterOptions);
48+
}
49+
50+
/// <inheritdoc />
51+
public Task<IStreamQueueCheckpointer<string>> Create(string partition)
52+
{
53+
return AzureTableStreamQueueCheckpointer.Create(
54+
_options,
55+
_providerName,
56+
partition,
57+
_clusterOptions.ServiceId.ToString(),
58+
_loggerFactory);
59+
}
60+
}
61+
}

0 commit comments

Comments
 (0)