mirror of
https://github.com/okieselbach/AutopilotMonitor.git
synced 2026-10-24 12:42:38 +00:00
feat(backend): M5.b IngestTelemetry endpoint — POST /api/agent/telemetry
V2 ingest endpoint consuming a heterogeneous batch of TelemetryItemDtos (Event + Signal + DecisionTransition in one JSON array). gzip request bodies are decoded transparently by the existing UseRequestDecompression middleware — new code only has to read a plain stream. - TelemetryItemDto (Shared): wire-format DTO mirroring the agent's TelemetryItem JSON shape so the backend can deserialise without linking agent assemblies. - IngestTelemetryFunction: security check (cert / rate / hardware whitelist) → device & version kill-switch short-circuits with DeviceBlocked=true body → body parse → tenant-header/body match → partition by Kind → persist via the appropriate repository → 200 with AdminAction + Actions control-signal fields populated from ISessionRepository (matching the /api/agent/ingest parity needed by the M4.6.ε agent-side UploadResult parser). - TelemetryPayloadParser: pure defensive parsing layer (wire → storage record). Malformed PayloadJson on a single item returns null; one poisonous item can't poison the batch. Projects index discriminators (IsTerminal, ClassifierVerdictId, ClassifierHypothesisLevel) off the DecisionTransition payload so the later queue-driven index fan-out reads them straight off the primary row. - EndpointAccessPolicyCatalog: POST agent/telemetry registered under DeviceOrBootstrapAuth (same policy as agent/ingest). Without this the fail-closed catalog rejects the route with 403. Scope boundary — Events get stored via StoreEventsBatchAsync only; the rule engine, app-install aggregation, SignalR notification, and vulnerability correlation remain on the legacy /api/agent/ingest path. Sharing that pipeline is tracked as M5.b.2 (extract EventIngestProcessor so both functions run the same downstream steps) — M7 rollout gates on completing it. Documented inline in PersistItemsAsync. 28 new tests: defensive null-handling per kind, typed-column extraction for Signal (with envelope fallback for missing SessionTraceOrdinal), DecisionTransition dead-end + classifier-verdict nesting, IsTerminal authoritative-list guard (Completed / Failed / WhiteGloveSealed / WhiteGloveCompletedPart2 — mirrors DecisionCore SessionStageExtensions), PartitionKey parser including malformed-shape rejection. The EndpointPolicyCatalogCompletenessTests guard the route registration. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Oliver Kieselbach
@@ -0,0 +1,42 @@
|
||||
using AutopilotMonitor.Functions.Functions.Ingest;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Tests for the pure helpers on <see cref="IngestTelemetryFunction"/>. The HTTP-trigger end
|
||||
/// of the function needs a live runtime harness to exercise; this file covers the deterministic
|
||||
/// bits (PartitionKey parsing, body/header TenantId reconciliation shape).
|
||||
/// </summary>
|
||||
public class IngestTelemetryFunctionTests
|
||||
{
|
||||
private const string TenantId = "a1b2c3d4-e5f6-7890-abcd-ef1234567890";
|
||||
private const string SessionId = "b2c3d4e5-f6a7-8901-bcde-f12345678901";
|
||||
|
||||
[Fact]
|
||||
public void TryParsePartitionKey_splits_tenant_and_session_on_single_underscore()
|
||||
{
|
||||
var ok = IngestTelemetryFunction.TryParsePartitionKey(
|
||||
$"{TenantId}_{SessionId}", out var tenant, out var session);
|
||||
|
||||
Assert.True(ok);
|
||||
Assert.Equal(TenantId, tenant);
|
||||
Assert.Equal(SessionId, session);
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData("")]
|
||||
[InlineData(null)]
|
||||
[InlineData("no-underscore-here")]
|
||||
[InlineData("_missing-tenant")]
|
||||
[InlineData("missing-session_")]
|
||||
[InlineData("too_many_parts_here")]
|
||||
public void TryParsePartitionKey_rejects_malformed_shapes(string? input)
|
||||
{
|
||||
var ok = IngestTelemetryFunction.TryParsePartitionKey(
|
||||
input!, out var tenant, out var session);
|
||||
|
||||
Assert.False(ok);
|
||||
Assert.Equal(string.Empty, tenant);
|
||||
Assert.Equal(string.Empty, session);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,207 @@
|
||||
using AutopilotMonitor.Functions.Functions.Ingest;
|
||||
using AutopilotMonitor.Shared.Models;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Pure parsing tests for <see cref="TelemetryPayloadParser"/>. Verifies the wire → storage
|
||||
/// mapping the <see cref="IngestTelemetryFunction"/> relies on: defensive null handling, typed
|
||||
/// column extraction, and the index-discriminator projection (IsTerminal, ClassifierVerdictId,
|
||||
/// ClassifierHypothesisLevel).
|
||||
/// </summary>
|
||||
public class TelemetryPayloadParserTests
|
||||
{
|
||||
private const string TenantId = "a1b2c3d4-e5f6-7890-abcd-ef1234567890";
|
||||
private const string SessionId = "b2c3d4e5-f6a7-8901-bcde-f12345678901";
|
||||
|
||||
private static TelemetryItemDto Dto(string kind, string payloadJson, long? traceOrdinal = null)
|
||||
=> new TelemetryItemDto
|
||||
{
|
||||
Kind = kind,
|
||||
PartitionKey = $"{TenantId}_{SessionId}",
|
||||
RowKey = "0000000001",
|
||||
TelemetryItemId = 1,
|
||||
SessionTraceOrdinal = traceOrdinal,
|
||||
PayloadJson = payloadJson,
|
||||
RequiresImmediateFlush = false,
|
||||
EnqueuedAtUtc = new DateTime(2026, 4, 21, 10, 0, 0, DateTimeKind.Utc),
|
||||
RetryCount = 0,
|
||||
};
|
||||
|
||||
// ============================================================ Defensive parsing
|
||||
|
||||
[Theory]
|
||||
[InlineData("")]
|
||||
[InlineData("{ this is not json }")]
|
||||
[InlineData("null")]
|
||||
public void ParseEvent_returns_null_for_missing_or_malformed_payload(string payload)
|
||||
{
|
||||
var result = TelemetryPayloadParser.ParseEvent(Dto("Event", payload), TenantId, SessionId);
|
||||
Assert.Null(result);
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData("")]
|
||||
[InlineData("{ this is not json }")]
|
||||
public void ParseSignal_returns_null_for_missing_or_malformed_payload(string payload)
|
||||
{
|
||||
var result = TelemetryPayloadParser.ParseSignal(Dto("Signal", payload), TenantId, SessionId);
|
||||
Assert.Null(result);
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData("")]
|
||||
[InlineData("{ this is not json }")]
|
||||
public void ParseTransition_returns_null_for_missing_or_malformed_payload(string payload)
|
||||
{
|
||||
var result = TelemetryPayloadParser.ParseTransition(Dto("DecisionTransition", payload), TenantId, SessionId);
|
||||
Assert.Null(result);
|
||||
}
|
||||
|
||||
// ============================================================ Event
|
||||
|
||||
[Fact]
|
||||
public void ParseEvent_deserialises_enrollment_event_shape()
|
||||
{
|
||||
var payload =
|
||||
"{" +
|
||||
"\"EventId\":\"evt-123\"," +
|
||||
"\"EventType\":\"test_event\"," +
|
||||
"\"Severity\":1," +
|
||||
"\"Source\":\"test\"," +
|
||||
"\"Phase\":2," +
|
||||
"\"Message\":\"hello\"," +
|
||||
"\"Sequence\":42," +
|
||||
"\"Timestamp\":\"2026-04-21T10:00:00Z\"," +
|
||||
"\"ReceivedAt\":\"2026-04-21T10:00:01Z\"" +
|
||||
"}";
|
||||
|
||||
var evt = TelemetryPayloadParser.ParseEvent(Dto("Event", payload), TenantId, SessionId);
|
||||
|
||||
Assert.NotNull(evt);
|
||||
Assert.Equal("evt-123", evt!.EventId);
|
||||
Assert.Equal("test_event", evt.EventType);
|
||||
Assert.Equal(42, evt.Sequence);
|
||||
}
|
||||
|
||||
// ============================================================ Signal
|
||||
|
||||
[Fact]
|
||||
public void ParseSignal_extracts_typed_columns_and_preserves_full_payload()
|
||||
{
|
||||
var payload =
|
||||
"{" +
|
||||
"\"SessionSignalOrdinal\":17," +
|
||||
"\"SessionTraceOrdinal\":117," +
|
||||
"\"Kind\":\"EspPhaseChanged\"," +
|
||||
"\"KindSchemaVersion\":2," +
|
||||
"\"OccurredAtUtc\":\"2026-04-21T10:00:00Z\"," +
|
||||
"\"SourceOrigin\":\"EspAndHelloTrackerAdapter\"," +
|
||||
"\"Evidence\":{\"Kind\":\"Raw\"}" +
|
||||
"}";
|
||||
|
||||
var rec = TelemetryPayloadParser.ParseSignal(Dto("Signal", payload), TenantId, SessionId);
|
||||
|
||||
Assert.NotNull(rec);
|
||||
Assert.Equal(TenantId, rec!.TenantId);
|
||||
Assert.Equal(SessionId, rec.SessionId);
|
||||
Assert.Equal(17L, rec.SessionSignalOrdinal);
|
||||
Assert.Equal(117L, rec.SessionTraceOrdinal);
|
||||
Assert.Equal("EspPhaseChanged", rec.Kind);
|
||||
Assert.Equal(2, rec.KindSchemaVersion);
|
||||
Assert.Equal(new DateTime(2026, 4, 21, 10, 0, 0, DateTimeKind.Utc), rec.OccurredAtUtc.ToUniversalTime());
|
||||
Assert.Equal("EspAndHelloTrackerAdapter", rec.SourceOrigin);
|
||||
Assert.Equal(payload, rec.PayloadJson);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ParseSignal_falls_back_to_envelope_SessionTraceOrdinal_when_payload_missing_field()
|
||||
{
|
||||
// Payload intentionally omits SessionTraceOrdinal; the envelope carries it.
|
||||
var payload = "{\"SessionSignalOrdinal\":1,\"Kind\":\"SessionStarted\"}";
|
||||
|
||||
var rec = TelemetryPayloadParser.ParseSignal(Dto("Signal", payload, traceOrdinal: 99), TenantId, SessionId);
|
||||
|
||||
Assert.NotNull(rec);
|
||||
Assert.Equal(99L, rec!.SessionTraceOrdinal);
|
||||
}
|
||||
|
||||
// ============================================================ DecisionTransition
|
||||
|
||||
[Fact]
|
||||
public void ParseTransition_extracts_core_columns_and_flags_taken_false_when_dead_end()
|
||||
{
|
||||
var payload =
|
||||
"{" +
|
||||
"\"StepIndex\":3," +
|
||||
"\"SessionTraceOrdinal\":10," +
|
||||
"\"SignalOrdinalRef\":9," +
|
||||
"\"OccurredAtUtc\":\"2026-04-21T10:00:00Z\"," +
|
||||
"\"Trigger\":\"EspExiting\"," +
|
||||
"\"FromStage\":\"EspInProgress\"," +
|
||||
"\"ToStage\":\"EspInProgress\"," +
|
||||
"\"Taken\":false," +
|
||||
"\"DeadEndReason\":\"hybrid_reboot_gate_blocking\"," +
|
||||
"\"ReducerVersion\":\"1.0.0\"" +
|
||||
"}";
|
||||
|
||||
var rec = TelemetryPayloadParser.ParseTransition(Dto("DecisionTransition", payload), TenantId, SessionId);
|
||||
|
||||
Assert.NotNull(rec);
|
||||
Assert.False(rec!.Taken);
|
||||
Assert.Equal("hybrid_reboot_gate_blocking", rec.DeadEndReason);
|
||||
Assert.False(rec.IsTerminal);
|
||||
Assert.Null(rec.ClassifierVerdictId);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ParseTransition_projects_IsTerminal_true_for_terminal_stages()
|
||||
{
|
||||
var terminalStages = new[] { "Completed", "Failed", "WhiteGloveSealed", "WhiteGloveCompletedPart2" };
|
||||
|
||||
foreach (var stage in terminalStages)
|
||||
{
|
||||
var payload =
|
||||
"{\"StepIndex\":1,\"SessionTraceOrdinal\":1,\"SignalOrdinalRef\":1," +
|
||||
$"\"Trigger\":\"t\",\"FromStage\":\"X\",\"ToStage\":\"{stage}\",\"Taken\":true," +
|
||||
"\"ReducerVersion\":\"1.0.0\",\"OccurredAtUtc\":\"2026-04-21T10:00:00Z\"}";
|
||||
|
||||
var rec = TelemetryPayloadParser.ParseTransition(Dto("DecisionTransition", payload), TenantId, SessionId);
|
||||
|
||||
Assert.NotNull(rec);
|
||||
Assert.True(rec!.IsTerminal, $"Expected {stage} to be terminal");
|
||||
}
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ParseTransition_extracts_classifier_verdict_nested_fields()
|
||||
{
|
||||
var payload =
|
||||
"{\"StepIndex\":1,\"SessionTraceOrdinal\":1,\"SignalOrdinalRef\":1," +
|
||||
"\"Trigger\":\"t\",\"FromStage\":\"X\",\"ToStage\":\"WhiteGloveSealed\",\"Taken\":true," +
|
||||
"\"ReducerVersion\":\"1.0.0\",\"OccurredAtUtc\":\"2026-04-21T10:00:00Z\"," +
|
||||
"\"ClassifierVerdict\":{\"ClassifierId\":\"whiteglove-sealing\",\"HypothesisLevel\":\"Strong\"}}";
|
||||
|
||||
var rec = TelemetryPayloadParser.ParseTransition(Dto("DecisionTransition", payload), TenantId, SessionId);
|
||||
|
||||
Assert.NotNull(rec);
|
||||
Assert.Equal("whiteglove-sealing", rec!.ClassifierVerdictId);
|
||||
Assert.Equal("Strong", rec.ClassifierHypothesisLevel);
|
||||
}
|
||||
|
||||
// ============================================================ IsTerminalStage
|
||||
|
||||
[Theory]
|
||||
[InlineData("Completed", true)]
|
||||
[InlineData("Failed", true)]
|
||||
[InlineData("WhiteGloveSealed", true)]
|
||||
[InlineData("WhiteGloveCompletedPart2", true)]
|
||||
[InlineData("EspInProgress", false)]
|
||||
[InlineData("AccountSetup", false)]
|
||||
[InlineData("", false)]
|
||||
[InlineData("completed", false)] // case-sensitive — enum names are PascalCase
|
||||
public void IsTerminalStage_matches_DecisionCore_authoritative_list(string stage, bool expected)
|
||||
{
|
||||
Assert.Equal(expected, TelemetryPayloadParser.IsTerminalStage(stage));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,345 @@
|
||||
using System.IO;
|
||||
using System.Net;
|
||||
using AutopilotMonitor.Functions.Security;
|
||||
using AutopilotMonitor.Functions.Services;
|
||||
using AutopilotMonitor.Shared.DataAccess;
|
||||
using AutopilotMonitor.Shared.Models;
|
||||
using Microsoft.Azure.Functions.Worker;
|
||||
using Microsoft.Azure.Functions.Worker.Http;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Newtonsoft.Json;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Functions.Ingest
|
||||
{
|
||||
/// <summary>
|
||||
/// V2 ingest endpoint. Consumes a heterogeneous batch of <see cref="TelemetryItemDto"/>s
|
||||
/// (Events + Signals + DecisionTransitions in a single JSON array, gzip-compressed on
|
||||
/// the wire — the <c>UseRequestDecompression</c> middleware decompresses before we
|
||||
/// parse). Plan §2.7a / §M5 / M4.6.ε.
|
||||
/// <para>
|
||||
/// <b>Routing by <see cref="TelemetryItemDto.Kind"/>:</b>
|
||||
/// <list type="bullet">
|
||||
/// <item><c>Event</c> → <see cref="ISessionRepository.StoreEventsBatchAsync"/> (storage only in this build — full event pipeline sharing with /api/agent/ingest is tracked as M5.b.2).</item>
|
||||
/// <item><c>Signal</c> → <see cref="ISignalRepository.StoreBatchAsync"/></item>
|
||||
/// <item><c>DecisionTransition</c> → <see cref="IDecisionTransitionRepository.StoreBatchAsync"/></item>
|
||||
/// </list>
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Response (M4.6.ε):</b> the agent parses <c>DeviceBlocked</c>/<c>UnblockAt</c>/
|
||||
/// <c>DeviceKillSignal</c>/<c>AdminAction</c>/<c>Actions</c> from the 2xx body and routes
|
||||
/// kill-switches through its <c>ServerActionDispatcher</c>. Populated from the same
|
||||
/// services the legacy /api/agent/ingest endpoint uses so behaviour is at parity.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public sealed class IngestTelemetryFunction
|
||||
{
|
||||
private readonly ILogger<IngestTelemetryFunction> _logger;
|
||||
private readonly ISessionRepository _sessionRepo;
|
||||
private readonly ISignalRepository _signalRepo;
|
||||
private readonly IDecisionTransitionRepository _transitionRepo;
|
||||
private readonly TenantConfigurationService _configService;
|
||||
private readonly RateLimitService _rateLimitService;
|
||||
private readonly AutopilotDeviceValidator _autopilotDeviceValidator;
|
||||
private readonly CorporateIdentifierValidator _corporateIdentifierValidator;
|
||||
private readonly DeviceAssociationValidator _deviceAssociationValidator;
|
||||
private readonly BootstrapSessionService _bootstrapSessionService;
|
||||
private readonly BlockedDeviceService _blockedDeviceService;
|
||||
private readonly BlockedVersionService _blockedVersionService;
|
||||
|
||||
public IngestTelemetryFunction(
|
||||
ILogger<IngestTelemetryFunction> logger,
|
||||
ISessionRepository sessionRepo,
|
||||
ISignalRepository signalRepo,
|
||||
IDecisionTransitionRepository transitionRepo,
|
||||
TenantConfigurationService configService,
|
||||
RateLimitService rateLimitService,
|
||||
AutopilotDeviceValidator autopilotDeviceValidator,
|
||||
CorporateIdentifierValidator corporateIdentifierValidator,
|
||||
DeviceAssociationValidator deviceAssociationValidator,
|
||||
BootstrapSessionService bootstrapSessionService,
|
||||
BlockedDeviceService blockedDeviceService,
|
||||
BlockedVersionService blockedVersionService)
|
||||
{
|
||||
_logger = logger;
|
||||
_sessionRepo = sessionRepo;
|
||||
_signalRepo = signalRepo;
|
||||
_transitionRepo = transitionRepo;
|
||||
_configService = configService;
|
||||
_rateLimitService = rateLimitService;
|
||||
_autopilotDeviceValidator = autopilotDeviceValidator;
|
||||
_corporateIdentifierValidator = corporateIdentifierValidator;
|
||||
_deviceAssociationValidator = deviceAssociationValidator;
|
||||
_bootstrapSessionService = bootstrapSessionService;
|
||||
_blockedDeviceService = blockedDeviceService;
|
||||
_blockedVersionService = blockedVersionService;
|
||||
}
|
||||
|
||||
[Function("IngestTelemetry")]
|
||||
public async Task<HttpResponseData> Run(
|
||||
[HttpTrigger(AuthorizationLevel.Anonymous, "post", Route = "agent/telemetry")] HttpRequestData req)
|
||||
{
|
||||
try
|
||||
{
|
||||
var tenantIdHeader = req.Headers.Contains("X-Tenant-Id")
|
||||
? req.Headers.GetValues("X-Tenant-Id").FirstOrDefault()
|
||||
: null;
|
||||
|
||||
if (string.IsNullOrEmpty(tenantIdHeader))
|
||||
{
|
||||
return await WriteErrorAsync(req, HttpStatusCode.BadRequest, "X-Tenant-Id header is required");
|
||||
}
|
||||
|
||||
var (validation, errorResponse) = await req.ValidateSecurityAsync(
|
||||
tenantIdHeader,
|
||||
_configService,
|
||||
_rateLimitService,
|
||||
_autopilotDeviceValidator,
|
||||
_corporateIdentifierValidator,
|
||||
_logger,
|
||||
bootstrapSessionService: _bootstrapSessionService,
|
||||
deviceAssociationValidator: _deviceAssociationValidator);
|
||||
|
||||
if (errorResponse != null) return errorResponse;
|
||||
|
||||
// Device + version kill-switches: short-circuit before parsing the body.
|
||||
var serialNumberHeader = req.Headers.Contains("X-Device-SerialNumber")
|
||||
? req.Headers.GetValues("X-Device-SerialNumber").FirstOrDefault()
|
||||
: null;
|
||||
var agentVersionHeader = req.Headers.Contains("X-Agent-Version")
|
||||
? req.Headers.GetValues("X-Agent-Version").FirstOrDefault()
|
||||
: null;
|
||||
|
||||
var killResponse = await CheckKillSwitchesAsync(
|
||||
req, tenantIdHeader, serialNumberHeader, agentVersionHeader);
|
||||
if (killResponse != null) return killResponse;
|
||||
|
||||
// Parse request body (already gzip-decompressed by middleware if Content-Encoding: gzip).
|
||||
List<TelemetryItemDto>? items;
|
||||
try
|
||||
{
|
||||
using var reader = new StreamReader(req.Body);
|
||||
var body = await reader.ReadToEndAsync();
|
||||
items = JsonConvert.DeserializeObject<List<TelemetryItemDto>>(body);
|
||||
}
|
||||
catch (JsonException ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "IngestTelemetry: malformed JSON body");
|
||||
return await WriteErrorAsync(req, HttpStatusCode.BadRequest, "Malformed JSON body");
|
||||
}
|
||||
|
||||
if (items == null || items.Count == 0)
|
||||
{
|
||||
return await WriteErrorAsync(req, HttpStatusCode.BadRequest, "No telemetry items provided");
|
||||
}
|
||||
|
||||
// Extract tenant+session from the first item's PartitionKey for body-vs-header tenant check
|
||||
// and for AdminAction / ServerAction lookups. All items in a batch belong to one session.
|
||||
if (!TryParsePartitionKey(items[0].PartitionKey, out var bodyTenantId, out var sessionId))
|
||||
{
|
||||
return await WriteErrorAsync(req, HttpStatusCode.BadRequest, "Malformed PartitionKey");
|
||||
}
|
||||
|
||||
if (!string.Equals(bodyTenantId, tenantIdHeader, StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
_logger.LogWarning(
|
||||
"IngestTelemetry: TenantId mismatch — header={Header}, body={Body}",
|
||||
tenantIdHeader, bodyTenantId);
|
||||
return await WriteErrorAsync(req, HttpStatusCode.Forbidden, "TenantId mismatch between header and payload");
|
||||
}
|
||||
|
||||
// Partition by Kind and persist.
|
||||
var (eventCount, signalCount, transitionCount, unknownCount) =
|
||||
await PersistItemsAsync(items, bodyTenantId, sessionId);
|
||||
|
||||
_logger.LogInformation(
|
||||
"IngestTelemetry: tenant={Tenant} session={Session} events={E} signals={S} transitions={T} unknown={U}",
|
||||
bodyTenantId, sessionId, eventCount, signalCount, transitionCount, unknownCount);
|
||||
|
||||
// Populate server→agent control signals (same services as /api/agent/ingest for parity).
|
||||
var adminAction = await TryReadAdminActionAsync(bodyTenantId, sessionId);
|
||||
var pendingActions = await _sessionRepo.FetchAndClearPendingActionsAsync(bodyTenantId, sessionId);
|
||||
|
||||
var response = req.CreateResponse(HttpStatusCode.OK);
|
||||
await response.WriteAsJsonAsync(new IngestEventsResponse
|
||||
{
|
||||
Success = true,
|
||||
EventsReceived = items.Count,
|
||||
EventsProcessed = eventCount + signalCount + transitionCount,
|
||||
Message = $"Stored {eventCount} events, {signalCount} signals, {transitionCount} transitions",
|
||||
ProcessedAt = DateTime.UtcNow,
|
||||
AdminAction = adminAction,
|
||||
Actions = pendingActions.Count > 0 ? pendingActions : null,
|
||||
});
|
||||
return response;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "IngestTelemetry: unhandled exception");
|
||||
return await WriteErrorAsync(req, HttpStatusCode.InternalServerError, "Internal server error");
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Runs device-serial and agent-version kill-switch checks. Returns a 200 response with
|
||||
/// <c>DeviceBlocked=true</c> (and optional <c>DeviceKillSignal=true</c>) if the caller
|
||||
/// should stop — or null if the request may proceed.
|
||||
/// </summary>
|
||||
private async Task<HttpResponseData?> CheckKillSwitchesAsync(
|
||||
HttpRequestData req, string tenantId, string? serialNumber, string? agentVersion)
|
||||
{
|
||||
if (!string.IsNullOrEmpty(serialNumber))
|
||||
{
|
||||
// Session-aware block: without body-parse we can't discriminate on SessionId,
|
||||
// so we use the tenant/serial blanket check. Session-scoped blocks still require
|
||||
// the agent to upload; the response just carries DeviceBlocked=true regardless of
|
||||
// session — tighter scoping lands once the body is parsed (M5.b.2 pipeline share).
|
||||
var (isBlocked, unblockAt, blockAction, _) =
|
||||
await _blockedDeviceService.IsBlockedAsync(tenantId, serialNumber);
|
||||
if (isBlocked)
|
||||
{
|
||||
var isKill = string.Equals(blockAction, "Kill", StringComparison.OrdinalIgnoreCase);
|
||||
_logger.LogWarning(
|
||||
"IngestTelemetry: {Action} device tenant={Tenant} serial={Serial} unblockAt={UnblockAt}",
|
||||
isKill ? "KILL" : "Block", tenantId, serialNumber, unblockAt);
|
||||
return await WriteDeviceBlockedAsync(req, isKill, unblockAt,
|
||||
isKill ? "Device has been issued a remote kill signal."
|
||||
: "Device is temporarily blocked by an administrator.");
|
||||
}
|
||||
}
|
||||
|
||||
if (!string.IsNullOrEmpty(agentVersion))
|
||||
{
|
||||
var (isVersionBlocked, versionAction, matchedPattern) =
|
||||
await _blockedVersionService.IsVersionBlockedAsync(agentVersion);
|
||||
if (isVersionBlocked)
|
||||
{
|
||||
var isKill = string.Equals(versionAction, "Kill", StringComparison.OrdinalIgnoreCase);
|
||||
_logger.LogWarning(
|
||||
"IngestTelemetry: version {Action} tenant={Tenant} agentVersion={AgentVersion} pattern={Pattern}",
|
||||
isKill ? "KILL" : "block", tenantId, agentVersion, matchedPattern);
|
||||
return await WriteDeviceBlockedAsync(req, isKill, null,
|
||||
isKill ? $"Agent version {agentVersion} has been issued a remote kill signal (pattern: {matchedPattern})."
|
||||
: $"Agent version {agentVersion} is blocked by administrator (pattern: {matchedPattern}).");
|
||||
}
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
private async Task<HttpResponseData> WriteDeviceBlockedAsync(
|
||||
HttpRequestData req, bool isKill, DateTime? unblockAt, string message)
|
||||
{
|
||||
var response = req.CreateResponse(HttpStatusCode.OK);
|
||||
await response.WriteAsJsonAsync(new IngestEventsResponse
|
||||
{
|
||||
Success = false,
|
||||
DeviceBlocked = true,
|
||||
DeviceKillSignal = isKill,
|
||||
UnblockAt = unblockAt,
|
||||
Message = message,
|
||||
ProcessedAt = DateTime.UtcNow,
|
||||
});
|
||||
return response;
|
||||
}
|
||||
|
||||
private async Task<string?> TryReadAdminActionAsync(string tenantId, string sessionId)
|
||||
{
|
||||
var session = await _sessionRepo.GetSessionAsync(tenantId, sessionId);
|
||||
if (session == null) return null;
|
||||
if (session.Status == SessionStatus.Succeeded) return "Succeeded";
|
||||
if (session.Status == SessionStatus.Failed) return "Failed";
|
||||
return null;
|
||||
}
|
||||
|
||||
private async Task<(int eventCount, int signalCount, int transitionCount, int unknownCount)>
|
||||
PersistItemsAsync(IReadOnlyList<TelemetryItemDto> items, string tenantId, string sessionId)
|
||||
{
|
||||
var events = new List<EnrollmentEvent>();
|
||||
var signals = new List<SignalRecord>();
|
||||
var transitions = new List<DecisionTransitionRecord>();
|
||||
var unknown = 0;
|
||||
|
||||
foreach (var item in items)
|
||||
{
|
||||
switch (item.Kind)
|
||||
{
|
||||
case "Event":
|
||||
var evt = TelemetryPayloadParser.ParseEvent(item, tenantId, sessionId);
|
||||
if (evt != null) events.Add(evt);
|
||||
break;
|
||||
case "Signal":
|
||||
var sig = TelemetryPayloadParser.ParseSignal(item, tenantId, sessionId);
|
||||
if (sig != null) signals.Add(sig);
|
||||
break;
|
||||
case "DecisionTransition":
|
||||
var tr = TelemetryPayloadParser.ParseTransition(item, tenantId, sessionId);
|
||||
if (tr != null) transitions.Add(tr);
|
||||
break;
|
||||
default:
|
||||
unknown++;
|
||||
_logger.LogWarning("IngestTelemetry: unknown Kind '{Kind}' (TelemetryItemId={Id})", item.Kind, item.TelemetryItemId);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
// Stamp authoritative TenantId/SessionId + ReceivedAt on events (same policy as legacy — never trust per-event agent values).
|
||||
var receivedAt = DateTime.UtcNow;
|
||||
foreach (var e in events)
|
||||
{
|
||||
e.TenantId = tenantId;
|
||||
e.SessionId = sessionId;
|
||||
e.ReceivedAt = receivedAt;
|
||||
}
|
||||
|
||||
var eventCount = 0;
|
||||
if (events.Count > 0)
|
||||
{
|
||||
var stored = await _sessionRepo.StoreEventsBatchAsync(events);
|
||||
eventCount = stored.Count;
|
||||
// NOTE: rule engine, app-install aggregation, SignalR notification, vulnerability
|
||||
// correlation NOT invoked here. Legacy /api/agent/ingest still owns that pipeline —
|
||||
// sharing lands in M5.b.2 via an extracted EventIngestProcessor service.
|
||||
}
|
||||
|
||||
var signalCount = await _signalRepo.StoreBatchAsync(signals);
|
||||
var transitionCount = await _transitionRepo.StoreBatchAsync(transitions);
|
||||
|
||||
return (eventCount, signalCount, transitionCount, unknown);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// PartitionKey convention is <c>{tenantId}_{sessionId}</c> — both GUIDs with dashes, so
|
||||
/// splitting on the single underscore between them is unambiguous. Returns false for any
|
||||
/// shape that doesn't match (malformed, extra parts, empty halves).
|
||||
/// </summary>
|
||||
internal static bool TryParsePartitionKey(string partitionKey, out string tenantId, out string sessionId)
|
||||
{
|
||||
tenantId = string.Empty;
|
||||
sessionId = string.Empty;
|
||||
if (string.IsNullOrEmpty(partitionKey)) return false;
|
||||
|
||||
var parts = partitionKey.Split('_');
|
||||
if (parts.Length != 2) return false;
|
||||
if (string.IsNullOrEmpty(parts[0]) || string.IsNullOrEmpty(parts[1])) return false;
|
||||
|
||||
tenantId = parts[0];
|
||||
sessionId = parts[1];
|
||||
return true;
|
||||
}
|
||||
|
||||
private static async Task<HttpResponseData> WriteErrorAsync(
|
||||
HttpRequestData req, HttpStatusCode status, string message)
|
||||
{
|
||||
var response = req.CreateResponse(status);
|
||||
await response.WriteAsJsonAsync(new IngestEventsResponse
|
||||
{
|
||||
Success = false,
|
||||
EventsReceived = 0,
|
||||
EventsProcessed = 0,
|
||||
Message = message,
|
||||
ProcessedAt = DateTime.UtcNow,
|
||||
});
|
||||
return response;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,146 @@
|
||||
using AutopilotMonitor.Shared.Models;
|
||||
using Newtonsoft.Json;
|
||||
using Newtonsoft.Json.Linq;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Functions.Ingest
|
||||
{
|
||||
/// <summary>
|
||||
/// Pure mapping layer from <see cref="TelemetryItemDto"/> wire format to backend storage
|
||||
/// records. All inputs are defensive — malformed <see cref="TelemetryItemDto.PayloadJson"/>
|
||||
/// returns <c>null</c> rather than throwing, so a single poisonous item can't take down
|
||||
/// the whole batch. Unit-tested against the canonical JSON shapes the agent produces.
|
||||
/// </summary>
|
||||
internal static class TelemetryPayloadParser
|
||||
{
|
||||
/// <summary>
|
||||
/// Parses an <c>Event</c> item's payload into an <see cref="EnrollmentEvent"/>.
|
||||
/// TenantId/SessionId on the event are authoritative from the caller; any agent-supplied
|
||||
/// values are overwritten by <see cref="IngestTelemetryFunction"/>.
|
||||
/// </summary>
|
||||
public static EnrollmentEvent? ParseEvent(TelemetryItemDto item, string tenantId, string sessionId)
|
||||
{
|
||||
if (string.IsNullOrEmpty(item?.PayloadJson)) return null;
|
||||
try
|
||||
{
|
||||
var evt = JsonConvert.DeserializeObject<EnrollmentEvent>(item.PayloadJson!);
|
||||
if (evt == null) return null;
|
||||
// Tenant/Session stamped by the function after return; here we only deserialise.
|
||||
return evt;
|
||||
}
|
||||
catch (JsonException)
|
||||
{
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Parses a <c>Signal</c> item's payload (agent-serialised DecisionSignal) into the
|
||||
/// flat <see cref="SignalRecord"/> storage shape. Extracts typed columns used for queries
|
||||
/// and indexes while preserving the full JSON in <see cref="SignalRecord.PayloadJson"/>
|
||||
/// for replay fidelity.
|
||||
/// </summary>
|
||||
public static SignalRecord? ParseSignal(TelemetryItemDto item, string tenantId, string sessionId)
|
||||
{
|
||||
if (string.IsNullOrEmpty(item?.PayloadJson)) return null;
|
||||
JObject root;
|
||||
try { root = JObject.Parse(item.PayloadJson!); }
|
||||
catch (JsonException) { return null; }
|
||||
|
||||
var ordinal = (long?)root["SessionSignalOrdinal"] ?? 0L;
|
||||
var traceOrdinal = (long?)root["SessionTraceOrdinal"] ?? item!.SessionTraceOrdinal ?? 0L;
|
||||
var kind = (string?)root["Kind"] ?? string.Empty;
|
||||
var schemaVersion = (int?)root["KindSchemaVersion"] ?? 0;
|
||||
var occurredAt = (DateTime?)root["OccurredAtUtc"] ?? item!.EnqueuedAtUtc;
|
||||
var sourceOrigin = (string?)root["SourceOrigin"] ?? string.Empty;
|
||||
|
||||
return new SignalRecord
|
||||
{
|
||||
TenantId = tenantId,
|
||||
SessionId = sessionId,
|
||||
SessionSignalOrdinal = ordinal,
|
||||
SessionTraceOrdinal = traceOrdinal,
|
||||
Kind = kind,
|
||||
KindSchemaVersion = schemaVersion,
|
||||
OccurredAtUtc = occurredAt,
|
||||
SourceOrigin = sourceOrigin,
|
||||
PayloadJson = item.PayloadJson!,
|
||||
};
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Parses a <c>DecisionTransition</c> item's payload into the flat
|
||||
/// <see cref="DecisionTransitionRecord"/> storage shape. Projects index-discriminator
|
||||
/// columns eagerly so the queue-driven index fan-out (M5.b) can read them off the
|
||||
/// primary row without re-parsing the JSON blob.
|
||||
/// </summary>
|
||||
public static DecisionTransitionRecord? ParseTransition(TelemetryItemDto item, string tenantId, string sessionId)
|
||||
{
|
||||
if (string.IsNullOrEmpty(item?.PayloadJson)) return null;
|
||||
JObject root;
|
||||
try { root = JObject.Parse(item.PayloadJson!); }
|
||||
catch (JsonException) { return null; }
|
||||
|
||||
var stepIndex = (int?)root["StepIndex"] ?? 0;
|
||||
var traceOrdinal = (long?)root["SessionTraceOrdinal"] ?? item!.SessionTraceOrdinal ?? 0L;
|
||||
var signalOrdinal = (long?)root["SignalOrdinalRef"] ?? 0L;
|
||||
var occurredAt = (DateTime?)root["OccurredAtUtc"] ?? item!.EnqueuedAtUtc;
|
||||
var trigger = (string?)root["Trigger"] ?? string.Empty;
|
||||
var fromStage = (string?)root["FromStage"] ?? string.Empty;
|
||||
var toStage = (string?)root["ToStage"] ?? string.Empty;
|
||||
var taken = (bool?)root["Taken"] ?? false;
|
||||
var deadEndReason = (string?)root["DeadEndReason"];
|
||||
var reducerVersion = (string?)root["ReducerVersion"] ?? string.Empty;
|
||||
|
||||
// Classifier verdict — optional nested object.
|
||||
string? verdictId = null, verdictLevel = null;
|
||||
if (root["ClassifierVerdict"] is JObject verdict)
|
||||
{
|
||||
verdictId = (string?)verdict["ClassifierId"] ?? (string?)verdict["Id"];
|
||||
verdictLevel = (string?)verdict["HypothesisLevel"] ?? (string?)verdict["Level"];
|
||||
}
|
||||
|
||||
return new DecisionTransitionRecord
|
||||
{
|
||||
TenantId = tenantId,
|
||||
SessionId = sessionId,
|
||||
StepIndex = stepIndex,
|
||||
SessionTraceOrdinal = traceOrdinal,
|
||||
SignalOrdinalRef = signalOrdinal,
|
||||
OccurredAtUtc = occurredAt,
|
||||
Trigger = trigger,
|
||||
FromStage = fromStage,
|
||||
ToStage = toStage,
|
||||
Taken = taken,
|
||||
DeadEndReason = deadEndReason,
|
||||
ReducerVersion = reducerVersion,
|
||||
IsTerminal = IsTerminalStage(toStage),
|
||||
ClassifierVerdictId = verdictId,
|
||||
ClassifierHypothesisLevel = verdictLevel,
|
||||
PayloadJson = item.PayloadJson!,
|
||||
};
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Terminal-stage classification mirrors the agent-side
|
||||
/// <c>SessionStageExtensions.IsTerminal()</c> (DecisionCore) so backend index queries can
|
||||
/// reach the same conclusion without linking DecisionCore. New terminal stages must be
|
||||
/// added in both places — enforced by the <c>SessionStageExtensions</c> contract test on
|
||||
/// the agent side.
|
||||
/// </summary>
|
||||
internal static bool IsTerminalStage(string toStage)
|
||||
{
|
||||
// Authoritative list mirrors DecisionCore SessionStageExtensions.IsTerminal().
|
||||
// Keep in sync — new terminal stages need both sites updated.
|
||||
switch (toStage)
|
||||
{
|
||||
case "Completed":
|
||||
case "Failed":
|
||||
case "WhiteGloveSealed":
|
||||
case "WhiteGloveCompletedPart2":
|
||||
return true;
|
||||
default:
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -105,6 +105,7 @@ public static class EndpointAccessPolicyCatalog
|
||||
// ── DeviceOrBootstrapAuth ───────────────────────────────────────
|
||||
new("POST", "agent/register-session", EndpointPolicy.DeviceOrBootstrapAuth),
|
||||
new("POST", "agent/ingest", EndpointPolicy.DeviceOrBootstrapAuth),
|
||||
new("POST", "agent/telemetry", EndpointPolicy.DeviceOrBootstrapAuth),
|
||||
new("GET", "agent/config", EndpointPolicy.DeviceOrBootstrapAuth),
|
||||
new("POST", "agent/upload-url", EndpointPolicy.DeviceOrBootstrapAuth),
|
||||
new("POST", "agent/error", EndpointPolicy.DeviceOrBootstrapAuth),
|
||||
|
||||
@@ -0,0 +1,44 @@
|
||||
using System;
|
||||
|
||||
namespace AutopilotMonitor.Shared.Models
|
||||
{
|
||||
/// <summary>
|
||||
/// Wire-format DTO for a single telemetry item sent by the V2 agent to
|
||||
/// <c>POST /api/agent/telemetry</c> (Plan §2.7a / §M5). The agent's in-memory class
|
||||
/// <c>AutopilotMonitor.Agent.V2.Core.Transport.Telemetry.TelemetryItem</c> is ctor-validated
|
||||
/// and immutable; this DTO mirrors the JSON shape so the backend can deserialise without
|
||||
/// depending on agent assemblies.
|
||||
/// <para>
|
||||
/// <b>Serialisation:</b> Newtonsoft.Json, PascalCase, <see cref="Kind"/> as string
|
||||
/// (<c>"Event"</c>/<c>"Signal"</c>/<c>"DecisionTransition"</c>) for forward-compat.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public sealed class TelemetryItemDto
|
||||
{
|
||||
/// <summary>
|
||||
/// Item type — routes to the destination table. Values: <c>Event</c>, <c>Signal</c>,
|
||||
/// <c>DecisionTransition</c>. Unknown values are rejected at parse time.
|
||||
/// </summary>
|
||||
public string Kind { get; set; } = string.Empty;
|
||||
|
||||
/// <summary>Azure-Table PartitionKey, typically <c>{tenantId}_{sessionId}</c>.</summary>
|
||||
public string PartitionKey { get; set; } = string.Empty;
|
||||
|
||||
public string RowKey { get; set; } = string.Empty;
|
||||
|
||||
/// <summary>Transport-cursor — monotonic per session across all item kinds.</summary>
|
||||
public long TelemetryItemId { get; set; }
|
||||
|
||||
/// <summary>Null for agent-global items (no session).</summary>
|
||||
public long? SessionTraceOrdinal { get; set; }
|
||||
|
||||
/// <summary>Already-serialised JSON of the wrapped payload (EnrollmentEvent / DecisionSignal / DecisionTransition).</summary>
|
||||
public string PayloadJson { get; set; } = string.Empty;
|
||||
|
||||
public bool RequiresImmediateFlush { get; set; }
|
||||
|
||||
public DateTime EnqueuedAtUtc { get; set; }
|
||||
|
||||
public int RetryCount { get; set; }
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user