# 13 · Edge IoT Telemetry Pipeline

**Format:** AWS CDK (TypeScript) + Python Lambda + Node.js simulator · **Est. cost:** ~$1–5/day while the simulator runs (Kinesis on-demand + IoT messages); near-zero idle

A connected-fleet backend: vehicles stream telemetry over MQTT, the **hot path** reacts in seconds, and the
**cold path** keeps cheap queryable history.

```
Trucks (MQTT + X.509) ─▶ AWS IoT Core ─▶ topic rule  SELECT *, topic(3) AS device_id ... 'fleet/+/+/telemetry'
                                          ├─ Kinesis (on-demand) ─▶ Lambda ─┬─ DynamoDB  "latest state per device"
                                          │                                 ├─ SNS + MQTT alerts/<device>  (overheat, low battery)
                                          │                                 └─ EMF metrics ─▶ CloudWatch dashboard "iot-fleet-live"
                                          └─ Firehose ─▶ S3 (gzip, dt=YYYY-MM-DD) ─▶ Athena  iot_telemetry.readings
```
> Why not Timestream? Amazon Timestream for LiveAnalytics no longer accepts new customers, so this uses
> DynamoDB for "now" and S3 + Athena for "history". It's cheaper and scales the same.

## Deploy
```bash
npm install
npx cdk deploy -c alertEmail=you@example.com -c tempMaxC=85
```

## Simulate a fleet
```bash
npm i @aws-sdk/client-iot @aws-sdk/client-iot-data-plane
node simulator/simulate.mjs --devices 25 --interval 2
```
Open the **iot-fleet-live** dashboard. Overheating spikes (about 1% of readings) trigger emails and `alerts/<device>` MQTT messages.

## Query history
```sql
SELECT device_id, max(temperature_c) AS peak_c, min(battery_pct) AS min_batt, count(*) AS readings
FROM iot_telemetry.readings
WHERE dt = date_format(current_date, '%Y-%m-%d')
GROUP BY device_id ORDER BY peak_c DESC LIMIT 10;
```

## Onboard a real device
```bash
aws iot create-thing --thing-name truck-101
aws iot create-keys-and-certificate --set-as-active --certificate-pem-outfile cert.pem --private-key-outfile key.pem
aws iot attach-policy --policy-name fleet-device-policy --target <certificateArn>
aws iot attach-thing-principal --thing-name truck-101 --principal <certificateArn>
```
The policy only lets `truck-101` publish to `fleet/*/truck-101/telemetry` (device identity = authorization).

## Clean up
```bash
npx cdk destroy
```

#!/usr/bin/env node
import * as cdk from 'aws-cdk-lib';
import { IotTelemetryStack } from '../lib/iot-telemetry-stack';

const app = new cdk.App();
new IotTelemetryStack(app, 'IotTelemetry', {
  env: { account: process.env.CDK_DEFAULT_ACCOUNT, region: process.env.CDK_DEFAULT_REGION },
  alertEmail: app.node.tryGetContext('alertEmail'),
  tempMaxC: Number(app.node.tryGetContext('tempMaxC') ?? 85),
});

import * as path from 'path';
import * as cdk from 'aws-cdk-lib';
import { Construct } from 'constructs';
import * as cloudwatch from 'aws-cdk-lib/aws-cloudwatch';
import * as dynamodb from 'aws-cdk-lib/aws-dynamodb';
import * as firehose from 'aws-cdk-lib/aws-kinesisfirehose';
import * as glue from 'aws-cdk-lib/aws-glue';
import * as iam from 'aws-cdk-lib/aws-iam';
import * as iot from 'aws-cdk-lib/aws-iot';
import * as kinesis from 'aws-cdk-lib/aws-kinesis';
import * as lambda from 'aws-cdk-lib/aws-lambda';
import * as sources from 'aws-cdk-lib/aws-lambda-event-sources';
import * as logs from 'aws-cdk-lib/aws-logs';
import * as s3 from 'aws-cdk-lib/aws-s3';
import * as sns from 'aws-cdk-lib/aws-sns';
import * as subs from 'aws-cdk-lib/aws-sns-subscriptions';

export interface IotTelemetryStackProps extends cdk.StackProps {
  alertEmail?: string;
  tempMaxC: number;
}

/**
 * Devices -> AWS IoT Core (MQTT, X.509) -> topic rule fan-out:
 *   hot path : Kinesis -> Lambda -> DynamoDB "latest state" + anomaly alerts + live metrics
 *   cold path: Firehose -> S3 (partitioned by day) -> Athena
 */
export class IotTelemetryStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props: IotTelemetryStackProps) {
    super(scope, id, props);

    // ---------------------------------------------------------------- storage
    const state = new dynamodb.TableV2(this, 'DeviceState', {
      tableName: 'iot-device-state',
      partitionKey: { name: 'device_id', type: dynamodb.AttributeType.STRING },
      billing: dynamodb.Billing.onDemand(),
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });
    const archive = new s3.Bucket(this, 'TelemetryArchive', {
      encryption: s3.BucketEncryption.S3_MANAGED,
      blockPublicAccess: s3.BlockPublicAccess.BLOCK_ALL,
      enforceSSL: true,
      lifecycleRules: [{ transitions: [{ storageClass: s3.StorageClass.INTELLIGENT_TIERING, transitionAfter: cdk.Duration.days(30) }] }],
      autoDeleteObjects: true,
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    const alerts = new sns.Topic(this, 'DeviceAlerts', { displayName: 'IoT device alerts' });
    if (props.alertEmail) alerts.addSubscription(new subs.EmailSubscription(props.alertEmail));

    // ---------------------------------------------------------------- hot path
    const stream = new kinesis.Stream(this, 'TelemetryStream', {
      streamMode: kinesis.StreamMode.ON_DEMAND,
      retentionPeriod: cdk.Duration.hours(24),
      encryption: kinesis.StreamEncryption.MANAGED,
    });

    const processor = new lambda.Function(this, 'Processor', {
      runtime: lambda.Runtime.PYTHON_3_12,
      architecture: lambda.Architecture.ARM_64,
      handler: 'processor.handler',
      code: lambda.Code.fromAsset(path.join(__dirname, '..', 'lambda')),
      timeout: cdk.Duration.seconds(60),
      memorySize: 256,
      environment: {
        STATE_TABLE: state.tableName,
        ALERT_TOPIC_ARN: alerts.topicArn,
        TEMP_MAX_C: String(props.tempMaxC),
        BATTERY_MIN_PCT: '15',
      },
      logGroup: new logs.LogGroup(this, 'ProcessorLogs', {
        retention: logs.RetentionDays.TWO_WEEKS,
        removalPolicy: cdk.RemovalPolicy.DESTROY,
      }),
    });
    processor.addEventSource(
      new sources.KinesisEventSource(stream, {
        startingPosition: lambda.StartingPosition.LATEST,
        batchSize: 500,
        maxBatchingWindow: cdk.Duration.seconds(2),
        bisectBatchOnError: true,
        retryAttempts: 3,
        reportBatchItemFailures: true,
      }),
    );
    state.grantWriteData(processor);
    alerts.grantPublish(processor);
    processor.addToRolePolicy(
      new iam.PolicyStatement({
        actions: ['iot:Publish'],
        resources: [`arn:${this.partition}:iot:${this.region}:${this.account}:topic/alerts/*`],
      }),
    );

    // ---------------------------------------------------------------- cold path
    const firehoseRole = new iam.Role(this, 'FirehoseRole', { assumedBy: new iam.ServicePrincipal('firehose.amazonaws.com') });
    archive.grantWrite(firehoseRole);
    const delivery = new firehose.CfnDeliveryStream(this, 'ArchiveDelivery', {
      deliveryStreamType: 'DirectPut',
      deliveryStreamEncryptionConfigurationInput: { keyType: 'AWS_OWNED_CMK' },
      extendedS3DestinationConfiguration: {
        bucketArn: archive.bucketArn,
        roleArn: firehoseRole.roleArn,
        prefix: 'telemetry/dt=!{timestamp:yyyy-MM-dd}/',
        errorOutputPrefix: 'errors/!{firehose:error-output-type}/dt=!{timestamp:yyyy-MM-dd}/',
        bufferingHints: { intervalInSeconds: 60, sizeInMBs: 5 },
        compressionFormat: 'GZIP',
      },
    });
    delivery.node.addDependency(firehoseRole);

    // ---------------------------------------------------------------- IoT Core
    const ruleRole = new iam.Role(this, 'IotRuleRole', { assumedBy: new iam.ServicePrincipal('iot.amazonaws.com') });
    stream.grantWrite(ruleRole);
    ruleRole.addToPolicy(new iam.PolicyStatement({ actions: ['firehose:PutRecord', 'firehose:PutRecordBatch'], resources: [delivery.attrArn] }));
    const ruleErrors = new logs.LogGroup(this, 'RuleErrors', {
      logGroupName: '/iot/telemetry-rule-errors',
      retention: logs.RetentionDays.TWO_WEEKS,
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });
    ruleErrors.grantWrite(ruleRole);

    new iot.CfnTopicRule(this, 'TelemetryRule', {
      ruleName: 'fleet_telemetry',
      topicRulePayload: {
        description: 'Fan out fleet telemetry to the hot and cold paths',
        awsIotSqlVersion: '2016-03-23',
        sql: "SELECT *, topic(2) AS fleet_id, topic(3) AS device_id, timestamp() AS received_at FROM 'fleet/+/+/telemetry'",
        actions: [
          { kinesis: { streamName: stream.streamName, partitionKey: '${topic(3)}', roleArn: ruleRole.roleArn } },
          { firehose: { deliveryStreamName: delivery.ref, separator: '\n', roleArn: ruleRole.roleArn } },
        ],
        errorAction: { cloudwatchLogs: { logGroupName: ruleErrors.logGroupName, roleArn: ruleRole.roleArn } },
      },
    });

    // Least-privilege device policy: a device may only publish its own telemetry.
    new iot.CfnPolicy(this, 'DevicePolicy', {
      policyName: 'fleet-device-policy',
      policyDocument: {
        Version: '2012-10-17',
        Statement: [
          {
            Effect: 'Allow',
            Action: 'iot:Connect',
            Resource: `arn:${this.partition}:iot:${this.region}:${this.account}:client/\${iot:Connection.Thing.ThingName}`,
          },
          {
            Effect: 'Allow',
            Action: 'iot:Publish',
            Resource: `arn:${this.partition}:iot:${this.region}:${this.account}:topic/fleet/*/\${iot:Connection.Thing.ThingName}/telemetry`,
          },
          {
            Effect: 'Allow',
            Action: ['iot:Subscribe'],
            Resource: `arn:${this.partition}:iot:${this.region}:${this.account}:topicfilter/alerts/\${iot:Connection.Thing.ThingName}`,
          },
          {
            Effect: 'Allow',
            Action: ['iot:Receive'],
            Resource: `arn:${this.partition}:iot:${this.region}:${this.account}:topic/alerts/\${iot:Connection.Thing.ThingName}`,
          },
        ],
      },
    });

    // ---------------------------------------------------------------- query history with Athena
    const db = new glue.CfnDatabase(this, 'Db', { catalogId: this.account, databaseInput: { name: 'iot_telemetry' } });
    const table = new glue.CfnTable(this, 'TelemetryTable', {
      catalogId: this.account,
      databaseName: 'iot_telemetry',
      tableInput: {
        name: 'readings',
        tableType: 'EXTERNAL_TABLE',
        partitionKeys: [{ name: 'dt', type: 'string' }],
        parameters: {
          'projection.enabled': 'true',
          'projection.dt.type': 'date',
          'projection.dt.format': 'yyyy-MM-dd',
          'projection.dt.range': '2025-01-01,NOW',
          'storage.location.template': `s3://${archive.bucketName}/telemetry/dt=\${dt}/`,
        },
        storageDescriptor: {
          location: `s3://${archive.bucketName}/telemetry/`,
          inputFormat: 'org.apache.hadoop.mapred.TextInputFormat',
          outputFormat: 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat',
          serdeInfo: { serializationLibrary: 'org.openx.data.jsonserde.JsonSerDe' },
          columns: [
            { name: 'fleet_id', type: 'string' },
            { name: 'device_id', type: 'string' },
            { name: 'ts', type: 'bigint' },
            { name: 'received_at', type: 'bigint' },
            { name: 'temperature_c', type: 'double' },
            { name: 'battery_pct', type: 'double' },
            { name: 'lat', type: 'double' },
            { name: 'lon', type: 'double' },
            { name: 'speed_kmh', type: 'double' },
          ],
        },
      },
    });
    table.node.addDependency(db);

    // ---------------------------------------------------------------- live dashboard (EMF metrics from the processor)
    const m = (name: string, stat: string) =>
      new cloudwatch.Metric({ namespace: 'IotFleet', metricName: name, statistic: stat, period: cdk.Duration.minutes(1) });
    new cloudwatch.Dashboard(this, 'FleetDashboard', {
      dashboardName: 'iot-fleet-live',
      widgets: [
        [
          new cloudwatch.GraphWidget({ title: 'Readings / min', left: [m('Readings', 'Sum')] }),
          new cloudwatch.GraphWidget({ title: 'Avg temperature °C', left: [m('TemperatureC', 'Average')], leftAnnotations: [{ value: props.tempMaxC, label: 'max' }] }),
          new cloudwatch.GraphWidget({ title: 'Anomalies', left: [m('Anomalies', 'Sum')] }),
          new cloudwatch.GraphWidget({ title: 'Pipeline lag (ms)', left: [stream.metricGetRecordsIteratorAgeMilliseconds({ statistic: 'Maximum' })] }),
        ],
      ],
    });

    new cdk.CfnOutput(this, 'StateTableName', { value: state.tableName });
    new cdk.CfnOutput(this, 'ArchiveBucketName', { value: archive.bucketName });
    new cdk.CfnOutput(this, 'DevicePolicyName', { value: 'fleet-device-policy' });
  }
}

"""Hot-path processor: Kinesis -> latest device state, anomaly alerts, live metrics (EMF)."""
import base64
import json
import os
import time
from decimal import Decimal

import boto3

TABLE = boto3.resource("dynamodb").Table(os.environ["STATE_TABLE"])
SNS = boto3.client("sns")
IOT = boto3.client("iot-data")
TOPIC = os.environ["ALERT_TOPIC_ARN"]
TEMP_MAX = float(os.environ.get("TEMP_MAX_C", "85"))
BATTERY_MIN = float(os.environ.get("BATTERY_MIN_PCT", "15"))


def handler(event, _context):
    failures, latest, temps, anomalies = [], {}, [], 0

    for rec in event["Records"]:
        try:
            r = json.loads(base64.b64decode(rec["kinesis"]["data"]))
            device = r["device_id"]
            temps.append(float(r.get("temperature_c", 0)))
            # keep only the newest reading per device in this batch
            if device not in latest or r.get("ts", 0) >= latest[device].get("ts", 0):
                latest[device] = r
            problems = []
            if float(r.get("temperature_c", 0)) > TEMP_MAX:
                problems.append(f"temperature {r['temperature_c']}°C > {TEMP_MAX}")
            if float(r.get("battery_pct", 100)) < BATTERY_MIN:
                problems.append(f"battery {r['battery_pct']}% < {BATTERY_MIN}")
            if problems:
                anomalies += 1
                alert(device, r, problems)
        except Exception as exc:  # report only the failed record; Kinesis retries it
            print(f"bad record {rec['kinesis']['sequenceNumber']}: {exc}")
            failures.append({"itemIdentifier": rec["kinesis"]["sequenceNumber"]})

    with TABLE.batch_writer(overwrite_by_pkeys=["device_id"]) as batch:
        for device, r in latest.items():
            batch.put_item(Item=json.loads(json.dumps({**r, "updated_at": int(time.time())}), parse_float=Decimal))

    emit_metrics(len(event["Records"]) - len(failures), temps, anomalies)
    return {"batchItemFailures": failures}


def alert(device, reading, problems):
    msg = {"device_id": device, "fleet_id": reading.get("fleet_id"), "problems": problems,
           "location": [reading.get("lat"), reading.get("lon")]}
    IOT.publish(topic=f"alerts/{device}", qos=1, payload=json.dumps(msg))  # tell the device
    SNS.publish(TopicArn=TOPIC, Subject=f"IoT anomaly: {device}"[:100], Message=json.dumps(msg, indent=2))


def emit_metrics(count, temps, anomalies):
    # CloudWatch Embedded Metric Format: structured log line -> metrics, no API calls.
    print(json.dumps({
        "_aws": {
            "Timestamp": int(time.time() * 1000),
            "CloudWatchMetrics": [{
                "Namespace": "IotFleet",
                "Dimensions": [[]],
                "Metrics": [
                    {"Name": "Readings", "Unit": "Count"},
                    {"Name": "TemperatureC", "Unit": "None"},
                    {"Name": "Anomalies", "Unit": "Count"},
                ],
            }],
        },
        "Readings": count,
        "TemperatureC": round(sum(temps) / len(temps), 2) if temps else 0,
        "Anomalies": anomalies,
    }))

#!/usr/bin/env node
// Simulates a fleet of delivery vehicles driving around a city and publishing telemetry
// to AWS IoT Core. Uses your AWS CLI credentials (IoT data plane) - fine for demos;
// real devices connect over MQTT with X.509 certificates and the fleet-device-policy.
//
//   npm i @aws-sdk/client-iot @aws-sdk/client-iot-data-plane
//   node simulator/simulate.mjs --devices 25 --interval 2 --region us-east-1
import { IoTClient, DescribeEndpointCommand } from '@aws-sdk/client-iot';
import { IoTDataPlaneClient, PublishCommand } from '@aws-sdk/client-iot-data-plane';

const arg = (name, dflt) => {
  const i = process.argv.indexOf(`--${name}`);
  return i > -1 ? process.argv[i + 1] : dflt;
};
const region = arg('region', process.env.AWS_REGION ?? 'us-east-1');
const devices = Number(arg('devices', 10));
const intervalSec = Number(arg('interval', 2));
const fleet = arg('fleet', 'austin');

const { endpointAddress } = await new IoTClient({ region }).send(new DescribeEndpointCommand({ endpointType: 'iot:Data-ATS' }));
const iot = new IoTDataPlaneClient({ region, endpoint: `https://${endpointAddress}` });

// Start every vehicle near downtown Austin, TX.
const fleetState = Array.from({ length: devices }, (_, i) => ({
  id: `truck-${String(i + 1).padStart(3, '0')}`,
  lat: 30.2672 + (Math.random() - 0.5) * 0.1,
  lon: -97.7431 + (Math.random() - 0.5) * 0.1,
  heading: Math.random() * 2 * Math.PI,
  temp: 40 + Math.random() * 10,
  battery: 60 + Math.random() * 40,
}));

console.log(`Publishing ${devices} devices every ${intervalSec}s to ${endpointAddress} (Ctrl+C to stop)`);
setInterval(async () => {
  await Promise.all(
    fleetState.map((d) => {
      d.heading += (Math.random() - 0.5) * 0.6;
      const speed = 20 + Math.random() * 50; // km/h
      d.lat += Math.cos(d.heading) * speed * 0.000005 * intervalSec;
      d.lon += Math.sin(d.heading) * speed * 0.000005 * intervalSec;
      d.temp += (Math.random() - 0.48) * 1.5 + (Math.random() < 0.01 ? 40 : 0); // rare overheating spike
      d.temp = Math.max(25, Math.min(d.temp, 120));
      d.battery = Math.max(0, d.battery - Math.random() * 0.2);
      const payload = {
        ts: Date.now(),
        temperature_c: Number(d.temp.toFixed(1)),
        battery_pct: Number(d.battery.toFixed(1)),
        lat: Number(d.lat.toFixed(6)),
        lon: Number(d.lon.toFixed(6)),
        speed_kmh: Number(speed.toFixed(1)),
      };
      return iot.send(new PublishCommand({ topic: `fleet/${fleet}/${d.id}/telemetry`, qos: 1, payload: JSON.stringify(payload) }));
    }),
  );
  process.stdout.write('.');
}, intervalSec * 1000);

{
  "app": "npx ts-node --prefer-ts-exts bin/app.ts",
  "watch": { "include": ["**"], "exclude": ["README.md", "cdk*.json", "**/*.d.ts", "**/*.js", "node_modules", "cdk.out"] },
  "context": {}
}

{
  "name": "13-iot-telemetry-pipeline",
  "version": "1.0.0",
  "private": true,
  "bin": { "app": "bin/app.js" },
  "scripts": {
    "build": "tsc",
    "synth": "cdk synth",
    "deploy": "cdk deploy --all",
    "destroy": "cdk destroy --all"
  },
  "devDependencies": {
    "@types/node": "^22.0.0",
    "aws-cdk": "^2.1000.0",
    "ts-node": "^10.9.2",
    "typescript": "~5.6.3"
  },
  "dependencies": {
    "aws-cdk-lib": "^2.200.0",
    "constructs": "^10.4.2"
  }
}

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "commonjs",
    "lib": ["es2022"],
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "noEmit": true,
    "types": ["node"]
  },
  "exclude": ["node_modules", "cdk.out"]
}