はじめに

弊社では Web アプリケーションや非同期処理用ジョブの一時データなどを取り扱う際に ElastiCache Redis をインメモリデータベースとして利用しています。

これまで ElastiCache Redis 自体の稼動状況の把握には CloudWatch メトリクスを中心とした数値ベースの情報のみを利用していましたが、今回ここにログ情報を加えることにしました。ログ取得に際し必要となった作業について扱います。

簡単のため、以降は特に断わりなく "ElastiCache Redis" を単に "Redis" と呼びます。

Redis ログについて

Redis ログには slow / engine の2種類があります。ElastiCache Redis の場合、ログのフォーマットは JSON およびプレーンテキストの2種類を選べますが、以下では JSON ログを前提とします。なお slow ログは Datadog Logs へのログ送信に際し注意が必要です（後述）。

slow ログ

「遅い処理」を記録するものです。詳細は Redis 公式のドキュメントを参照ください。

何をもって遅い処理とするかは ElastiCache Redis の場合、パラメタグループ内slowlog-log-slower-than にてマイクロ秒単位で決定されます。また slow ログの収集は ElastiCache Redis 6.0 以上のバージョンから可能です。

ログ例は ElastiCache 公式ドキュメントで示される通り、以下のようになります。S3 へ保存されるログもこの形式になります。記録対象のログ1件に対し JSON オブジェクトが1件対応します。

{
  "CacheClusterId": "logslowxxxxmsxj", 
  "CacheNodeId": "0001", 
  "Id": 296, 
  "Timestamp": 1605631822, 
  "Duration (us)": 0, 
  "Command": "GET ... (1 more arguments)", 
  "ClientAddress": "192.168.12.104:55452", 
  "ClientName": "logslowxxxxmsxj##" 
}

各要素は以下のような意味合いになります。前掲ドキュメントの内容を意訳し、適宜補足を加えています。

CacheClusterId：Redis クラスタ名
CacheNodeId：Redis ノード名（上記クラスタ内にあるはず）
Id：各ログ固有の識別子
Timestamp： 当該ログが Redis 内で記録された際の 日時
- Datadog でログが処理された時刻とは異なるので注意
Duration (us)：ログで記録された処理に要した処理時間（マイクロ秒）
Command：Redis で処理されたコマンドの内容
ClientAddress：Redis にコマンドを発行したクライアントの IP
- 通常これは ECS タスクなどが持つ VPC 内 IP になるはず
ClientName：Redis にコマンドを発行したクライアントの名前

engine ログ

Redis そのものの稼動状況を示すログになります。engine ログの収集は ElastiCache Redis 6.2 以上のバージョンから可能です。これも ElastiCache 公式ドキュメントから例を引用します。S3 側に保存される状況は slow ログと同様です。

例：

{ 
  "CacheClusterId": "xxxxxxxxxzy-engine-log-test", 
  "CacheNodeId": "0001", 
  "LogLevel": "VERBOSE", 
  "Role": "M", 
  "Time": "12 Nov 2020 01:28:57.994 UTC", 
  "Message": "Replica is waiting for next BGSAVE before synchronizing with the primary. Check back later" 
}

内容については以下となります。

CacheClusterId：Redis クラスタ名
CacheNodeId：Redis ノード名（上記クラスタ内にあるはず）
LogLevel：ログの重要性の度合い
- VERBOSE：雑多なやつ
- NOTICE：見ておくほうがよいやつ
- WARNING：真剣に見ておくほうがよいやつ
Time： 当該ログが Redis 内で記録された際の 日時
- Datadog でログが処理された時刻とは異なるので注意
Role：ログを出力した主体
- M：プライマリノード
- S：レプリカ（セカンダリ）ノード
- C：ディスクにデータが書き出されたとき
  - より詳細には RDB / AOF 処理が発生したとき（Redis 公式ドキュメントの "Redis persistence" が詳しい）
- X：フェイルオーバが発生したとき
  - より詳細には Redis Sentinel が動いたとき（Redis 公式ドキュメントの "High availability with Redis Sentinel " が詳しい）

設定方針

既に稼動中の Redis クラスタ（レプリケーショングループ）はそのまま利用し、ログ取得のための設定を追加で入れます。ログの出力先については S3 バケット及び Datadog Logs を選定しました。これは以下のような事情によります。

Datadog Logs：弊社ではログ管理（集約 / 検索 / 監視 etc.）を Datadog で一元化しており、Redis ログも Datadog 上で取り扱えると都合がよい
S3：基本的には Datadog 上でログを取り扱うが、Datadog に万一の事態が生じた場合に Athena 等でログを扱えるようにしておきたい

いっぽうで ElastiCache ログ取得に関する AWS のドキュメントによればログ出力先には Data Firehose と CloudWatch Logs の2つの選択肢があります。どちらが適切かという話になりますが、以下のとおり Data Firehose を使っておくのがベターなようです。

Redis ログ出力先	メリット	デメリット
Data Firehose	● S3 をバックアップ先に指定することで S3 へのログ転送も達成可（ドキュメント） ● Datadog が公式にData Firehose によるログ転送を解説している（ドキュメント）	● CloudWatch Logs へはログが飛ばない
CloudWatch Logs	● S3 と Datadog Logs に加え CloudWatch Logs でのログ取扱が可能になる（CloudWatch Logs Insights が使える）	● S3 / Datadog Logs 向けにサブスクリプションフィルタと別途ログ送信用リソースの用意が必要になる

よって本稿では Redis ログを Data Firehose へ出力し、Data Firehose から S3 および Datadog Logs へログを送信するような構成をとることとしました。

構成

構成図は以下の通りです。slow / engine 各ログは最終目的地が同一の S3 バケット / Datadog Logs であることから、Data Firehose ストリームは特に分けない構成としました。S3 バケットは Redis ログ用に新設したものを使います。

いきなり Data Firehose に Lambda 関数がくっつく構成となっていますが、これには事情があります。

Datadog Logs 向けに slow ログを修正する

当初、上記構成のうち Lambda 関数を設けず（つまり各ログを無加工の状態で） Redis ログを送信した際、slow ログのみ Datadog Logs でログとして取り扱えない状況となりました。Data Firehose から Datadog へのログ送信は問題無い様子だったので、Datadog がログを正しく処理できていない模様です。

トラブルシュートに難儀したのですが、ヒントは Datadog ドキュメントのうち Log Management -> Log Configuration -> Pipelines にありました。関連する箇所を同ドキュメントから要約します：

JSON ログ中に Timestamp という attribute がある場合、Datadog Logs はその情報をログの時刻情報として扱う
- 取り扱える日付のフォーマットは ISO8601 / ミリ秒精度の UNIX 時間 / RFC3164
JSON ログ中に存在する Message の内容をログメッセージ本体として扱う

ここで各ログの内容を見てみましょう。前掲のログ例を引用します。

slow ログ

{
  "CacheClusterId": "logslowxxxxmsxj", 
  "CacheNodeId": "0001", 
  "Id": 296, 
  "Timestamp": 1605631822, 
  "Duration (us)": 0, 
  "Command": "GET ... (1 more arguments)", 
  "ClientAddress": "192.168.12.104:55452", 
  "ClientName": "logslowxxxxmsxj##" 
}

engine ログ

{ 
  "CacheClusterId": "xxxxxxxxxzy-engine-log-test", 
  "CacheNodeId": "0001", 
  "LogLevel": "VERBOSE", 
  "Role": "M", 
  "Time": "12 Nov 2020 01:28:57.994 UTC", 
  "Message": "Replica is waiting for next BGSAVE before synchronizing with the primary. Check back later" 
}

2ログを見比べると以下がわかります。

slow ログ
- Timestamp に UNIX 時間（ミリ秒精度ではない）と解釈できる値が入っている
- Message が無い
- Command に Redis へ発行されたコマンドが記録されている
engine ログ
- Timestamp は無い
- Message がある

engine ログが Datadog Logs で処理できているので、slow ログを engine ログと同様の内容にするのが手っ取り早いでしょう。つまり以下のようにすると光明が見えそうです。

{
 "CacheClusterId": "logslowxxxxmsxj", 
  "CacheNodeId": "0001", 
  "Id": 296, 
  "Time": 1605631822,                              # 元 `Timestamp`
  "Duration (us)": 0, 
  "Message": "GET ... (1 more arguments)",         # 元 `Command`
  "ClientAddress": "192.168.12.104:55452", 
  "ClientName": "logslowxxxxmsxj##" 
}

Data Firehose はストリーム内を流れるデータを Lambda 関数を使い処理できる機能があります。弊社では以下のような TypeScript コードを Node.js ランタイムを使う Lambda 関数として整備し、上記変換処理を行わせることにしました。

コードはこちらを参照

import { Buffer } from 'buffer';
import { DateTime } from 'luxon';

console.log('Loading function');

interface InputRecord {
  recordId: string;
  data: string;
}

interface Event {
  records: InputRecord[];
}

interface OutputRecord {
  recordId: string;
  result: string;
  data: string;
}

interface Output {
  records: OutputRecord[];
}

export const handler = async (event: Event): Promise<Output> => {
  const output = event.records.map((record, index): OutputRecord => {
    const decoded = Buffer.from(record.data, 'base64').toString('utf-8');
    console.log(`Decoded data: ${decoded}`);

    /*
     * event として Data Firehose から受け取ったペイロードが JSON ログでない場合は後続処理をさせない
     * ElastiCache ログは Data Firehose へ JSON の形式で送るよう設定しているので通常はここに落ちないはず
     */
    let parsed: any;
    try {
      parsed = JSON.parse(decoded);
    } catch (error) {
      console.error(`Error parsing JSON for record ${record.recordId}:`, error);
      return {
        recordId: record.recordId,
        result: 'Ok',
        data: record.data,
      };
    }

    /*
     * Timestamp というエントリは Datadog Logs が処理する上で不適。以下のように直す
     * （参考：https://docs.datadoghq.com/logs/log_configuration/pipelines/?tab=date）
     *
     * - Time というエントリに改名
     * - 元々 Timestamp として記録されていた epoch 秒を engine ログと同じような日付フォーマットの文字列に変換
     *   - engine ログと互換性のある内容にしたい
     */
    if ('Timestamp' in parsed) {
      const dt = DateTime.fromSeconds(parsed.Timestamp, { zone: 'utc' });
      parsed.Time = dt.toFormat("dd MMM yyyy HH:mm:ss.SSS 'UTC'");
      delete parsed.Timestamp;
    }

    /*
     * Message というエントリが無いと Datadog Logs 上でログとして受け付けてくれない
     * 拾いたい情報である "Command" の内容を Message に改名する
     * これも engine ログと同様の措置
     * （参考：https://docs.datadoghq.com/logs/log_configuration/pipelines/?tab=message）
     */
    if ('Command' in parsed) {
      parsed.Message = parsed.Command;
      delete parsed.Command;
    }

    console.log(`Transformed data: ${JSON.stringify(parsed)}`);
    const encoded = Buffer.from(JSON.stringify(parsed)).toString('base64');

    return {
      recordId: record.recordId,
      result: 'Ok',
      data: encoded,
    };
  });

  console.log(`Processing completed. Successful records: ${output.length}`);
  return { records: output };
};

インフラ構成

Terraform コード例を示します。実際に利用しているコードから要所を抜粋 / 省略したものとなります。

コードはこちらを参照

# Data Firehose 内でのログ処理に使用する Lambda 関数のソースコード
# Terraform とは別の経路で事前に S3 バケットへアップロードしておく
data "aws_s3_object" "lambda_source_archive" {
  bucket = var.bucket_name
  key    = "transform-elasticache-logs/artifact.zip"
}

# Lambda 関数にアタッチされる IAM ロールで使用する IAM ポリシ
# 当該関数内ではシンプルに JSON データをいじるのみで AWS サービスを触らない
# 最低限の権限（CloudWatch Logs へのログ書き出し）をもつ IAM ポリシだけ参照できるようにする
data "aws_iam_policy" "lambda_basic_role" {
  name = "AWSLambdaBasicExecutionRole"
}

# Lambda 関数本体
module "lambda_transform_elasticache_logs" {
  source  = "terraform-aws-modules/lambda/aws"
  version = "7.20.2"

  function_name  = "${var.env}-transform-elasticache-logs"
  description    = "Lambda function for transforming slow log from Elasticache to appropriate format suitable for Datadog Logs"
  handler        = "main.handler"
  runtime        = "nodejs22.x"
  timeout        = 60
  publish        = true
  attach_policy  = true
  create_package = false
  policy         = data.aws_iam_policy.lambda_basic_role.arn

  s3_existing_package = {
    bucket     = var.bucket_name
    key        = "transform-elasticache-logs/artifact.zip"
    version_id = data.aws_s3_object.lambda_source_archive.version_id
  }

  assume_role_policy_statements = {
    account_root = {
      effect  = "Allow"
      actions = ["sts:AssumeRole"]
      principals = {
        account_principal = {
          type        = "AWS"
          identifiers = ["arn:aws:iam::${data.aws_caller_identity.self.account_id}:root"]
        }
      }
    }
  }

  tags = {
    Name = "${var.env}-transform-elasticache-logs"
  }
}

# Data Firehose 向け principal を assume する
data "aws_iam_policy_document" "elasticache_log_assuming" {
  statement {
    actions = ["sts:AssumeRole"]
    effect  = "Allow"

    principals {
      type        = "Service"
      identifiers = ["firehose.amazonaws.com"]
    }
  }
}

# Data Firehose が所定の S3 バケットにログを流し込めるようにする
data "aws_iam_policy_document" "elasticache_log_permission" {
  statement {
    effect = "Allow"

    actions = [
      "s3:AbortMultipartUpload",
      "s3:GetBucketLocation",
      "s3:GetObject",
      "s3:ListBucket",
      "s3:ListBucketMultipartUploads",
      "s3:PutObject",
    ]

    resources = [
      aws_s3_bucket.main.arn,
      "${aws_s3_bucket.main.arn}/*",
    ]
  }

  # ログ送信のためには CloudWatch Logs 関連の権限も必要そう
  statement {
    effect = "Allow"

    actions = [
      "logs:CreateLogStream",
      "logs:PutLogEvents",
      "logs:CreateLogGroup",
    ]

    resources = ["*"]
  }
}

# ElastiCache ログを Data Firehose 経由で配信する為の IAM ロール
resource "aws_iam_role" "main" {
  name               = "${var.env}-elasticache-log-role"
  assume_role_policy = data.aws_iam_policy_document.elasticache_log_assuming.json
}

# ElastiCache ログを Data Firehose 経由で配信する為の IAM ロールにポリシーをアタッチ
resource "aws_iam_role_policy" "main" {
  name   = "${var.env}-elasticache-log-policy"
  role   = aws_iam_role.main.id
  policy = data.aws_iam_policy_document.elasticache_log_permission.json
}

# ログ処理用 Data Firehose ストリーム
resource "aws_kinesis_firehose_delivery_stream" "main" {
  depends_on = [
    aws_s3_bucket.main,
  ]

  name = "${var.env}-redis-log"

  # S3 と Datadog Logs とにログを送信する
  # S3 にログを送るのみであれば extended_s3_configuration が使えるが、その場合 Datadog 向けの設定と共存不可
  destination = "http_endpoint"

  http_endpoint_configuration {
    url  = "https://aws-kinesis-http-intake.logs.datadoghq.com/v1/input"
    name = "Datadog Log"

    # 以下 `DATADOG_API_KEY` は適切な値に置き換える
    access_key = DATADOG_API_KEY

    role_arn = aws_iam_role.main.arn

    # 今回は Datadog Logs と同内容のログを S3 にも置きたいという目的につき、全ログを S3 に置く設定とした
    s3_backup_mode = "AllData"

    request_configuration {
      content_encoding = "GZIP"

      # これら値を設定しておくことで Datadog Logs 上でタグが付与される
      common_attributes {
        name  = "env"
        value = var.env
      }
    }

    s3_configuration {
      role_arn   = aws_iam_role.main.arn
      bucket_arn = aws_s3_bucket.main.arn
      prefix     = "redis/"
    }

    # ログ整形用 Lambda 関数定義
    processing_configuration {
      enabled = "true"
      processors {
        type = "Lambda"
        parameters {
          parameter_name  = "LambdaArn"
          parameter_value = module.lambda_transform_elasticache_logs.lambda_function_arn
        }
      }
    }
  }

  tags = {
    "LogDeliveryEnabled" = "true"
  }
}

# Redis クラスタ
resource "aws_elasticache_replication_group" "main" {
  depends_on = [
    aws_kinesis_firehose_delivery_stream.elasticache_log,
  ]

  replication_group_id       = "${var.env}-redis"
  description                = "Redis replication group"
  engine                     = var.elasticache.engine
  engine_version             = var.elasticache.engine_version
  node_type                  = var.elasticache.instance_type
  port                       = var.elasticache.port
  subnet_group_name          = var.elasticache.subnet_group_name
  parameter_group_name       = aws_elasticache_parameter_group.main.name
  num_cache_clusters         = var.elasticache.instance_num
  at_rest_encryption_enabled = true
  apply_immediately          = true

  security_group_ids = var.elasticache.security_group_ids

  # slow ログ（後述）
  log_delivery_configuration {
    destination      = aws_kinesis_firehose_delivery_stream.main.name
    destination_type = "kinesis-firehose"
    log_format       = "json"
    log_type         = "slow-log"
  }

  # engine ログ（後述）
  log_delivery_configuration {
    destination      = aws_kinesis_firehose_delivery_stream.main.name
    destination_type = "kinesis-firehose"
    log_format       = "json"
    log_type         = "engine-log"
  }
}

# Redis パラメタグループ
# ログ取得は Redis 6.x 系で利用出来る機能につき、パラメタグループも Redis 6.x 系を対象にする
resource "aws_elasticache_parameter_group" "main" {
  name   = "${var.env}-redis6"
  family = "redis6.x"

  /*
   どのくらい処理に時間が掛かれば slow な処理として slow log へ出すかを制御する
   マイクロ秒単位で設定する。デフォルトは 10000（= 10ミリ秒）
   */
  parameter {
    name  = "slowlog-log-slower-than"
    value = 10000
  }
}

# ログ保存先 S3 バケット
resource "aws_s3_bucket" "main" {
  bucket = "${var.env}-elasticache-log"
}

必要になるリソースは以下の通りです。

当然用意が必要なもの
- S3 バケット
- Redis クラスタ（レプリケーショングループ）
- Data Firehose ストリーム
注意が必要なもの
- Data Firehose が使用する IAM ロールと権限認可のための IAM ポリシ
  - firehose.amazonaws.com な principal を assume できるよう設定
  - S3 バケット / オブジェクトを変更含め操作する為の権限を設定
  - Data Firehose は転送エラーを CloudWatch Logs へ記録するので CloudWatch Logs 向けの権限も設定
- Redis パラメタグループ
  - slowlog-log-slower-than というパラメタで slow ログに記録する対象となる処理時間の閾値を設定（マイクロ秒単位）
  - ここで指定した時間以内で終わる処理はログに記録されないので注意
- Lambda 関数
  - 「Datadog Logs 向けに slow ログを修正する」の節で述べた通り

slowlog-log-slower-than パラメタによって slow ログとしての記録がなされるか否かについては見落しがちなので注意が必要です。遅い処理が存在しない場合ログにも記録されません。Redis / Data Firehose / S3 各所の設定は問題ないはずなのに何故ログが出ないのだと悩む羽目にならないよう、適当な閾値を設定しておきましょう。

Datadog Logs の風景

実際に Datadog Logs 上で収集されたログをお見せできればと思います（所々伏せ字なのはご容赦ください）

slow ログ	engine ログ

既に扱った通り、slow / engine ログはログ中に自身が slow / engine ログである旨を示す情報を持っていません。各ログに固有の情報でログの種類を判定するのが手軽です。弊社では以下のように @Role の有無でログ判別を行うような view を Datadog Logs へそれぞれ追加し、運用しています。

slow ログ：service:aws -@Role:*
engine ログ：service:aws @Role:*

Data Firehose + Lambda によりログ内容の調整が可能な構成になっていることから、必要に応じてログ中へ @type: slow といったような値の盛り込みも検討できる状態ではあります。これは運用を続けてみてからの判断となるでしょう。

おわりに

ここまで本稿を読んで頂き有り難うございました。Data Firehose を使用して ElastiCache Redis ログを S3 及び Datadog Logs で取り扱う為の設定について扱いました。 engine ログと slow ログとで Datadog Logs へのログ送信に際し検討すべきことに差が出るのは想定外でしたが、最終的には狙った結果を得ることが出来ました。

ログは取得を始めて直ちに何らか嬉しさが生じるようなものではありませんが、月日が経過し記録と実績が蓄積されゆく中で自ずとその価値や用途が見えてくるものとは成り得るはずです。不確定要素の多い将来への投資としての一助となれば幸いです。

MNTSQ 株式会社 SRE 秋本