IoT at Scale: Device Management, Data Ingestion, and Production Architecture

// MQTT Connect Packet Structure
// Fixed Header (2 bytes minimum)
// Variable Header:
//   - Protocol Name (4 bytes): "MQTT"
//   - Protocol Level (1 byte): 4 (MQTT 3.1.1) or 5 (MQTT 5.0)
//   - Connect Flags (1 byte)
//   - Keep Alive (2 bytes)
// Payload:
//   - Client Identifier
//   - Will Topic (optional)
//   - Will Message (optional)
//   - Username (optional)
//   - Password (optional)

# MQTT QoS Implementation with Paho

import paho.mqtt.client as mqtt

# QoS 0 - At most once (fire and forget)
client.publish("sensors/temperature", payload, qos=0)

# QoS 1 - At least once (acknowledged)
client.publish("sensors/temperature", payload, qos=1)
# Broker acknowledges with PUBACK

# QoS 2 - Exactly once (four-way handshake)
client.publish("sensors/temperature", payload, qos=2)
# Broker: PUBREC -> Client: PUBREL -> Broker: PUBCOMP

# Callback for QoS 1/2
def on_publish(client, userdata, mid):
    print(f"Message {mid} published")

# MQTT 5.0 Enhanced Features

# 1. User Properties
client.publish(
    "sensors/temperature",
    payload='{"value": 25.5}',
    qos=1,
    properties={
        'User-Property': [('device-id', 'sensor-001'), ('location', 'warehouse-a')]
    }
)

# 2. Message Expiry
client.publish(
    "sensors/commands",
    payload=command,
    message_expiry_interval=300  # 5 minutes
)

# 3. Topic Aliases (reduce bandwidth)
# Client sends: topic alias 1 -> maps to "sensors/+/temperature"
# Subsequent messages just send alias

# 4. Negative Acknowledgment (NACK)
# Broker can reject messages with reason codes

# Device Certificate Generation
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa

def generate_device_certificate(device_id, ca_key, ca_cert):
    """Generate device certificate signed by CA"""
    
    # Generate device key pair
    device_key = rsa.generate_private_key(
        public_exponent=65537,
        key_size=2048
    )
    
    # Create CSR
    csr = x509.CertificateSigningRequestBuilder().subject_name(
        x509.Name([
            x509.NameAttribute(x509.oid.NameOID.COMMON_NAME, device_id),
            x509.NameAttribute(x509.oid.NameOID.ORGANIZATION_NAME, "Acme IoT"),
        ])
    ).sign(device_key, hashes.SHA256())
    
    # Sign with CA
    device_cert = x509.CertificateBuilder().subject_name(
        csr.subject
    ).issuer_name(
        ca_cert.subject
    ).public_key(
        csr.public_key()
    ).serial_number(
        x509.random_serial_number()
    ).not_valid_before(
        datetime.datetime.utcnow()
    ).not_valid_after(
        datetime.datetime.utcnow() + datetime.timedelta(days=365)
    ).add_extension(
        x509.SubjectAlternativeName([
            x509.DNSName(f"{device_id}.devices.example.com"),
        ]),
        critical=False,
    ).sign(ca_key, hashes.SHA256())
    
    return device_cert, device_key

# JIT Provisioning with AWS IoT Core
import boto3
import json

iot_client = boto3.client('iot')

def handle_device_connect(event):
    """Called when device first connects"""
    
    # Extract certificate info
    cert = event['certificate']
    device_id = cert.get('commonName')
    
    # Check if device already provisioned
    try:
        iot_client.describe_thing(thingName=device_id)
        return  # Already exists
    except iot_client.exceptions.ResourceNotFoundException:
        pass
    
    # Provision device dynamically
    # 1. Create thing
    thing = iot_client.create_thing(
        thingName=device_id,
        thingTypeName='SensorDevice',
        attributePayload={
            'attributes': {
                'device_type': 'temperature_sensor',
                'firmware_version': '1.0.0',
                'location': cert.get('attributes', {}).get('location', 'unknown')
            }
        }
    )
    
    # 2. Attach certificate to thing
    iot_client.attach_thing_principal(
        thingName=device_id,
        principal=cert['certificateArn']
    )
    
    # 3. Apply default policy
    iot_client.attach_principal_policy(
        policyName='DefaultDevicePolicy',
        principal=cert['certificateArn']
    )
    
    print(f"Provisioned new device: {device_id}")

# Kafka Consumer for IoT Data
from kafka import KafkaConsumer
import json
import time

class IoTConsumer:
    def __init__(self, brokers, topic_group):
        self.consumer = KafkaConsumer(
            'iot.sensors.*',  # Wildcard topic
            bootstrap_servers=brokers,
            group_id=topic_group,
            auto_offset_reset='latest',
            enable_auto_commit=False,
            max_poll_records=1000,
            fetch_max_bytes=10*1024*1024,  # 10MB
            max_in_flight_requests_per_connection=5
        )
        
    def process_messages(self):
        """Process messages with backpressure"""
        
        while True:
            # Poll with timeout
            messages = self.consumer.poll(timeout_ms=1000)
            
            if not messages:
                continue
                
            batch = []
            
            for topic_partition, records in messages.items():
                for record in records:
                    # Parse sensor data
                    data = json.loads(record.value)
                    batch.append(self.process_record(data))
            
            # Batch write to database
            if batch:
                self.write_batch(batch)
                
            # Commit offsets
            self.consumer.commit()
    
    def process_record(self, data):
        """Process individual sensor reading"""
        return {
            'device_id': data['device_id'],
            'timestamp': data['timestamp'],
            'sensor_type': data['sensor_type'],
            'value': data['value'],
            'quality': self.check_quality(data)
        }

# TimescaleDB Schema for IoT Data

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, Float, String, DateTime, Index
from sqlalchemy.dialects.postgresql import JSONB
from datetime import datetime

Base = declarative_base()

class SensorReading(Base):
    __tablename__ = 'sensor_readings'
    
    id = Column(Integer, primary_key=True)
    device_id = Column(String(50), nullable=False)
    timestamp = Column(DateTime, nullable=False, default=datetime.utcnow)
    sensor_type = Column(String(30), nullable=False)
    value = Column(Float, nullable=False)
    quality = Column(String(10), default='good')  # good, suspect, bad
    metadata = Column(JSONB, default={})
    
    # Hypertable configuration (TimescaleDB)
    __table_args__ = (
        Index('idx_device_time', 'device_id', 'timestamp'),
        # Convert to hypertable
        # SELECT create_hypertable('sensor_readings', 'timestamp')
    )

# Query with time aggregation
QUERY_HOURLY_AVG = """
    SELECT 
        device_id,
        time_bucket('1 hour', timestamp) AS bucket,
        AVG(value) as avg_value,
        MIN(value) as min_value,
        MAX(value) as max_value,
        COUNT(*) as sample_count
    FROM sensor_readings
    WHERE timestamp >= :start_time AND timestamp < :end_time
    AND device_id = :device_id
    GROUP BY device_id, bucket
    ORDER BY bucket
"""

# Apache Flink for Real-time IoT Analytics
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.functions import MapFunction, KeyedProcessFunction
from pyflink.datastream import TimeCharacteristic

env = StreamExecutionEnvironment.get_execution_environment()
env.set_stream_time_characteristic(TimeCharacteristic.EventTime)

class SensorAnalytics(KeyedProcessFunction):
    """Calculate rolling statistics per device"""
    
    def __init__(self, window_size_seconds=300):
        self.window_size = window_size_seconds * 1000
        
    def process_element(self, value, ctx):
        # Get current watermark
        current_watermark = ctx.timer_service().current_watermark()
        
        # Register timer for window end
        event_time = value.timestamp
        window_end = (event_time // self.window_size + 1) * self.window_size
        
        ctx.timer_service().register_event_time_timer(window_end)
        
    def on_timer(self, timestamp, ctx):
        # Calculate aggregated statistics
        readings = ctx.get_state()
        
        if readings:
            values = list(readings.get())
            result = {
                'device_id': ctx.get_current_key(),
                'window_start': timestamp - self.window_size,
                'window_end': timestamp,
                'avg': sum(values) / len(values),
                'min': min(values),
                'max': max(values),
                'count': len(values)
            }
            yield result
            
            # Clear state
            readings.clear()

{
  "device": {
    "device_id": "sensor-001",
    "device_type": "temperature_sensor",
    "firmware_version": "2.1.0",
    "hardware_version": "v3.2",
    "manufacturer": "Acme Sensors",
    "model": "TMP-5000",
    "status": "active",
    "tags": ["warehouse", "zone-a", "critical"],
    "properties": {
      "battery_level": 85,
      "signal_strength": -45,
      "last_seen": "2026-02-22T10:30:00Z"
    },
    "shadow": {
      "desired": {
        "reporting_interval": 60,
        "temperature_threshold": 25.0
      },
      "reported": {
        "reporting_interval": 60,
        "firmware_version": "2.1.0"
      },
      "delta": {}
    }
  }
}

# OTA Update Management
import hashlib
import boto3

class OTAManager:
    def __init__(self, bucket):
        self.s3 = boto3.client('s3')
        self.bucket = bucket
        
    def upload_firmware(self, device_type, version, file_path):
        """Upload firmware file"""
        
        with open(file_path, 'rb') as f:
            firmware_data = f.read()
        
        # Calculate checksums
        sha256 = hashlib.sha256(firmware_data).hexdigest()
        md5 = hashlib.md5(firmware_data).hexdigest()
        
        key = f"firmware/{device_type}/{version}/firmware.bin"
        
        self.s3.upload_file(
            file_path,
            self.bucket,
            key,
            ExtraArgs={
                'Metadata': {
                    'sha256': sha256,
                    'md5': md5,
                    'version': version
                }
            }
        )
        
        return {
            'url': f"s3://{self.bucket}/{key}",
            'sha256': sha256,
            'version': version,
            'size': len(firmware_data)
        }
    
    def create_fleet_campaign(self, devices, firmware_info):
        """Create OTA campaign for device group"""
        
        campaign = {
            'name': f"Update {firmware_info['version']}",
            'firmware': firmware_info,
            'target_devices': devices,
            'rollout_strategy': {
                'type': 'gradual',
                'initial_percentage': 5,
                'increment_percentage': 10,
                'interval_minutes': 60,
                'pause_on_failure': True
            }
        }
        
        return campaign

# AWS IoT Device Shadow
import json
import boto3
from datetime import datetime

iot_data = boto3.client('iot-data')

def update_device_shadow(thing_name, desired_state, reported_state):
    """Update device shadow with desired and reported state"""
    
    payload = {
        'state': {
            'desired': desired_state,
            'reported': reported_state
        },
        'metadata': {
            'desired': {k: {'timestamp': int(datetime.utcnow().timestamp())} 
                       for k in desired_state},
            'reported': {k: {'timestamp': int(datetime.utcnow().timestamp())} 
                        for k in reported_state}
        },
        'version': None  # Let service handle versioning
    }
    
    response = iot_data.update_thing_shadow(
        thingName=thing_name,
        payload=json.dumps(payload)
    )
    
    return json.loads(response['payload'].read())

def get_device_shadow(thing_name):
    """Get current device shadow"""
    
    response = iot_data.get_thing_shadow(thingName=thing_name)
    return json.loads(response['payload'].read())

# Mutual TLS (mTLS) Configuration
# mosquitto.conf

# CA and certificate configuration
cafile /etc/mosquitto/certs/ca.crt
certfile /etc/mosquitto/certs/server.crt
keyfile /etc/mosquitto/certs/server.key

# Require client certificates
require_certificate true
use_identity_as_username true

# TLS version requirements
tls_version tlsv1.3

# Cipher suites
ciphers ECDHE-RSA-AES256-GCM-SHA384:ECDHE-RSA-AES128-GCM-SHA256

# AWS IoT Policy
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "iot:Connect"
      ],
      "Resource": [
        "arn:aws:iot:us-east-1:123456789012:client/${iot:Connection.Thing.ThingName}"
      ]
    },
    {
      "Effect": "Allow",
      "Action": [
        "iot:Publish",
        "iot:Subscribe"
      ],
      "Resource": [
        "arn:aws:iot:us-east-1:123456789012:topic/${iot:Connection.Thing.ThingName}/telemetry",
        "arn:aws:iot:us-east-1:123456789012:topic/${iot:Connection.Thing.ThingName}/commands"
      ]
    },
    {
      "Effect": "Allow",
      "Action": ["iot:Receive"],
      "Resource": [
        "arn:aws:iot:us-east-1:123456789012:topic/device/${iot:Connection.Thing.ThingName}/*"
      ]
    }
  ]
}

# Kubernetes IoT Platform Components
apiVersion: v1
kind: ConfigMap
metadata:
  name: iot-platform-config
data:
  MQTT_PORT: "1883"
  MQTT_TLS_PORT: "8883"
  KAFKA_BROKERS: "kafka:9092"
  INFLUXDB_URL: "http://influxdb:8086"
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: mqtt-broker
spec:
  replicas: 3
  selector:
    matchLabels:
      app: mqtt-broker
  template:
    spec:
      containers:
      - name: emqx
        image: emqx/emqx:5.0
        ports:
        - containerPort: 1883
        - containerPort: 8883
        - containerPort: 8083
        envFrom:
        - configMapRef:
            name: iot-platform-config
        resources:
          limits:
            memory: "2Gi"
            cpu: "2000m"
---
apiVersion: v1
kind: Service
metadata:
  name: mqtt-broker
spec:
  type: ClusterIP
  ports:
  - port: 1883
    targetPort: 1883
    protocol: TCP
  selector:
    app: mqtt-broker

# Horizontal Scaling with Message Partitioning

# Kafka topic partitioning by device_id
# Partition key: device_id -> hash % num_partitions

# For 1M devices, 100 partitions = 10K devices per partition
# Consumer group with 100 consumers = each handles 10 partitions

from kafka import TopicPartition

# Assign partitions to consumers
def assign_partitions(consumer_group, num_partitions):
    partitions = [TopicPartition('iot.sensors', i) 
                  for i in range(num_partitions)]
    
    # Rebalance strategy: sticky assignor
    # Keeps partitions assigned to same consumer after rebalance
    
    return partitions

# Scaling trigger based on message lag
def check_scaling_needed(consumer_group):
    consumer = KafkaConsumer(
        'iot.sensors',
        group_id=consumer_group,
        metadata.max.age.ms=5000
    )
    
    # Get lag per partition
    end_offsets = consumer.end_offsets(consumer.assignment())
    committed = consumer.committed(consumer.assignment())
    
    lag = {tp: end_offsets[tp] - committed[tp] 
           for tp in consumer.assignment()}
    
    # If any partition > 10000 messages behind, scale
    return any(v > 10000 for v in lag.values())