MQTTで受け付けたIoTデータをLambdaでGrafana Cloudに送りつけて可視化する

{"endpoint":"xxxxx","token":"xxxxx"}

aws ssm put-parameter \
  --name "/grafana/otel/secrets" \
  --type "SecureString" \
  --value "{"endpoint":"xxxxx","token":"xxxxx"}"

import * as iot from '@aws-cdk/aws-iot-alpha';
import * as iotActions from '@aws-cdk/aws-iot-actions-alpha';
import * as lambda from "aws-cdk-lib/aws-lambda";
import * as nodejs from "aws-cdk-lib/aws-lambda-nodejs";
import * as logs from "aws-cdk-lib/aws-logs";
import * as ssm from "aws-cdk-lib/aws-ssm";
import * as cdk from 'aws-cdk-lib';

// 簡易的な実装で失礼します。。。
const app = new cdk.App();
const stack = new cdk.Stack(app, 'PushIotMetricsToGrafanaMimirStack');

/** Lambda関数の定義 */
const fn = new nodejs.NodejsFunction(stack, 'Function', {
  entry: './src/app.Function.ts',
  runtime: lambda.Runtime.NODEJS_22_X,
  architecture: lambda.Architecture.ARM_64,
})
// SSMパラメータストアに対するアクセス許可
ssm.StringParameter
  .fromSecureStringParameterAttributes(stack, 'GrafanaOtelSecrets', {
    parameterName: '/grafana/otel/secrets',
    version: 1,
  })
  .grantRead(fn);

/** エラーログの定義 */
const errorLog = new logs.LogGroup(stack, "ErrorLog", {
  removalPolicy: cdk.RemovalPolicy.DESTROY,
});

/** MQTTのハンドリングについて定義 */
new iot.TopicRule(stack, 'TopicRule', {
  sql: iot.IotSql.fromStringAsVer20160323("SELECT * FROM 'PushIotMetricsToGrafanaMimir'"),
  actions: [new iotActions.LambdaFunctionAction(fn)],
  errorAction: new iotActions.CloudWatchLogsAction(errorLog),
});

import { getParameter } from "@aws-lambda-powertools/parameters/ssm";
import { ValueType } from "@opentelemetry/api";
import { OTLPMetricExporter } from "@opentelemetry/exporter-metrics-otlp-http";
import { MeterProvider, PeriodicExportingMetricReader } from "@opentelemetry/sdk-metrics";
import { resourceFromAttributes } from "@opentelemetry/resources";
import { ATTR_SERVICE_NAME } from "@opentelemetry/semantic-conventions";
import * as v from 'valibot';

/** MQTTのペイロードのスキーマ */
const mqttPayloadSchema = v.object({
  deviceId: v.string(),
  label: v.string(),
  value: v.number(),
});

/** パラメーターストアに保存したJSONのスキーマ */
const secretSchema = v.object({
  endpoint: v.string(),
  token: v.string(),
});

export const handler = async (event: unknown) => {
  /** MQTTのスキーマ確認 */
  const payload = v.parse(mqttPayloadSchema, event);

  /** パラメーターストアから秘密情報を取得 */
  const rawSecret = await getParameter("/grafana/otel/secrets", {
    transform: "json",
    decrypt: true,
    maxAge: 10 * 60, // seconds
  });
  const secret = v.parse(secretSchema, rawSecret)

  // OpenTelemetryを用いたデータ送信処理

  const meterProvider = new MeterProvider({
    resource: resourceFromAttributes({
      [ATTR_SERVICE_NAME]: "test_service",
    }),
    readers: [
      new PeriodicExportingMetricReader({
        exporter: new OTLPMetricExporter({
          /**
           * OTELエンドポイント(`https://${host}/otlp`)に加えて、
           * MimirのAPIリファレンスに記載されているOTELのpath(`/otlp/v1/metrics`)を合わせて指定する
           * @see https://grafana.com/docs/mimir/latest/references/http-api/#otlp
           */
          url: `${secret.endpoint}/v1/metrics`,
          headers: {
            Authorization: `Basic ${secret.token}`,
          },
        }),
        exportIntervalMillis: 1000,
      }),
    ],
  });
  const meter = meterProvider.getMeter("test", "1.0.0");

  const inputGauge = meter.createGauge(`test_service_${payload.label}`, {
    valueType: ValueType.DOUBLE,
  });

  inputGauge.record(payload.value, {
    deviceId: payload.deviceId,
  });

  await meterProvider.shutdown();
}

pnpm cdk deploy

aws iot-data publish \
  --topic="PushIotMetricsToGrafanaMimir" \
  --payload='{"deviceId": "test_device_A","label": "rpm","value": 101}' \
  --cli-binary-format=raw-in-base64-out