return DeviceClient.Create(result.AssignedHub,
authentication,
new ITransportSettings[]
{
new AmqpTransportSettings(TransportType.Amqp_Tcp_Only)
{
PrefetchCount = 0,
AmqpConnectionPoolSettings = new AmqpConnectionPoolSettings()
{
Pooling = true,
}
}
}
);
My first attempt failed as I hadn’t configured “TransportType.Amqp_Tcp_Only” which would have allowed the AMQP implementation to fallback to other protocols which don’t support pooling.
Exception caused by not using TransportType.Amqp_Tcp_Only
I then deployed the updated code and ran my 1000 device stress test (note the different x axis scales)
Number of connections with pooling
This confirmed what I found in the Azure.AMQP source code
/// <summary>
/// The default size of the pool
/// </summary>
/// <remarks>
/// Allows up to 100,000 devices
/// </remarks>
/// private const uint DefaultPoolSize = 100;
For a PoC the DIY cache was ok but I wanted to replace it with something more robust like the .Net ObjectCache which is in the System.Runtime.Caching namespace.
I started by replacing the ConcurrentDictionary declaration
static readonly ConcurrentDictionary<string, DeviceClient> DeviceClients = new ConcurrentDictionary<string, DeviceClient>();
Then, where there were compiler errors I updated the method call.
// See if the device has already been provisioned or is being provisioned on another thread.
if (DeviceClients.Add(registrationId, deviceContext, cacheItemPolicy))
{
log.LogInformation("RegID:{registrationId} Device provisioning start", registrationId);
...
One difference I found was that ObjectCache throws an exception if the value is null. I was using a null value to indicate that the Device Provisioning Service(DPS) process had been initiated on another thread and was underway.
The Azure Key Vault is intended for securing sensitive information like connection strings so I added one to my resource group.
Azure Key Vault overview and basic metrics
I wrote a wrapper which resolves configuration settings based on the The Things Network(TTN) application identifier and port information in the uplink message payload. The resolve methods start by looking for configuration for the applicationId and port (separated by a – ), then the applicationId and then finally falling back to a default value. This functionality is used for AzureIoTHub connection strings, DPS IDScopes, DPS Enrollment Group Symmetric Keys, and is also used to format the cache keys.
public class ApplicationConfiguration
{
const string DpsGlobaDeviceEndpointDefault = "global.azure-devices-provisioning.net";
private IConfiguration Configuration;
public void Initialise( )
{
// Check that KeyVault URI is configured in environment variables. Not a lot we can do if it isn't....
if (Configuration == null)
{
string keyVaultUri = Environment.GetEnvironmentVariable("KeyVaultURI");
if (string.IsNullOrWhiteSpace(keyVaultUri))
{
throw new ApplicationException("KeyVaultURI environment variable not set");
}
// Load configuration from KeyVault
Configuration = new ConfigurationBuilder()
.AddEnvironmentVariables()
.AddAzureKeyVault(keyVaultUri)
.Build();
}
}
public string DpsGlobaDeviceEndpointResolve()
{
string globaDeviceEndpoint = Configuration.GetSection("DPSGlobaDeviceEndpoint").Value;
if (string.IsNullOrWhiteSpace(globaDeviceEndpoint))
{
globaDeviceEndpoint = DpsGlobaDeviceEndpointDefault;
}
return globaDeviceEndpoint;
}
public string ConnectionStringResolve(string applicationId, int port)
{
// Check to see if there is application + port specific configuration
string connectionString = Configuration.GetSection($"AzureIotHubConnectionString-{applicationId}-{port}").Value;
if (!string.IsNullOrWhiteSpace(connectionString))
{
return connectionString;
}
// Check to see if there is application specific configuration, otherwise run with default
connectionString = Configuration.GetSection($"AzureIotHubConnectionString-{applicationId}").Value;
if (!string.IsNullOrWhiteSpace(connectionString))
{
return connectionString;
}
// get the default as not a specialised configuration
connectionString = Configuration.GetSection("AzureIotHubConnectionStringDefault").Value;
return connectionString;
}
public string DpsIdScopeResolve(string applicationId, int port)
{
// Check to see if there is application + port specific configuration
string idScope = Configuration.GetSection($"DPSIDScope-{applicationId}-{port}").Value;
if (!string.IsNullOrWhiteSpace(idScope))
{
return idScope;
}
// Check to see if there is application specific configuration, otherwise run with default
idScope = Configuration.GetSection($"DPSIDScope-{applicationId}").Value;
if (!string.IsNullOrWhiteSpace(idScope))
{
return idScope;
}
// get the default as not a specialised configuration
idScope = Configuration.GetSection("DPSIDScopeDefault").Value;
if (string.IsNullOrWhiteSpace(idScope))
{
throw new ApplicationException($"DPSIDScope configuration invalid");
}
return idScope;
}
In the Azure Key Vault “Access Policies” I configured an “Application Access Policy” so my Azure TTNAzureIoTHubMessageV2Processor function identity could retrieve secrets.
Azure Key Vault Secrets
I kept on making typos in the secret names and types which was frustrating.
Azure Key Vault secret
While debugging in Visual Studio you may need to configure the Azure Identity so the application can access the Azure Key Vault.
The next “learning” was that if you forget to enable “always on” the caching won’t work and your application will call the DPS way more often than expected.
Azure Application “always on configuration
The next “learning” was if your soak test sends 24000 messages it will start to fail just after you go out to get a coffee because of the 8000 msgs/day limit on the free version of IoT Hub.
Azure IoT Hub Free tier 8000 messages/day limit
After these “learnings” the application appeared to be working and every so often a message would briefly appear in Azure Storage Explorer queue view.
Azure storage explorer view of uplink messages queue
The console test application simulated 1000 devices sending 24 messages every so often and took roughly 8 hours to complete.
Message generator finished
In the Azure IoT Hub telemetry 24000 messages had been received after roughly 8 hours confirming the test rig was working as expected.
The notch was another “learning”, if you go and do some gardening then after roughly 40 minutes of inactivity your desktop PC will go into power save mode and the test client will stop sending messages.
The caching of settings appeared to be work as there were only a couple of requests to my Azure Key Vault where sensitive information like connection strings, symmetric keys etc. are stored.
Memory consumption did look to bad and topped out at roughly 120M.
In the application logging you can see the 1000 calls to DPS at the beginning (the yellow dependency events) then the regular processing of messages.
Application Insights logging
Even with the “learnings” the testing went pretty well overall. I do need to run the test rig for longer and with even more simulated devices.
In the last couple of posts I had been building an Azure Function with a QueueTrigger to process the uplink messages. The function used custom bindings so that the CloudQueueMessage could be accessed, and load the Azure Storage account plus queue name from configuration. I’m still using classes generated by JSON2CSharp (with minimal modifications) for deserialising the payloads with JSON.Net.
The message processor Azure Function uses a ConcurrentCollection to store AzureDeviceClient objects constructed using the information returned by the Azure Device Provisioning Service(DPS). This is so the DPS doesn’t have to be called for the connection details for every message.(When the Azure function is restarted the dictionary of DeviceClient objects has to be repopulated). If there is a backlog of messages the message processor can process more than a dozen messages concurrently so the telemetry events displayed in an application like Azure IoT Central can arrive out of order.
The solution uses DPS Group Enrollment with Symmetric Key Attestation so Azure IoT Hub devices can be “automagically” created when a message from a new device is processed. The processing code is multi-thread and relies on many error conditions being handled by the Azure Function retry mechanism. After a number of failed retries the messages are moved to a poison queue. Azure Storage Explorer is a good tool for viewing payloads and moving poison messages back to the processing queue.
public static class UplinkMessageProcessor
{
static readonly ConcurrentDictionary<string, DeviceClient> DeviceClients = new ConcurrentDictionary<string, DeviceClient>();
[FunctionName("UplinkMessageProcessor")]
public static async Task Run(
[QueueTrigger("%UplinkQueueName%", Connection = "AzureStorageConnectionString")]
CloudQueueMessage cloudQueueMessage, // Used to get CloudQueueMessage.Id for logging
Microsoft.Azure.WebJobs.ExecutionContext context,
ILogger log)
{
PayloadV5 payloadObect;
DeviceClient deviceClient = null;
DeviceProvisioningServiceSettings deviceProvisioningServiceConfig;
string environmentName = Environment.GetEnvironmentVariable("ENVIRONMENT");
// Load configuration for DPS. Refactor approach and store securely...
var configuration = new ConfigurationBuilder()
.SetBasePath(context.FunctionAppDirectory)
.AddJsonFile($"appsettings.json")
.AddJsonFile($"appsettings.{environmentName}.json")
.AddEnvironmentVariables()
.Build();
// Load configuration for DPS. Refactor approach and store securely...
try
{
deviceProvisioningServiceConfig = (DeviceProvisioningServiceSettings)configuration.GetSection("DeviceProvisioningService").Get<DeviceProvisioningServiceSettings>(); ;
}
catch (Exception ex)
{
log.LogError(ex, $"Configuration loading failed");
throw;
}
// Deserialise uplink message from Azure storage queue
try
{
payloadObect = JsonConvert.DeserializeObject<PayloadV5>(cloudQueueMessage.AsString);
}
catch (Exception ex)
{
log.LogError(ex, $"MessageID:{cloudQueueMessage.Id} uplink message deserialisation failed");
throw;
}
// Extract the device ID as it's used lots of places
string registrationID = payloadObect.hardware_serial;
// Construct the prefix used in all the logging
string messagePrefix = $"MessageID: {cloudQueueMessage.Id} DeviceID:{registrationID} Counter:{payloadObect.counter} Application ID:{payloadObect.app_id}";
log.LogInformation($"{messagePrefix} Uplink message device processing start");
// See if the device has already been provisioned
if (DeviceClients.TryAdd(registrationID, deviceClient))
{
log.LogInformation($"{messagePrefix} Device provisioning start");
string enrollmentGroupSymmetricKey = deviceProvisioningServiceConfig.EnrollmentGroupSymmetricKeyDefault;
// figure out if custom mapping for TTN applicationID
if (deviceProvisioningServiceConfig.ApplicationEnrollmentGroupMapping != null)
{
deviceProvisioningServiceConfig.ApplicationEnrollmentGroupMapping.GetValueOrDefault(payloadObect.app_id, deviceProvisioningServiceConfig.EnrollmentGroupSymmetricKeyDefault);
}
// Do DPS magic first time device seen
await DeviceRegistration(log, messagePrefix, deviceProvisioningServiceConfig.GlobalDeviceEndpoint, deviceProvisioningServiceConfig.ScopeID, enrollmentGroupSymmetricKey, registrationID);
}
// Wait for the Device Provisioning Service to complete on this or other thread
log.LogInformation($"{messagePrefix} Device provisioning polling start");
if (!DeviceClients.TryGetValue(registrationID, out deviceClient))
{
log.LogError($"{messagePrefix} Device provisioning polling TryGet before while failed");
throw new ApplicationException($"{messagePrefix} Device provisioning polling TryGet before while failed");
}
while (deviceClient == null)
{
log.LogInformation($"{messagePrefix} provisioning polling delay");
await Task.Delay(deviceProvisioningServiceConfig.DeviceProvisioningPollingDelay);
if (!DeviceClients.TryGetValue(registrationID, out deviceClient))
{
log.LogError($"{messagePrefix} Device provisioning polling TryGet while loop failed");
throw new ApplicationException($"{messagePrefix} Device provisioning polling TryGet while loopfailed");
}
}
// Assemble the JSON payload to send to Azure IoT Hub/Central.
log.LogInformation($"{messagePrefix} Payload assembly start");
JObject telemetryEvent = new JObject();
try
{
JObject payloadFields = (JObject)payloadObect.payload_fields;
telemetryEvent.Add("HardwareSerial", payloadObect.hardware_serial);
telemetryEvent.Add("Retry", payloadObect.is_retry);
telemetryEvent.Add("Counter", payloadObect.counter);
telemetryEvent.Add("DeviceID", payloadObect.dev_id);
telemetryEvent.Add("ApplicationID", payloadObect.app_id);
telemetryEvent.Add("Port", payloadObect.port);
telemetryEvent.Add("PayloadRaw", payloadObect.payload_raw);
telemetryEvent.Add("ReceivedAt", payloadObect.metadata.time);
// If the payload has been unpacked in TTN backend add fields to telemetry event payload
if (payloadFields != null)
{
foreach (JProperty child in payloadFields.Children())
{
EnumerateChildren(telemetryEvent, child);
}
}
}
catch (Exception ex)
{
if (DeviceClients.TryRemove(registrationID, out deviceClient))
{
log.LogWarning($"{messagePrefix} TryRemove payload assembly failed");
}
log.LogError(ex, $"{messagePrefix} Payload assembly failed");
throw;
}
// Send the message to Azure IoT Hub/Azure IoT Central
log.LogInformation($"{messagePrefix} Payload SendEventAsync start");
try
{
using (Message ioTHubmessage = new Message(Encoding.ASCII.GetBytes(JsonConvert.SerializeObject(telemetryEvent))))
{
// Ensure the displayed time is the acquired time rather than the uploaded time. esp. importan for messages that end up in poison queue
ioTHubmessage.Properties.Add("iothub-creation-time-utc", payloadObect.metadata.time.ToString("s", CultureInfo.InvariantCulture));
await deviceClient.SendEventAsync(ioTHubmessage);
}
}
catch (Exception ex)
{
if (DeviceClients.TryRemove(registrationID, out deviceClient))
{
log.LogWarning($"{messagePrefix} TryRemove SendEventAsync failed");
}
log.LogError(ex, $"{messagePrefix} SendEventAsync failed");
throw;
}
log.LogInformation($"{messagePrefix} Uplink message device processing completed");
}
}
There is also support for using a specific GroupEnrollment based on the application_id in the uplink message payload.
In addition to the appsettings.json there is configuration for application insights, uplink message queue name and Azure Storage connection strings. The “Environment” setting is important as it specifies what appsettings.json file should be used if code is being debugged etc..
The deployed solution application consists of Azure IoTHub and DPS instances. There are two Azure functions, one for putting the messages from the TTN into a queue the other is for processing them. The Azure Functions are hosted in an Azure AppService plan.
Azure solution deployment
An Azure Storage account is used for the queue and Azure Function synchronisation information and Azure Application Insights is used to monitor the solution.
For development and testing being able to provision an individual device is really useful, though for Azure IoT Central it is not easy (especially with the deprecation of DPS-KeyGen). With an Azure IoT Hub device connection strings are available in the portal which is convenient but not terribly scalable.
Initially the enrollment group had no registration records so I ran my command-line application to generate group enrollment keys for one of my devices.
Device registration before running my command line application
Then I ran the command-line application with my scopeID, registrationID (LoRaWAN deviceEUI) and the device group enrollment key I had generated in the previous step.
Registering a device and sending a message to the my Azure IoT Hub
After running the command line application the device was visible in the enrollment group registration records.
Device registration after running my command line application
Provisioning a device with an individual enrollment has a different workflow. I had to run my command-line application with the RegistrationID, ScopeID, and one of the symmetric keys from the DPS individual enrollment device configuration.
DPS Individual enrollment configuration
A major downside to an individual enrollment is either the primary or the secondary symmetric key for the device has to be deployed on the device which could be problematic if the device has no secure storage.
With a group enrollment only the registration ID and the derived symmetric key have to be deployed on the device which is more secure.
Registering a device and sending a message to the my Azure IoT Hub
In Azure IoT Explorer I could see messages from both my group and individually enrolled devices arriving at my Azure IoT hub
After some initial issues I found DPS was quite reliable and surprisingly easy to configure. I did find the DPS ProvisioningDeviceClient.RegisterAsync method sometimes took several seconds to execute which may have some ramifications when my application is doing this on demand.
