Message's body is miscategorized as XML when the Content-Type header is missing

src/Directory.Packages.props

@@ -8,7 +8,7 @@
     <PackageVersion Include="AWSSDK.CloudWatch" Version="4.0.1" />
     <PackageVersion Include="AWSSDK.SecurityToken" Version="4.0.0.4" />
     <PackageVersion Include="Azure.Identity" Version="1.13.2" />
-    <PackageVersion Include="Azure.Monitor.Query" Version="1.6.0" />
+    <PackageVersion Include="Azure.Monitor.Query.Metrics" Version="1.0.0" />
     <PackageVersion Include="Azure.ResourceManager.ServiceBus" Version="1.1.0" />
     <PackageVersion Include="ByteSize" Version="2.1.2" />
     <PackageVersion Include="Caliburn.Micro" Version="4.0.230" />
@@ -17,8 +17,8 @@
     <PackageVersion Include="Fody" Version="6.9.1" />
     <PackageVersion Include="GitHubActionsTestLogger" Version="2.4.1" />
     <PackageVersion Include="HdrHistogram" Version="2.5.0" />
-    <PackageVersion Include="Microsoft.AspNetCore.Mvc.Testing" Version="8.0.17" />
-    <PackageVersion Include="Microsoft.AspNetCore.SignalR.Client" Version="8.0.17" />
+    <PackageVersion Include="Microsoft.AspNetCore.Mvc.Testing" Version="8.0.21" />
+    <PackageVersion Include="Microsoft.AspNetCore.SignalR.Client" Version="8.0.21" />
     <PackageVersion Include="Microsoft.Extensions.DependencyInjection" Version="8.0.1" />
     <PackageVersion Include="Microsoft.Extensions.DependencyModel" Version="8.0.2" />
     <PackageVersion Include="Microsoft.Extensions.Hosting" Version="8.0.1" />
@@ -42,7 +42,7 @@
     <PackageVersion Include="NServiceBus.Metrics" Version="5.0.1" />
     <PackageVersion Include="NServiceBus.Metrics.ServiceControl" Version="5.0.0" />
     <PackageVersion Include="NServiceBus.Persistence.NonDurable" Version="2.0.1" />
-    <PackageVersion Include="NServiceBus.RabbitMQ" Version="10.1.3" />
+    <PackageVersion Include="NServiceBus.RabbitMQ" Version="10.1.4" />
     <PackageVersion Include="NServiceBus.SagaAudit" Version="5.0.2" />
     <PackageVersion Include="NServiceBus.Testing" Version="9.0.1" />
     <PackageVersion Include="NServiceBus.Transport.AzureServiceBus" Version="5.0.2" />

src/ServiceControl.Audit.Persistence.RavenDB/UnitOfWork/RavenAuditIngestionUnitOfWork.cs

@@ -35,7 +35,7 @@ public async Task RecordProcessedMessage(ProcessedMessage processedMessage, Read
             if (!body.IsEmpty)
             {
                 await using var stream = new ReadOnlyStream(body);
-                var contentType = processedMessage.Headers.GetValueOrDefault(Headers.ContentType, "text/xml");
+                var contentType = processedMessage.Headers.GetValueOrDefault(Headers.ContentType, "text/plain");
 
                 await bodyStorage.Store(processedMessage.Id, contentType, body.Length, stream, cancellationToken);
             }

src/ServiceControl.Audit.Persistence.Tests/AuditTests.cs

@@ -80,7 +80,7 @@ await IngestProcessedMessagesAudits(
         [Test]
         public async Task Can_roundtrip_message_body()
         {
-            string expectedContentType = "text/xml";
+            string expectedContentType = "text/plain";
             var unitOfWork = await StartAuditUnitOfWork(1);
 
             var body = new byte[100];

src/ServiceControl.Audit.Persistence/BodyStorageEnricher.cs

@@ -21,7 +21,7 @@ public async ValueTask StoreAuditMessageBody(ReadOnlyMemory<byte> body, Processe
                 return;
             }
 
-            var contentType = GetContentType(processedMessage.Headers, "text/xml");
+            var contentType = GetContentType(processedMessage.Headers, "text/plain");
             processedMessage.MessageMetadata.Add("ContentType", contentType);
 
             var stored = await TryStoreBody(body, processedMessage, bodySize, contentType, cancellationToken);

src/ServiceControl.Audit.UnitTests/BodyStorage/BodyStorageEnricherTests.cs

@@ -92,7 +92,7 @@ public async Task Should_store_body_in_metadata_when_below_large_object_heap_and
                 [Headers.MessageId] = "someid",
                 ["ServiceControl.Retry.UniqueMessageId"] = "someid",
                 [Headers.ProcessingEndpoint] = "someendpoint",
-                [Headers.ContentType] = "text/xml"
+                [Headers.ContentType] = "text/plain"
             };
 
             var message = new ProcessedMessage(headers, metadata);

src/ServiceControl.Persistence.RavenDB/UnitOfWork/RavenRecoverabilityIngestionUnitOfWork.cs

@@ -33,7 +33,7 @@ public Task RecordFailedProcessingAttempt(
             List<FailedMessage.FailureGroup> groups)
         {
             var uniqueMessageId = context.Headers.UniqueId();
-            var contentType = GetContentType(context.Headers, "text/xml");
+            var contentType = GetContentType(context.Headers, "text/plain");
             var bodySize = context.Body.Length;
 
             processingAttempt.MessageMetadata.Add("ContentType", contentType);

src/ServiceControl.Transports.ASBS/AzureQuery.cs

@@ -12,8 +12,8 @@ namespace ServiceControl.Transports.ASBS;
 using Azure.Core;
 using Azure.Core.Pipeline;
 using Azure.Identity;
-using Azure.Monitor.Query;
-using Azure.Monitor.Query.Models;
+using Azure.Monitor.Query.Metrics;
+using Azure.Monitor.Query.Metrics.Models;
 using Azure.ResourceManager;
 using Azure.ResourceManager.Resources;
 using Azure.ResourceManager.ServiceBus;
@@ -26,10 +26,12 @@ public class AzureQuery(ILogger<AzureQuery> logger, TimeProvider timeProvider, T
     : BrokerThroughputQuery(logger, "AzureServiceBus")
 {
     string serviceBusName = string.Empty;
-    MetricsQueryClient? client;
     ArmClient? armClient;
-    string? resourceId;
+    TokenCredential? credential;
+    ResourceIdentifier? resourceId;
     ArmEnvironment armEnvironment;
+    MetricsClientAudience metricsClientAudience;
+    MetricsClient? metricsClient;
 
     protected override void InitializeCore(ReadOnlyDictionary<string, string> settings)
     {
@@ -102,7 +104,7 @@ protected override void InitializeCore(ReadOnlyDictionary<string, string> settin
             Diagnostics.AppendLine("Client secret set");
         }
 
-        (armEnvironment, var metricsQueryAudience) = GetEnvironment();
+        (armEnvironment, metricsClientAudience) = GetEnvironment();
 
         if (managementUrl == null)
         {
@@ -118,28 +120,17 @@ protected override void InitializeCore(ReadOnlyDictionary<string, string> settin
             return;
         }
 
-        TokenCredential clientCredentials;
         if (connectionSettings.AuthenticationMethod is TokenCredentialAuthentication tokenCredentialAuthentication)
         {
             Diagnostics.AppendLine("Attempting to use managed identity");
-            clientCredentials = tokenCredentialAuthentication.Credential;
+            credential = tokenCredentialAuthentication.Credential;
         }
         else
         {
-            clientCredentials = new ClientSecretCredential(tenantId, clientId, clientSecret);
+            credential = new ClientSecretCredential(tenantId, clientId, clientSecret);
         }
 
-        client = new MetricsQueryClient(armEnvironment.Endpoint, clientCredentials,
-            new MetricsQueryClientOptions
-            {
-                Audience = metricsQueryAudience,
-                Transport = new HttpClientTransport(
-                    new HttpClient(new SocketsHttpHandler
-                    {
-                        PooledConnectionIdleTimeout = TimeSpan.FromMinutes(2)
-                    }))
-            });
-        armClient = new ArmClient(clientCredentials, subscriptionId,
+        armClient = new ArmClient(credential, subscriptionId,
             new ArmClientOptions
             {
                 Environment = armEnvironment,
@@ -152,31 +143,31 @@ protected override void InitializeCore(ReadOnlyDictionary<string, string> settin
 
         return;
 
-        (ArmEnvironment armEnvironment, MetricsQueryAudience metricsQueryAudience) GetEnvironment()
+        (ArmEnvironment armEnvironment, MetricsClientAudience metricsClientAudience) GetEnvironment()
         {
             if (managementUrlParsed == null)
             {
-                return (ArmEnvironment.AzurePublicCloud, MetricsQueryAudience.AzurePublicCloud);
+                return (ArmEnvironment.AzurePublicCloud, MetricsClientAudience.AzurePublicCloud);
             }
 
             if (managementUrlParsed == ArmEnvironment.AzurePublicCloud.Endpoint)
             {
-                return (ArmEnvironment.AzurePublicCloud, MetricsQueryAudience.AzurePublicCloud);
+                return (ArmEnvironment.AzurePublicCloud, MetricsClientAudience.AzurePublicCloud);
             }
 
             if (managementUrlParsed == ArmEnvironment.AzureChina.Endpoint)
             {
-                return (ArmEnvironment.AzureChina, MetricsQueryAudience.AzureChina);
+                return (ArmEnvironment.AzureChina, MetricsClientAudience.AzureChina);
             }
 
             if (managementUrlParsed == ArmEnvironment.AzureGermany.Endpoint)
             {
-                return (ArmEnvironment.AzureGermany, MetricsQueryAudience.AzurePublicCloud);
+                return (ArmEnvironment.AzureGermany, MetricsClientAudience.AzurePublicCloud);
             }
 
             if (managementUrlParsed == ArmEnvironment.AzureGovernment.Endpoint)
             {
-                return (ArmEnvironment.AzureGovernment, MetricsQueryAudience.AzureGovernment);
+                return (ArmEnvironment.AzureGovernment, MetricsClientAudience.AzureGovernment);
             }
 
             string options = string.Join(", ",
@@ -187,7 +178,7 @@ protected override void InitializeCore(ReadOnlyDictionary<string, string> settin
                 }.Select(armEnvironment => $"\"{armEnvironment.Endpoint}\""));
             InitialiseErrors.Add($"Management url configuration is invalid, available options are {options}");
 
-            return (ArmEnvironment.AzurePublicCloud, MetricsQueryAudience.AzurePublicCloud);
+            return (ArmEnvironment.AzurePublicCloud, MetricsClientAudience.AzurePublicCloud);
         }
     }
 
@@ -229,7 +220,6 @@ public override async IAsyncEnumerable<QueueThroughput> GetThroughputPerDay(IBro
         while (currentDate <= endDate)
         {
             data.Add(currentDate, new QueueThroughput { TotalThroughput = 0, DateUTC = currentDate });
-
             currentDate = currentDate.AddDays(1);
         }
 
@@ -244,23 +234,67 @@ public override async IAsyncEnumerable<QueueThroughput> GetThroughputPerDay(IBro
         }
     }
 
+    async Task<MetricsClient> InitializeMetricsClient(CancellationToken cancellationToken = default)
+    {
+        if (resourceId is null || armClient is null || credential is null)
+        {
+            throw new InvalidOperationException("AzureQuery has not been initialized correctly.");
+        }
+
+        var serviceBusNamespaceResource = await armClient
+            .GetServiceBusNamespaceResource(resourceId).GetAsync(cancellationToken)
+                ?? throw new Exception($"Could not find ServiceBus with resource Id: \"{resourceId}\"");
+
+        // Determine the region of the namespace
+        var regionName = serviceBusNamespaceResource.Value.Data.Location.Name;
+
+        // Build the regional Azure Monitor Metrics endpoint from the audience
+        var metricsEndpoint = BuildMetricsEndpoint(metricsClientAudience, regionName);
+
+        // CreateNewOnMetadataUpdateAttribute the MetricsClient for this namespace
+        return new MetricsClient(
+            metricsEndpoint,
+            credential!,
+            new MetricsClientOptions
+            {
+                Audience = metricsClientAudience,
+                Transport = new HttpClientTransport(
+                    new HttpClient(new SocketsHttpHandler
+                    {
+                        PooledConnectionIdleTimeout = TimeSpan.FromMinutes(2)
+                    }))
+            });
+    }
+
+    static Uri BuildMetricsEndpoint(MetricsClientAudience audience, string regionName)
+    {
+        var region = regionName.ToLowerInvariant();
+        var builder = new UriBuilder(audience.ToString());
+        builder.Host = $"{region}.{builder.Host}";
+        return builder.Uri;
+    }
+
     async Task<IReadOnlyList<MetricValue>> GetMetrics(string queueName, DateOnly startTime, DateOnly endTime,
         CancellationToken cancellationToken = default)
     {
-        var response = await client!.QueryResourceAsync(resourceId,
-            new[] { "CompleteMessage" },
-            new MetricsQueryOptions
+        metricsClient ??= await InitializeMetricsClient(cancellationToken);
+
+        var response = await metricsClient.QueryResourcesAsync(
+            [resourceId!],
+            ["CompleteMessage"],
+            "Microsoft.ServiceBus/namespaces",
+            new MetricsQueryResourcesOptions
             {
                 Filter = $"EntityName eq '{queueName}'",
-                TimeRange = new QueryTimeRange(startTime.ToDateTime(TimeOnly.MinValue, DateTimeKind.Utc), endTime.ToDateTime(TimeOnly.MaxValue, DateTimeKind.Utc)),
+                TimeRange = new MetricsQueryTimeRange(startTime.ToDateTime(TimeOnly.MinValue, DateTimeKind.Utc), endTime.ToDateTime(TimeOnly.MaxValue, DateTimeKind.Utc)),
                 Granularity = TimeSpan.FromDays(1)
             },
             cancellationToken);
 
         var metricValues =
-            response.Value.Metrics.FirstOrDefault()?.TimeSeries.FirstOrDefault()?.Values ?? [];
+            response.Value.Values.FirstOrDefault()?.Metrics.FirstOrDefault()?.TimeSeries.FirstOrDefault()?.Values ?? [];
 
-        return metricValues;
+        return metricValues.AsReadOnly();
     }
 
     public override async IAsyncEnumerable<IBrokerQueue> GetQueueNames(
@@ -269,15 +303,14 @@ public override async IAsyncEnumerable<IBrokerQueue> GetQueueNames(
         var validNamespaces = await GetValidNamespaceNames(cancellationToken);
 
         SubscriptionResource? subscription = await armClient!.GetDefaultSubscriptionAsync(cancellationToken);
-        var namespaces =
-            subscription.GetServiceBusNamespacesAsync(cancellationToken);
+        var namespaces = subscription.GetServiceBusNamespacesAsync(cancellationToken);
 
-        await foreach (var serviceBusNamespaceResource in namespaces.WithCancellation(
-                           cancellationToken))
+        await foreach (var serviceBusNamespaceResource in namespaces.WithCancellation(cancellationToken))
         {
             if (validNamespaces.Contains(serviceBusNamespaceResource.Data.Name))
             {
                 resourceId = serviceBusNamespaceResource.Id;
+
                 await foreach (var queue in serviceBusNamespaceResource.GetServiceBusQueues()
                                    .WithCancellation(cancellationToken))
                 {
@@ -372,4 +405,4 @@ public static class AzureServiceBusSettings
         public static readonly string ManagementUrl = "ASB/ManagementUrl";
         public static readonly string ManagementUrlDescription = "Azure management URL";
     }
-}
+}

src/ServiceControl.Transports.ASBS/ServiceControl.Transports.ASBS.csproj

@@ -12,7 +12,7 @@
 
   <ItemGroup>
     <PackageReference Include="Azure.Identity" />
-    <PackageReference Include="Azure.Monitor.Query" />
+    <PackageReference Include="Azure.Monitor.Query.Metrics" />
     <PackageReference Include="Azure.ResourceManager.ServiceBus" />
     <PackageReference Include="DnsClient" />
     <PackageReference Include="NServiceBus.CustomChecks" />