この記事について

Lambda API でエラーが発生したときに、CloudWatchLogs にいったんエラーを集積して、cron で動く他の Lambda で定期的にエラーを集計してその件数を slack へ通知していた。

これが突如一切のエラーを報告しなくなった。
その訳と取った対策についての記事。

時間がない場合経緯の箇条書き部分と結論だけ読めばよい。

問題のコード

エラー発生時に slack 通知用のエラー内容を専用のロググループに書き込んでおくことで定期的に slack に通知する。

Lambda API は ruby で作っているが、通知用の Lambda は Node.js で作ってあるというちぐはぐ具合だが気にしない。

ログ書き込み部分

Lambda API でエラーが発生したときに send_api_error_event に exeption を渡して実行すると指定のロググループにログを書き込む。

一応、ある程度汎用的にするために同一日付・同一インスタンス使用中は同じログストリームに書き込むことを期待したコードにしているが、API 実行ごとに異なるセッションになり @error_log_stream_uuid が変わるため今回の使用方法ではログ1つごとにログストリームが作成される。

class LogSender
  def send_event(log_event, log_group_name, log_stream_name)
    return if log_event.nil?
    params = {
      log_group_name: log_group_name,
      log_stream_name: log_stream_name,
      log_events: [log_event],
      sequence_token: latest_log_stream_sequence_token(log_group_name, log_stream_name)
    }

    cwl_client.put_log_events(params)
  end

  def send_api_error_event(exp)
    now = Time.now

    log_group_name = 'error-notify-to-slack'

    log_stream_date = now.strftime('%F') # yyyy-MM-dd

    # インスタンス変数に無ければ作成
    @error_log_stream_uuid ||= SecureRandom.uuid

    log_stream_name = "#{log_stream_date}/#{@error_log_stream_uuid}"

    log_event = {
      message: JSON.generate({
        title: "api-error #{適当なタイトル}",
        message: exp.message,
        stack_trace: exp.backtrace
      }),
      timestamp: (now.to_f * 1000).to_i
    }

    send_event(log_event, log_group_name, log_stream_name)
  end

  private

  def cwl_client
    @cwl_client ||= Aws::CloudWatchLogs::Client.new
  end

  # ログストリームの最新のシーケンストークンを取得
  def latest_log_stream_sequence_token(log_group_name, log_stream_name)
    describe_log_stream(log_group_name, log_stream_name).upload_sequence_token
  rescue Aws::CloudWatchLogs::Errors
    # ログストリームが無ければ作成
    cwl_client.create_log_stream(
      log_group_name: log_group_name,
      log_stream_name: log_stream_name
    )
    # 新しいログストリームにはシーケンストークンが存在しないので、 nil を返す
    nil
  end

  # ログストリームを取得
  def describe_log_stream(log_group_name, log_stream_name)
    log_stream = cwl_client.describe_log_streams(
      log_group_name: log_group_name,
      log_stream_name_prefix: log_stream_name,
      limit: 1
    ).log_streams.first
    log_stream || (raise NoLogStreamError)
  end
end

# ストリーム見つからないエラー
class NoLogStreamError < StandardError
  include Aws::CloudWatchLogs::Errors
end

通知 lambda 本体

Serverless Framework でデプロイを行っている。
EventBridge で EXECUTION_INTERVAL_MIN ごとに実行されて指定期間のエラーをタイトルで集約して slack へ報告する。

handler.js

'use strict';

const AWS = require('aws-sdk');
const request = require('request-promise');

// for Date class
const SECOND = 1000;
const MINUTE = 60 * SECOND;
// ログを取得する際に最大でページングする回数
const MAXIMUM_FETCH_COUNT = 100;
const ERROR_COLOR = '#dc143c';

// Config
const slackToken = process.env.SLACK_TOKEN;
const conversationId = process.env.CHANNEL_ID;
const logGroupName = process.env.LOG_GROUP;
const executionIntervalMin = process.env.EXECUTION_INTERVAL_MIN * MINUTE;
// 直近のログはストリームに入っていない可能性があるため適宜取得終了時間を調整する
const cwlClient = new AWS.CloudWatchLogs({apiVersion: '2014-03-28'});


module.exports.slack = async () => {
  const now = Date.now();
  // バッチ実行間隔で時間を丸める
  // ex. (5分毎)9:32->9:30 ex2. (10分毎)16:16 -> 16:10
  const logEndAt = Math.floor(now / (executionIntervalMin)) * executionIntervalMin
  const logStartAt = logEndAt - executionIntervalMin;
  // 指定したロググループのログを取得
  console.log("logStartAt, logEndAt: ", new Date(logStartAt), new Date(logEndAt));
  const logs = await getLogs(logGroupName, logStartAt, logEndAt);
  if (logs.length < 1) {
    console.log('Log not found.');
    return {
      statusCode: 200,
    };
  };
  // 取得したログをSlackAPIの形式に加工
  const attachments = summarizeLogs(logs).map(error => {
      return {
        title: error.title + ` ${error.count}件`,
        text: error.message,
        footer: error.stack_trace,
        color: ERROR_COLOR,
      }
    });
  // Slackへ通知
  await postMessage(slackToken, conversationId, attachments);

  return {
    statusCode: 200,
  };
};

const postMessage = (token, channel, attachments) => {
  const options = {
    uri: 'https://slack.com/api/chat.postMessage',
    method: 'POST',
    headers: makeSlackHeaders(token),
    json: {channel, attachments, text: "Errors" },
  };

  return request(options);
};

const makeSlackHeaders = token =>
  ({
    'Content-type': 'application/json',
    'Authorization': `Bearer ${token}`,
  });

const getLogs = async(logGroupName, startAt, endAt) => {
  let logs = [];
  let nextToken = null;
  for (let i = 0; i < MAXIMUM_FETCH_COUNT; i++) {
    let res = await filterLogEvents(logGroupName, startAt, endAt, nextToken);
    logs = logs.concat(res.events);
    if (!res.nextToken) {
      break;
    }
    nextToken = res.nextToken;
  }
  return logs;
}

const filterLogEvents = (logGroupName, startAt, endAt, nextToken) => {
  return cwlClient.filterLogEvents({
    logGroupName: logGroupName,
    endTime: endAt,
    startTime: startAt,
    nextToken: nextToken,
  }).promise();
}

// 取得したログをtitle毎で集計
const summarizeLogs = logs => {
  return logs.reduce((acc, current) => {
    let msg;
    try {
      msg = JSON.parse(current.message);
    } catch(e) {
      msg = {
        title: current.message.split("\n")[0],
        message: current.message,
      }
    };
    let found = acc.find(element => element.title === msg.title);
    if (found) {
      found.count++;
    } else {
      acc.push({
        count: 1,
        ...msg
      });
    };
    return acc;
  }, []);
}

serverless.yml

service: error-notifier
plugins:
  - serverless-offline
custom:
  serverless-offline:
    resourceRoutes: true

provider:
  name: aws
  runtime: nodejs12.x
  iamManagedPolicies:
    - arn:aws:iam::aws:policy/CloudWatchLogsReadOnlyAccess
    - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole

functions:
  slack:
    handler: handler.slack
    timeout: 20
    environment:
      SLACK_TOKEN: ${env:SLACK_TOKEN}
      CHANNEL_ID: ${env:CHANNEL_ID}
      LOG_GROUP: ${env:LOG_GROUP}
      EXECUTION_INTERVAL_MIN: ${env:EXECUTION_INTERVAL_MIN}
    events:
      - schedule: cron(1/${env:EXECUTION_INTERVAL_MIN}, *, *, *, ?, *)

export SLACK_TOKEN=XXXXX
export CHANNEL_ID=YYYYY
export LOG_GROUP=error-notify-to-slack
export EXECUTION_INTERVAL_MIN=2

sls deploy --stage production --region YOUR_PROJECT_REGION

経緯

ここから長いので発見・調査・対処の経緯を箇条書きにする。

ある時期を境に通知用 Lambda がすべてタイムアウト(20秒)していた。ログの数が爆増したころからである。
Lambda のタイムアウトを伸ばして(1分)動かしたら正常終了したが、"Log not found." と返してきた。
ログを探し続けないために filterLogEvents のループ数の上限(100回)が定められていたが、ループ中に指定期間のログを発見できなかったらしい。
集積しているエラーの数が爆増してから問題が発生しているので、単純なログ数の増大が影響を及ぼしていると考え、ログの保持期間の経過を待ってみたが、改善しなかった。
確認したところログイベントが破棄されても、空になったログストリームは残っていた。この時点で問題は単純なログの数ではなくログストリームの数であると考えた。
ログストリームでの絞り込みを行っていない場合、まずログストリームを取得して、その中から条件に合ったログイベントを捜索するような挙動を見せていた。結果として関係のないログストリームをたっぷり回収してきてログイベントを探したが指定期間のログが発見できない、という行為をループ限界まで繰り返していた。
ログを検索する際にログストリーム名で絞り込みを行うことで無駄なログストリームの回収を抑制して一旦の解決を見た。

問題の発生

ある日、slack へエラー通知が一切来なくなったことに気が付く。
流石にエラーが一切起きていないわけはないと思われるし、事実開発フェーズに staging 環境で発生したエラーの通知も来なかった。
どうやら通知の調子が悪いらしい。

通知用 Lambda のメトリクスを確認したところ、ある時期から Duration が20秒へ張り付き Error count and success rate (%) がエラー100%を記録していた。

完全に不調

現象が始まったのは非常にアクセスの多い案件が稼働した後ごろかららしい。
ログの数が爆増したことが問題ではないかと考えた。

問題の調査

とりあえずタイムアウトを伸ばす

何かわからんがタイムアウトしているなら Lambda の生存時間を延ばしてみる。
Lambda は稼働時間で課金されるので、調査が終わったらもったいないからまた短命にしよう。

というわけで寿命を１分に伸ばした。

Lambda は40秒ほどをかけて正常終了するようになった。

それでも通知は来ない

が、通知はまだ来ない。

通知 Lambda 自身のログを確認すると、Log not found. と稼働のたびにログが残っていた。

filterLogEvents は1度の呼び出しでロググループの全ログを回収することはできず、容量や件数に制限がある。
検索対象とできるログが残っている場合 token が返却されるので、その token とともに再度 filterLogEvents を実行することで続きのログを獲得できる。
件のコードでは、いつまでもログを探し続けないように filterLogEvents にループの上限数が定められていた。(const MAXIMUM_FETCH_COUNT = 100;)

つまりログの発見できない呼び出しを100回行いあきらめて Log not found.
という至極無駄な行為がなされている。

実際、ローカルで同等のコードを実行し1アクセスごとにレスポンスを確認すると以下のような空っぽの返却が来ていた。

{
  events: [],
  searchedLogStreams: [],
  nextToken: XXXXX
}

よくわからないので時間に任せる

CloudWatchLogs ではロググループごとにログの保持期間を指定できる。
ログが多すぎて問題があるなら一旦時に任せてログが少なくなるのを待ってみることにした。

要するに調査が行き詰まって半ばあきらめた形である。(また過大なアクセスがあったらどうする気なのだろうか。)

そしてその日がやってきて、ロググループの保存されているバイト数がガクッと下がったことを確認した後に通知を稼働させた。

が、slack に通知は来なかった。
事態に変化はなく、いまだに無意味な100ループを行っていた。

ログストリームは空になっても削除されない

ログを書き込む際にログストリーム名にログの日時を記載していたのだが、どう見ても保持期間外のログストリームが生存している。

どうやらログストリームは空になっても削除されないようである。
(よく考えれば日時の概念があるのはログイベントのみでログストリームには日時は関係ない & もともと空のログストリームを作成することも可能なので当たり前のことではある。)

log group ⊃ log stream ⊃ log event の関係だが、保持期間を過ぎて破棄されるのは log event のみで log stream は残る。

つまり問題は単純なログイベントの数ではなくログストリームの数であることが分かった。

ログストリームで絞り込む

filterLogEvents をコールする際、startTimeとendTimeを指定しているので問題ないと思っていたが、前述の通りログストリームには時間の概念がないため、取得してみないと指定期間のログイベントが含まれるかどうかがわからず、無駄なログストリームを回収して「ログイベントなかったッス」と延々と繰り返しているらしい。

そこで、幸いログストリーム名には日付をつけていたのでこれで絞り込みを行うようにした。
logStreamNamePrefix を指定する。

handler.js

'use strict';

const AWS = require('aws-sdk');
const request = require('request-promise');

// for Date class
const SECOND = 1000;
const MINUTE = 60 * SECOND;
// ログを取得する際に最大でページングする回数
const MAXIMUM_FETCH_COUNT = 100;
const ERROR_COLOR = '#dc143c';

// Config
const slackToken = process.env.SLACK_TOKEN;
const conversationId = process.env.CHANNEL_ID;
const logGroupName = process.env.LOG_GROUP;
const executionIntervalMin = process.env.EXECUTION_INTERVAL_MIN * MINUTE;
// 直近のログはストリームに入っていない可能性があるため適宜取得終了時間を調整する
const cwlClient = new AWS.CloudWatchLogs({apiVersion: '2014-03-28'});


module.exports.slack = async () => {
  const now = Date.now();
  // バッチ実行間隔で時間を丸める
  // ex. (5分毎)9:32->9:30 ex2. (10分毎)16:16 -> 16:10
  const logEndAt = Math.floor(now / (executionIntervalMin)) * executionIntervalMin
  const logStartAt = logEndAt - executionIntervalMin;
  // 指定したロググループのログを取得
  console.log("logStartAt, logEndAt: ", new Date(logStartAt), new Date(logEndAt));
  const logs = await getLogs(logGroupName, logStartAt, logEndAt);
  if (logs.length < 1) {
    console.log('Log not found.');
    return {
      statusCode: 200,
    };
  };
  // 取得したログをSlackAPIの形式に加工
  const attachments = summarizeLogs(logs).map(error => {
      return {
        title: error.title + ` ${error.count}件`,
        text: error.message,
        footer: error.stack_trace,
        color: ERROR_COLOR,
      }
    });
  // Slackへ通知
  await postMessage(slackToken, conversationId, attachments);

  return {
    statusCode: 200,
  };
};

const postMessage = (token, channel, attachments) => {
  const options = {
    uri: 'https://slack.com/api/chat.postMessage',
    method: 'POST',
    headers: makeSlackHeaders(token),
    json: {channel, attachments, text: "Errors" },
  };

  return request(options);
};

const makeSlackHeaders = token =>
  ({
    'Content-type': 'application/json',
    'Authorization': `Bearer ${token}`,
  });

const getLogs = async(logGroupName, startAt, endAt) => {
  let logs = [];
  let nextToken = null;
  const logStreamNamePrefix = formatDate(startAt);
  for (let i = 0; i < MAXIMUM_FETCH_COUNT; i++) {
    let res = await filterLogEvents(logGroupName, startAt, endAt, nextToken, logStreamNamePrefix);
    logs = logs.concat(res.events);
    if (!res.nextToken) {
      break;
    }
    nextToken = res.nextToken;
  }
  return logs;
}

const filterLogEvents = (logGroupName, startAt, endAt, nextToken, logStreamNamePrefix) => {
  return cwlClient.filterLogEvents({
    logGroupName: logGroupName,
    endTime: endAt,
    startTime: startAt,
    nextToken: nextToken,
    logStreamNamePrefix: logStreamNamePrefix
  }).promise();
}

// 取得したログをtitle毎で集計
const summarizeLogs = logs => {
  return logs.reduce((acc, current) => {
    let msg;
    try {
      msg = JSON.parse(current.message);
    } catch(e) {
      msg = {
        title: current.message.split("\n")[0],
        message: current.message,
      }
    };
    let found = acc.find(element => element.title === msg.title);
    if (found) {
      found.count++;
    } else {
      acc.push({
        count: 1,
        ...msg
      });
    };
    return acc;
  }, []);
}

const formatDate = (unix) => {
  const date = new Date(unix);
  const year = date.getFullYear();
  // getMonth は 0 から返すので +1
  // 0埋めして桁数を合わせるために00つけて末尾2桁回収
  const month = ('00' + (date.getMonth()+1)).slice(-2);
  const day = ('00' + date.getDate()).slice(-2);

  return `${year}-${month}-${day}`;
}

これでついに通知が回復した。

ログストリーム名の指定は回収前、期間指定は回収後に行われる。
無駄なログストリームの回収は検索効率を下げ、必要な FilterLogEvents のコール数を増加させる。
以上のように考えられる。

今回はこれで解決としたが、日付絞り込みなため1日のエラーログが何十万、何百万といった数になったら当日限りで問題が再発するかもしれない。
(1日にそこまでエラーが出る時点で本当の問題は別にある気がするが。)

結論

ログの保持期間を過ぎても削除されるのはログイベントのみでログストリームは空になっても残る。
filterLogEvents は対象になるログストリームが多いと極端に検索効率が下がる。
なので適切にログストリームに対する絞り込みを行う必要がある。
また、ログストリーム数の無用な増大は避けるのがよい。