Skip to content

metrics-collector - split D2C messages greater than limit #6904

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
LiamDavey wants to merge 9 commits into
base: main
Choose a base branch
from
+59 −17
Open

metrics-collector - split D2C messages greater than limit #6904

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
885559f
metrics-collector - split D2C messages before sending upstream if mes…
LiamDavey Feb 13, 2023
b8e880d
Merge branch 'Azure:main' into split-metrics
LiamDavey May 30, 2023
dc9ff45
Merge branch 'Azure:main' into split-metrics
LiamDavey Sep 11, 2023
f1e3100
Merge branch 'Azure:main' into split-metrics
LiamDavey Dec 14, 2023
73c741d
reduced metrics-collector max message size to prevent errors
LiamDavey Dec 14, 2023
a653189
Merge branch 'Azure:main' into split-metrics
LiamDavey Feb 18, 2024
46f0f31
Merge branch 'Azure:main' into split-metrics
LiamDavey Nov 20, 2024
8d618c8
reduce split msg size to fix still breaching limit rarely
LiamDavey Nov 20, 2024
b9a2459
Merge branch 'Azure:main' into split-metrics
LiamDavey Jul 1, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion edge-modules/metrics-collector/src/Constants.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,5 +13,6 @@ public static class Constants
public const string DefaultLogAnalyticsWorkspaceDomainPrefixOds = ".ods.opinsights.";
public const string DefaultLogAnalyticsWorkspaceDomainPrefixOms = ".oms.opinsights.";
public const string ProductInfo = "IoTEdgeMetricsCollectorModule";
public static readonly int MaxMessageSize = 225000;
}
}
}
73 changes: 57 additions & 16 deletions edge-modules/metrics-collector/src/IotHubUpload/IotHubUpload.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,25 +31,14 @@ public async Task<bool> PublishAsync(IEnumerable<Metric> metrics, CancellationTo
{
Preconditions.CheckNotNull(metrics, nameof(metrics));
IEnumerable<ExportMetric> outputMetrics = metrics.Select(m => new ExportMetric(m));
List<IEnumerable<ExportMetric>> splitMetrics = new List<IEnumerable<ExportMetric>> { outputMetrics };
IEnumerable<Message> messagesToSend = BatchAndBuildMessages(splitMetrics);

string outputString = JsonConvert.SerializeObject(outputMetrics);

if (Settings.Current.TransformForIoTCentral)
{
outputString = Transform(outputMetrics);
}

byte[] metricsData = Encoding.UTF8.GetBytes(outputString);
if (Settings.Current.CompressForUpload)
foreach (Message metricsMessage in messagesToSend)
{
metricsData = Compression.CompressToGzip(metricsData);
await this.ModuleClientWrapper.SendMessageAsync("metricOutput", metricsMessage);
}

Message metricsMessage = new Message(metricsData);
metricsMessage.Properties[IdentifierPropertyName] = Constants.IoTUploadMessageIdentifier;

await this.ModuleClientWrapper.SendMessageAsync("metricOutput", metricsMessage);

return true;
}
catch (Exception e)
Expand All @@ -59,6 +48,58 @@ public async Task<bool> PublishAsync(IEnumerable<Metric> metrics, CancellationTo
}
}

private byte[] serializeMetrics(IEnumerable<ExportMetric> outputMetrics)
{
string outputString = JsonConvert.SerializeObject(outputMetrics);

if (Settings.Current.TransformForIoTCentral)
{
outputString = Transform(outputMetrics);
}

byte[] metricsData = Encoding.UTF8.GetBytes(outputString);

if (Settings.Current.CompressForUpload)
{
metricsData = Compression.CompressToGzip(metricsData);
}

return metricsData;
}

private IEnumerable<Message> BatchAndBuildMessages(List<IEnumerable<ExportMetric>> splitMetrics)
{

List<Message> messageList = new List<Message>();
byte[] metricsData = serializeMetrics(splitMetrics.Last());
if (metricsData.Length > Constants.MaxMessageSize)
{
LoggerUtil.Writer.LogInformation($"IoT message is {metricsData.Length} bytes, splitting messages...");
List<IEnumerable<ExportMetric>> splitAgainMetrics = new List<IEnumerable<ExportMetric>>();
foreach (IEnumerable<ExportMetric> metrics in splitMetrics)
{
ExportMetric[] metricsarray = metrics.ToArray();
ExportMetric[] firstArray = metricsarray.Take(metricsarray.Length / 2).ToArray();
splitAgainMetrics.Add(firstArray);
ExportMetric[] secondArray = metricsarray.Skip(metricsarray.Length / 2).ToArray();
splitAgainMetrics.Add(secondArray);
}

return BatchAndBuildMessages(splitAgainMetrics);
}
else
{
foreach (IEnumerable<ExportMetric> metrics in splitMetrics)
{
metricsData = serializeMetrics(metrics);
Copy link
Member

@damonbarry damonbarry Sep 3, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A list of metrics that is close to but not exceeding the size limit will always be serialized twice, including possibly transforming the data for IoT Central and compressing the data. If the list of metrics exceeds the size limit, then the entire list will be serialized, followed by half the list, etc., until all sub-lists are within the size, then each sublist is serialized again. So we're going from O(n) to O(n log n) performance. This will likely be a noticeable (and possibly unacceptable) performance dip for many users.

Message metricsMessage = new Message(metricsData);
metricsMessage.Properties[IdentifierPropertyName] = Constants.IoTUploadMessageIdentifier;
messageList.Add(metricsMessage);
}
return messageList;
}
}

private string Transform(IEnumerable<ExportMetric> metrics)
{
DateTime timeGeneratedUtc = DateTime.MaxValue;
Expand Down Expand Up @@ -267,4 +308,4 @@ public ExportMetric(Metric baseMetric)
}
}
}
}
}