-
Notifications
You must be signed in to change notification settings - Fork 5.2k
[WebJobs.Extensions.Queues/[WebJobs.Extensions.Blobs] Scaler APIs #35358
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Merged
Merged
Changes from 5 commits
Commits
Show all changes
12 commits
Select commit
Hold shift + click to select a range
70af0f2
[WebJobs.Extensions.ServiceBus] Scaler APIs
alrod a229616
Moving Microsoft.Azure.WebJobs.Host.Storage to Packages.Data.props
alrod 3f514b0
Fixing AzureComponentFactoryWrapper build error
alrod 2702088
Some fixes
alrod 5111275
Adding public API
alrod 9f98309
Disabling BlobScaleHostEndToEndTests
alrod 4014d1a
Adding "Ignore" attrribute
alrod 611acda
Adding mocking blob storage test
alrod e835a40
Adding logs
alrod 61f17ff
Use build-in credentials
alrod e0b0dc2
Adding test logs to the output on exception
alrod aeb5a8c
remove test code
alrod File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
2 changes: 1 addition & 1 deletion
2
...Jobs.Extensions.Clients/samples/Microsoft.Azure.WebJobs.Extensions.Clients.Samples.csproj
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
33 changes: 33 additions & 0 deletions
33
...xtensions/Microsoft.Azure.WebJobs.Extensions.Clients/tests/shared/TestComponentFactory.cs
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,33 @@ | ||
| // Copyright (c) Microsoft Corporation. All rights reserved. | ||
| // Licensed under the MIT License. | ||
|
|
||
| using System; | ||
| using Azure.Core; | ||
| using Microsoft.Extensions.Azure; | ||
| using Microsoft.Extensions.Configuration; | ||
|
|
||
| namespace Microsoft.Azure.WebJobs.Host.TestCommon | ||
| { | ||
| public class TestComponentFactory : AzureComponentFactory | ||
| { | ||
| private readonly AzureComponentFactory _factory; | ||
| private readonly TokenCredential _tokenCredential; | ||
|
|
||
| public TestComponentFactory(AzureComponentFactory factory, TokenCredential tokenCredential) | ||
| { | ||
| _factory = factory; | ||
| _tokenCredential = tokenCredential; | ||
| } | ||
|
|
||
| public override TokenCredential CreateTokenCredential(IConfiguration configuration) | ||
| { | ||
| return _tokenCredential != null ? _tokenCredential : _factory.CreateTokenCredential(configuration); | ||
| } | ||
|
|
||
| public override object CreateClientOptions(Type optionsType, object serviceVersion, IConfiguration configuration) | ||
| => _factory.CreateClientOptions(optionsType, serviceVersion, configuration); | ||
|
|
||
| public override object CreateClient(Type clientType, IConfiguration configuration, TokenCredential credential, object clientOptions) | ||
| => _factory.CreateClient(clientType, configuration, credential, clientOptions); | ||
| } | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
146 changes: 146 additions & 0 deletions
146
...crosoft.Azure.WebJobs.Extensions.Storage.Blobs/src/Listeners/BlobScalerMonitorProvider.cs
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,146 @@ | ||
| // Copyright (c) Microsoft Corporation. All rights reserved. | ||
| // Licensed under the MIT License. | ||
|
|
||
| using System; | ||
| using System.Linq; | ||
| using System.Text; | ||
| using System.Threading; | ||
| using System.Threading.Tasks; | ||
| using Azure.Storage.Blobs; | ||
| using Azure.Storage.Blobs.Specialized; | ||
| using Microsoft.Azure.WebJobs.Host.Scale; | ||
| using Microsoft.Extensions.Azure; | ||
| using Microsoft.Extensions.Configuration; | ||
| using Microsoft.Extensions.DependencyInjection; | ||
| using Microsoft.Extensions.Logging; | ||
| using Newtonsoft.Json; | ||
|
|
||
| namespace Microsoft.Azure.WebJobs.Extensions.Storage.Blobs.Listeners | ||
| { | ||
| internal class BlobScalerMonitorProvider : IScaleMonitorProvider | ||
| { | ||
| private readonly IScaleMonitor _scaleMonitor; | ||
|
|
||
| public BlobScalerMonitorProvider(IServiceProvider serviceProvider, TriggerMetadata triggerMetadata) | ||
| { | ||
| AzureComponentFactory azureComponentFactory = null; | ||
| if ((triggerMetadata.Properties != null) && (triggerMetadata.Properties.TryGetValue(nameof(AzureComponentFactory), out object value))) | ||
|
alrod marked this conversation as resolved.
|
||
| { | ||
| azureComponentFactory = value as AzureComponentFactory; | ||
| } | ||
| else | ||
| { | ||
| azureComponentFactory = serviceProvider.GetService<AzureComponentFactory>(); | ||
| } | ||
| IConfiguration configuration = serviceProvider.GetService<IConfiguration>(); | ||
| AzureEventSourceLogForwarder logForwarder = serviceProvider.GetService<AzureEventSourceLogForwarder>(); | ||
| var factory = serviceProvider.GetService<ILoggerFactory>(); | ||
| BlobMetadata blobMetadata = JsonConvert.DeserializeObject<BlobMetadata>(triggerMetadata.Metadata.ToString()); | ||
| BlobServiceClientProvider blobServiceClientProvider = new BlobServiceClientProvider(configuration, azureComponentFactory, logForwarder, factory.CreateLogger<BlobServiceClient>()); | ||
| BlobServiceClient blobServiceClient = blobServiceClientProvider.Get(blobMetadata.Connection, serviceProvider.GetRequiredService<INameResolver>()); | ||
| _scaleMonitor = new ZeroToOneScaleMonitor(triggerMetadata.FunctionName, blobServiceClient, factory); | ||
| } | ||
|
|
||
| public IScaleMonitor GetMonitor() | ||
| { | ||
| return _scaleMonitor; | ||
| } | ||
|
|
||
| private class ZeroToOneScaleMonitor : IScaleMonitor<ScaleMetrics> | ||
| { | ||
| private readonly ScaleMonitorDescriptor _scaleMonitorDescriptor; | ||
| private readonly Lazy<Task<BlobLogListener>> _blobLogListener; | ||
| private readonly ILogger _logger; | ||
| private int _threadSafeWritesDetectedValue; | ||
|
|
||
| public ZeroToOneScaleMonitor(string functionId, BlobServiceClient blobServiceClient, ILoggerFactory loggerFactory) | ||
| { | ||
| _scaleMonitorDescriptor = new ScaleMonitorDescriptor(functionId, functionId); | ||
| _blobLogListener = new(() => BlobLogListener.CreateAsync(blobServiceClient, loggerFactory.CreateLogger<BlobListener>(), CancellationToken.None)); | ||
|
alrod marked this conversation as resolved.
Outdated
|
||
| _logger = loggerFactory.CreateLogger<ZeroToOneScaleMonitor>(); | ||
| } | ||
|
|
||
| public ScaleMonitorDescriptor Descriptor => _scaleMonitorDescriptor; | ||
|
|
||
| public async Task<ScaleMetrics> GetMetricsAsync() | ||
| { | ||
| // if new blob were detected we want to GetScaleStatus return scale out vote at least once | ||
| if (Interlocked.Equals(_threadSafeWritesDetectedValue, 1)) | ||
| { | ||
| _logger.LogInformation($"New writes were detectd but GetScaleStatus was not called. Waiting GetScaleStatus to call."); | ||
| return new ScaleMetrics(); | ||
| } | ||
|
|
||
| var blobLogListener = await _blobLogListener.Value.ConfigureAwait(false); | ||
| BlobWithContainer<BlobBaseClient>[] recentWrites = (await blobLogListener.GetRecentBlobWritesAsync(CancellationToken.None).ConfigureAwait(false)).ToArray(); | ||
| if (recentWrites.Length > 0) | ||
| { | ||
| StringBuilder stringBuilder = new StringBuilder(); | ||
| foreach (var write in recentWrites) | ||
| { | ||
| stringBuilder.Append($"'{write.BlobClient.Name}', "); | ||
| if (stringBuilder.Length > 1000) | ||
| { | ||
| stringBuilder.Append("[truncated]"); | ||
| break; | ||
| } | ||
| } | ||
| _logger.LogInformation($"'{recentWrites.Length}' recent writes were detected for '{_scaleMonitorDescriptor.FunctionId}': {stringBuilder}"); | ||
| Interlocked.CompareExchange(ref _threadSafeWritesDetectedValue, 1, 0); | ||
| } | ||
| else | ||
| { | ||
| _logger.LogInformation($"No recent writes were detected for '{_scaleMonitorDescriptor.FunctionId}'"); | ||
| Interlocked.CompareExchange(ref _threadSafeWritesDetectedValue, 0, 1); | ||
| } | ||
| return new ScaleMetrics(); | ||
| } | ||
|
|
||
| public ScaleStatus GetScaleStatus(ScaleStatusContext context) | ||
| { | ||
| return GetScaleStatusCore(context.WorkerCount); | ||
| } | ||
|
|
||
| public ScaleStatus GetScaleStatus(ScaleStatusContext<ScaleMetrics> context) | ||
| { | ||
| return GetScaleStatusCore(context.WorkerCount); | ||
| } | ||
|
|
||
| private ScaleStatus GetScaleStatusCore(int workerCount) | ||
| { | ||
| // if there is at least one worker we assume all the blobs are added to internal queue and we need to ScaleIn | ||
| if (workerCount > 0) | ||
| { | ||
| // Set to 0 if there is an active worker | ||
| Interlocked.CompareExchange(ref _threadSafeWritesDetectedValue, 0, 1); | ||
| } | ||
|
|
||
| ScaleVote vote = ScaleVote.None; | ||
| if (workerCount == 0 && _threadSafeWritesDetectedValue == 1) | ||
| { | ||
| vote = ScaleVote.ScaleOut; | ||
| } | ||
| else if (workerCount > 0 && _threadSafeWritesDetectedValue == 0) | ||
| { | ||
| vote = ScaleVote.ScaleIn; | ||
| } | ||
| else if (workerCount == 0 && _threadSafeWritesDetectedValue == 0) | ||
| { | ||
| vote = ScaleVote.None; | ||
| } | ||
| _logger.LogInformation($"Current vote is '{vote}', active workers is '{workerCount}' for '{_scaleMonitorDescriptor.FunctionId}'"); | ||
|
|
||
| return new ScaleStatus() | ||
| { | ||
| Vote = vote | ||
| }; | ||
| } | ||
| } | ||
|
|
||
| internal class BlobMetadata | ||
| { | ||
| [JsonProperty] | ||
| public string Connection { get; set; } | ||
| } | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.