Skip to content

Scope orphaned-event cleanup to recent ingest window to eliminate 8-hour ES CPU spikes #2279

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.

Sign up for GitHub

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

Open
Copilot wants to merge 4 commits into
base: main
Choose a base branch
from
+111 −12
Open

Scope orphaned-event cleanup to recent ingest window to eliminate 8-hour ES CPU spikes #2279

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
a23d502
Initial plan
Copilot May 31, 2026
794182e
Scope orphan cleanup event scans to recent data
Copilot Jun 1, 2026
7f50a3a
Use injected time provider for orphan cleanup cutoff and tests
Copilot Jun 1, 2026
a4386e7
Merge remote-tracking branch 'origin/main' into copilot/investigate-c…
niemyjski Jul 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 51 additions & 12 deletions src/Exceptionless.Core/Jobs/CleanupOrphanedDataJob.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ namespace Exceptionless.Core.Jobs;
[Job(Description = "Deletes orphaned data.", IsContinuous = false)]
public class CleanupOrphanedDataJob : JobWithLockBase, IHealthCheck
{
internal static readonly TimeSpan OrphanedEventLookback = TimeSpan.FromDays(3);
private readonly ExceptionlessElasticConfiguration _config;
private readonly ElasticsearchClient _elasticClient;
private readonly IStackRepository _stackRepository;
Expand Down Expand Up @@ -58,21 +59,28 @@ ILoggerFactory loggerFactory

protected override async Task<JobResult> RunInternalAsync(JobContext context)
{
await DeleteOrphanedEventsByStackAsync(context);
await DeleteOrphanedEventsByProjectAsync(context);
await DeleteOrphanedEventsByOrganizationAsync(context);
var orphanedEventCutoffUtc = GetOrphanedEventCutoffUtc();
await DeleteOrphanedEventsByStackAsync(context, orphanedEventCutoffUtc);
await DeleteOrphanedEventsByProjectAsync(context, orphanedEventCutoffUtc);
await DeleteOrphanedEventsByOrganizationAsync(context, orphanedEventCutoffUtc);

await FixDuplicateStacks(context);

return JobResult.Success;
}

public async Task DeleteOrphanedEventsByStackAsync(JobContext context)
public Task DeleteOrphanedEventsByStackAsync(JobContext context)
{
return DeleteOrphanedEventsByStackAsync(context, GetOrphanedEventCutoffUtc());
}

private async Task DeleteOrphanedEventsByStackAsync(JobContext context, DateTime orphanedEventCutoffUtc)
{
// get approximate number of unique stack ids
var stackCardinality = await _elasticClient.SearchAsync<PersistentEvent>(s => s
.Indices(GetEventIndexPattern())
.Size(0)
.Query(q => RecentEventQuery(q, orphanedEventCutoffUtc))
.AddAggregation("cardinality_stack_id", a => a.Cardinality(c => c.Field(f => f.StackId).PrecisionThreshold(40000))));

double? uniqueStackIdCount = stackCardinality.Aggregations?.GetCardinality("cardinality_stack_id")?.Value;
Expand All @@ -93,6 +101,7 @@ public async Task DeleteOrphanedEventsByStackAsync(JobContext context)
var stackIdTerms = await _elasticClient.SearchAsync<PersistentEvent>(s => s
.Indices(GetEventIndexPattern())
.Size(0)
.Query(q => RecentEventQuery(q, orphanedEventCutoffUtc))
.AddAggregation("terms_stack_id", a => a.Terms(c => c.Field(f => f.StackId).Include(new TermsInclude(batchNumber, buckets)).Size(batchSize * 2))));

string[] stackIds = stackIdTerms.Aggregations?.GetStringTerms("terms_stack_id")?.Buckets.Select(b => b.Key.ToString()!).ToArray() ?? [];
Expand All @@ -115,18 +124,26 @@ public async Task DeleteOrphanedEventsByStackAsync(JobContext context)
_logger.LogInformation("{BatchNumber}/{BatchCount}: Found {OrphanedEventCount} orphaned events from missing stacks {MissingStackIds} out of {StackIdCount}", batchNumber, buckets, missingStackIds.Length, missingStackIds, stackIds.Length);
await _elasticClient.DeleteByQueryAsync<PersistentEvent>(r => r
.Indices(GetEventIndexPattern())
.Query(q => q.Terms(t => t.Field(f => f.StackId).Terms(new TermsQueryField(missingStackIds.Select(FieldValueHelper.ToFieldValue).ToList())))));
.Query(q => q.Bool(b => b.Filter(
f => f.Terms(t => t.Field(e => e.StackId).Terms(new TermsQueryField(missingStackIds.Select(FieldValueHelper.ToFieldValue).ToList()))),
f => RecentEventQuery(f, orphanedEventCutoffUtc)))));
}

_logger.LogInformation("Found {OrphanedEventCount} orphaned events from missing stacks out of {StackIdCount}", totalOrphanedEventCount, totalStackIds);
_logger.LogInformation("Found {OrphanedEventCount} orphaned events from missing stacks out of {StackIdCount} since {OrphanedEventCutoffUtc}", totalOrphanedEventCount, totalStackIds, orphanedEventCutoffUtc);
}

public Task DeleteOrphanedEventsByProjectAsync(JobContext context)
{
return DeleteOrphanedEventsByProjectAsync(context, GetOrphanedEventCutoffUtc());
}

public async Task DeleteOrphanedEventsByProjectAsync(JobContext context)
private async Task DeleteOrphanedEventsByProjectAsync(JobContext context, DateTime orphanedEventCutoffUtc)
{
// get approximate number of unique project ids
var projectCardinality = await _elasticClient.SearchAsync<PersistentEvent>(s => s
.Indices(GetEventIndexPattern())
.Size(0)
.Query(q => RecentEventQuery(q, orphanedEventCutoffUtc))
.AddAggregation("cardinality_project_id", a => a.Cardinality(c => c.Field(f => f.ProjectId).PrecisionThreshold(40000))));

double? uniqueProjectIdCount = projectCardinality.Aggregations?.GetCardinality("cardinality_project_id")?.Value;
Expand All @@ -147,6 +164,7 @@ public async Task DeleteOrphanedEventsByProjectAsync(JobContext context)
var projectIdTerms = await _elasticClient.SearchAsync<PersistentEvent>(s => s
.Indices(GetEventIndexPattern())
.Size(0)
.Query(q => RecentEventQuery(q, orphanedEventCutoffUtc))
.AddAggregation("terms_project_id", a => a.Terms(c => c.Field(f => f.ProjectId).Include(new TermsInclude(batchNumber, buckets)).Size(batchSize * 2))));

string[] projectIds = projectIdTerms.Aggregations?.GetStringTerms("terms_project_id")?.Buckets.Select(b => b.Key.ToString()!).ToArray() ?? [];
Expand All @@ -169,18 +187,26 @@ public async Task DeleteOrphanedEventsByProjectAsync(JobContext context)
_logger.LogInformation("{BatchNumber}/{BatchCount}: Found {OrphanedEventCount} orphaned events from missing projects {MissingProjectIds} out of {ProjectIdCount}", batchNumber, buckets, missingProjectIds.Length, missingProjectIds, projectIds.Length);
await _elasticClient.DeleteByQueryAsync<PersistentEvent>(r => r
.Indices(GetEventIndexPattern())
.Query(q => q.Terms(t => t.Field(f => f.ProjectId).Terms(new TermsQueryField(missingProjectIds.Select(FieldValueHelper.ToFieldValue).ToList())))));
.Query(q => q.Bool(b => b.Filter(
f => f.Terms(t => t.Field(e => e.ProjectId).Terms(new TermsQueryField(missingProjectIds.Select(FieldValueHelper.ToFieldValue).ToList()))),
f => RecentEventQuery(f, orphanedEventCutoffUtc)))));
}

_logger.LogInformation("Found {OrphanedEventCount} orphaned events from missing projects out of {ProjectIdCount}", totalOrphanedEventCount, totalProjectIds);
_logger.LogInformation("Found {OrphanedEventCount} orphaned events from missing projects out of {ProjectIdCount} since {OrphanedEventCutoffUtc}", totalOrphanedEventCount, totalProjectIds, orphanedEventCutoffUtc);
}

public async Task DeleteOrphanedEventsByOrganizationAsync(JobContext context)
public Task DeleteOrphanedEventsByOrganizationAsync(JobContext context)
{
return DeleteOrphanedEventsByOrganizationAsync(context, GetOrphanedEventCutoffUtc());
}

private async Task DeleteOrphanedEventsByOrganizationAsync(JobContext context, DateTime orphanedEventCutoffUtc)
{
// get approximate number of unique organization ids
var organizationCardinality = await _elasticClient.SearchAsync<PersistentEvent>(s => s
.Indices(GetEventIndexPattern())
.Size(0)
.Query(q => RecentEventQuery(q, orphanedEventCutoffUtc))
.AddAggregation("cardinality_organization_id", a => a.Cardinality(c => c.Field(f => f.OrganizationId).PrecisionThreshold(40000))));

double? uniqueOrganizationIdCount = organizationCardinality.Aggregations?.GetCardinality("cardinality_organization_id")?.Value;
Expand All @@ -201,6 +227,7 @@ public async Task DeleteOrphanedEventsByOrganizationAsync(JobContext context)
var organizationIdTerms = await _elasticClient.SearchAsync<PersistentEvent>(s => s
.Indices(GetEventIndexPattern())
.Size(0)
.Query(q => RecentEventQuery(q, orphanedEventCutoffUtc))
.AddAggregation("terms_organization_id", a => a.Terms(c => c.Field(f => f.OrganizationId).Include(new TermsInclude(batchNumber, buckets)).Size(batchSize * 2))));

string[] organizationIds = organizationIdTerms.Aggregations?.GetStringTerms("terms_organization_id")?.Buckets.Select(b => b.Key.ToString()!).ToArray() ?? [];
Expand All @@ -223,10 +250,12 @@ public async Task DeleteOrphanedEventsByOrganizationAsync(JobContext context)
_logger.LogInformation("{BatchNumber}/{BatchCount}: Found {OrphanedEventCount} orphaned events from missing organizations {MissingOrganizationIds} out of {OrganizationIdCount}", batchNumber, buckets, missingOrganizationIds.Length, missingOrganizationIds, organizationIds.Length);
await _elasticClient.DeleteByQueryAsync<PersistentEvent>(r => r
.Indices(GetEventIndexPattern())
.Query(q => q.Terms(t => t.Field(f => f.OrganizationId).Terms(new TermsQueryField(missingOrganizationIds.Select(FieldValueHelper.ToFieldValue).ToList())))));
.Query(q => q.Bool(b => b.Filter(
f => f.Terms(t => t.Field(e => e.OrganizationId).Terms(new TermsQueryField(missingOrganizationIds.Select(FieldValueHelper.ToFieldValue).ToList()))),
f => RecentEventQuery(f, orphanedEventCutoffUtc)))));
}

_logger.LogInformation("Found {OrphanedEventCount} orphaned events from missing organizations out of {OrganizationIdCount}", totalOrphanedEventCount, totalOrganizationIds);
_logger.LogInformation("Found {OrphanedEventCount} orphaned events from missing organizations out of {OrganizationIdCount} since {OrphanedEventCutoffUtc}", totalOrphanedEventCount, totalOrganizationIds, orphanedEventCutoffUtc);
}

public async Task FixDuplicateStacks(JobContext context)
Expand Down Expand Up @@ -401,6 +430,16 @@ private Task RenewLockAsync(JobContext context)
return context.RenewLockAsync();
}

private DateTime GetOrphanedEventCutoffUtc()
{
return _timeProvider.GetUtcNow().UtcDateTime.Subtract(OrphanedEventLookback);
}

private static QueryDescriptor<PersistentEvent> RecentEventQuery(QueryDescriptor<PersistentEvent> query, DateTime orphanedEventCutoffUtc)
{
return query.Range(r => r.Date(d => d.Field(e => e.CreatedUtc).Gte(orphanedEventCutoffUtc)));
}

private string GetEventIndexPattern()
{
return $"{_config.Events.VersionedName}-*";
Expand Down
60 changes: 60 additions & 0 deletions tests/Exceptionless.Tests/Jobs/CleanupDataJobTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -323,6 +323,66 @@ public async Task CanDeleteOrphanedEventsByStack()
Assert.Equal(5000, eventCount);
}

[Fact]
public async Task RunAsync_OrphanedEventsAtAndBeforeLookbackBoundary_DeletesOnlyEventsWithinLookback()
{
TimeProvider.SetUtcNow(DateTimeOffset.UtcNow);

var organization = await _organizationRepository.AddAsync(
_organizationData.GenerateSampleOrganization(_billingManager, _plans),
o => o.ImmediateConsistency());
var project = await _projectRepository.AddAsync(
_projectData.GenerateSampleProject(),
o => o.ImmediateConsistency());
var stack = await _stackRepository.AddAsync(
_stackData.GenerateSampleStack(),
o => o.ImmediateConsistency());

var cutoffUtc = TimeProvider.GetUtcNow().UtcDateTime.Subtract(CleanupOrphanedDataJob.OrphanedEventLookback);
var beforeCutoffUtc = cutoffUtc.AddMilliseconds(-1);
var validEvent = _eventData.GenerateEvent(organization.Id, project.Id, stack.Id);

string missingStackId = ObjectId.GenerateNewId().ToString();
var stackOrphanAtCutoff = _eventData.GenerateEvent(organization.Id, project.Id, missingStackId);
stackOrphanAtCutoff.CreatedUtc = cutoffUtc;
var stackOrphanBeforeCutoff = _eventData.GenerateEvent(organization.Id, project.Id, missingStackId);
stackOrphanBeforeCutoff.CreatedUtc = beforeCutoffUtc;

string missingProjectId = ObjectId.GenerateNewId().ToString();
var projectOrphanAtCutoff = _eventData.GenerateEvent(organization.Id, missingProjectId, stack.Id);
projectOrphanAtCutoff.CreatedUtc = cutoffUtc;
var projectOrphanBeforeCutoff = _eventData.GenerateEvent(organization.Id, missingProjectId, stack.Id);
projectOrphanBeforeCutoff.CreatedUtc = beforeCutoffUtc;

string missingOrganizationId = ObjectId.GenerateNewId().ToString();
var organizationOrphanAtCutoff = _eventData.GenerateEvent(missingOrganizationId, project.Id, stack.Id);
organizationOrphanAtCutoff.CreatedUtc = cutoffUtc;
var organizationOrphanBeforeCutoff = _eventData.GenerateEvent(missingOrganizationId, project.Id, stack.Id);
organizationOrphanBeforeCutoff.CreatedUtc = beforeCutoffUtc;

await _eventRepository.AddAsync([
validEvent,
stackOrphanAtCutoff,
stackOrphanBeforeCutoff,
projectOrphanAtCutoff,
projectOrphanBeforeCutoff,
organizationOrphanAtCutoff,
organizationOrphanBeforeCutoff
], o => o.ImmediateConsistency());

await GetService<CleanupOrphanedDataJob>().RunAsync(TestCancellationToken);

var remainingEvents = await _eventRepository.GetAllAsync(o => o.PageLimit(10).ImmediateConsistency());
Assert.Equal(4, remainingEvents.Total);
Assert.Contains(remainingEvents.Documents, e => e.Id == validEvent.Id);
Assert.Contains(remainingEvents.Documents, e => e.Id == stackOrphanBeforeCutoff.Id);
Assert.Contains(remainingEvents.Documents, e => e.Id == projectOrphanBeforeCutoff.Id);
Assert.Contains(remainingEvents.Documents, e => e.Id == organizationOrphanBeforeCutoff.Id);
Assert.DoesNotContain(remainingEvents.Documents, e => e.Id == stackOrphanAtCutoff.Id);
Assert.DoesNotContain(remainingEvents.Documents, e => e.Id == projectOrphanAtCutoff.Id);
Assert.DoesNotContain(remainingEvents.Documents, e => e.Id == organizationOrphanAtCutoff.Id);
}

[Fact]
public async Task CanCleanupSuspendedTokens_MultiTenant_OnlySuspendedOrganizationTokensAffected()
{
Expand Down
Loading