AWS CloudWatchEventsの時間指定をCDKでラクにする。

CloudWatchEventsってUTCでしか動かないじゃないですか。
いつもそれを忘れて時間指定失敗するので、CDKでJST→UTCに変換することにしたんですよ。

CDKでCloudWatchEventsを定義する

取り敢えずCloudWatchEventsの定義のとこだけ。
これをよしなにStepFunctionsとかと組み合わせるなり何なりする。

# バッチスケジュールをJSTで定義する（cron式で * にする所は空でよい）
props['schedule'] = {
    'minute': 40,
    'hour': 9,
    'week_day': 'MON-FRI'
}

# JSTをUTCに変換する
if props['schedule'].get('hour') is not None:
    props['schedule']['hour'] = props['schedule']['hour'] - 9
    if props['schedule']['hour'] < 0:
        props['schedule']['hour'] += 24
        if props['schedule'].get('day') is not None:
            props['schedule']['day'] = props['schedule']['day'] - 1

# JST 月初の朝9時までに動作する場合は、月末っぽい日付で動作させてLambda等で判定する
if props['schedule'].get('day') == 0:
    props['schedule']['day'] = '28-31'

# props['schedule']の中身をstrに変換する
for key in props['schedule'].keys():
    props['schedule'][key] = str(props['schedule'][key])

# Cloud Watch Events
Rule = events.Rule(
    app, f"Batch-{id}",
    schedule=events.Schedule.cron(**props['schedule'])
)

月初の9:00より前（UTC月末）の判定Lambda

今がJSTの月初（1日）かどうかだけ判定してTrueかFalseが返る。
StepFunctionsに組み込むならこれの返り値を判定して次に流すか終わるかすればよいし、単純な処理ならこのLambda内にてif result:で続けて処理してしまってもよいかと思います。

import datetime
import calendar


def lambda_handler(event, context):
    print('===Start Lambda===')
    print(event)

    dt_now = datetime.datetime.now()
    monthrange = calendar.monthrange(dt_now.year, dt_now.month)[1]

    result = dt_now.day == monthrange

    print(result)

    return result

※printは動作確認のためだけに書いてるので要らなければ消してよい。

おわり

何かもっといい方法とかご存知の賢者様はご指摘いただけると嬉しいです。