mirror of
https://github.com/okieselbach/AutopilotMonitor.git
synced 2026-10-24 20:53:06 +00:00
refactor(v2core): M4.4.2 backend telemetry uploader HTTP skeleton
Concrete IBackendTelemetryUploader implementation for the Plan §2.7a
POST /api/telemetry/batch endpoint. Skeleton only — the live backend
endpoint lands in M5; this milestone only needs the client-side contract
testable against HttpMessageHandler fakes.
Design:
- New class BackendTelemetryUploader in Transport/Telemetry/ (NOT added as a
method on the M2-copied Legacy BackendApiClient, which is slated for
removal in M4.5 / v11.1+). Lives next to IBackendTelemetryUploader;
clean layer, no legacy cross-dependency.
- HttpClient injected (not self-constructed). mTLS client-cert wiring stays
with the orchestrator-builder in M4.4.5 via
HttpClientHandler.ClientCertificates; the uploader itself is cert-agnostic,
which makes it trivially testable with a custom handler.
- Ctor deps are flat: baseUrl, tenantId, optional hardware/bootstrap/version
headers. No AgentConfiguration coupling — orchestrator extracts values.
- No retry/back-off here; TelemetryUploadOrchestrator (M4.1) owns that.
- Body: JSON array of TelemetryItem via Newtonsoft (PascalCase; Kind as
string via existing StringEnumConverter annotation). No gzip — evaluate
in M5 once bandwidth profile is known.
- HTTP status -> UploadResult mapping:
2xx -> Ok
401 / 403 -> Permanent ("unauthorized: ...")
408 / 429 -> Transient (rate limit / request timeout)
other 4xx -> Permanent (client bug; retry pointless)
5xx -> Transient
HttpRequestException-> Transient ("network: ...")
TaskCanceledException (HttpClient internal timeout, not caller CTS)
-> Transient ("timeout: ...")
OperationCanceledException with caller-CTS signaled -> rethrow
Empty batch short-circuits with Ok (no HTTP call) to avoid wasted
round-trips when a concurrent drain emptied the spool mid-upload.
- Security headers: X-Tenant-Id (pre-body auth check), hardware headers
(X-Device-Manufacturer/Model/SerialNumber for whitelist + Autopilot
validation), optional X-Bootstrap-Token (pre-MDM OOBE), X-Agent-Version
for kill-switch + diagnostics.
New Constants.ApiEndpoints.TelemetryBatch = "/api/telemetry/batch" — shared
with M5 backend-side routing.
+26 tests (1313 -> 1339 green): URL + header shape, JSON body wire-format
(Kind as string, TelemetryItemId/PK/RK/PayloadJson round-trip), empty-batch
short-circuit, full HTTP status-code -> UploadResult matrix (2xx/401/403/
4xx/408/429/5xx), HttpRequestException + timeout mapping, caller
cancellation propagation, null-arg rejection.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
+307
@@ -0,0 +1,307 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Net;
|
||||
using System.Net.Http;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using AutopilotMonitor.Agent.V2.Core.Transport.Telemetry;
|
||||
using Newtonsoft.Json.Linq;
|
||||
using Xunit;
|
||||
|
||||
namespace AutopilotMonitor.Agent.V2.Core.Tests.Transport
|
||||
{
|
||||
public sealed class BackendTelemetryUploaderTests
|
||||
{
|
||||
private const string BaseUrl = "https://backend.test";
|
||||
private const string ExpectedEndpoint = "https://backend.test/api/telemetry/batch";
|
||||
private static readonly DateTime At = new DateTime(2026, 4, 20, 10, 0, 0, DateTimeKind.Utc);
|
||||
|
||||
private static TelemetryItem NewEventItem(long id, string sessionId = "S1", string tenantId = "T1") =>
|
||||
new TelemetryItem(
|
||||
kind: TelemetryItemKind.Event,
|
||||
partitionKey: $"{tenantId}_{sessionId}",
|
||||
rowKey: $"20260420100000000_{id:D10}",
|
||||
telemetryItemId: id,
|
||||
sessionTraceOrdinal: id,
|
||||
payloadJson: $"{{\"EventType\":\"test\",\"Sequence\":{id}}}",
|
||||
requiresImmediateFlush: false,
|
||||
enqueuedAtUtc: At);
|
||||
|
||||
private static TelemetryItem NewSignalItem(long id) =>
|
||||
new TelemetryItem(
|
||||
kind: TelemetryItemKind.Signal,
|
||||
partitionKey: "T1_S1",
|
||||
rowKey: id.ToString("D10"),
|
||||
telemetryItemId: id,
|
||||
sessionTraceOrdinal: id,
|
||||
payloadJson: "{\"Kind\":\"SessionStarted\"}",
|
||||
requiresImmediateFlush: false,
|
||||
enqueuedAtUtc: At);
|
||||
|
||||
private sealed class Rig : IDisposable
|
||||
{
|
||||
public RecordingHttpMessageHandler Handler { get; } = new RecordingHttpMessageHandler();
|
||||
public HttpClient Http { get; }
|
||||
public BackendTelemetryUploader Sut { get; }
|
||||
|
||||
public Rig(
|
||||
string tenantId = "T1",
|
||||
string? manufacturer = "Dell",
|
||||
string? model = "Latitude",
|
||||
string? serial = "ABC123",
|
||||
string? bootstrap = null,
|
||||
string? version = "11.0.0")
|
||||
{
|
||||
Http = new HttpClient(Handler) { Timeout = TimeSpan.FromSeconds(30) };
|
||||
Sut = new BackendTelemetryUploader(
|
||||
httpClient: Http,
|
||||
baseUrl: BaseUrl,
|
||||
tenantId: tenantId,
|
||||
manufacturer: manufacturer,
|
||||
model: model,
|
||||
serialNumber: serial,
|
||||
bootstrapToken: bootstrap,
|
||||
agentVersion: version);
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
Http.Dispose();
|
||||
Handler.Dispose();
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================ Request shape
|
||||
|
||||
[Fact]
|
||||
public async Task UploadBatchAsync_posts_to_api_telemetry_batch_with_expected_headers()
|
||||
{
|
||||
using var rig = new Rig();
|
||||
rig.Handler.QueueStatus(HttpStatusCode.OK);
|
||||
|
||||
var result = await rig.Sut.UploadBatchAsync(new[] { NewEventItem(1) }, CancellationToken.None);
|
||||
|
||||
Assert.True(result.Success);
|
||||
Assert.Single(rig.Handler.Captured);
|
||||
var req = rig.Handler.Captured[0];
|
||||
Assert.Equal(HttpMethod.Post, req.Method);
|
||||
Assert.Equal(ExpectedEndpoint, req.RequestUri!.ToString());
|
||||
|
||||
Assert.True(req.TryGetHeader("X-Tenant-Id", out var tid));
|
||||
Assert.Equal("T1", tid);
|
||||
Assert.True(req.TryGetHeader("X-Device-Manufacturer", out var mf));
|
||||
Assert.Equal("Dell", mf);
|
||||
Assert.True(req.TryGetHeader("X-Device-Model", out var mo));
|
||||
Assert.Equal("Latitude", mo);
|
||||
Assert.True(req.TryGetHeader("X-Device-SerialNumber", out var sn));
|
||||
Assert.Equal("ABC123", sn);
|
||||
Assert.True(req.TryGetHeader("X-Agent-Version", out var ver));
|
||||
Assert.Equal("11.0.0", ver);
|
||||
|
||||
Assert.True(req.TryGetHeader("Content-Type", out var ct));
|
||||
Assert.StartsWith("application/json", ct);
|
||||
|
||||
Assert.False(req.Headers.ContainsKey("X-Bootstrap-Token"));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Bootstrap_token_is_attached_only_when_provided()
|
||||
{
|
||||
using var rig = new Rig(bootstrap: "bs-abc");
|
||||
rig.Handler.QueueStatus(HttpStatusCode.OK);
|
||||
|
||||
await rig.Sut.UploadBatchAsync(new[] { NewEventItem(1) }, CancellationToken.None);
|
||||
|
||||
var req = rig.Handler.Captured[0];
|
||||
Assert.True(req.TryGetHeader("X-Bootstrap-Token", out var tok));
|
||||
Assert.Equal("bs-abc", tok);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Hardware_headers_are_omitted_when_values_are_null()
|
||||
{
|
||||
using var rig = new Rig(manufacturer: null, model: null, serial: null);
|
||||
rig.Handler.QueueStatus(HttpStatusCode.OK);
|
||||
|
||||
await rig.Sut.UploadBatchAsync(new[] { NewEventItem(1) }, CancellationToken.None);
|
||||
|
||||
var req = rig.Handler.Captured[0];
|
||||
Assert.False(req.Headers.ContainsKey("X-Device-Manufacturer"));
|
||||
Assert.False(req.Headers.ContainsKey("X-Device-Model"));
|
||||
Assert.False(req.Headers.ContainsKey("X-Device-SerialNumber"));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Body_is_JSON_array_of_TelemetryItems_with_Kind_as_string()
|
||||
{
|
||||
using var rig = new Rig();
|
||||
rig.Handler.QueueStatus(HttpStatusCode.OK);
|
||||
|
||||
var batch = new[] { NewEventItem(1), NewSignalItem(2) };
|
||||
await rig.Sut.UploadBatchAsync(batch, CancellationToken.None);
|
||||
|
||||
var body = rig.Handler.Captured[0].Body!;
|
||||
var parsed = JArray.Parse(body);
|
||||
Assert.Equal(2, parsed.Count);
|
||||
Assert.Equal("Event", (string?)parsed[0]["Kind"]);
|
||||
Assert.Equal(1, (long)parsed[0]["TelemetryItemId"]!);
|
||||
Assert.Equal("Signal", (string?)parsed[1]["Kind"]);
|
||||
Assert.Equal(2, (long)parsed[1]["TelemetryItemId"]!);
|
||||
Assert.Equal("T1_S1", (string?)parsed[0]["PartitionKey"]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Empty_batch_short_circuits_without_hitting_the_network()
|
||||
{
|
||||
using var rig = new Rig();
|
||||
var result = await rig.Sut.UploadBatchAsync(Array.Empty<TelemetryItem>(), CancellationToken.None);
|
||||
|
||||
Assert.True(result.Success);
|
||||
Assert.Empty(rig.Handler.Captured);
|
||||
}
|
||||
|
||||
// ============================================================ Status mapping
|
||||
|
||||
[Theory]
|
||||
[InlineData(HttpStatusCode.OK)]
|
||||
[InlineData(HttpStatusCode.Created)]
|
||||
[InlineData(HttpStatusCode.Accepted)]
|
||||
[InlineData(HttpStatusCode.NoContent)]
|
||||
public async Task Success_2xx_maps_to_Ok(HttpStatusCode status)
|
||||
{
|
||||
using var rig = new Rig();
|
||||
rig.Handler.QueueStatus(status);
|
||||
|
||||
var result = await rig.Sut.UploadBatchAsync(new[] { NewEventItem(1) }, CancellationToken.None);
|
||||
|
||||
Assert.True(result.Success);
|
||||
Assert.Null(result.ErrorReason);
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(HttpStatusCode.Unauthorized)]
|
||||
[InlineData(HttpStatusCode.Forbidden)]
|
||||
public async Task Auth_failure_401_403_maps_to_Permanent_unauthorized(HttpStatusCode status)
|
||||
{
|
||||
using var rig = new Rig();
|
||||
rig.Handler.QueueStatus(status, body: "device not whitelisted");
|
||||
|
||||
var result = await rig.Sut.UploadBatchAsync(new[] { NewEventItem(1) }, CancellationToken.None);
|
||||
|
||||
Assert.False(result.Success);
|
||||
Assert.False(result.IsTransient);
|
||||
Assert.Contains("unauthorized", result.ErrorReason);
|
||||
Assert.Contains(((int)status).ToString(), result.ErrorReason);
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(HttpStatusCode.BadRequest)] // 400
|
||||
[InlineData(HttpStatusCode.NotFound)] // 404
|
||||
[InlineData(HttpStatusCode.Conflict)] // 409
|
||||
[InlineData(HttpStatusCode.UnsupportedMediaType)] // 415
|
||||
public async Task Other_4xx_maps_to_Permanent(HttpStatusCode status)
|
||||
{
|
||||
using var rig = new Rig();
|
||||
rig.Handler.QueueStatus(status);
|
||||
|
||||
var result = await rig.Sut.UploadBatchAsync(new[] { NewEventItem(1) }, CancellationToken.None);
|
||||
|
||||
Assert.False(result.Success);
|
||||
Assert.False(result.IsTransient);
|
||||
Assert.Contains(((int)status).ToString(), result.ErrorReason);
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(HttpStatusCode.RequestTimeout)] // 408
|
||||
[InlineData((HttpStatusCode)429)] // TooManyRequests
|
||||
public async Task Rate_limit_and_408_map_to_Transient(HttpStatusCode status)
|
||||
{
|
||||
using var rig = new Rig();
|
||||
rig.Handler.QueueStatus(status);
|
||||
|
||||
var result = await rig.Sut.UploadBatchAsync(new[] { NewEventItem(1) }, CancellationToken.None);
|
||||
|
||||
Assert.False(result.Success);
|
||||
Assert.True(result.IsTransient);
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(HttpStatusCode.InternalServerError)] // 500
|
||||
[InlineData(HttpStatusCode.BadGateway)] // 502
|
||||
[InlineData(HttpStatusCode.ServiceUnavailable)] // 503
|
||||
[InlineData(HttpStatusCode.GatewayTimeout)] // 504
|
||||
public async Task Server_5xx_maps_to_Transient(HttpStatusCode status)
|
||||
{
|
||||
using var rig = new Rig();
|
||||
rig.Handler.QueueStatus(status);
|
||||
|
||||
var result = await rig.Sut.UploadBatchAsync(new[] { NewEventItem(1) }, CancellationToken.None);
|
||||
|
||||
Assert.False(result.Success);
|
||||
Assert.True(result.IsTransient);
|
||||
}
|
||||
|
||||
// ============================================================ Exception handling
|
||||
|
||||
[Fact]
|
||||
public async Task HttpRequestException_network_error_maps_to_Transient()
|
||||
{
|
||||
using var rig = new Rig();
|
||||
rig.Handler.QueueThrow(new HttpRequestException("dns failure"));
|
||||
|
||||
var result = await rig.Sut.UploadBatchAsync(new[] { NewEventItem(1) }, CancellationToken.None);
|
||||
|
||||
Assert.False(result.Success);
|
||||
Assert.True(result.IsTransient);
|
||||
Assert.Contains("network", result.ErrorReason);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task HttpClient_timeout_maps_to_Transient()
|
||||
{
|
||||
// HttpClient surfaces its own internal timeout as TaskCanceledException with
|
||||
// CancellationToken.IsCancellationRequested == false.
|
||||
using var rig = new Rig();
|
||||
rig.Handler.QueueThrow(new TaskCanceledException("The operation was canceled."));
|
||||
|
||||
var result = await rig.Sut.UploadBatchAsync(new[] { NewEventItem(1) }, CancellationToken.None);
|
||||
|
||||
Assert.False(result.Success);
|
||||
Assert.True(result.IsTransient);
|
||||
Assert.Contains("timeout", result.ErrorReason);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Caller_cancellation_propagates_as_OperationCanceledException()
|
||||
{
|
||||
using var rig = new Rig();
|
||||
using var cts = new CancellationTokenSource();
|
||||
cts.Cancel();
|
||||
|
||||
await Assert.ThrowsAnyAsync<OperationCanceledException>(() =>
|
||||
rig.Sut.UploadBatchAsync(new[] { NewEventItem(1) }, cts.Token));
|
||||
}
|
||||
|
||||
// ============================================================ Misuse
|
||||
|
||||
[Fact]
|
||||
public void Constructor_rejects_null_or_empty_required_args()
|
||||
{
|
||||
var http = new HttpClient();
|
||||
Assert.Throws<ArgumentNullException>(() =>
|
||||
new BackendTelemetryUploader(null!, BaseUrl, "T1"));
|
||||
Assert.Throws<ArgumentException>(() =>
|
||||
new BackendTelemetryUploader(http, "", "T1"));
|
||||
Assert.Throws<ArgumentException>(() =>
|
||||
new BackendTelemetryUploader(http, BaseUrl, ""));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task UploadBatchAsync_rejects_null_items()
|
||||
{
|
||||
using var rig = new Rig();
|
||||
await Assert.ThrowsAsync<ArgumentNullException>(() =>
|
||||
rig.Sut.UploadBatchAsync(null!, CancellationToken.None));
|
||||
}
|
||||
}
|
||||
}
|
||||
+104
@@ -0,0 +1,104 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Net;
|
||||
using System.Net.Http;
|
||||
using System.Text;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
|
||||
namespace AutopilotMonitor.Agent.V2.Core.Tests.Transport
|
||||
{
|
||||
/// <summary>
|
||||
/// Test <see cref="HttpMessageHandler"/> that captures every outgoing request (including the
|
||||
/// body string, because the default handler disposes content after <c>SendAsync</c> returns)
|
||||
/// and returns responses from a scripted queue.
|
||||
/// </summary>
|
||||
internal sealed class RecordingHttpMessageHandler : HttpMessageHandler
|
||||
{
|
||||
private readonly Queue<Func<HttpRequestMessage, HttpResponseMessage>> _script =
|
||||
new Queue<Func<HttpRequestMessage, HttpResponseMessage>>();
|
||||
private readonly List<CapturedRequest> _captured = new List<CapturedRequest>();
|
||||
|
||||
public IReadOnlyList<CapturedRequest> Captured => _captured;
|
||||
|
||||
public RecordingHttpMessageHandler QueueStatus(HttpStatusCode status, string? body = null)
|
||||
{
|
||||
_script.Enqueue(_ =>
|
||||
{
|
||||
var resp = new HttpResponseMessage(status);
|
||||
if (body != null) resp.Content = new StringContent(body, Encoding.UTF8, "application/json");
|
||||
return resp;
|
||||
});
|
||||
return this;
|
||||
}
|
||||
|
||||
public RecordingHttpMessageHandler QueueThrow(Exception ex)
|
||||
{
|
||||
_script.Enqueue(_ => throw ex);
|
||||
return this;
|
||||
}
|
||||
|
||||
protected override async Task<HttpResponseMessage> SendAsync(
|
||||
HttpRequestMessage request,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
// Real HttpClient throws TaskCanceledException when the caller's token is already
|
||||
// signaled, before touching the network. Mirror that here so cancellation tests
|
||||
// don't silently succeed with a default 200.
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
|
||||
string? body = null;
|
||||
if (request.Content != null)
|
||||
{
|
||||
body = await request.Content.ReadAsStringAsync().ConfigureAwait(false);
|
||||
}
|
||||
|
||||
_captured.Add(new CapturedRequest(request, body));
|
||||
|
||||
if (_script.Count == 0)
|
||||
{
|
||||
return new HttpResponseMessage(HttpStatusCode.OK);
|
||||
}
|
||||
|
||||
return _script.Dequeue()(request);
|
||||
}
|
||||
|
||||
internal sealed class CapturedRequest
|
||||
{
|
||||
public CapturedRequest(HttpRequestMessage request, string? body)
|
||||
{
|
||||
Method = request.Method;
|
||||
RequestUri = request.RequestUri;
|
||||
Headers = new Dictionary<string, string[]>(StringComparer.OrdinalIgnoreCase);
|
||||
foreach (var h in request.Headers)
|
||||
{
|
||||
Headers[h.Key] = System.Linq.Enumerable.ToArray(h.Value);
|
||||
}
|
||||
if (request.Content != null)
|
||||
{
|
||||
foreach (var h in request.Content.Headers)
|
||||
{
|
||||
Headers[h.Key] = System.Linq.Enumerable.ToArray(h.Value);
|
||||
}
|
||||
}
|
||||
Body = body;
|
||||
}
|
||||
|
||||
public HttpMethod Method { get; }
|
||||
public Uri? RequestUri { get; }
|
||||
public Dictionary<string, string[]> Headers { get; }
|
||||
public string? Body { get; }
|
||||
|
||||
public bool TryGetHeader(string name, out string value)
|
||||
{
|
||||
if (Headers.TryGetValue(name, out var values) && values.Length > 0)
|
||||
{
|
||||
value = values[0];
|
||||
return true;
|
||||
}
|
||||
value = string.Empty;
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+182
@@ -0,0 +1,182 @@
|
||||
#nullable enable
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Net;
|
||||
using System.Net.Http;
|
||||
using System.Net.Http.Headers;
|
||||
using System.Text;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using AutopilotMonitor.Shared;
|
||||
using Newtonsoft.Json;
|
||||
|
||||
namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
|
||||
{
|
||||
/// <summary>
|
||||
/// Produktions-<see cref="IBackendTelemetryUploader"/>. Plan §2.7a / §4.x M4.4.2.
|
||||
/// <para>
|
||||
/// HTTP-Skelett gegen <c>POST /api/telemetry/batch</c>. Sendet einen Batch von
|
||||
/// <see cref="TelemetryItem"/>s als JSON-Array; Backend routet pro Item nach
|
||||
/// <see cref="TelemetryItemKind"/> in die Ziel-Tabelle (Events / Signals /
|
||||
/// DecisionTransitions) und dedupliziert via (<c>PartitionKey</c>, <c>RowKey</c>).
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Retry/Back-off</b> übernimmt <see cref="TelemetryUploadOrchestrator"/> (M4.1) — dieser
|
||||
/// Uploader macht genau einen HTTP-Call und mapped die Response auf
|
||||
/// <see cref="UploadResult"/>.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>mTLS</b>: Client-Zertifikat wird über den vom Caller eingerichteten
|
||||
/// <see cref="HttpClient"/> (<see cref="HttpClientHandler.ClientCertificates"/>) gesendet —
|
||||
/// dieser Uploader ist cert-agnostisch. Tests injizieren einen <see cref="HttpClient"/>
|
||||
/// mit Fake-<see cref="HttpMessageHandler"/>.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public sealed class BackendTelemetryUploader : IBackendTelemetryUploader
|
||||
{
|
||||
private readonly HttpClient _httpClient;
|
||||
private readonly string _endpointUrl;
|
||||
private readonly string _tenantId;
|
||||
private readonly string? _manufacturer;
|
||||
private readonly string? _model;
|
||||
private readonly string? _serialNumber;
|
||||
private readonly string? _bootstrapToken;
|
||||
private readonly string? _agentVersion;
|
||||
|
||||
public BackendTelemetryUploader(
|
||||
HttpClient httpClient,
|
||||
string baseUrl,
|
||||
string tenantId,
|
||||
string? manufacturer = null,
|
||||
string? model = null,
|
||||
string? serialNumber = null,
|
||||
string? bootstrapToken = null,
|
||||
string? agentVersion = null)
|
||||
{
|
||||
_httpClient = httpClient ?? throw new ArgumentNullException(nameof(httpClient));
|
||||
if (string.IsNullOrEmpty(baseUrl)) throw new ArgumentException("BaseUrl is mandatory.", nameof(baseUrl));
|
||||
if (string.IsNullOrEmpty(tenantId)) throw new ArgumentException("TenantId is mandatory.", nameof(tenantId));
|
||||
|
||||
_endpointUrl = baseUrl.TrimEnd('/') + Constants.ApiEndpoints.TelemetryBatch;
|
||||
_tenantId = tenantId;
|
||||
_manufacturer = manufacturer;
|
||||
_model = model;
|
||||
_serialNumber = serialNumber;
|
||||
_bootstrapToken = bootstrapToken;
|
||||
_agentVersion = agentVersion;
|
||||
}
|
||||
|
||||
public async Task<UploadResult> UploadBatchAsync(
|
||||
IReadOnlyList<TelemetryItem> items,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
if (items == null) throw new ArgumentNullException(nameof(items));
|
||||
|
||||
// Empty batch: no-op success. Avoids burning a round-trip when the spool pulls an
|
||||
// empty page (possible in race conditions after a concurrent drain succeeded).
|
||||
if (items.Count == 0) return UploadResult.Ok();
|
||||
|
||||
var body = SerializeBatch(items);
|
||||
using (var request = new HttpRequestMessage(HttpMethod.Post, _endpointUrl))
|
||||
{
|
||||
request.Content = new StringContent(body, Encoding.UTF8);
|
||||
request.Content.Headers.ContentType = new MediaTypeHeaderValue("application/json");
|
||||
AddSecurityHeaders(request);
|
||||
|
||||
HttpResponseMessage response;
|
||||
try
|
||||
{
|
||||
response = await _httpClient.SendAsync(request, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
// Caller-initiated cancellation — propagate; the orchestrator's drain
|
||||
// loop will stop entirely rather than schedule a retry.
|
||||
throw;
|
||||
}
|
||||
catch (TaskCanceledException ex)
|
||||
{
|
||||
// HttpClient timeout (not caller cancellation) surfaces as TaskCanceledException
|
||||
// with cancellationToken.IsCancellationRequested == false.
|
||||
return UploadResult.Transient($"timeout: {ex.Message}");
|
||||
}
|
||||
catch (HttpRequestException ex)
|
||||
{
|
||||
// DNS failure, socket error, etc. — network-layer is retryable.
|
||||
return UploadResult.Transient($"network: {ex.Message}");
|
||||
}
|
||||
|
||||
using (response)
|
||||
{
|
||||
return await MapResponseAsync(response).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static async Task<UploadResult> MapResponseAsync(HttpResponseMessage response)
|
||||
{
|
||||
var statusCode = (int)response.StatusCode;
|
||||
|
||||
if (response.IsSuccessStatusCode) return UploadResult.Ok();
|
||||
|
||||
var reason = await TryReadReasonAsync(response).ConfigureAwait(false);
|
||||
var shortReason = $"http {statusCode}{(string.IsNullOrEmpty(reason) ? string.Empty : ": " + reason)}";
|
||||
|
||||
switch (response.StatusCode)
|
||||
{
|
||||
case HttpStatusCode.Unauthorized: // 401
|
||||
case HttpStatusCode.Forbidden: // 403
|
||||
return UploadResult.Permanent($"unauthorized: {shortReason}");
|
||||
|
||||
case HttpStatusCode.RequestTimeout: // 408
|
||||
case (HttpStatusCode)429: // TooManyRequests — enum available net48 SP1+, cast for safety
|
||||
return UploadResult.Transient(shortReason);
|
||||
}
|
||||
|
||||
if (statusCode >= 500 && statusCode <= 599) return UploadResult.Transient(shortReason);
|
||||
if (statusCode >= 400 && statusCode <= 499) return UploadResult.Permanent(shortReason);
|
||||
|
||||
// Unexpected range (1xx, 3xx). Treat as transient — orchestrator retries and logs.
|
||||
return UploadResult.Transient(shortReason);
|
||||
}
|
||||
|
||||
private static async Task<string> TryReadReasonAsync(HttpResponseMessage response)
|
||||
{
|
||||
try
|
||||
{
|
||||
var body = await response.Content.ReadAsStringAsync().ConfigureAwait(false);
|
||||
if (string.IsNullOrEmpty(body)) return string.Empty;
|
||||
return body.Length <= 200 ? body : body.Substring(0, 200);
|
||||
}
|
||||
catch
|
||||
{
|
||||
return string.Empty;
|
||||
}
|
||||
}
|
||||
|
||||
private static string SerializeBatch(IReadOnlyList<TelemetryItem> items)
|
||||
{
|
||||
// Newtonsoft PascalCase default + TelemetryItemKind carries [StringEnumConverter] →
|
||||
// Backend routes by string "Event"/"Signal"/"DecisionTransition". Wire-format
|
||||
// matches Plan §4.x M5 contract (PascalCase, StringEnum, ISO-8601 UTC).
|
||||
return JsonConvert.SerializeObject(items, Formatting.None);
|
||||
}
|
||||
|
||||
private void AddSecurityHeaders(HttpRequestMessage request)
|
||||
{
|
||||
// X-Tenant-Id lets the backend run auth checks before parsing the body.
|
||||
request.Headers.Add("X-Tenant-Id", _tenantId);
|
||||
|
||||
// Hardware headers for whitelist/Autopilot validation.
|
||||
if (!string.IsNullOrEmpty(_manufacturer)) request.Headers.Add("X-Device-Manufacturer", _manufacturer);
|
||||
if (!string.IsNullOrEmpty(_model)) request.Headers.Add("X-Device-Model", _model);
|
||||
if (!string.IsNullOrEmpty(_serialNumber)) request.Headers.Add("X-Device-SerialNumber", _serialNumber);
|
||||
|
||||
// Bootstrap token auth for pre-MDM agents (cert-less OOBE phase).
|
||||
if (!string.IsNullOrEmpty(_bootstrapToken)) request.Headers.Add("X-Bootstrap-Token", _bootstrapToken);
|
||||
|
||||
// Agent version for version-based kill-switch + server-side diagnostics.
|
||||
if (!string.IsNullOrEmpty(_agentVersion)) request.Headers.Add("X-Agent-Version", _agentVersion);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -139,6 +139,12 @@ namespace AutopilotMonitor.Shared
|
||||
public const string ReseedFromGitHub = "/api/rules/reseed-from-github";
|
||||
public const string ValidateBootstrapCode = "/api/bootstrap/validate/{code}";
|
||||
|
||||
// V2 Agent generic telemetry transport (Plan §2.7a).
|
||||
// Backend routes per TelemetryItem.Kind to Events / Signals / DecisionTransitions tables.
|
||||
// Single endpoint for Event + Signal + DecisionTransition batch ingest; replaces the
|
||||
// per-type IngestEvents path once the Legacy Agent is decommissioned (v11.1+).
|
||||
public const string TelemetryBatch = "/api/telemetry/batch";
|
||||
|
||||
// MCP/Agent API search endpoints
|
||||
public const string SearchSessions = "/api/search/sessions";
|
||||
public const string SearchSessionsByEvent = "/api/search/sessions-by-event";
|
||||
|
||||
Reference in New Issue
Block a user