kakakakakku blog

Weekly Tech Blog: Keep on Learning!

LocalStack を使って CloudWatch Logs サブスクリプションフィルタをローカル環境で試す

Amazon CloudWatch Logs サブスクリプションフィルタを使ってログを AWS Lambda に流す構成を AWS アカウントにデプロイする前に LocalStack にデプロイして確認してみた❗

docs.aws.amazon.com

LocalStack は Amazon CloudWatch Logs サブスクリプションフィルタもサポートしている👌しかし注意点はあって,サブスクリプションフィルタに設定するフィルタパターン(JSON フィルタ・正規表現フィルタなど)は LocalStack Pro でサポートされている💡よって,今回はフィルタなし(すべてのログを流す)で試す.

docs.localstack.cloud

あくまでサンプルとして以下の構成を LocalStack を使ってローカル環境にデプロイする.

サンプルコード

👾 template.yaml(AWS SAM テンプレート)

今回は以下のように AWS SAM テンプレートを実装した📝AWS Lambda 関数は log-senderlog-receiver の2つを準備して,log-sender の Amazon CloudWatch Logs ロググループにサブスクリプションフィルタを設定してある👌

AWSTemplateFormatVersion: 2010-09-09
Transform: AWS::Serverless-2016-10-31

Resources:
  LogSenderFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: log-sender
      CodeUri: src/
      Handler: log-sender.lambda_handler
      Runtime: python3.12
      Architectures:
        - x86_64
  LogSenderLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: /aws/lambda/log-sender
  LogSenderSubscriptionFilter:
    Type: AWS::Logs::SubscriptionFilter
    Properties:
      LogGroupName: !Ref LogSenderLogGroup
      FilterPattern: ""
      DestinationArn: !GetAtt LogReceiverFunction.Arn
  LogReceiverFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: log-receiver
      CodeUri: src/
      Handler: log-receiver.lambda_handler
      Runtime: python3.12
      Architectures:
        - x86_64

👾 log-sender.py

log-sender ではシンプルに {"id": "27a1cc30-b4c4-4192-9db9-d19962fe8f33", "message": "sample message"} のように UUID と固定メッセージを含んだ構造化ログ (JSON) を出力する👌

import json
import uuid


def main():
    print(
        json.dumps(
            {
                'id': str(uuid.uuid4()),
                'message': 'sample message',
            }
        )
    )


def lambda_handler(event, context):
    main()


if __name__ == '__main__':
    main()

👾 log-receiver.py

log-receiver はサブスクリプションフィルタから流れてきたログをそのままログに出力する.サブスクリプションフィルタ経由だとログは GZIP 圧縮と Base64 エンコードで変換された状態になるけど,今回は Powertools for AWS Lambda (Python) の Event Source Data Classes で CloudWatchLogsEventCloudWatchLogsDecodedData を使ってお手軽に実装した👏 便利〜 \( 'ω')/

docs.powertools.aws.dev

import json

from aws_lambda_powertools.utilities.data_classes import CloudWatchLogsEvent
from aws_lambda_powertools.utilities.data_classes.cloud_watch_logs_event import CloudWatchLogsDecodedData


def main(event):
    event = CloudWatchLogsEvent(event)
    decompressed: CloudWatchLogsDecodedData = event.parse_logs_data()
    logs = decompressed.log_events
    for log in logs:
        print(log.message)


def lambda_handler(event, context):
    main(event)


if __name__ == '__main__':
    with open('./events/event.json', 'r') as f:
        event = json.load(f)

    main(event)

ちなみに @event_source(data_class=CloudWatchLogsEvent) のようにデコレータを使えば event = CloudWatchLogsEvent(event) という値の詰め直しは不要になるけど,Lambda コンテキストに依存していてローカル開発がしにくく採用しなかった.前に紹介した Powertools for AWS Lambda (Python) の Validation でも Lambda コンテキスト依存を避ける実装を紹介していたりする💡

kakakakakku.hatenablog.com

動作確認

samlocal コマンドでビルド・デプロイをして,awslocal コマンドで AWS Lambda 関数を実行した.

$ samlocal build
$ samlocal deploy

$ awslocal lambda invoke --function-name log-sender outfile

LocalStack の Amazon CloudWatch Logs を確認すると,期待通りに log-senderlog-receiver どちらにも同じログが出ていた👌

log-sender

{"id": "27a1cc30-b4c4-4192-9db9-d19962fe8f33", "message": "sample message"}

log-receiver

{"id": "27a1cc30-b4c4-4192-9db9-d19962fe8f33", "message": "sample message"}

localstack-utils: 単体テスト実行時に使い捨て可能な LocalStack を起動しよう

LocalStack から公式に提供されている localstack-utils を使うと,pytest など Python で単体テストを実行するときに一時的な(使い捨て可能な)LocalStack 環境を起動できる🌍

docs.localstack.cloud

ちなみに僕は普段仕事で testcontainers-pythonLocalStackContainer を使ってて最高に便利なんだけど,結果的に localstack-utils もほとんど同じように使うことができた👌

kakakakakku.hatenablog.com

現在最新は localstack-utils 1.0.1 だった.

github.com

pypi.org

サンプルコード

👾 app.py

まずは Amazon DynamoDB の Forum テーブルからアイテムを取得する search_forum() 関数を今回のテスト対象とする.Forum テーブルというのは Amazon DynamoDB のドキュメントに載っているサンプルでそのまま使うことにした.

そして,boto3 client は環境変数 ENV によって3種類作れるようにしてある👌

  • local: ローカル開発用 (LocalStack)
  • test: テスト用 (localstack-utils / LocalStack)
  • その他: 実際の AWS アカウント

ちなみに localstack-utils のデフォルト設定では 4566 ポートで LocalStack を起動する.それだとローカル開発用の LocalStack とポートが競合してしまうため,今回は 14566 ポートで起動することにした.

import os

import boto3

TABLE_NAME = 'Forum'

if os.environ['ENV'] == 'local':
    dynamodb = boto3.client('dynamodb', endpoint_url='http://localhost:4566')
elif os.environ['ENV'] == 'test':
    dynamodb = boto3.client('dynamodb', endpoint_url='http://localhost:14566')
else:
    dynamodb = boto3.client('dynamodb')


def search_forum(name):
    return dynamodb.get_item(
        TableName=TABLE_NAME,
        Key={
            'Name': {'S': name},
        },
    )

👾 test_app.py

pytest 実行時に呼び出すフィクスチャとして @pytest.fixture デコレータで _setup() 関数を実装した.localstack-utils には LocalStack を起動する startup_localstack() 関数と LocalStack を停止する stop_localstack() 関数が実装されているため,それを _setup() 関数内で実行している.14566 ポートで起動する設定もしてある👌

そして,Amazon DynamoDB テーブル Forum を作って,サンプルデータ(アイテム)を1つ登録している.最後にテストケースとしては search_forum() 関数を呼び出して「アイテムを取得できる場合」「アイテムを取得できない場合」を確認している✔️

import boto3
import pytest
from app import search_forum
from localstack_utils.localstack import startup_localstack, stop_localstack

TABLE_NAME = 'Forum'


@pytest.fixture(scope='module', autouse=True)
def _setup():
    startup_localstack(gateway_listen='0.0.0.0:14566')

    dynamodb = boto3.client('dynamodb', endpoint_url='http://localhost:14566')

    dynamodb.create_table(
        TableName=TABLE_NAME,
        KeySchema=[
            {
                'AttributeName': 'Name',
                'KeyType': 'HASH',
            }
        ],
        AttributeDefinitions=[
            {
                'AttributeName': 'Name',
                'AttributeType': 'S',
            }
        ],
        BillingMode='PAY_PER_REQUEST',
    )

    item = {
        'Name': {'S': 'Amazon DynamoDB'},
        'Category': {'S': 'Amazon Web Services'},
        'Threads': {'N': '2'},
        'Messages': {'N': '4'},
        'Views': {'N': '1000'},
    }

    dynamodb.put_item(TableName=TABLE_NAME, Item=item)

    yield

    stop_localstack()


def test_search_forum():
    item = search_forum('Amazon DynamoDB')
    assert item['Item']['Category']['S'] == 'Amazon Web Services'
    assert item['Item']['Views']['N'] == '1000'

    item = search_forum('Amazon S3')
    assert 'Item' not in item

動作確認

期待通り実行できた👏

$ ENV=test pytest -p no:warnings
=========================================================================================== test session starts ============================================================================================
(中略)
configfile: pyproject.toml
collected 1 item

tests/test_app.py .                                                                                                                                                                                  [100%]

============================================================================================ 1 passed in 4.97s =============================================================================================

詳しくはコード参照

localstack-utils にはドキュメントというドキュメントはなく,オプションなど詳しくは GitHub のコードを確認する必要がある.例えば今回は startup_localstack() 関数に gateway_listen を設定したけど,他にも image_name / tag / pro なども設定できる👌

github.com

まとめ

localstack-utils を使って Python で単体テストを実行するときに一時的な(使い捨て可能な)LocalStack 環境を活用しよう❗️

Apache Ivy で Guava を取得するときに javadoc.jar と sources.jar を除外する

Apache Ant と Apache Ivy を使って Maven Repository から Guava を取得すると不要な javadoc.jarsources.jar も一緒に取得されてしまうときの解決策を2つメモしておく〜📝

再現確認

まず,Apache Ant の build.xml にデフォルト設定の ivy:retrieve タスクを書いておく.

<target name="resolve">
    <ivy:retrieve/>
</target>

そして,Apache Ivy の ivy.xml には Maven Repository に載っている dependency タグを書く.ちなみに今回は取得するファイル数を減らすために意図的に transitive="false" を付けておく👌

<ivy-module version="2.0">
    <info organisation="sandbox" module="module"/>
    <dependencies>
        <dependency org="com.google.guava" name="guava" rev="33.2.1-jre" transitive="false"/>
    </dependencies>
</ivy-module>

ant resolve コマンドを実行すると,以下の3ファイルが取得されてしまう😇

guava-33.2.1-jre-javadoc.jar
guava-33.2.1-jre-sources.jar
guava-33.2.1-jre.jar

解決策1

解決策の一つは build.xmlivy:retrieve タスクに type="jar, bundle" を設定しておくこと.そうすると guava-33.2.1-jre.jar のみを取得できる❗️

<ivy:retrieve type="jar, bundle"/>

ant.apache.org

解決策2

もう一つの解決策は ivy.xmldependency タグの中に artifact タグを設定しておくこと.同じく guava-33.2.1-jre.jar のみを取得できる❗️

<ivy-module version="2.0">
    <info organisation="sandbox" module="module"/>
    <dependencies>
        <dependency org="com.google.guava" name="guava" rev="33.2.1-jre" transitive="false">
            <artifact name="guava"/>
        </dependency>
    </dependencies>
</ivy-module>

ant.apache.org

LocalStack を使って Amazon Transcribe をローカル環境で操作する

Amazon Transcribe で Speech-to-Text を実現するときに,LocalStack を使えば Amazon Transcribe を「AWS アカウントを使わずにローカル環境で」動かせる👌LocalStack 自体は仕事でもプライベートでも使ってるけど,Amazon Transcribe は今まで試したことがなくて,今回試してみた❗️

Amazon Transcribe API のサポート状況

もちろん LocalStack がすべての Amazon Transcribe API をサポートしているわけではないけど,カバレッジは以下で確認できる.StartTranscriptionJob / ListTranscriptionJobs / GetTranscriptionJob はサポートされてて基本的な操作はできそう.

docs.localstack.cloud

Vosk

そもそも Amazon Transcribe の Speech-to-Text の仕組みは公開されていないのでは?と疑問に感じるけど,LocalStack の内部では Vosk が使われているとドキュメントに載っていた💡英語・日本語・ドイツ語など,多くの言語をサポートしている.もちろん Vosk の精度に依存せず,LocalStack をインタフェース確認として使うのが良いと思う👌

alphacephei.com

注意点

最初に Amazon Transcribe を試したところ,TranscriptionJob(文字起こしジョブ)を実行すると以下のエラーが出てしまって困ったけど,何やら関連する issue もあって,プラットフォームが関係するようだった.今回は Apple M3 を使っているため localstack:latest-amd64 イメージで LocalStack を起動したら問題なく動くようになった👌試すときに注意が必要かなと🚨

cannot load library '/var/lib/localstack/lib/python-packages/lib/python3.11/site-packages/vosk/libvosk.so': libatomic.so.1: cannot open shared object file: No such file or directory

試す

まずは AWS CloudFormation で Amazon S3 バケットを作る.LocalStack は AWS CloudFormation / Amazon S3 もサポートしている❗️

AWSTemplateFormatVersion: '2010-09-09'

Resources:
  S3Bucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: transcribe-sandbox

awslocal コマンドで AWS CloudFormation スタックをデプロイして,transcribe-sandbox バケットを確認した👌

$ awslocal cloudformation deploy --stack-name transcribe-sandbox --template-file templates/template.yaml

Waiting for changeset to be created..
Waiting for stack create/update to complete
Successfully created/updated stack - transcribe-sandbox

$ awslocal s3api list-buckets | jq -r '.Buckets[].Name'
transcribe-sandbox

次に AWS CLI で Speech-to-Text をする mp3 を transcribe-sandbox バケットにアップロードする.今回はサンプルとして効果音ラボ「よろしくお願いします(冷静な魔術師)」を使う.準備完了〜 \( 'ω')/

$ awslocal s3 cp ./mp3/wizard-greeting1.mp3 s3://transcribe-sandbox/wizard-greeting1.mp3
upload: mp3/wizard-greeting1.mp3 to s3://transcribe-sandbox/wizard-greeting1.mp3

そして今度は Python (boto3) スクリプトで Amazon Transcribe の TranscriptionJob(文字起こしジョブ)を実行する❗️

import uuid

import boto3

transcribe = boto3.client('transcribe', endpoint_url='http://localhost:4566')

response = transcribe.start_transcription_job(
    TranscriptionJobName=f'job-{str(uuid.uuid4())}',
    Media={
        'MediaFileUri': 's3://transcribe-sandbox/wizard-greeting1.mp3',
    },
    LanguageCode='ja-JP',
)

print('TranscriptionJobName: ' + response['TranscriptionJob']['TranscriptionJobName'])
print('TranscriptionJobStatus: ' + response['TranscriptionJob']['TranscriptionJobStatus'])

実行すると TranscriptionJobName などのレスポンスを確認できる.

TranscriptionJobName: job-b9b2d46a-6621-4890-b09f-08fb3d5f059e
TranscriptionJobStatus: IN_PROGRESS

少し待つと IN_PROGRESS から COMPLETED になる.LocalStack の Resource Browser(マネジメントコンソール)で TranscriptionJob(文字起こしジョブ)の詳細を確認できる.

もちろん AWS CLI の transcribe get-transcription-job コマンドでも確認できる👌

$ awslocal transcribe get-transcription-job --transcription-job job-b9b2d46a-6621-4890-b09f-08fb3d5f059e | jq .
{
  "TranscriptionJob": {
    "TranscriptionJobName": "job-b9b2d46a-6621-4890-b09f-08fb3d5f059e",
    "TranscriptionJobStatus": "COMPLETED",
    "LanguageCode": "ja-JP",
    "MediaSampleRateHertz": 44100,
    "MediaFormat": "wav",
    "Media": {
      "MediaFileUri": "s3://transcribe-sandbox/wizard-greeting1.mp3"
    },
    "Transcript": {
      "TranscriptFileUri": "http://s3.localhost.localstack.cloud:4566/transcribe-sandbox/job-b9b2d46a-6621-4890-b09f-08fb3d5f059e.json?AWSAccessKeyId=__internal_call__&Signature=pihlxHmUVQOeBuoHSjUCy1Yz0Gs%3D&Expires=1718111742"
    },
    "StartTime": "2024-06-11T21:38:34.119979+09:00",
    "CreationTime": "2024-06-11T21:38:34.119496+09:00",
    "CompletionTime": "2024-06-11T21:40:32.940129+09:00"
  }
}

最終的に transcribe-sandbox バケットにアップロードされた job-b9b2d46a-6621-4890-b09f-08fb3d5f059e.json を確認すると,期待通りに「よろしくお願いします」を Speech-to-Text できていた👏

{
  "jobName": "job-b9b2d46a-6621-4890-b09f-08fb3d5f059e",
  "status": "COMPLETED",
  "results": {
    "transcripts": [
      {
        "transcript": "よろしく お 願い し ます"
      }
    ],
    "items": [
      {
        "start_time": 0.0,
        "end_time": 0.48,
        "type": "pronunciation",
        "alternatives": [
          {
            "confidence": 0.816937,
            "content": "よろしく"
          }
        ]
      },
      {
        "start_time": 0.48,
        "end_time": 0.57,
        "type": "pronunciation",
        "alternatives": [
          {
            "confidence": 0.984702,
            "content": ""
          }
        ]
      },
      {
        "start_time": 0.57,
        "end_time": 0.87,
        "type": "pronunciation",
        "alternatives": [
          {
            "confidence": 0.984702,
            "content": "願い"
          }
        ]
      },
      {
        "start_time": 0.87,
        "end_time": 0.99,
        "type": "pronunciation",
        "alternatives": [
          {
            "confidence": 1.0,
            "content": ""
          }
        ]
      },
      {
        "start_time": 0.99,
        "end_time": 1.23,
        "type": "pronunciation",
        "alternatives": [
          {
            "confidence": 1.0,
            "content": "ます"
          }
        ]
      }
    ]
  }
}

関連記事

kakakakakku.hatenablog.com

kakakakakku.hatenablog.com

アプリケーションの認可ソリューションを体験できる「Amazon Verified Permissions workshop」

Amazon Verified Permissions を学ぶため「Amazon Verified Permissions workshop」を実施してみた💡3時間ほどで Amazon Verified Permissions の基本的な概念・マネジメントコンソール操作から,実際にサーバーレスアプリケーションに組み込んだときの実装例も学べてとても良かった❗️

boto3 の操作だと,通常の is_authorized() 以外に is_authorized_with_token()batch_is_authorized() も出てくる💡もし Amazon Verified Permissions をまだ試したことがないけど気になってる人がいたらおすすめ.個人的な実施メモをまとめておこうと思う \( 'ω')/

catalog.workshops.aws

Lab 1 - Amazon Verified Permissions Console Demo

まず,マネジメントコンソールで Amazon Verified Permissions に入門する.ポリシーストア・ポリシーを設定して,テストベンチを使って認可リクエストの動作確認をする👌

alice は LIST#000001 を ReadList できる🙆‍♂

alice は LIST#000002 を ReadList できない🙅‍♂

さらに認可仕様のスキーマを設定する.今回は TinyTodo という TODO リストを管理するアプリケーションを題材にしているため「タスク」「タスクリスト」を操作するアクションをスキーマに設定している✅ スキーマによってポリシーを厳格に設定できることを確認し,またポリシーテンプレートも活用することでポリシーを再利用できるようにした.Lab 1 で Amazon Verified Permissions の基本的な流れを掴むことができた.

スキーマ(アクション図)

Lab 2 - Application Built on Amazon Verified Permissions

Lab 2 からは Amazon Verified Permissions を TinyTodo アプリケーションに組み込んでいく❗️Amazon Verified Permissions 以外は AWS CloudFormation で簡単に構築できる.サーバーレスアーキテクチャに理解があれば良いと思うけど,もし Amazon API Gateway / AWS Lambda / Amazon Cognito / Amazon DynamoDB に慣れていなければ,ワークショップを進める前にデプロイされたサービスの設定などをザッと確認しておくと進めやすくなると思う👌

TinyTodo アーキテクチャ図は Lab 2 から引用

あとは API を実行してタスク一覧の取得・新しいタスクの登録を試す.boto3 で is_authorized() を実行する実装の流れも理解できた.

boto3.amazonaws.com

$ curl -X POST "$WS_API_ENDPOINT/task/create" \
>   -H "Content-Type: application/json" \
>   -H "Authorization: Bearer $WS_ACCESS_TOKEN_ALICE" \
>   -d '{
quote>         "listId": 1,
quote>         "name": "Task 7 - Execute The Project Plan",
quote>         "description": "Once you have got your project plan in place, it is time to execute it—and bring your project to life!"
quote>       }' | jq
{
  "taskId": 7
}

$ curl -X POST "$WS_API_ENDPOINT/task/create" \
>   -H "Content-Type: application/json" \
>   -H "Authorization: Bearer $WS_ACCESS_TOKEN_ALICE" \
>   -d '{
quote>         "listId": 2,
quote>         "name": "Task 7 - Execute The Project Plan",
quote>         "description": "Once you have got your project plan in place, it is time to execute it—and bring your project to life!"
quote>       }' | jq
{
  "message": "Access denied -- permissions check failed"
}

さらに Lab 2 の最後は boto3 で create_policy() を実行して,アプリケーションからポリシーを追加する実装もあって参考になった❗️へぇ〜

boto3.amazonaws.com

Lab 3 - Using Identity Sources

Amazon Cognito User Pool の「カスタム属性」として project: Project-Alpha を追加して,認証時の ID Token (JWT) に含まれる custom:project をそのまま認可リクエストに使う流れを体験した.そして,boto3 で is_authorized_with_token() を実行すると ID Token (JWT) をそのまま渡せることも学べた❗️

ID Token (JWT) を渡せるとは言え,boto3 のドキュメントに At this time, Verified Permissions accepts tokens from only Amazon Cognito. と書いてあるのは注意点かなと💡

boto3.amazonaws.com

Lab 4 - Authorization Improvements

最後の Lab 4 ではもともと意図的に組み込まれていた実装上の誤りを修正しつつ,タスクリストの共有機能の動作確認を進めていく.そして,Amazon Verified Permissions への認可リクエストが過剰に増えてしまうことを AWS X-Ray で確認しつつ,boto3 の batch_is_authorized() を使って改善する.

boto3.amazonaws.com

batch_is_authorized() 活用前

batch_is_authorized() 活用後

認可リクエストをまとめて実行できる機能が提供されているのも知らなくて勉強になった❗️2023年11月にリリースされた機能だったのか〜 \( 'ω')/

aws.amazon.com

まとめ

Amazon Verified Permissions に実践入門できるワークショップ「Amazon Verified Permissions workshop」を試した👌 Amazon Verified Permissions の基本的な概念・マネジメントコンソール操作から,実際にサーバーレスアプリケーションに組み込んだときの実装例も学べてとても良かった❗️

今回は TinyTodo アプリケーションに組み込んだけど,実際に仕事で Amazon Verified Permissions を活用する場合は認可モデルの設計が良し悪しを決めるのは間違いなく,ドキュメントにモデリングの指針がまとまっているため,あわせて読んでおくと良さそう📝

docs.aws.amazon.com