ecsクラスタにあるコンテナたちのヘルスチェックを AWS Lambda(boto3)で実装してみた

概要

ECSクラスタ内でバッチ処理を行うコンテナが常駐しているのですが、
ここ最近Fargateのメモリを食い潰してて、よろしくない挙動をしていました。

問題はつくりにあったのですが、それより問題だったのはヘルスチェックが実装されていなかったこと。

そこで既存のコンテナたちを変更することなく、
boto3を利用したヘルスチェックを実装することにしました。

図にしてみると

元々内部でログ検知の仕組みがLambdaで動いているので、アラート自体はそちらに任せることにしました。
ヘルスチェックのログを別関数でチェックしてもらうイメージ。

やったこと

とりあえずLambdaだけで構築なので、Serverless Frameworkでやりました。
ただ一点厄介事がありました・・（詳細は後述

lambda実装

# coding: UTF-8
import datetime
import time
import boto3
import os

from src.log import get_logger

logger = get_logger()


def lambda_handler(event, context):
    logger.info("START")
    client = boto3.client('ecs')

    clusterName = "hoge-cluster"
    # ECSサービス一覧を取得
    serviceList = client.list_services(
        cluster=clusterName,
    )

    # ECSサービス詳細取得
    describeService = client.describe_services(
        cluster=clusterName,
        services=serviceList["serviceArns"],
    )

    # ログ取得
    logGroupName = "/ecs/hoge-service"
    queryId = get_log_queryId(logGroupName)
    logResults = get_log_insights(queryId)

    for service in describeService['services']:
        # タスク定義取得
        task = client.describe_task_definition(
            taskDefinition=service["taskDefinition"]
        )

        containerDefinition = task['taskDefinition']['containerDefinitions'][0]
        taskName = containerDefinition['name']

        # 対象タスクのログが取得できない場合エラー
        if (logResults is not None):
            if (taskName in logResults["results"]):
                logger.error(f"{taskName}タスクとまってます")
            else:
                logger.info(f"{taskName}タスク動いてます")

    logger.info("END")


def get_log_queryId(logGroupName):
    """
    cwl insightsで取得するログのQueryIdを返す
    直近5分間のログを検索範囲とする
    """
    # logstream 取得
    five_minutes_ago = datetime.datetime.now() - datetime.timedelta(minutes=5)
    startTime = five_minutes_ago.replace(second=0, microsecond=0)
    endTime = startTime + \
        datetime.timedelta(minutes=5) - datetime.timedelta(milliseconds=1)

    # cwl insights
    client = boto3.client('logs')
    start_query_res = client.start_query(
        logGroupName=logGroupName,
        startTime=int(startTime.timestamp()),
        endTime=int(endTime.timestamp()),
        queryString="""
        fields @ logStream
        | parse @ logStream '*/*/*' as a, task, b
        | stats count() as logcount by task
        """,
        limit=20
    )

    # cwl insightsでQueryId発行
    queryId = start_query_res['queryId']
    logger.info(f"queryIdは{queryId}")

    return queryId


def get_log_insights(queryId):
    """
    QueryIdを利用してcwl insightsを取得
    """

    client = boto3.client('logs')
    response = None
    count = 1

    while response is None or response['status'] == 'Running':
        # 10回試して取得出来ない場合は処理終了
        if (count > 10):
            break

        print('Waiting for query to complete ...')
        time.sleep(1)
        response = client.get_query_results(
            queryId=queryId
        )
        count += 1

    return response

serverlessのプラグインが使えない・・

slsで速攻で実装だ〜と思ったのですが、つまづきました・・・
サブスクリプションフィルタの設定はpluginがあったのでいけると思ったんですが、うまくいかず。
見た所メンテがされておらず、色々試したが断念・・・
LambdaだけでCfn用意するのは微妙だったので今回は見送り。

サブスクリプションフィルタの設定は手動で実行としました。。

デプロイ

GitHub Actions上でsls deploy実行させています
最近はghでリリースする事を覚えて楽しんでます🍕

gh release create 202103xx.1 -t "healthcheck release" -n "gogo"

所感：CDK書こうと思った

TypeScript書いてみたいだけです。
こんな感じでサブスクリプションフィルタの設定出来るので↓

    //subscripition filter設定
    fn.logGroup.addSubscriptionFilter("hogeSubscriptionFilter",{
      destination: new logs_destinations.LambdaDestination(destFn),
      filterPattern: logs.FilterPattern.allTerms("?WARNING", "?ERROR"),
    })

以下参照↓
https://docs.aws.amazon.com/cdk/api/latest/docs/aws-logs-readme.html#subscriptions-and-destinations

CDKも今後活用していきたいと思います！

参考ドキュメント

https://qiita.com/chii-08/items/e20651e7912596e9a556
https://docs.aws.amazon.com/cdk/api/latest/docs/aws-logs-readme.html#subscriptions-and-destinations
https://cli.github.com/