タイトル通りのLambda関数を作ろうと思いましたがPythonだとrequestsモジュール等をレイヤーにして
読み込まないといけないとか色々面倒に感じたのでボツに。
(結局のところ、Rubyで書きました、aws-sdk読み込むだけで特に面倒な手順はないので・・・)

自分の思い出としてこちらに残します。

※動作確認はしていないので、参考程度にお願いします。
※ご指摘、改善点あればコメントください。

仕様

大雑把に説明するとDocumentDBクラスターとDocumentDBクラスターに紐づくDocumentDBインスタンスのステータスをチェックし、「available」以外のステータスのものがあればメールを送信します。

前提

監視対象のAWS DocumentDBクラスターが存在していること

事前準備

本筋ではないので割愛します。

Lambda関数の作成
Lambda関数のロールにポリシー追加(AmazonDocDBReadOnlyAccess, AmazonSNSFullAccess)
requestsモジュール等を動作させるためにレイヤーを作る(こちらの方の記事が参考になります)
Lambda関数の環境変数の設定(TOPIC_ARN, DB_CLUSTER_IDENTIFIER)
Amazon SNSのトピック作成
Amazon SNSのトピックにサブスクリプション追加(メールの送信先)

今時メールじゃなくてもSlackやTeams連携の方が利便性は高いかもしれませんね。

実際のソース

lambda_function.py

# -*- coding: utf-8 -*-

import os
import boto3
import requests
import logging

# Lambda関数の環境変数に設定した値を取得
# AWS SNSトピックのARN(メール送信先)
TOPIC_ARN = os.environ['TOPIC_ARN']
# DocumentDBクラスター識別子(ステータス取得対象のクラスターを指定)
DB_CLUSTER_IDENTIFIER = os.environ['DB_CLUSTER_IDENTIFIER']

def lambda_handler(event, context):
    # ロガーの設定
    logger = logging.getLogger()
    logger.setLevel(logging.INFO)

    # DocumentDBのクライアント
    docdb_client = boto3.client('docdb')
    # AWS SNSのクライアント
    sns_client = boto3.client('sns')

    # ステータスの結果保存のための配列
    db_statuses = []

    try:
        logger.info('Start >>> get DocumentDB status.')
        # dbクラスターの情報取得
        cluster_info = docdb_client.describe_db_clusters(DBClusterIdentifier=DB_CLUSTER_IDENTIFIER)
        cluster_status = cluster_info['DBClusters'][0]['Status']
        db_statuses.append(
            {
                DB_CLUSTER_IDENTIFIER : cluster_status
            }
        )

        instances = cluster_info['DBClusters'][0]['DBClusterMembers']

        for instance in instances:
            instance_id = instance['DBInstanceIdentifier']
            instance_info = docdb_client.describe_db_instances(DBInstanceIdentifier=instance_id)
            status = instance_info['DBInstances'][0]['DBInstanceStatus']

            db_statuses.append(
                {
                    instance_id : status
                }
            )

        # 1件でも'available'以外の場合、通知処理実行
        # ※'available'以外にも通知しないステータスを追加する場合は、in以降の配列にステータスの文字列を追加する
        is_notify = not all(list(i.values())[0] in ['available'] for i in db_statuses)

        if is_notify:
            # 通知処理の実行

            # メールに記載する内容を作成
            subject = '【AWS DocumentDB】An instance with an abnormal status has been detected.'
            message = ''
            for status_info in db_statuses:
                key = list(status_info.keys())[0]
                value = status_info[key]
                message += f'{key}: {value}\n'

            # メール送信
            sns_client.publish(
                TopicArn = TOPIC_ARN,
                Subject = subject,
                Message = message,
            )

            logger.info('DocumentDB status list↓')
            logger.info(message)
            logger.info('Completed successfully.')

    except Exception as e:
        logger.error(e)

    finally:
        logger.info('End >>> get DocumentDB status.')

※作成したレイヤーの追加は忘れずに！

あとはCloudWatch Eventsにルール追加

例えばcron式に「*/10 * * * ? *」を設定することで10分毎にDocumentDBのステータスをチェックします。
もし異常があればメールが飛んでくるはずです。

ターゲットとして、監視対象のクラスター識別子を渡してあげるようにすれば、Lambda関数内の引数eventで参照することができるのでLambda関数はそのままにルールを追加して環境毎に監視することも可能です。

「available」以外の監視でいいのか？

「available」以外という条件では、定期メンテナンスの場合もメールが飛んでくることがあると思うので
定期メンテナンスの場合の「maintenance」も足して

is_notify = not all(list(i.values())[0] in ['available', 'maintenance'] for i in db_statuses)

とするのもいいかもしれません。

参考

Amazon DocumentDB インスタンスのステータスのモニタリング
 Boto3/Docs/Available services/DocDB
Boto3/Docs/Available services/SNS

AWS DocumentDBのステータスを通知するLambda関数

仕様

前提

事前準備

実際のソース

あとはCloudWatch Eventsにルール追加

「available」以外の監視でいいのか？

参考