IoT MQ Jumpstart Lightning demo

Link to video: https://youtu.be/0Hk4t17SxRc

Start with

• Regular Azure IoT Operations deployment using the codespace path
• Event Grid namespace
  • MQTT broker enabled, with at least one topic space with telemetry/#
  • Routing enabled to Event Hub
• Event Hub namespace
  • Two event hubs, one for direct connection, one for routing from Event Grid
  • Both capture enabled
• Storage account
  • ADLS Gen2 enabled
  • Three containers, one for Event Hub capture, one for Event Grid capture, and one for ADLS Gen2 direct connection

Open k9s in a bottom left terminal

k9s

Get the managed identity of the Arc extension and store it as a variable

principalId=$(az k8s-extension show --resource-group $RESOURCE_GROUP --cluster-name $CLUSTER_NAME --name mq --cluster-type connectedClusters --query identity.principalId -o tsv)

Assign RBAC permissions on Azure resources

az role assignment create --assignee $principalId --role "Azure Event Hubs Data Sender" --scope /subscriptions/<SUB_ID>/resourcegroups/<RG>/providers/Microsoft.EventHub/namespaces/<NAME>

az role assignment create --assignee $principalId --role "Azure Event Hubs Data Receiver" --scope /subscriptions/<SUB_ID>/resourcegroups/<RG>/providers/Microsoft.EventHub/namespaces/<NAME>

az role assignment create --assignee $principalId --role "EventGrid TopicSpaces Publisher" --scope /subscriptions/<SUB_ID>/resourceGroups/<RG>/providers/Microsoft.EventGrid/namespaces/<NAME>

az role assignment create --assignee $principalId --role "EventGrid TopicSpaces Subscriber" --scope /subscriptions/<SUB_ID>/resourceGroups/<RG>/providers/Microsoft.EventGrid/namespaces/<NAME>

az role assignment create --assignee $principalId --role "Storage Blob Data Contributor" --scope /subscriptions/<SUB_ID>/resourceGroups/<RG>/providers/Microsoft.Storage/storageAccounts/<NAME>

Deploy listener and three connectors

kubectl apply -f demo.yaml

Notice the lack of credentials in the YAML file. The credentials are managed by the Arc extension.

apiVersion: mq.iotoperations.azure.com/v1beta1
kind: BrokerListener
metadata:
  name: non-tls-listener
  namespace: azure-iot-operations
spec:
  brokerRef: broker
  serviceType: loadBalancer
  serviceName: quick-connect
  authenticationEnabled: false
  authorizationEnabled: false
  port: 1883
---
apiVersion: mq.iotoperations.azure.com/v1beta1
kind: MqttBridgeConnector
metadata:
  name: bridge
  namespace: azure-iot-operations
spec:
  image: 
    repository: mcr.microsoft.com/azureiotoperations/mqttbridge
    tag: 0.2.0-preview
    pullPolicy: IfNotPresent
  protocol: v5
  bridgeInstances: 1
  logLevel: debug
  remoteBrokerConnection:
    endpoint: <NAME>.<REGION>.ts.eventgrid.azure.net:8883
    tls:
      tlsEnabled: true
    authentication:
      systemAssignedManagedIdentity:
        audience: https://eventgrid.azure.net
  localBrokerConnection:
    endpoint: aio-mq-dmqtt-frontend:8883
    tls:
      tlsEnabled: true
      trustedCaCertificateConfigMap: aio-ca-trust-bundle-test-only
    authentication:
      kubernetes: {}
---
apiVersion: mq.iotoperations.azure.com/v1beta1
kind: MqttBridgeTopicMap
metadata:
  name: bridge-tp
  namespace: azure-iot-operations 
spec:
  mqttBridgeConnectorRef: bridge
  routes:
    - direction: local-to-remote
      name: publish
      source: telemetry
      target: telemetry
      qos: 1
---
apiVersion: mq.iotoperations.azure.com/v1beta1
kind: DataLakeConnector
metadata:
  name: dlv2
  namespace: azure-iot-operations
spec:
  protocol: v5
  image:
    repository: mcr.microsoft.com/azureiotoperations/datalake
    tag: 0.2.0-preview
    pullPolicy: IfNotPresent
  instances: 1
  logLevel: debug
  databaseFormat: delta
  target:
    datalakeStorage:
      endpoint: https://<NAME>.blob.core.windows.net
      authentication:
        systemAssignedManagedIdentity:
          audience: https://<NAME>.blob.core.windows.net
  localBrokerConnection:
    endpoint: aio-mq-dmqtt-frontend:8883
    tls:
      tlsEnabled: true
      trustedCaCertificateConfigMap: aio-ca-trust-bundle-test-only
    authentication:
      kubernetes: {}
---
apiVersion: mq.iotoperations.azure.com/v1beta1
kind: DataLakeConnectorTopicMap
metadata:
  name: dlv2-tp
  namespace: azure-iot-operations
spec:
  dataLakeConnectorRef: dlv2
  mapping:
    allowedLatencySecs: 1
    messagePayloadType: json
    maxMessagesPerBatch: 10
    clientId: id
    mqttSourceTopic: telemetry
    qos: 1
    table:
      tableName: telemetry
      schema:
      - name: temperature
        format: float64
        optional: false
        mapping: ambient.temperature
      - name: humidity
        format: int32
        optional: false
        mapping: ambient.humidity
---
apiVersion: mq.iotoperations.azure.com/v1beta1
kind: KafkaConnector
metadata:
  name: eh
  namespace: azure-iot-operations
spec:
  image:
    pullPolicy: IfNotPresent
    repository: mcr.microsoft.com/azureiotoperations/kafka
    tag: 0.2.0-preview
  instances: 1
  kafkaConnection:
    # Port 9093 is Event Hubs' Kakfa endpoint
    endpoint: <NAME>.servicebus.windows.net:9093
    tls:
      tlsEnabled: true
    authentication:
      enabled: true
      authType:
        systemAssignedManagedIdentity:
          audience: https://<NAME>.servicebus.windows.net
  localBrokerConnection:
    endpoint: aio-mq-dmqtt-frontend:8883
    tls:
      tlsEnabled: true
      trustedCaCertificateConfigMap: aio-ca-trust-bundle-test-only
    authentication:
      kubernetes: {}
---

apiVersion: mq.iotoperations.azure.com/v1beta1
kind: KafkaConnectorTopicMap
metadata:
  name: eh-tp
  namespace: azure-iot-operations
spec:
  kafkaConnectorRef: eh
  copyMqttProperties: true
  routes:
    - mqttToKafka:
        name: demo
        mqttTopic: telemetry
        kafkaTopic: telemetry
        kafkaAcks: one
        qos: 1

Open mqttui in bottom right terminal

mqttui telemetry

Send a message to the local broker

while read -r line; do echo "$line"; sleep 2; done < sim-sensor.json | mosquitto_pub -h localhost -t telemetry -l -d -q 1 -i sim-sensor

The simulated temperature and humidity data look like this sim-sensor.json

{"ambient": {"temperature": 21.42, "humidity": 35}}
{"ambient": {"temperature": 20.99, "humidity": 22}}
{"ambient": {"temperature": 21.389, "humidity": 38}}
{"ambient": {"temperature": 21.56, "humidity": 33}}
{"ambient": {"temperature": 22.927, "humidity": 23}}
{"ambient": {"temperature": 22.478, "humidity": 39}}
{"ambient": {"temperature": 22.654, "humidity": 21}}
...

Data flow diagram

graph LR
    s[Sensor]-->mq[IoT MQ];
    subgraph Kubernetes
      mq-->c[Cloud connectors];
    end
    c-->e[Event Hub];
    c-->b[Event Grid];
    subgraph Azure
      e-->d1[Storage container `eh-capture`];
      b-->d2[Storage container `eg-capture`];
      c-->adls[Data Lake];
    end

Loading

Cleanup

kubectl delete -f demo.yaml