Skip to content
This repository was archived by the owner on Mar 31, 2026. It is now read-only.

Commit 7892e44

Browse files
joelverhagenjoelverhagen
committed
Improve admin panel performance by using batch queries
1 parent f945fdf commit 7892e44

5 files changed

Lines changed: 147 additions & 30 deletions

File tree

‎src/Website/Models/ViewModelFactory.cs‎

Lines changed: 22 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -56,12 +56,8 @@ public async Task<AdminViewModel> GetAdminViewModelAsync()
5656
{
5757
var workQueueTask = GetQueueAsync(QueueType.Work);
5858
var expandQueueTask = GetQueueAsync(QueueType.Expand);
59-
60-
var catalogScanTasks = CatalogScanCursorService
61-
.StartableDriverTypes
62-
.Select(GetCatalogScanAsync)
63-
.ToList();
64-
59+
var cursorsTask = _catalogScanCursorService.GetCursorsAsync();
60+
var catalogScansTask = _catalogScanStorageService.GetAllLatestIndexScansAsync(HistoryCount);
6561
var isWorkflowRunningTask = _workflowService.IsWorkflowRunningAsync();
6662
var timerStatesTask = _timerExecutionService.GetStateAsync();
6763
var workflowRunsTask = GetWorkflowRunsAsync();
@@ -71,25 +67,39 @@ public async Task<AdminViewModel> GetAdminViewModelAsync()
7167
await Task.WhenAll(
7268
workQueueTask,
7369
expandQueueTask,
70+
catalogScansTask,
71+
cursorsTask,
7472
isWorkflowRunningTask,
7573
timerStatesTask,
7674
workflowRunsTask,
7775
kustoIngestionsTask,
7876
catalogCommitTimestampTask);
7977

80-
var catalogScans = await Task.WhenAll(catalogScanTasks);
81-
82-
// Calculate the cursor age.
78+
// Build the catalog scan (driver) view models
8379
var catalogCommitTimestamp = await catalogCommitTimestampTask;
84-
foreach (var catalogScan in catalogScans)
80+
var catalogScans = new List<CatalogScanViewModel>();
81+
foreach ((var driverType, var latestScans) in catalogScansTask.Result.OrderBy(x => x.Key))
8582
{
86-
var min = catalogScan.Cursor.Value;
83+
var cursor = cursorsTask.Result[driverType];
84+
var nextCommitTimestampForCursor = await _catalogCommitTimestampProvider.GetNextAsync(cursor.Value);
85+
86+
var min = cursor.Value;
8787
if (min < CatalogClient.NuGetOrgMin)
8888
{
8989
min = CatalogClient.NuGetOrgMin;
9090
}
9191

92-
catalogScan.CursorAge = catalogCommitTimestamp - min;
92+
catalogScans.Add(new CatalogScanViewModel
93+
{
94+
Cursor = cursor,
95+
DefaultMax = nextCommitTimestampForCursor ?? CatalogClient.NuGetOrgFirstCommit,
96+
CursorAge = catalogCommitTimestamp - min,
97+
DriverType = driverType,
98+
LatestScans = latestScans,
99+
SupportsReprocess = _catalogScanService.SupportsReprocess(driverType),
100+
OnlyLatestLeavesSupport = _catalogScanService.GetOnlyLatestLeavesSupport(driverType),
101+
IsEnabled = _catalogScanService.IsEnabled(driverType),
102+
});
93103
}
94104

95105
// Calculate the next default max, which supports processing the catalog one commit at a time.
@@ -161,23 +171,5 @@ private MoveQueueMessagesState GetMoveQueueMessagesState(QueueType queueType, bo
161171
return MoveQueueMessagesState.None;
162172
}
163173
}
164-
165-
private async Task<CatalogScanViewModel> GetCatalogScanAsync(CatalogScanDriverType driverType)
166-
{
167-
var latestScans = await _catalogScanStorageService.GetLatestIndexScansAsync(driverType, HistoryCount);
168-
var cursor = await _catalogScanCursorService.GetCursorAsync(driverType);
169-
var nextCommitTimestamp = await _catalogCommitTimestampProvider.GetNextAsync(cursor.Value);
170-
171-
return new CatalogScanViewModel
172-
{
173-
DriverType = driverType,
174-
Cursor = cursor,
175-
LatestScans = latestScans,
176-
SupportsReprocess = _catalogScanService.SupportsReprocess(driverType),
177-
OnlyLatestLeavesSupport = _catalogScanService.GetOnlyLatestLeavesSupport(driverType),
178-
IsEnabled = _catalogScanService.IsEnabled(driverType),
179-
DefaultMax = nextCommitTimestamp ?? CatalogClient.NuGetOrgFirstCommit,
180-
};
181-
}
182174
}
183175
}

‎src/Worker.Logic/CatalogScan/CatalogScanCursorService.cs‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -173,6 +173,13 @@ public async Task SetAllCursorsAsync(DateTimeOffset value)
173173
}
174174
}
175175

176+
public async Task<Dictionary<CatalogScanDriverType, CursorTableEntity>> GetCursorsAsync()
177+
{
178+
var nameToType = StartableDriverTypes.ToDictionary(GetCursorName);
179+
var cursors = await _cursorStorageService.GetOrCreateAllAsync(nameToType.Keys.ToList());
180+
return cursors.ToDictionary(x => nameToType[x.Name]);
181+
}
182+
176183
public async Task<CursorTableEntity> GetCursorAsync(CatalogScanDriverType driverType)
177184
{
178185
return await _cursorStorageService.GetOrCreateAsync(GetCursorName(driverType));

‎src/Worker.Logic/CatalogScan/CatalogScanStorageService.cs‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -216,6 +216,37 @@ public async Task<IReadOnlyList<CatalogIndexScan>> GetIndexScansAsync()
216216
.ToListAsync(_telemetryClient.StartQueryLoopMetrics());
217217
}
218218

219+
public async Task<Dictionary<CatalogScanDriverType, List<CatalogIndexScan>>> GetAllLatestIndexScansAsync(int maxEntities)
220+
{
221+
var pks = CatalogScanCursorService.StartableDriverTypes.Select(x => x.ToString()).ToList();
222+
var minPk = pks.Min(StringComparer.Ordinal);
223+
var maxPk = pks.Max(StringComparer.Ordinal);
224+
225+
var table = await GetIndexScanTableAsync();
226+
var query = table.QueryAsync<CatalogIndexScan>(x => x.PartitionKey.CompareTo(minPk) >= 0 && x.PartitionKey.CompareTo(maxPk) <= 0);
227+
228+
var output = CatalogScanCursorService.StartableDriverTypes.ToDictionary(x => x, x => new List<CatalogIndexScan>());
229+
var completed = 0;
230+
231+
await foreach (var item in query)
232+
{
233+
if (output.TryGetValue(item.DriverType, out var list) && list.Count < maxEntities)
234+
{
235+
list.Add(item);
236+
if (list.Count == maxEntities)
237+
{
238+
completed++;
239+
if (completed == output.Count)
240+
{
241+
break;
242+
}
243+
}
244+
}
245+
}
246+
247+
return output;
248+
}
249+
219250
public async Task<IReadOnlyList<CatalogIndexScan>> GetLatestIndexScansAsync(CatalogScanDriverType driverType, int maxEntities)
220251
{
221252
return await (await GetIndexScanTableAsync())

‎src/Worker.Logic/Cursors/CursorStorageService.cs‎

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,9 @@
11
// Copyright (c) .NET Foundation. All rights reserved.
22
// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
33

4+
using System;
5+
using System.Collections.Generic;
6+
using System.Linq;
47
using System.Threading.Tasks;
58
using Azure.Data.Tables;
69
using Microsoft.Extensions.Logging;
@@ -29,6 +32,52 @@ public async Task InitializeAsync()
2932
await (await GetTableAsync()).CreateIfNotExistsAsync(retry: true);
3033
}
3134

35+
public async Task<IReadOnlyList<CursorTableEntity>> GetOrCreateAllAsync(IReadOnlyList<string> names)
36+
{
37+
if (names.Count == 0)
38+
{
39+
return Array.Empty<CursorTableEntity>();
40+
}
41+
else if (names.Count == 1)
42+
{
43+
return new List<CursorTableEntity> { await GetOrCreateAsync(names[0]) };
44+
}
45+
46+
var table = await GetTableAsync();
47+
var min = names.Min(StringComparer.Ordinal);
48+
var max = names.Max(StringComparer.Ordinal);
49+
50+
var nameToExisting = await table
51+
.QueryAsync<CursorTableEntity>(c => c.PartitionKey == string.Empty && c.RowKey.CompareTo(min) >= 0 && c.RowKey.CompareTo(max) <= 0)
52+
.ToDictionaryAsync(x => x.Name);
53+
54+
var output = new List<CursorTableEntity>();
55+
var unique = new HashSet<string>();
56+
var batch = new MutableTableTransactionalBatch(table);
57+
58+
foreach (var name in names)
59+
{
60+
if (unique.Add(name))
61+
{
62+
if (nameToExisting.TryGetValue(name, out var existing))
63+
{
64+
output.Add(existing);
65+
}
66+
else
67+
{
68+
var cursor = new CursorTableEntity(name);
69+
_logger.LogInformation("Creating cursor {Name} to timestamp {Value:O}.", name, cursor.Value);
70+
batch.AddEntity(cursor);
71+
output.Add(cursor);
72+
}
73+
}
74+
}
75+
76+
await batch.SubmitBatchIfNotEmptyAsync();
77+
78+
return output;
79+
}
80+
3281
public async Task<CursorTableEntity> GetOrCreateAsync(string name)
3382
{
3483
var table = await GetTableAsync();

‎test/Worker.Logic.Test/Cursors/CursorStorageServiceTest.cs‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,44 @@ namespace NuGet.Insights.Worker
1818
{
1919
public class CursorStorageServiceTest : IClassFixture<CursorStorageServiceTest.Fixture>, IAsyncLifetime
2020
{
21+
public class TheGetOrCreateAllAsyncMethod : CursorStorageServiceTest
22+
{
23+
public TheGetOrCreateAllAsyncMethod(Fixture fixture, ITestOutputHelper output) : base(fixture, output)
24+
{
25+
}
26+
27+
[Fact]
28+
public async Task ReturnsExistingCursors()
29+
{
30+
var value = new DateTimeOffset(2020, 1, 1, 12, 30, 0, TimeSpan.Zero);
31+
var cursorA = await Target.GetOrCreateAsync(CursorName + "A");
32+
var cursorB = await Target.GetOrCreateAsync(CursorName + "B");
33+
34+
var cursors = await Target.GetOrCreateAllAsync(new[] { cursorB.Name, cursorA.Name });
35+
36+
Assert.Equal(2, cursors.Count);
37+
Assert.Equal(cursorB.Name, cursors[0].Name);
38+
Assert.Equal(cursorB.ETag, cursors[0].ETag);
39+
Assert.Equal(cursorA.Name, cursors[1].Name);
40+
Assert.Equal(cursorA.ETag, cursors[1].ETag);
41+
}
42+
43+
[Fact]
44+
public async Task CreateNewCursors()
45+
{
46+
var value = new DateTimeOffset(2020, 1, 1, 12, 30, 0, TimeSpan.Zero);
47+
48+
var cursors = await Target.GetOrCreateAllAsync(new[] { CursorName + "B", CursorName + "A" });
49+
50+
var entities = await GetEntitiesAsync<CursorTableEntity>();
51+
Assert.Equal(2, cursors.Count);
52+
Assert.Equal(entities[1].Name, cursors[0].Name);
53+
Assert.Equal(entities[1].ETag, cursors[0].ETag);
54+
Assert.Equal(entities[0].Name, cursors[1].Name);
55+
Assert.Equal(entities[0].ETag, cursors[1].ETag);
56+
}
57+
}
58+
2159
public class TheGetOrCreateAsyncMethod : CursorStorageServiceTest
2260
{
2361
public TheGetOrCreateAsyncMethod(Fixture fixture, ITestOutputHelper output) : base(fixture, output)

0 commit comments

Comments
 (0)