概要

前回メディアのログをOpenSearchに集約させたことをお話しました。

cuebic.hatenablog.com

このとき課題としてストレージに上限があり、溜まり続けると新しいログが転送できないという問題がありました。
OpenSearchのAPIを叩くことで削除することが可能ですが、手動でAPIリクエストを送るのは面倒です。

curl  -XDELETE https://endpoint.com/<index>

今回EventBridge + Lambdaで自動で古いインデックスを削除できるようにしましたので、紹介いたします。
(シェルスクリプトのままLambdaに実装もできなくはないですが、他のエンジニアにも読めるように~~見よう見まねで~~Pythonで書いてみました)

アーキテクチャ構成図

かなりシンプルなアーキテクチャ構成図ですが使っているものは実際にこれだけです。
EventBridgeが毎日深夜1時にトリガーをかけてLambdaを実行させます。

Lambdaソースコード

import requests
import json
import os
import boto3
from datetime import date,timedelta
import datetime
import calendar
from requests.exceptions import Timeout

def lambda_handler(event, context): 
    
    #パラメーターストアからOpenSearch 認証情報を取得
    ssm = boto3.client('ssm','ap-northeast-1')
    OpenSearch_user     = ssm.get_parameter(
        Name='/opensearch/cuebic-media-log/user',
        WithDecryption=True
    )["Parameter"]["Value"]
    
    OpenSearch_password = ssm.get_parameter(
        Name='/opensearch/cuebic-media-log/pass',
        WithDecryption=True
    )["Parameter"]["Value"]
 
    # 過去日付情報取得
    ut = datetime.datetime.now()
    ago_30 = ut-timedelta(days=30)
    ago_90 = ut-timedelta(days=90)
    
    # 対象URLリスト
    index_list = [ ~~~~~~ ]

    # タイムアウト値設定(コネクションタイムアウト3.0秒、リードタイムアウト7.5秒)
    connect_timeout = 3.0
    read_timeout    = 7.5
    try:
        for index in index_list:
            #30日過ぎたログをクローズさせる
            response = requests.post('https://endpoint.com/'+index+ago_30.strftime('%Y%m%d')+'/_close', auth=(OpenSearch_user, OpenSearch_password), timeout=(connect_timeout, read_timeout))
            #90日すぎたログを削除する
            response = requests.delete('https://endpoint.com/'+index+ago_90.strftime('%Y%m%d'), auth=(OpenSearch_user, OpenSearch_password), timeout=(connect_timeout, read_timeout))

        
    except Timeout:
        print('Timeout')
        return

    except:
        print('Unkown exception')
        return
    
    return {
        "statusCode": 200,
        "headers": {
            "Content-Type": "application/json"
        },
        "body": json.dumps({
            "message": response.text
        })
    }

ソースコード解説

LambdaがOpenSearchへアクセスする際に使用する認証情報はSystems Managerのパラメーターストアを利用しています。

Secrets Managerと違ってパラメーターのローテーション機能はありませんが、無料で使えるので今回はこちらを採用しました。

    ssm = boto3.client('ssm','ap-northeast-1')
    OpenSearch_user     = ssm.get_parameter(
        Name='/opensearch/cuebic-media-log/user',
        WithDecryption=True
    )["Parameter"]["Value"]

各種ログのインデックス名には日付を付けていますので、インデックス名の日付が30日以上前のものはクローズにし画面上から検索できないようにする。90日以上前のものはOpenSearchから完全に削除するようにしています。

基本的にインデックス名はログ名+yyyymmddでしたので、ログ名部分を配列変数にしてfor構文でAPIを実行するようにしています。

課題

今の所うまくいっているので問題にはなっていませんが、メッセージ出力が最後に実行された responseに対する結果のみを出力しますのでもし実行に失敗した場合のログ調査ができない問題があります。

    return {
        "statusCode": 200,
        "headers": {
            "Content-Type": "application/json"
        },
        "body": json.dumps({
            "message": response.text #リストの最後に格納されているインデックス削除の結果しか出力されない
        })
    }

ログを配列に格納すれば解決できそうな気がしますので、Pythonに詳しい人に相談してみようと思います。

所感

OpenSearchに溜まり続けるインデックスの削除をEventBridge + Lambdaで自動化してみました。OpenSearchの料金は事前に確保したEBSのストレージサイズに応じてコストが上がるので、事前確保分を抑えて定期的に削除する運用にすれば良い改善になるのではないかなと思います。

おまけ

curlでAPIを実行する方法をどうやってPythonで実行すればいいのか最初わからなかったですが、こちらのコンバーターサイトを使えばcurlコマンドの内容をGoやPythonに変換してくれるのでプログラミングの苦手な私には重宝しました。みなさんも使ってみてください。

curlconverter.com

CUEBiC TEC BLOG

キュービックTECチームの技術ネタを投稿しております。

OpenSearchの古いインデックスを定期的に削除できるようにした

概要

アーキテクチャ構成図

Lambdaソースコード

ソースコード解説

課題

所感

おまけ