mirror of
https://github.com/okieselbach/AutopilotMonitor.git
synced 2026-10-24 12:42:38 +00:00
refactor(backend): split EventIngestProcessor into thematic partials
The single 1060-LoC file was hard to navigate; partial-class split mirrors the legacy TableStorageService/IngestEventsFunction pattern and keeps each file focused on one concern. - EventIngestProcessor.cs (491) — orchestrator + ctor + DI + ProcessEventsAsync + EventIngestResult DTO - EventIngestProcessor.Classification.cs (212) — ClassifyEvents, IsPeriodicOrStallEvent, UpdateSessionStatusAsync - EventIngestProcessor.Notifications.cs (168) — SendWebhookNotificationsAsync, BuildSignalRMessages - EventIngestProcessor.AppInstall.cs (171) — AggregateAppInstallEvent - EventIngestProcessor.RuleStats.cs (86) — RecordGatherRuleStatsAsync, RecordAnalyzeRuleStatsAsync Pure mechanical refactor — code is verbatim, only file boundaries moved. All 596 backend tests green. No changes to behaviour or to legacy /api/agent/ingest. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Oliver Kieselbach
@@ -0,0 +1,171 @@
|
||||
using AutopilotMonitor.Functions.Functions.Ingest;
|
||||
using AutopilotMonitor.Shared.Models;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Services
|
||||
{
|
||||
/// <summary>
|
||||
/// Per-app install summary aggregation — folds a batch of <c>app_install_*</c> +
|
||||
/// <c>download_progress</c> + <c>do_telemetry</c> events into an
|
||||
/// <see cref="AppInstallSummary"/> keyed by app name. Verbatim copy of the legacy helper;
|
||||
/// see <see cref="EventIngestProcessor"/> for the duplication rationale.
|
||||
/// </summary>
|
||||
public sealed partial class EventIngestProcessor
|
||||
{
|
||||
private void AggregateAppInstallEvent(EnrollmentEvent evt, string tenantId, string sessionId, Dictionary<string, AppInstallAggregationState> summaries)
|
||||
{
|
||||
bool isRelevant =
|
||||
evt.EventType == "app_install_started" || evt.EventType == "app_install_start" ||
|
||||
evt.EventType == "app_install_completed" || evt.EventType == "app_install_complete" ||
|
||||
evt.EventType == "app_install_failed" ||
|
||||
evt.EventType == "app_download_started" ||
|
||||
evt.EventType == "app_install_skipped" ||
|
||||
evt.EventType == "download_progress" ||
|
||||
evt.EventType == "do_telemetry";
|
||||
|
||||
if (!isRelevant) return;
|
||||
|
||||
var appName = evt.Data?.ContainsKey("appName") == true ? evt.Data["appName"]?.ToString()?.Trim() : null;
|
||||
if (string.IsNullOrEmpty(appName)) return;
|
||||
|
||||
if (!summaries.TryGetValue(appName, out var state))
|
||||
{
|
||||
state = new AppInstallAggregationState
|
||||
{
|
||||
Summary = new AppInstallSummary
|
||||
{
|
||||
AppName = appName,
|
||||
SessionId = sessionId,
|
||||
TenantId = tenantId,
|
||||
StartedAt = evt.Timestamp
|
||||
}
|
||||
};
|
||||
summaries[appName] = state;
|
||||
}
|
||||
|
||||
var summary = state.Summary;
|
||||
|
||||
if (evt.Data != null)
|
||||
{
|
||||
if (evt.Data.TryGetValue("appVersion", out var appVersionObj))
|
||||
{
|
||||
var appVersion = appVersionObj?.ToString();
|
||||
if (!string.IsNullOrWhiteSpace(appVersion))
|
||||
summary.AppVersion = appVersion.Trim();
|
||||
}
|
||||
if (evt.Data.TryGetValue("appType", out var appTypeObj))
|
||||
{
|
||||
var appType = appTypeObj?.ToString();
|
||||
if (!string.IsNullOrWhiteSpace(appType))
|
||||
summary.AppType = appType.Trim();
|
||||
}
|
||||
if (evt.Data.TryGetValue("attemptNumber", out var attemptObj) &&
|
||||
int.TryParse(attemptObj?.ToString(), out var attempt) && attempt > 0)
|
||||
{
|
||||
summary.AttemptNumber = Math.Max(summary.AttemptNumber, attempt);
|
||||
}
|
||||
if (evt.Data.TryGetValue("installerPhase", out var phaseObj))
|
||||
{
|
||||
var phase = phaseObj?.ToString();
|
||||
if (!string.IsNullOrWhiteSpace(phase))
|
||||
summary.InstallerPhase = phase.Trim();
|
||||
}
|
||||
if (evt.Data.TryGetValue("exitCode", out var exitCodeObj) &&
|
||||
int.TryParse(exitCodeObj?.ToString(), out var exitCode))
|
||||
{
|
||||
summary.ExitCode = exitCode;
|
||||
}
|
||||
if (evt.Data.TryGetValue("detectionResult", out var detectionObj))
|
||||
{
|
||||
var detection = detectionObj?.ToString();
|
||||
if (!string.IsNullOrWhiteSpace(detection))
|
||||
summary.DetectionResult = detection.Trim();
|
||||
}
|
||||
}
|
||||
|
||||
switch (evt.EventType)
|
||||
{
|
||||
case "app_install_started":
|
||||
case "app_install_start":
|
||||
if (!state.InstallStartedAt.HasValue || evt.Timestamp < state.InstallStartedAt.Value)
|
||||
state.InstallStartedAt = evt.Timestamp;
|
||||
if (summary.Status == "InProgress" || summary.Status == string.Empty)
|
||||
summary.Status = "InProgress";
|
||||
break;
|
||||
|
||||
case "app_download_started":
|
||||
if (!state.DownloadStartedAt.HasValue || evt.Timestamp < state.DownloadStartedAt.Value)
|
||||
state.DownloadStartedAt = evt.Timestamp;
|
||||
if (summary.Status == "InProgress" || summary.Status == string.Empty)
|
||||
summary.Status = "InProgress";
|
||||
break;
|
||||
|
||||
case "app_install_completed":
|
||||
case "app_install_complete":
|
||||
summary.Status = "Succeeded";
|
||||
summary.CompletedAt = evt.Timestamp;
|
||||
if (summary.StartedAt != DateTime.MinValue)
|
||||
summary.DurationSeconds = Math.Max(1, EventTimestampValidator.SafeDurationSeconds(summary.StartedAt, evt.Timestamp));
|
||||
break;
|
||||
|
||||
case "app_install_failed":
|
||||
summary.Status = "Failed";
|
||||
summary.CompletedAt = evt.Timestamp;
|
||||
if (summary.StartedAt != DateTime.MinValue)
|
||||
summary.DurationSeconds = Math.Max(1, EventTimestampValidator.SafeDurationSeconds(summary.StartedAt, evt.Timestamp));
|
||||
summary.FailureCode = evt.Data?.ContainsKey("errorCode") == true
|
||||
? evt.Data["errorCode"]?.ToString() ?? string.Empty : string.Empty;
|
||||
summary.FailureMessage = evt.Data?.ContainsKey("errorMessage") == true
|
||||
? evt.Data["errorMessage"]?.ToString() ?? string.Empty : evt.Message ?? string.Empty;
|
||||
break;
|
||||
|
||||
case "app_install_skipped":
|
||||
if (summary.Status == "InProgress")
|
||||
summary.Status = "Succeeded";
|
||||
break;
|
||||
|
||||
case "download_progress":
|
||||
var bytesKey = evt.Data?.ContainsKey("bytesDownloaded") == true ? "bytesDownloaded"
|
||||
: evt.Data?.ContainsKey("bytes_downloaded") == true ? "bytes_downloaded" : null;
|
||||
if (bytesKey != null && long.TryParse(evt.Data![bytesKey]?.ToString(), out var bytes))
|
||||
summary.DownloadBytes = Math.Max(summary.DownloadBytes, bytes);
|
||||
break;
|
||||
|
||||
case "do_telemetry":
|
||||
if (evt.Data != null)
|
||||
{
|
||||
if (evt.Data.ContainsKey("doFileSize") && long.TryParse(evt.Data["doFileSize"]?.ToString(), out var doFs))
|
||||
{
|
||||
summary.DownloadBytes = Math.Max(summary.DownloadBytes, doFs);
|
||||
summary.DoFileSize = doFs;
|
||||
}
|
||||
if (evt.Data.ContainsKey("doTotalBytesDownloaded") && long.TryParse(evt.Data["doTotalBytesDownloaded"]?.ToString(), out var doTotalDl))
|
||||
summary.DoTotalBytesDownloaded = doTotalDl;
|
||||
if (evt.Data.ContainsKey("doBytesFromPeers") && long.TryParse(evt.Data["doBytesFromPeers"]?.ToString(), out var doPeers))
|
||||
summary.DoBytesFromPeers = doPeers;
|
||||
if (evt.Data.ContainsKey("doBytesFromHttp") && long.TryParse(evt.Data["doBytesFromHttp"]?.ToString(), out var doHttp))
|
||||
summary.DoBytesFromHttp = doHttp;
|
||||
if (evt.Data.ContainsKey("doPercentPeerCaching") && int.TryParse(evt.Data["doPercentPeerCaching"]?.ToString(), out var doPct))
|
||||
summary.DoPercentPeerCaching = doPct;
|
||||
if (evt.Data.ContainsKey("doDownloadMode") && int.TryParse(evt.Data["doDownloadMode"]?.ToString(), out var doMode))
|
||||
summary.DoDownloadMode = doMode;
|
||||
if (evt.Data.ContainsKey("doDownloadDuration"))
|
||||
{
|
||||
var doDurStr = evt.Data["doDownloadDuration"]?.ToString() ?? string.Empty;
|
||||
summary.DoDownloadDuration = doDurStr;
|
||||
if (TimeSpan.TryParse(doDurStr, out var doDurTs) && doDurTs.TotalSeconds >= 1)
|
||||
summary.DownloadDurationSeconds = Math.Max(summary.DownloadDurationSeconds, (int)doDurTs.TotalSeconds);
|
||||
}
|
||||
if (evt.Data.ContainsKey("doBytesFromLanPeers") && long.TryParse(evt.Data["doBytesFromLanPeers"]?.ToString(), out var doLan))
|
||||
summary.DoBytesFromLanPeers = doLan;
|
||||
if (evt.Data.ContainsKey("doBytesFromGroupPeers") && long.TryParse(evt.Data["doBytesFromGroupPeers"]?.ToString(), out var doGroup))
|
||||
summary.DoBytesFromGroupPeers = doGroup;
|
||||
if (evt.Data.ContainsKey("doBytesFromInternetPeers") && long.TryParse(evt.Data["doBytesFromInternetPeers"]?.ToString(), out var doInet))
|
||||
summary.DoBytesFromInternetPeers = doInet;
|
||||
}
|
||||
break;
|
||||
}
|
||||
|
||||
IngestEventsFunction.RecalculateAppDurations(state);
|
||||
}
|
||||
}
|
||||
}
|
||||
+212
@@ -0,0 +1,212 @@
|
||||
using AutopilotMonitor.Functions.Functions.Ingest;
|
||||
using AutopilotMonitor.Shared.Models;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Services
|
||||
{
|
||||
/// <summary>
|
||||
/// Event classification + session-status transitions. Verbatim copy of the legacy
|
||||
/// <c>IngestEventsFunction</c> helpers (see the class-level comment on
|
||||
/// <see cref="EventIngestProcessor"/> for the copy-duplication rationale).
|
||||
/// </summary>
|
||||
public sealed partial class EventIngestProcessor
|
||||
{
|
||||
private EventClassification ClassifyEvents(List<EnrollmentEvent> storedEvents)
|
||||
{
|
||||
var classification = new EventClassification();
|
||||
|
||||
foreach (var evt in storedEvents)
|
||||
{
|
||||
if (!classification.EarliestEventTimestamp.HasValue || evt.Timestamp < classification.EarliestEventTimestamp.Value)
|
||||
classification.EarliestEventTimestamp = evt.Timestamp;
|
||||
if (!classification.LatestEventTimestamp.HasValue || evt.Timestamp > classification.LatestEventTimestamp.Value)
|
||||
classification.LatestEventTimestamp = evt.Timestamp;
|
||||
|
||||
switch (evt.EventType)
|
||||
{
|
||||
case "phase_changed":
|
||||
case "esp_phase_changed":
|
||||
classification.LastPhaseChangeEvent = evt;
|
||||
break;
|
||||
case "enrollment_complete":
|
||||
classification.CompletionEvent = evt;
|
||||
break;
|
||||
case "gather_rules_collection_completed":
|
||||
classification.GatherCompletionEvent = evt;
|
||||
break;
|
||||
case "enrollment_failed":
|
||||
classification.FailureEvent = evt;
|
||||
break;
|
||||
case "diagnostics_uploaded":
|
||||
classification.DiagnosticsUploadedEvent = evt;
|
||||
break;
|
||||
case "whiteglove_complete":
|
||||
classification.WhiteGloveEvent = evt;
|
||||
break;
|
||||
case "whiteglove_resumed":
|
||||
classification.WhiteGloveResumedEvent = evt;
|
||||
break;
|
||||
case "whiteglove_started":
|
||||
classification.WhiteGloveStartedEvent = evt;
|
||||
break;
|
||||
case "esp_failure":
|
||||
classification.EspFailureEvent = evt;
|
||||
break;
|
||||
case "device_location":
|
||||
classification.DeviceLocationEvent = evt;
|
||||
break;
|
||||
case "session_stalled":
|
||||
classification.SessionStalledEvent = evt;
|
||||
break;
|
||||
case "script_completed":
|
||||
case "script_failed":
|
||||
var scriptType = evt.Data?.ContainsKey("scriptType") == true
|
||||
? evt.Data["scriptType"]?.ToString() : null;
|
||||
if (string.Equals(scriptType, "platform", StringComparison.OrdinalIgnoreCase))
|
||||
classification.PlatformScriptCount++;
|
||||
else if (string.Equals(scriptType, "remediation", StringComparison.OrdinalIgnoreCase))
|
||||
classification.RemediationScriptCount++;
|
||||
break;
|
||||
}
|
||||
|
||||
if (!IsPeriodicOrStallEvent(evt.EventType))
|
||||
classification.HasNonPeriodicRealEvent = true;
|
||||
|
||||
AggregateAppInstallEvent(evt, storedEvents[0].TenantId!, storedEvents[0].SessionId!, classification.AppInstallUpdates);
|
||||
}
|
||||
|
||||
return classification;
|
||||
}
|
||||
|
||||
private static bool IsPeriodicOrStallEvent(string? eventType) => eventType switch
|
||||
{
|
||||
"performance_snapshot" => true,
|
||||
"agent_metrics_snapshot" => true,
|
||||
"performance_collector_stopped" => true,
|
||||
"agent_metrics_collector_stopped" => true,
|
||||
"stall_probe_check" => true,
|
||||
"stall_probe_result" => true,
|
||||
"session_stalled" => true,
|
||||
"modern_deployment_log" => true,
|
||||
_ => false
|
||||
};
|
||||
|
||||
private async Task<(bool statusTransitioned, bool whiteGloveStatusTransitioned, string? failureReason)>
|
||||
UpdateSessionStatusAsync(IngestEventsRequest request, string sessionPrefix, EventClassification c)
|
||||
{
|
||||
bool statusTransitioned = false;
|
||||
bool whiteGloveStatusTransitioned = false;
|
||||
string? failureReason = null;
|
||||
|
||||
if (c.CompletionEvent != null)
|
||||
{
|
||||
statusTransitioned = await _sessionRepo.UpdateSessionStatusAsync(
|
||||
request.TenantId, request.SessionId, SessionStatus.Succeeded, c.CompletionEvent.Phase,
|
||||
completedAt: c.CompletionEvent.Timestamp,
|
||||
earliestEventTimestamp: c.EarliestEventTimestamp, latestEventTimestamp: c.LatestEventTimestamp);
|
||||
_logger.LogInformation("{SessionPrefix} Status: Succeeded (transitioned={Transitioned})", sessionPrefix, statusTransitioned);
|
||||
}
|
||||
else if (c.FailureEvent != null)
|
||||
{
|
||||
failureReason = c.FailureEvent.Data?.ContainsKey("errorCode") == true
|
||||
? $"{c.FailureEvent.Message} ({c.FailureEvent.Data["errorCode"]})"
|
||||
: c.FailureEvent.Message;
|
||||
|
||||
statusTransitioned = await _sessionRepo.UpdateSessionStatusAsync(
|
||||
request.TenantId, request.SessionId, SessionStatus.Failed, c.FailureEvent.Phase, failureReason,
|
||||
completedAt: c.FailureEvent.Timestamp,
|
||||
earliestEventTimestamp: c.EarliestEventTimestamp, latestEventTimestamp: c.LatestEventTimestamp);
|
||||
_logger.LogWarning("{SessionPrefix} Status: Failed - {FailureReason} (transitioned={Transitioned})", sessionPrefix, failureReason, statusTransitioned);
|
||||
}
|
||||
else if (c.EspFailureEvent != null)
|
||||
{
|
||||
failureReason = c.EspFailureEvent.Message ?? "ESP failure (backend fallback)";
|
||||
statusTransitioned = await _sessionRepo.UpdateSessionStatusAsync(
|
||||
request.TenantId, request.SessionId, SessionStatus.Failed, c.EspFailureEvent.Phase, failureReason,
|
||||
completedAt: c.EspFailureEvent.Timestamp,
|
||||
earliestEventTimestamp: c.EarliestEventTimestamp, latestEventTimestamp: c.LatestEventTimestamp);
|
||||
_logger.LogWarning("{SessionPrefix} Status: Failed via esp_failure fallback - {FailureReason} (transitioned={Transitioned})",
|
||||
sessionPrefix, failureReason, statusTransitioned);
|
||||
}
|
||||
else if (c.GatherCompletionEvent != null)
|
||||
{
|
||||
await _sessionRepo.UpdateSessionStatusAsync(
|
||||
request.TenantId, request.SessionId, SessionStatus.Succeeded, c.GatherCompletionEvent.Phase,
|
||||
completedAt: c.GatherCompletionEvent.Timestamp,
|
||||
earliestEventTimestamp: c.EarliestEventTimestamp, latestEventTimestamp: c.LatestEventTimestamp);
|
||||
_logger.LogInformation("{SessionPrefix} Status: Succeeded (gather_rules)", sessionPrefix);
|
||||
}
|
||||
else if (c.WhiteGloveEvent != null)
|
||||
{
|
||||
whiteGloveStatusTransitioned = await _sessionRepo.UpdateSessionStatusAsync(
|
||||
request.TenantId, request.SessionId, SessionStatus.Pending, EnrollmentPhase.AppsDevice,
|
||||
earliestEventTimestamp: c.EarliestEventTimestamp, latestEventTimestamp: c.LatestEventTimestamp,
|
||||
isPreProvisioned: true, isUserDriven: false);
|
||||
|
||||
if (!whiteGloveStatusTransitioned)
|
||||
{
|
||||
_logger.LogWarning("{SessionPrefix} WhiteGlove UpdateSessionStatusAsync failed, attempting unconditional fallback for IsPreProvisioned + Status", sessionPrefix);
|
||||
try
|
||||
{
|
||||
await _sessionRepo.SetSessionPreProvisionedAsync(request.TenantId, request.SessionId, true, SessionStatus.Pending, isUserDriven: false);
|
||||
whiteGloveStatusTransitioned = true;
|
||||
_logger.LogInformation("{SessionPrefix} WhiteGlove fallback succeeded: IsPreProvisioned + Status=Pending set via unconditional merge", sessionPrefix);
|
||||
}
|
||||
catch (Exception fallbackEx)
|
||||
{
|
||||
_logger.LogError(fallbackEx, "{SessionPrefix} WhiteGlove fallback SetSessionPreProvisionedAsync also failed", sessionPrefix);
|
||||
}
|
||||
}
|
||||
|
||||
_logger.LogInformation("{SessionPrefix} Status: Pending (WhiteGlove pre-provisioning complete, transitioned={Transitioned})", sessionPrefix, whiteGloveStatusTransitioned);
|
||||
}
|
||||
else if (c.WhiteGloveResumedEvent != null)
|
||||
{
|
||||
var currentSession = await _sessionRepo.GetSessionAsync(request.TenantId, request.SessionId);
|
||||
if (currentSession?.Status == SessionStatus.Pending)
|
||||
{
|
||||
await _sessionRepo.UpdateSessionStatusAsync(
|
||||
request.TenantId, request.SessionId, SessionStatus.InProgress, c.WhiteGloveResumedEvent.Phase,
|
||||
earliestEventTimestamp: c.EarliestEventTimestamp, latestEventTimestamp: c.LatestEventTimestamp,
|
||||
isUserDriven: true, resumedAt: c.WhiteGloveResumedEvent.Timestamp);
|
||||
_logger.LogInformation("{SessionPrefix} Status: InProgress (WhiteGlove Part 2 resumed, IsUserDriven=true)", sessionPrefix);
|
||||
}
|
||||
else
|
||||
{
|
||||
_logger.LogInformation("{SessionPrefix} WhiteGlove resumed skipped, session already {Status}", sessionPrefix, currentSession?.Status);
|
||||
}
|
||||
}
|
||||
|
||||
if (c.WhiteGloveStartedEvent != null)
|
||||
{
|
||||
_logger.LogInformation("{SessionPrefix} whiteglove_started detected (soft signal — not setting IsPreProvisioned, awaiting whiteglove_complete)", sessionPrefix);
|
||||
}
|
||||
|
||||
if (c.SessionStalledEvent != null)
|
||||
{
|
||||
var stalledReason = "Agent reported stall after 60min without progress (stall_probe)";
|
||||
var stalledTransitioned = await _sessionRepo.UpdateSessionStatusAsync(
|
||||
request.TenantId, request.SessionId, SessionStatus.Stalled,
|
||||
earliestEventTimestamp: c.EarliestEventTimestamp, latestEventTimestamp: c.LatestEventTimestamp,
|
||||
stalledAt: c.SessionStalledEvent.Timestamp, failureReason: stalledReason);
|
||||
if (stalledTransitioned)
|
||||
_logger.LogWarning("{SessionPrefix} Status: Stalled (agent-reported via session_stalled event)", sessionPrefix);
|
||||
}
|
||||
else if (c.HasNonPeriodicRealEvent && !statusTransitioned && !whiteGloveStatusTransitioned)
|
||||
{
|
||||
var currentSession = await _sessionRepo.GetSessionAsync(request.TenantId, request.SessionId);
|
||||
if (currentSession?.Status == SessionStatus.Stalled)
|
||||
{
|
||||
var healed = await _sessionRepo.UpdateSessionStatusAsync(
|
||||
request.TenantId, request.SessionId, SessionStatus.InProgress,
|
||||
earliestEventTimestamp: c.EarliestEventTimestamp, latestEventTimestamp: c.LatestEventTimestamp,
|
||||
clearStalledAt: true, clearFailureReason: true);
|
||||
if (healed)
|
||||
_logger.LogInformation("{SessionPrefix} Status: InProgress (healed from Stalled by new real event)", sessionPrefix);
|
||||
}
|
||||
}
|
||||
|
||||
return (statusTransitioned, whiteGloveStatusTransitioned, failureReason);
|
||||
}
|
||||
}
|
||||
}
|
||||
+168
@@ -0,0 +1,168 @@
|
||||
using System.Linq;
|
||||
using AutopilotMonitor.Functions.Functions.Ingest;
|
||||
using AutopilotMonitor.Functions.Services.Notifications;
|
||||
using AutopilotMonitor.Shared.Models;
|
||||
using AutopilotMonitor.Shared.Models.Notifications;
|
||||
using Microsoft.Azure.Functions.Worker;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Services
|
||||
{
|
||||
/// <summary>
|
||||
/// External + real-time notification plumbing: outgoing webhook alerts (Teams/Slack-style)
|
||||
/// and SignalR messages for the live UI push. Verbatim copy of the legacy helpers; see
|
||||
/// <see cref="EventIngestProcessor"/> for the duplication rationale.
|
||||
/// </summary>
|
||||
public sealed partial class EventIngestProcessor
|
||||
{
|
||||
private async Task SendWebhookNotificationsAsync(
|
||||
IngestEventsRequest request, string sessionPrefix, EventClassification c,
|
||||
SessionSummary? updatedSession, bool statusTransitioned, bool whiteGloveStatusTransitioned, string? failureReason,
|
||||
List<RuleResult> ruleResults)
|
||||
{
|
||||
var tenantConfig = await _configService.GetConfigurationAsync(request.TenantId);
|
||||
var (webhookUrl, providerTypeInt) = tenantConfig.GetEffectiveWebhookConfig();
|
||||
|
||||
if (string.IsNullOrEmpty(webhookUrl) || providerTypeInt == 0)
|
||||
return;
|
||||
|
||||
var providerType = (WebhookProviderType)providerTypeInt;
|
||||
var sessionUrl = updatedSession != null
|
||||
? $"https://www.autopilotmonitor.com/session/{request.TenantId}/{request.SessionId}"
|
||||
: null;
|
||||
|
||||
if (statusTransitioned && (c.CompletionEvent != null || c.FailureEvent != null))
|
||||
{
|
||||
var notifySuccess = c.CompletionEvent != null && tenantConfig.GetEffectiveNotifyOnSuccess();
|
||||
var notifyFailure = c.FailureEvent != null && tenantConfig.GetEffectiveNotifyOnFailure();
|
||||
if (notifySuccess || notifyFailure)
|
||||
{
|
||||
var duration = updatedSession?.DurationSeconds != null
|
||||
? TimeSpan.FromSeconds(updatedSession.DurationSeconds.Value)
|
||||
: (TimeSpan?)null;
|
||||
|
||||
if (updatedSession?.IsPreProvisioned == true && updatedSession?.ResumedAt != null)
|
||||
{
|
||||
var completionTime = c.CompletionEvent?.Timestamp ?? c.FailureEvent?.Timestamp;
|
||||
if (completionTime.HasValue)
|
||||
duration = completionTime.Value - updatedSession.ResumedAt.Value;
|
||||
}
|
||||
|
||||
var alert = NotificationAlertBuilder.BuildEnrollmentAlert(
|
||||
updatedSession?.DeviceName,
|
||||
updatedSession?.SerialNumber,
|
||||
updatedSession?.Manufacturer,
|
||||
updatedSession?.Model,
|
||||
success: c.CompletionEvent != null,
|
||||
failureReason: failureReason,
|
||||
duration: duration,
|
||||
sessionUrl: sessionUrl);
|
||||
NotificationAlertBuilder.AddRuleResultSections(alert, ruleResults);
|
||||
|
||||
_ = _webhookNotificationService.SendNotificationAsync(webhookUrl, providerType, alert)
|
||||
.ContinueWith(t => _logger.LogWarning(t.Exception?.InnerException,
|
||||
"Fire-and-forget webhook notification failed"), TaskContinuationOptions.OnlyOnFaulted);
|
||||
}
|
||||
}
|
||||
|
||||
if (whiteGloveStatusTransitioned && c.WhiteGloveEvent != null && tenantConfig.GetEffectiveNotifyOnSuccess())
|
||||
{
|
||||
var duration = updatedSession?.DurationSeconds != null
|
||||
? TimeSpan.FromSeconds(updatedSession.DurationSeconds.Value)
|
||||
: (TimeSpan?)null;
|
||||
|
||||
var alert = NotificationAlertBuilder.BuildWhiteGloveAlert(
|
||||
updatedSession?.DeviceName,
|
||||
updatedSession?.SerialNumber,
|
||||
updatedSession?.Manufacturer,
|
||||
updatedSession?.Model,
|
||||
success: true,
|
||||
duration: duration,
|
||||
sessionUrl: sessionUrl);
|
||||
NotificationAlertBuilder.AddRuleResultSections(alert, ruleResults);
|
||||
|
||||
_ = _webhookNotificationService.SendNotificationAsync(webhookUrl, providerType, alert)
|
||||
.ContinueWith(t => _logger.LogWarning(t.Exception?.InnerException,
|
||||
"Fire-and-forget webhook notification failed"), TaskContinuationOptions.OnlyOnFaulted);
|
||||
}
|
||||
|
||||
if (c.EspFailureEvent != null && updatedSession?.IsPreProvisioned == true && tenantConfig.GetEffectiveNotifyOnFailure())
|
||||
{
|
||||
var duration = updatedSession?.DurationSeconds != null
|
||||
? TimeSpan.FromSeconds(updatedSession.DurationSeconds.Value)
|
||||
: (TimeSpan?)null;
|
||||
|
||||
var alert = NotificationAlertBuilder.BuildWhiteGloveAlert(
|
||||
updatedSession?.DeviceName,
|
||||
updatedSession?.SerialNumber,
|
||||
updatedSession?.Manufacturer,
|
||||
updatedSession?.Model,
|
||||
success: false,
|
||||
duration: duration,
|
||||
sessionUrl: sessionUrl);
|
||||
NotificationAlertBuilder.AddRuleResultSections(alert, ruleResults);
|
||||
|
||||
_ = _webhookNotificationService.SendNotificationAsync(webhookUrl, providerType, alert)
|
||||
.ContinueWith(t => _logger.LogWarning(t.Exception?.InnerException,
|
||||
"Fire-and-forget webhook notification failed"), TaskContinuationOptions.OnlyOnFaulted);
|
||||
}
|
||||
}
|
||||
|
||||
private SignalRMessageAction[] BuildSignalRMessages(
|
||||
IngestEventsRequest request, SessionSummary? updatedSession, int processedCount,
|
||||
List<RuleResult> newRuleResults)
|
||||
{
|
||||
object? sessionDelta = updatedSession != null ? new {
|
||||
updatedSession.CurrentPhase,
|
||||
updatedSession.CurrentPhaseDetail,
|
||||
updatedSession.Status,
|
||||
updatedSession.FailureReason,
|
||||
updatedSession.EventCount,
|
||||
updatedSession.DurationSeconds,
|
||||
updatedSession.CompletedAt,
|
||||
updatedSession.DiagnosticsBlobName,
|
||||
updatedSession.IsPreProvisioned
|
||||
} : null;
|
||||
|
||||
var summaryMessage = new SignalRMessageAction("newevents")
|
||||
{
|
||||
GroupName = $"tenant-{request.TenantId}",
|
||||
Arguments = new object[] { new {
|
||||
sessionId = request.SessionId,
|
||||
tenantId = request.TenantId,
|
||||
eventCount = processedCount,
|
||||
sessionUpdate = sessionDelta
|
||||
} }
|
||||
};
|
||||
|
||||
var slimRuleResults = newRuleResults.Count > 0
|
||||
? newRuleResults.Select(r => new {
|
||||
r.ResultId,
|
||||
r.RuleId,
|
||||
r.RuleTitle,
|
||||
r.Severity,
|
||||
r.Category,
|
||||
r.ConfidenceScore,
|
||||
r.Explanation,
|
||||
r.Remediation,
|
||||
r.RelatedDocs,
|
||||
r.MatchedConditions,
|
||||
r.DetectedAt
|
||||
}).ToList<object>()
|
||||
: null;
|
||||
|
||||
var eventsMessage = new SignalRMessageAction("eventStream")
|
||||
{
|
||||
GroupName = $"session-{request.TenantId}-{request.SessionId}",
|
||||
Arguments = new object[] { new {
|
||||
sessionId = request.SessionId,
|
||||
tenantId = request.TenantId,
|
||||
newEventCount = processedCount,
|
||||
newRuleResults = slimRuleResults
|
||||
} }
|
||||
};
|
||||
|
||||
return new[] { summaryMessage, eventsMessage };
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
using System.Linq;
|
||||
using AutopilotMonitor.Shared.Models;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Services
|
||||
{
|
||||
/// <summary>
|
||||
/// Daily rule-fire/evaluation telemetry (per-tenant + global aggregate rows). Verbatim
|
||||
/// copy of the legacy helpers; see <see cref="EventIngestProcessor"/> for the duplication
|
||||
/// rationale. Both methods are fire-and-forget — failures are logged, never thrown.
|
||||
/// </summary>
|
||||
public sealed partial class EventIngestProcessor
|
||||
{
|
||||
private async Task RecordGatherRuleStatsAsync(string tenantId, List<EnrollmentEvent> events)
|
||||
{
|
||||
try
|
||||
{
|
||||
var today = DateTime.UtcNow.ToString("yyyy-MM-dd");
|
||||
|
||||
var gatherEvents = events.Where(e =>
|
||||
e.Data != null &&
|
||||
e.Data.ContainsKey("ruleId") &&
|
||||
e.Source == "GatherRuleExecutor").ToList();
|
||||
|
||||
if (gatherEvents.Count == 0) return;
|
||||
|
||||
var seen = new HashSet<string>(StringComparer.OrdinalIgnoreCase);
|
||||
foreach (var evt in gatherEvents)
|
||||
{
|
||||
var ruleId = evt.Data!["ruleId"]?.ToString();
|
||||
if (string.IsNullOrEmpty(ruleId) || !seen.Add(ruleId)) continue;
|
||||
|
||||
var ruleTitle = evt.Data.ContainsKey("ruleTitle") ? evt.Data["ruleTitle"]?.ToString() ?? "" : "";
|
||||
|
||||
await _metricsRepo.IncrementRuleStatAsync(
|
||||
today, tenantId, ruleId, "gather",
|
||||
ruleTitle, "", "",
|
||||
fired: true, confidenceScore: null);
|
||||
|
||||
await _metricsRepo.IncrementRuleStatAsync(
|
||||
today, "global", ruleId, "gather",
|
||||
ruleTitle, "", "",
|
||||
fired: true, confidenceScore: null);
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Failed to record gather rule stats (non-fatal)");
|
||||
}
|
||||
}
|
||||
|
||||
private async Task RecordAnalyzeRuleStatsAsync(string tenantId, AnalysisOutcome outcome)
|
||||
{
|
||||
try
|
||||
{
|
||||
var today = DateTime.UtcNow.ToString("yyyy-MM-dd");
|
||||
var firedRuleIds = new HashSet<string>(outcome.Results.Select(r => r.RuleId));
|
||||
|
||||
foreach (var rule in outcome.EvaluatedRules)
|
||||
{
|
||||
var fired = firedRuleIds.Contains(rule.RuleId);
|
||||
int? confidence = null;
|
||||
if (fired)
|
||||
{
|
||||
var result = outcome.Results.FirstOrDefault(r => r.RuleId == rule.RuleId);
|
||||
confidence = result?.ConfidenceScore;
|
||||
}
|
||||
|
||||
await _metricsRepo.IncrementRuleStatAsync(
|
||||
today, tenantId, rule.RuleId, "analyze",
|
||||
rule.Title, rule.Category, rule.Severity,
|
||||
fired, confidence);
|
||||
|
||||
await _metricsRepo.IncrementRuleStatAsync(
|
||||
today, "global", rule.RuleId, "analyze",
|
||||
rule.Title, rule.Category, rule.Severity,
|
||||
fired, confidence);
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Failed to record analyze rule stats (non-fatal)");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -5,7 +5,6 @@ using AutopilotMonitor.Functions.Services.Notifications;
|
||||
using AutopilotMonitor.Functions.Services.Vulnerability;
|
||||
using AutopilotMonitor.Shared.DataAccess;
|
||||
using AutopilotMonitor.Shared.Models;
|
||||
using AutopilotMonitor.Shared.Models.Notifications;
|
||||
using Microsoft.ApplicationInsights;
|
||||
using Microsoft.Azure.Functions.Worker;
|
||||
using Microsoft.Extensions.Logging;
|
||||
@@ -33,8 +32,18 @@ namespace AutopilotMonitor.Functions.Services
|
||||
/// (state-less, stable, no behaviour) — the duplication is scoped to the logic that
|
||||
/// actually runs.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Split across partials for readability — this file owns the orchestrator (ctor + DI +
|
||||
/// <see cref="ProcessEventsAsync"/>); thematic helpers live in siblings:
|
||||
/// <c>.Classification.cs</c> (<c>ClassifyEvents</c>, <c>IsPeriodicOrStallEvent</c>,
|
||||
/// <c>UpdateSessionStatusAsync</c>), <c>.Notifications.cs</c>
|
||||
/// (<c>SendWebhookNotificationsAsync</c>, <c>BuildSignalRMessages</c>),
|
||||
/// <c>.RuleStats.cs</c> (<c>RecordGatherRuleStatsAsync</c>,
|
||||
/// <c>RecordAnalyzeRuleStatsAsync</c>), <c>.AppInstall.cs</c>
|
||||
/// (<c>AggregateAppInstallEvent</c>).
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public sealed class EventIngestProcessor
|
||||
public sealed partial class EventIngestProcessor
|
||||
{
|
||||
private readonly ILogger<EventIngestProcessor> _logger;
|
||||
private readonly ISessionRepository _sessionRepo;
|
||||
@@ -465,586 +474,6 @@ namespace AutopilotMonitor.Functions.Services
|
||||
};
|
||||
}
|
||||
|
||||
// --------------------------------------------------------------------------- Helpers
|
||||
// Everything below is a verbatim copy of IngestEventsFunction's private instance helpers.
|
||||
// Keep the two copies in sync until /api/agent/ingest is decommissioned.
|
||||
|
||||
private EventClassification ClassifyEvents(List<EnrollmentEvent> storedEvents)
|
||||
{
|
||||
var classification = new EventClassification();
|
||||
|
||||
foreach (var evt in storedEvents)
|
||||
{
|
||||
if (!classification.EarliestEventTimestamp.HasValue || evt.Timestamp < classification.EarliestEventTimestamp.Value)
|
||||
classification.EarliestEventTimestamp = evt.Timestamp;
|
||||
if (!classification.LatestEventTimestamp.HasValue || evt.Timestamp > classification.LatestEventTimestamp.Value)
|
||||
classification.LatestEventTimestamp = evt.Timestamp;
|
||||
|
||||
switch (evt.EventType)
|
||||
{
|
||||
case "phase_changed":
|
||||
case "esp_phase_changed":
|
||||
classification.LastPhaseChangeEvent = evt;
|
||||
break;
|
||||
case "enrollment_complete":
|
||||
classification.CompletionEvent = evt;
|
||||
break;
|
||||
case "gather_rules_collection_completed":
|
||||
classification.GatherCompletionEvent = evt;
|
||||
break;
|
||||
case "enrollment_failed":
|
||||
classification.FailureEvent = evt;
|
||||
break;
|
||||
case "diagnostics_uploaded":
|
||||
classification.DiagnosticsUploadedEvent = evt;
|
||||
break;
|
||||
case "whiteglove_complete":
|
||||
classification.WhiteGloveEvent = evt;
|
||||
break;
|
||||
case "whiteglove_resumed":
|
||||
classification.WhiteGloveResumedEvent = evt;
|
||||
break;
|
||||
case "whiteglove_started":
|
||||
classification.WhiteGloveStartedEvent = evt;
|
||||
break;
|
||||
case "esp_failure":
|
||||
classification.EspFailureEvent = evt;
|
||||
break;
|
||||
case "device_location":
|
||||
classification.DeviceLocationEvent = evt;
|
||||
break;
|
||||
case "session_stalled":
|
||||
classification.SessionStalledEvent = evt;
|
||||
break;
|
||||
case "script_completed":
|
||||
case "script_failed":
|
||||
var scriptType = evt.Data?.ContainsKey("scriptType") == true
|
||||
? evt.Data["scriptType"]?.ToString() : null;
|
||||
if (string.Equals(scriptType, "platform", StringComparison.OrdinalIgnoreCase))
|
||||
classification.PlatformScriptCount++;
|
||||
else if (string.Equals(scriptType, "remediation", StringComparison.OrdinalIgnoreCase))
|
||||
classification.RemediationScriptCount++;
|
||||
break;
|
||||
}
|
||||
|
||||
if (!IsPeriodicOrStallEvent(evt.EventType))
|
||||
classification.HasNonPeriodicRealEvent = true;
|
||||
|
||||
AggregateAppInstallEvent(evt, storedEvents[0].TenantId!, storedEvents[0].SessionId!, classification.AppInstallUpdates);
|
||||
}
|
||||
|
||||
return classification;
|
||||
}
|
||||
|
||||
private static bool IsPeriodicOrStallEvent(string? eventType) => eventType switch
|
||||
{
|
||||
"performance_snapshot" => true,
|
||||
"agent_metrics_snapshot" => true,
|
||||
"performance_collector_stopped" => true,
|
||||
"agent_metrics_collector_stopped" => true,
|
||||
"stall_probe_check" => true,
|
||||
"stall_probe_result" => true,
|
||||
"session_stalled" => true,
|
||||
"modern_deployment_log" => true,
|
||||
_ => false
|
||||
};
|
||||
|
||||
private async Task<(bool statusTransitioned, bool whiteGloveStatusTransitioned, string? failureReason)>
|
||||
UpdateSessionStatusAsync(IngestEventsRequest request, string sessionPrefix, EventClassification c)
|
||||
{
|
||||
bool statusTransitioned = false;
|
||||
bool whiteGloveStatusTransitioned = false;
|
||||
string? failureReason = null;
|
||||
|
||||
if (c.CompletionEvent != null)
|
||||
{
|
||||
statusTransitioned = await _sessionRepo.UpdateSessionStatusAsync(
|
||||
request.TenantId, request.SessionId, SessionStatus.Succeeded, c.CompletionEvent.Phase,
|
||||
completedAt: c.CompletionEvent.Timestamp,
|
||||
earliestEventTimestamp: c.EarliestEventTimestamp, latestEventTimestamp: c.LatestEventTimestamp);
|
||||
_logger.LogInformation("{SessionPrefix} Status: Succeeded (transitioned={Transitioned})", sessionPrefix, statusTransitioned);
|
||||
}
|
||||
else if (c.FailureEvent != null)
|
||||
{
|
||||
failureReason = c.FailureEvent.Data?.ContainsKey("errorCode") == true
|
||||
? $"{c.FailureEvent.Message} ({c.FailureEvent.Data["errorCode"]})"
|
||||
: c.FailureEvent.Message;
|
||||
|
||||
statusTransitioned = await _sessionRepo.UpdateSessionStatusAsync(
|
||||
request.TenantId, request.SessionId, SessionStatus.Failed, c.FailureEvent.Phase, failureReason,
|
||||
completedAt: c.FailureEvent.Timestamp,
|
||||
earliestEventTimestamp: c.EarliestEventTimestamp, latestEventTimestamp: c.LatestEventTimestamp);
|
||||
_logger.LogWarning("{SessionPrefix} Status: Failed - {FailureReason} (transitioned={Transitioned})", sessionPrefix, failureReason, statusTransitioned);
|
||||
}
|
||||
else if (c.EspFailureEvent != null)
|
||||
{
|
||||
failureReason = c.EspFailureEvent.Message ?? "ESP failure (backend fallback)";
|
||||
statusTransitioned = await _sessionRepo.UpdateSessionStatusAsync(
|
||||
request.TenantId, request.SessionId, SessionStatus.Failed, c.EspFailureEvent.Phase, failureReason,
|
||||
completedAt: c.EspFailureEvent.Timestamp,
|
||||
earliestEventTimestamp: c.EarliestEventTimestamp, latestEventTimestamp: c.LatestEventTimestamp);
|
||||
_logger.LogWarning("{SessionPrefix} Status: Failed via esp_failure fallback - {FailureReason} (transitioned={Transitioned})",
|
||||
sessionPrefix, failureReason, statusTransitioned);
|
||||
}
|
||||
else if (c.GatherCompletionEvent != null)
|
||||
{
|
||||
await _sessionRepo.UpdateSessionStatusAsync(
|
||||
request.TenantId, request.SessionId, SessionStatus.Succeeded, c.GatherCompletionEvent.Phase,
|
||||
completedAt: c.GatherCompletionEvent.Timestamp,
|
||||
earliestEventTimestamp: c.EarliestEventTimestamp, latestEventTimestamp: c.LatestEventTimestamp);
|
||||
_logger.LogInformation("{SessionPrefix} Status: Succeeded (gather_rules)", sessionPrefix);
|
||||
}
|
||||
else if (c.WhiteGloveEvent != null)
|
||||
{
|
||||
whiteGloveStatusTransitioned = await _sessionRepo.UpdateSessionStatusAsync(
|
||||
request.TenantId, request.SessionId, SessionStatus.Pending, EnrollmentPhase.AppsDevice,
|
||||
earliestEventTimestamp: c.EarliestEventTimestamp, latestEventTimestamp: c.LatestEventTimestamp,
|
||||
isPreProvisioned: true, isUserDriven: false);
|
||||
|
||||
if (!whiteGloveStatusTransitioned)
|
||||
{
|
||||
_logger.LogWarning("{SessionPrefix} WhiteGlove UpdateSessionStatusAsync failed, attempting unconditional fallback for IsPreProvisioned + Status", sessionPrefix);
|
||||
try
|
||||
{
|
||||
await _sessionRepo.SetSessionPreProvisionedAsync(request.TenantId, request.SessionId, true, SessionStatus.Pending, isUserDriven: false);
|
||||
whiteGloveStatusTransitioned = true;
|
||||
_logger.LogInformation("{SessionPrefix} WhiteGlove fallback succeeded: IsPreProvisioned + Status=Pending set via unconditional merge", sessionPrefix);
|
||||
}
|
||||
catch (Exception fallbackEx)
|
||||
{
|
||||
_logger.LogError(fallbackEx, "{SessionPrefix} WhiteGlove fallback SetSessionPreProvisionedAsync also failed", sessionPrefix);
|
||||
}
|
||||
}
|
||||
|
||||
_logger.LogInformation("{SessionPrefix} Status: Pending (WhiteGlove pre-provisioning complete, transitioned={Transitioned})", sessionPrefix, whiteGloveStatusTransitioned);
|
||||
}
|
||||
else if (c.WhiteGloveResumedEvent != null)
|
||||
{
|
||||
var currentSession = await _sessionRepo.GetSessionAsync(request.TenantId, request.SessionId);
|
||||
if (currentSession?.Status == SessionStatus.Pending)
|
||||
{
|
||||
await _sessionRepo.UpdateSessionStatusAsync(
|
||||
request.TenantId, request.SessionId, SessionStatus.InProgress, c.WhiteGloveResumedEvent.Phase,
|
||||
earliestEventTimestamp: c.EarliestEventTimestamp, latestEventTimestamp: c.LatestEventTimestamp,
|
||||
isUserDriven: true, resumedAt: c.WhiteGloveResumedEvent.Timestamp);
|
||||
_logger.LogInformation("{SessionPrefix} Status: InProgress (WhiteGlove Part 2 resumed, IsUserDriven=true)", sessionPrefix);
|
||||
}
|
||||
else
|
||||
{
|
||||
_logger.LogInformation("{SessionPrefix} WhiteGlove resumed skipped, session already {Status}", sessionPrefix, currentSession?.Status);
|
||||
}
|
||||
}
|
||||
|
||||
if (c.WhiteGloveStartedEvent != null)
|
||||
{
|
||||
_logger.LogInformation("{SessionPrefix} whiteglove_started detected (soft signal — not setting IsPreProvisioned, awaiting whiteglove_complete)", sessionPrefix);
|
||||
}
|
||||
|
||||
if (c.SessionStalledEvent != null)
|
||||
{
|
||||
var stalledReason = "Agent reported stall after 60min without progress (stall_probe)";
|
||||
var stalledTransitioned = await _sessionRepo.UpdateSessionStatusAsync(
|
||||
request.TenantId, request.SessionId, SessionStatus.Stalled,
|
||||
earliestEventTimestamp: c.EarliestEventTimestamp, latestEventTimestamp: c.LatestEventTimestamp,
|
||||
stalledAt: c.SessionStalledEvent.Timestamp, failureReason: stalledReason);
|
||||
if (stalledTransitioned)
|
||||
_logger.LogWarning("{SessionPrefix} Status: Stalled (agent-reported via session_stalled event)", sessionPrefix);
|
||||
}
|
||||
else if (c.HasNonPeriodicRealEvent && !statusTransitioned && !whiteGloveStatusTransitioned)
|
||||
{
|
||||
var currentSession = await _sessionRepo.GetSessionAsync(request.TenantId, request.SessionId);
|
||||
if (currentSession?.Status == SessionStatus.Stalled)
|
||||
{
|
||||
var healed = await _sessionRepo.UpdateSessionStatusAsync(
|
||||
request.TenantId, request.SessionId, SessionStatus.InProgress,
|
||||
earliestEventTimestamp: c.EarliestEventTimestamp, latestEventTimestamp: c.LatestEventTimestamp,
|
||||
clearStalledAt: true, clearFailureReason: true);
|
||||
if (healed)
|
||||
_logger.LogInformation("{SessionPrefix} Status: InProgress (healed from Stalled by new real event)", sessionPrefix);
|
||||
}
|
||||
}
|
||||
|
||||
return (statusTransitioned, whiteGloveStatusTransitioned, failureReason);
|
||||
}
|
||||
|
||||
private async Task SendWebhookNotificationsAsync(
|
||||
IngestEventsRequest request, string sessionPrefix, EventClassification c,
|
||||
SessionSummary? updatedSession, bool statusTransitioned, bool whiteGloveStatusTransitioned, string? failureReason,
|
||||
List<RuleResult> ruleResults)
|
||||
{
|
||||
var tenantConfig = await _configService.GetConfigurationAsync(request.TenantId);
|
||||
var (webhookUrl, providerTypeInt) = tenantConfig.GetEffectiveWebhookConfig();
|
||||
|
||||
if (string.IsNullOrEmpty(webhookUrl) || providerTypeInt == 0)
|
||||
return;
|
||||
|
||||
var providerType = (WebhookProviderType)providerTypeInt;
|
||||
var sessionUrl = updatedSession != null
|
||||
? $"https://www.autopilotmonitor.com/session/{request.TenantId}/{request.SessionId}"
|
||||
: null;
|
||||
|
||||
if (statusTransitioned && (c.CompletionEvent != null || c.FailureEvent != null))
|
||||
{
|
||||
var notifySuccess = c.CompletionEvent != null && tenantConfig.GetEffectiveNotifyOnSuccess();
|
||||
var notifyFailure = c.FailureEvent != null && tenantConfig.GetEffectiveNotifyOnFailure();
|
||||
if (notifySuccess || notifyFailure)
|
||||
{
|
||||
var duration = updatedSession?.DurationSeconds != null
|
||||
? TimeSpan.FromSeconds(updatedSession.DurationSeconds.Value)
|
||||
: (TimeSpan?)null;
|
||||
|
||||
if (updatedSession?.IsPreProvisioned == true && updatedSession?.ResumedAt != null)
|
||||
{
|
||||
var completionTime = c.CompletionEvent?.Timestamp ?? c.FailureEvent?.Timestamp;
|
||||
if (completionTime.HasValue)
|
||||
duration = completionTime.Value - updatedSession.ResumedAt.Value;
|
||||
}
|
||||
|
||||
var alert = NotificationAlertBuilder.BuildEnrollmentAlert(
|
||||
updatedSession?.DeviceName,
|
||||
updatedSession?.SerialNumber,
|
||||
updatedSession?.Manufacturer,
|
||||
updatedSession?.Model,
|
||||
success: c.CompletionEvent != null,
|
||||
failureReason: failureReason,
|
||||
duration: duration,
|
||||
sessionUrl: sessionUrl);
|
||||
NotificationAlertBuilder.AddRuleResultSections(alert, ruleResults);
|
||||
|
||||
_ = _webhookNotificationService.SendNotificationAsync(webhookUrl, providerType, alert)
|
||||
.ContinueWith(t => _logger.LogWarning(t.Exception?.InnerException,
|
||||
"Fire-and-forget webhook notification failed"), TaskContinuationOptions.OnlyOnFaulted);
|
||||
}
|
||||
}
|
||||
|
||||
if (whiteGloveStatusTransitioned && c.WhiteGloveEvent != null && tenantConfig.GetEffectiveNotifyOnSuccess())
|
||||
{
|
||||
var duration = updatedSession?.DurationSeconds != null
|
||||
? TimeSpan.FromSeconds(updatedSession.DurationSeconds.Value)
|
||||
: (TimeSpan?)null;
|
||||
|
||||
var alert = NotificationAlertBuilder.BuildWhiteGloveAlert(
|
||||
updatedSession?.DeviceName,
|
||||
updatedSession?.SerialNumber,
|
||||
updatedSession?.Manufacturer,
|
||||
updatedSession?.Model,
|
||||
success: true,
|
||||
duration: duration,
|
||||
sessionUrl: sessionUrl);
|
||||
NotificationAlertBuilder.AddRuleResultSections(alert, ruleResults);
|
||||
|
||||
_ = _webhookNotificationService.SendNotificationAsync(webhookUrl, providerType, alert)
|
||||
.ContinueWith(t => _logger.LogWarning(t.Exception?.InnerException,
|
||||
"Fire-and-forget webhook notification failed"), TaskContinuationOptions.OnlyOnFaulted);
|
||||
}
|
||||
|
||||
if (c.EspFailureEvent != null && updatedSession?.IsPreProvisioned == true && tenantConfig.GetEffectiveNotifyOnFailure())
|
||||
{
|
||||
var duration = updatedSession?.DurationSeconds != null
|
||||
? TimeSpan.FromSeconds(updatedSession.DurationSeconds.Value)
|
||||
: (TimeSpan?)null;
|
||||
|
||||
var alert = NotificationAlertBuilder.BuildWhiteGloveAlert(
|
||||
updatedSession?.DeviceName,
|
||||
updatedSession?.SerialNumber,
|
||||
updatedSession?.Manufacturer,
|
||||
updatedSession?.Model,
|
||||
success: false,
|
||||
duration: duration,
|
||||
sessionUrl: sessionUrl);
|
||||
NotificationAlertBuilder.AddRuleResultSections(alert, ruleResults);
|
||||
|
||||
_ = _webhookNotificationService.SendNotificationAsync(webhookUrl, providerType, alert)
|
||||
.ContinueWith(t => _logger.LogWarning(t.Exception?.InnerException,
|
||||
"Fire-and-forget webhook notification failed"), TaskContinuationOptions.OnlyOnFaulted);
|
||||
}
|
||||
}
|
||||
|
||||
private SignalRMessageAction[] BuildSignalRMessages(
|
||||
IngestEventsRequest request, SessionSummary? updatedSession, int processedCount,
|
||||
List<RuleResult> newRuleResults)
|
||||
{
|
||||
object? sessionDelta = updatedSession != null ? new {
|
||||
updatedSession.CurrentPhase,
|
||||
updatedSession.CurrentPhaseDetail,
|
||||
updatedSession.Status,
|
||||
updatedSession.FailureReason,
|
||||
updatedSession.EventCount,
|
||||
updatedSession.DurationSeconds,
|
||||
updatedSession.CompletedAt,
|
||||
updatedSession.DiagnosticsBlobName,
|
||||
updatedSession.IsPreProvisioned
|
||||
} : null;
|
||||
|
||||
var summaryMessage = new SignalRMessageAction("newevents")
|
||||
{
|
||||
GroupName = $"tenant-{request.TenantId}",
|
||||
Arguments = new object[] { new {
|
||||
sessionId = request.SessionId,
|
||||
tenantId = request.TenantId,
|
||||
eventCount = processedCount,
|
||||
sessionUpdate = sessionDelta
|
||||
} }
|
||||
};
|
||||
|
||||
var slimRuleResults = newRuleResults.Count > 0
|
||||
? newRuleResults.Select(r => new {
|
||||
r.ResultId,
|
||||
r.RuleId,
|
||||
r.RuleTitle,
|
||||
r.Severity,
|
||||
r.Category,
|
||||
r.ConfidenceScore,
|
||||
r.Explanation,
|
||||
r.Remediation,
|
||||
r.RelatedDocs,
|
||||
r.MatchedConditions,
|
||||
r.DetectedAt
|
||||
}).ToList<object>()
|
||||
: null;
|
||||
|
||||
var eventsMessage = new SignalRMessageAction("eventStream")
|
||||
{
|
||||
GroupName = $"session-{request.TenantId}-{request.SessionId}",
|
||||
Arguments = new object[] { new {
|
||||
sessionId = request.SessionId,
|
||||
tenantId = request.TenantId,
|
||||
newEventCount = processedCount,
|
||||
newRuleResults = slimRuleResults
|
||||
} }
|
||||
};
|
||||
|
||||
return new[] { summaryMessage, eventsMessage };
|
||||
}
|
||||
|
||||
private async Task RecordGatherRuleStatsAsync(string tenantId, List<EnrollmentEvent> events)
|
||||
{
|
||||
try
|
||||
{
|
||||
var today = DateTime.UtcNow.ToString("yyyy-MM-dd");
|
||||
|
||||
var gatherEvents = events.Where(e =>
|
||||
e.Data != null &&
|
||||
e.Data.ContainsKey("ruleId") &&
|
||||
e.Source == "GatherRuleExecutor").ToList();
|
||||
|
||||
if (gatherEvents.Count == 0) return;
|
||||
|
||||
var seen = new HashSet<string>(StringComparer.OrdinalIgnoreCase);
|
||||
foreach (var evt in gatherEvents)
|
||||
{
|
||||
var ruleId = evt.Data!["ruleId"]?.ToString();
|
||||
if (string.IsNullOrEmpty(ruleId) || !seen.Add(ruleId)) continue;
|
||||
|
||||
var ruleTitle = evt.Data.ContainsKey("ruleTitle") ? evt.Data["ruleTitle"]?.ToString() ?? "" : "";
|
||||
|
||||
await _metricsRepo.IncrementRuleStatAsync(
|
||||
today, tenantId, ruleId, "gather",
|
||||
ruleTitle, "", "",
|
||||
fired: true, confidenceScore: null);
|
||||
|
||||
await _metricsRepo.IncrementRuleStatAsync(
|
||||
today, "global", ruleId, "gather",
|
||||
ruleTitle, "", "",
|
||||
fired: true, confidenceScore: null);
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Failed to record gather rule stats (non-fatal)");
|
||||
}
|
||||
}
|
||||
|
||||
private async Task RecordAnalyzeRuleStatsAsync(string tenantId, AnalysisOutcome outcome)
|
||||
{
|
||||
try
|
||||
{
|
||||
var today = DateTime.UtcNow.ToString("yyyy-MM-dd");
|
||||
var firedRuleIds = new HashSet<string>(outcome.Results.Select(r => r.RuleId));
|
||||
|
||||
foreach (var rule in outcome.EvaluatedRules)
|
||||
{
|
||||
var fired = firedRuleIds.Contains(rule.RuleId);
|
||||
int? confidence = null;
|
||||
if (fired)
|
||||
{
|
||||
var result = outcome.Results.FirstOrDefault(r => r.RuleId == rule.RuleId);
|
||||
confidence = result?.ConfidenceScore;
|
||||
}
|
||||
|
||||
await _metricsRepo.IncrementRuleStatAsync(
|
||||
today, tenantId, rule.RuleId, "analyze",
|
||||
rule.Title, rule.Category, rule.Severity,
|
||||
fired, confidence);
|
||||
|
||||
await _metricsRepo.IncrementRuleStatAsync(
|
||||
today, "global", rule.RuleId, "analyze",
|
||||
rule.Title, rule.Category, rule.Severity,
|
||||
fired, confidence);
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Failed to record analyze rule stats (non-fatal)");
|
||||
}
|
||||
}
|
||||
|
||||
private void AggregateAppInstallEvent(EnrollmentEvent evt, string tenantId, string sessionId, Dictionary<string, AppInstallAggregationState> summaries)
|
||||
{
|
||||
bool isRelevant =
|
||||
evt.EventType == "app_install_started" || evt.EventType == "app_install_start" ||
|
||||
evt.EventType == "app_install_completed" || evt.EventType == "app_install_complete" ||
|
||||
evt.EventType == "app_install_failed" ||
|
||||
evt.EventType == "app_download_started" ||
|
||||
evt.EventType == "app_install_skipped" ||
|
||||
evt.EventType == "download_progress" ||
|
||||
evt.EventType == "do_telemetry";
|
||||
|
||||
if (!isRelevant) return;
|
||||
|
||||
var appName = evt.Data?.ContainsKey("appName") == true ? evt.Data["appName"]?.ToString()?.Trim() : null;
|
||||
if (string.IsNullOrEmpty(appName)) return;
|
||||
|
||||
if (!summaries.TryGetValue(appName, out var state))
|
||||
{
|
||||
state = new AppInstallAggregationState
|
||||
{
|
||||
Summary = new AppInstallSummary
|
||||
{
|
||||
AppName = appName,
|
||||
SessionId = sessionId,
|
||||
TenantId = tenantId,
|
||||
StartedAt = evt.Timestamp
|
||||
}
|
||||
};
|
||||
summaries[appName] = state;
|
||||
}
|
||||
|
||||
var summary = state.Summary;
|
||||
|
||||
if (evt.Data != null)
|
||||
{
|
||||
if (evt.Data.TryGetValue("appVersion", out var appVersionObj))
|
||||
{
|
||||
var appVersion = appVersionObj?.ToString();
|
||||
if (!string.IsNullOrWhiteSpace(appVersion))
|
||||
summary.AppVersion = appVersion.Trim();
|
||||
}
|
||||
if (evt.Data.TryGetValue("appType", out var appTypeObj))
|
||||
{
|
||||
var appType = appTypeObj?.ToString();
|
||||
if (!string.IsNullOrWhiteSpace(appType))
|
||||
summary.AppType = appType.Trim();
|
||||
}
|
||||
if (evt.Data.TryGetValue("attemptNumber", out var attemptObj) &&
|
||||
int.TryParse(attemptObj?.ToString(), out var attempt) && attempt > 0)
|
||||
{
|
||||
summary.AttemptNumber = Math.Max(summary.AttemptNumber, attempt);
|
||||
}
|
||||
if (evt.Data.TryGetValue("installerPhase", out var phaseObj))
|
||||
{
|
||||
var phase = phaseObj?.ToString();
|
||||
if (!string.IsNullOrWhiteSpace(phase))
|
||||
summary.InstallerPhase = phase.Trim();
|
||||
}
|
||||
if (evt.Data.TryGetValue("exitCode", out var exitCodeObj) &&
|
||||
int.TryParse(exitCodeObj?.ToString(), out var exitCode))
|
||||
{
|
||||
summary.ExitCode = exitCode;
|
||||
}
|
||||
if (evt.Data.TryGetValue("detectionResult", out var detectionObj))
|
||||
{
|
||||
var detection = detectionObj?.ToString();
|
||||
if (!string.IsNullOrWhiteSpace(detection))
|
||||
summary.DetectionResult = detection.Trim();
|
||||
}
|
||||
}
|
||||
|
||||
switch (evt.EventType)
|
||||
{
|
||||
case "app_install_started":
|
||||
case "app_install_start":
|
||||
if (!state.InstallStartedAt.HasValue || evt.Timestamp < state.InstallStartedAt.Value)
|
||||
state.InstallStartedAt = evt.Timestamp;
|
||||
if (summary.Status == "InProgress" || summary.Status == string.Empty)
|
||||
summary.Status = "InProgress";
|
||||
break;
|
||||
|
||||
case "app_download_started":
|
||||
if (!state.DownloadStartedAt.HasValue || evt.Timestamp < state.DownloadStartedAt.Value)
|
||||
state.DownloadStartedAt = evt.Timestamp;
|
||||
if (summary.Status == "InProgress" || summary.Status == string.Empty)
|
||||
summary.Status = "InProgress";
|
||||
break;
|
||||
|
||||
case "app_install_completed":
|
||||
case "app_install_complete":
|
||||
summary.Status = "Succeeded";
|
||||
summary.CompletedAt = evt.Timestamp;
|
||||
if (summary.StartedAt != DateTime.MinValue)
|
||||
summary.DurationSeconds = Math.Max(1, EventTimestampValidator.SafeDurationSeconds(summary.StartedAt, evt.Timestamp));
|
||||
break;
|
||||
|
||||
case "app_install_failed":
|
||||
summary.Status = "Failed";
|
||||
summary.CompletedAt = evt.Timestamp;
|
||||
if (summary.StartedAt != DateTime.MinValue)
|
||||
summary.DurationSeconds = Math.Max(1, EventTimestampValidator.SafeDurationSeconds(summary.StartedAt, evt.Timestamp));
|
||||
summary.FailureCode = evt.Data?.ContainsKey("errorCode") == true
|
||||
? evt.Data["errorCode"]?.ToString() ?? string.Empty : string.Empty;
|
||||
summary.FailureMessage = evt.Data?.ContainsKey("errorMessage") == true
|
||||
? evt.Data["errorMessage"]?.ToString() ?? string.Empty : evt.Message ?? string.Empty;
|
||||
break;
|
||||
|
||||
case "app_install_skipped":
|
||||
if (summary.Status == "InProgress")
|
||||
summary.Status = "Succeeded";
|
||||
break;
|
||||
|
||||
case "download_progress":
|
||||
var bytesKey = evt.Data?.ContainsKey("bytesDownloaded") == true ? "bytesDownloaded"
|
||||
: evt.Data?.ContainsKey("bytes_downloaded") == true ? "bytes_downloaded" : null;
|
||||
if (bytesKey != null && long.TryParse(evt.Data![bytesKey]?.ToString(), out var bytes))
|
||||
summary.DownloadBytes = Math.Max(summary.DownloadBytes, bytes);
|
||||
break;
|
||||
|
||||
case "do_telemetry":
|
||||
if (evt.Data != null)
|
||||
{
|
||||
if (evt.Data.ContainsKey("doFileSize") && long.TryParse(evt.Data["doFileSize"]?.ToString(), out var doFs))
|
||||
{
|
||||
summary.DownloadBytes = Math.Max(summary.DownloadBytes, doFs);
|
||||
summary.DoFileSize = doFs;
|
||||
}
|
||||
if (evt.Data.ContainsKey("doTotalBytesDownloaded") && long.TryParse(evt.Data["doTotalBytesDownloaded"]?.ToString(), out var doTotalDl))
|
||||
summary.DoTotalBytesDownloaded = doTotalDl;
|
||||
if (evt.Data.ContainsKey("doBytesFromPeers") && long.TryParse(evt.Data["doBytesFromPeers"]?.ToString(), out var doPeers))
|
||||
summary.DoBytesFromPeers = doPeers;
|
||||
if (evt.Data.ContainsKey("doBytesFromHttp") && long.TryParse(evt.Data["doBytesFromHttp"]?.ToString(), out var doHttp))
|
||||
summary.DoBytesFromHttp = doHttp;
|
||||
if (evt.Data.ContainsKey("doPercentPeerCaching") && int.TryParse(evt.Data["doPercentPeerCaching"]?.ToString(), out var doPct))
|
||||
summary.DoPercentPeerCaching = doPct;
|
||||
if (evt.Data.ContainsKey("doDownloadMode") && int.TryParse(evt.Data["doDownloadMode"]?.ToString(), out var doMode))
|
||||
summary.DoDownloadMode = doMode;
|
||||
if (evt.Data.ContainsKey("doDownloadDuration"))
|
||||
{
|
||||
var doDurStr = evt.Data["doDownloadDuration"]?.ToString() ?? string.Empty;
|
||||
summary.DoDownloadDuration = doDurStr;
|
||||
if (TimeSpan.TryParse(doDurStr, out var doDurTs) && doDurTs.TotalSeconds >= 1)
|
||||
summary.DownloadDurationSeconds = Math.Max(summary.DownloadDurationSeconds, (int)doDurTs.TotalSeconds);
|
||||
}
|
||||
if (evt.Data.ContainsKey("doBytesFromLanPeers") && long.TryParse(evt.Data["doBytesFromLanPeers"]?.ToString(), out var doLan))
|
||||
summary.DoBytesFromLanPeers = doLan;
|
||||
if (evt.Data.ContainsKey("doBytesFromGroupPeers") && long.TryParse(evt.Data["doBytesFromGroupPeers"]?.ToString(), out var doGroup))
|
||||
summary.DoBytesFromGroupPeers = doGroup;
|
||||
if (evt.Data.ContainsKey("doBytesFromInternetPeers") && long.TryParse(evt.Data["doBytesFromInternetPeers"]?.ToString(), out var doInet))
|
||||
summary.DoBytesFromInternetPeers = doInet;
|
||||
}
|
||||
break;
|
||||
}
|
||||
|
||||
IngestEventsFunction.RecalculateAppDurations(state);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
||||
Reference in New Issue
Block a user