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 23 commits into
base: main
Choose a base branch
from
+668 −51
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 12 commits
Commits
Show all changes
23 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
e3f7655
Merge origin/main into copilot/investigate-cluster-cpu-spikes
niemyjski Jul 16, 2026
c953187
Harden recent orphan event cleanup
niemyjski Jul 16, 2026
b0910ef
Fix orphan cleanup index coverage
niemyjski Jul 16, 2026
d11819a
Keep orphan cleanup health current
niemyjski Jul 16, 2026
55afd53
Merge remote-tracking branch 'origin/main' into copilot/investigate-c…
niemyjski Jul 26, 2026
835f9e5
Harden orphan cleanup cutoff contract
niemyjski Jul 26, 2026
c247cbf
Cover orphan cleanup discovery window
niemyjski Jul 26, 2026
c9ebc8c
Align orphan cleanup health with schedule
niemyjski Jul 26, 2026
d0bcc2b
Record orphan cleanup health after success
niemyjski Jul 26, 2026
3b36131
Merge remote-tracking branch 'origin/main' into copilot/investigate-c…
niemyjski Jul 30, 2026
6136e0d
Close orphan cleanup coverage gaps
niemyjski Jul 30, 2026
2d12721
Merge remote-tracking branch 'origin/main' into copilot/investigate-c…
niemyjski Jul 30, 2026
0333421
Reject partial orphan cleanup searches
niemyjski Jul 30, 2026
82d671d
Merge branch 'main' into copilot/investigate-cluster-cpu-spikes
niemyjski Jul 30, 2026
f4e4399
Report orphan cleanup deletions accurately
niemyjski Jul 30, 2026
5a53ae9
Track active orphan cleanup health
niemyjski Jul 30, 2026
f24d4f5
Merge remote-tracking branch 'origin/main' into copilot/investigate-c…
niemyjski Aug 5, 2026
19b09a3
docs: record PR 2279 dogfood evidence
niemyjski Aug 5, 2026
206ea76
docs: update PR 2279 CI evidence
niemyjski Aug 5, 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
72 changes: 57 additions & 15 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,8 @@ namespace Exceptionless.Core.Jobs;
[Job(Description = "Deletes orphaned data.", IsContinuous = false)]
public class CleanupOrphanedDataJob : JobWithLockBase, IHealthCheck
{
private static readonly TimeSpan HealthCheckWindow = TimeSpan.FromHours(9);
private 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 +60,30 @@ ILoggerFactory loggerFactory

protected override async Task<JobResult> RunInternalAsync(JobContext context)
{
await DeleteOrphanedEventsByStackAsync(context);
await DeleteOrphanedEventsByProjectAsync(context);
await DeleteOrphanedEventsByOrganizationAsync(context);
_lastRun = _timeProvider.GetUtcNow().UtcDateTime;
Comment thread
niemyjski marked this conversation as resolved.
Outdated
Comment thread
niemyjski marked this conversation as resolved.
Outdated

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))
Comment thread
niemyjski marked this conversation as resolved.
.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 +104,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 +127,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);
Comment thread
niemyjski marked this conversation as resolved.
Outdated
}

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

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 +167,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 +190,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 +230,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 +253,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 +433,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 All @@ -411,9 +453,9 @@ public Task<HealthCheckResult> CheckHealthAsync(HealthCheckContext context, Canc
if (!_lastRun.HasValue)
return Task.FromResult(HealthCheckResult.Healthy("Job has not been run yet."));

if (_timeProvider.GetUtcNow().UtcDateTime.Subtract(_lastRun.Value) > TimeSpan.FromMinutes(65))
return Task.FromResult(HealthCheckResult.Unhealthy("Job has not run in the last 65 minutes."));
if (_timeProvider.GetUtcNow().UtcDateTime.Subtract(_lastRun.Value) > HealthCheckWindow)
return Task.FromResult(HealthCheckResult.Unhealthy("Job has not run in the last 9 hours."));

return Task.FromResult(HealthCheckResult.Healthy("Job has run in the last 65 minutes."));
return Task.FromResult(HealthCheckResult.Healthy("Job has run in the last 9 hours."));
}
}
Loading
Loading