ソースに絡まるエスカルゴ

貧弱プログラマの外部記憶装置です。

【AWS/python】ElastiCache Redisへの疎通確認をpythonで行う

AWS python

　AWSでEC2やECSコンテナに何かアプリを実行させ、ElastiCache Redisのキャッシュを使って処理するような場合が結構あります。
　AWS上でリソースを構築した際にちゃんとできているかの疎通確認を行いたい場合がありますが、案外そのやり方がすんなり見つからなくて苦労しました。

　なので今回はそんなElastiCache Redisへの疎通確認をpythonで行う場合の備忘録となります。詳しい解説は参考資料に挙げているページ様を参照してください。

　また、AWSリソース自体を構築する説明はしません。AWS上にリソースが構築済みであとは疎通確認をするだけという前提で始めるので注意してください。


　では、始めます。


０：前提
　冒頭でも少し書きましたが、以下の状態である前提で話を進めます。

  • EC2またはECSコンテナからElastiCache Redisへの疎通確認とする
  • EC2またはECSコンテナからElastiCache RedisへのセキュリティグループなどAWSのリソース設定が正しく構築されている


１：実行環境構築
　EC2ならAnsible、ECSならDockerfileによるイメージ作成となるかと思いますが、まずはpythonとpipの環境を構築します。
　そしてElastiCache Redisへアクセスするためのライブラリであるredisの最新版をpipでインストールします。

pip install --upgrade redis

　これで実行環境の構築は終了です。


２：疎通確認のコード（クラスターなし）
　ElastiCache Redisでクラスターなし構成の場合は以下のpythonコードをコピペし、Redisのエンドポイントの部分を書き換えます。

・redis_non_cluster.py

# -*- coding: utf-8 -*-
import redis

# Redisの接続情報
redis_host = "clustercfg.xxxxxxxxxxxxxxxxxxx.cache.amazonaws.com" # Redisのエンドポイント
redis_port = 6379
timeout    = 30  # タイムアウト時間（秒）

# Redisクライアントを初期化
redis_client = redis.StrictRedis(
  host=redis_host,
  port=redis_port,
  decode_responses=True,
  ssl=True,               # Redis Cache側でat_rest_encryption_enabledやtransit_encryption_enabledがTrueならSSL接続になるのでTrueにする必要がある
  socket_timeout=timeout  # タイムアウト時間を設定
)

try:
  # PING コマンドで接続確認
  if redis_client.ping():
    print("Redisへの接続が成功しました。")

    # データをキャッシュに保存
    redis_client.set("key", "hoge")
    # キャッシュからデータを取得
    value = redis_client.get("key")
    # 以下の内容が表示されれば接続できている(タイミングによっては失敗するが何度かやって表示されればOK)
    print(f"Retrieved value: {value}")
    # キャッシュから削除
    redis_client.delete("key")
  else:
    print("Redisへのpingに失敗しました。")
except redis.ConnectionError as e:
  print(f"エラーが発生しました。: {e}")

　上記のpythonコードを実行して疎通に成功すると以下のような表示になります。

Redisへの接続が成功しました。
Retrieved value: hoge


３：疎通確認のコード（クラスター有効）
　ElastiCache Redisでクラスター有効構成の場合は以下のpythonコードをコピペし、Redisのエンドポイントの部分を書き換えます。

・redis_cluster.py

# -*- coding: utf-8 -*-
import redis

# Redisクラスタの接続情報
redis_host = "clustercfg.xxxxxxxxxxxxxxxxxxx.cache.amazonaws.com" # Redisのエンドポイント
redis_port = 6379
timeout    = 30  # タイムアウト時間（秒）

# Redisライブラリのclusterモジュールを使用
redis_client = redis.RedisCluster(
  host=redis_host,
  port=redis_port,
  decode_responses=True,
  ssl=True,               # Redis Cache側でat_rest_encryption_enabledやtransit_encryption_enabledがTrueならSSL接続になるのでTrueにする必要がある
  socket_timeout=timeout  # タイムアウト時間を設定
)

try:
  # 疎通確認
  if redis_client.ping():
    print("Redisクラスタへの接続が成功しました。")
    # データをキャッシュに保存
    redis_client.set("key", "hoge")
    # キャッシュからデータを取得
    value = redis_client.get("key")
    print(f"Retrieved value: {value}")
    # キャッシュから削除
    redis_client.delete("key")
  else:
    print("Redisクラスタへのpingに失敗しました。")
except Exception as e:
  print(f"エラーが発生しました。: {e}")

　基本的には２と同じですが、クラスター用の「RedisCluster」モジュールを使用するようにしています。

　こちらも同じようにpythonコードを実行して疎通に成功すると以下のような表示になります。

Redisクラスタへの接続が成功しました。
Retrieved value: hoge


　以上がElastiCache Redisへの疎通確認をpythonで行う方法になります。

　最近AWSを触ることが出てきているのですが、案外こういう疎通の具体的な方法がぱっと見つからなかったりしたので記事にしてみた次第です。

　AWSやTerraformに関するリソース構築についてもどこか時間があれば記事にしていきたいのですが、ひとまず自分が躓いたところをちょっとずつ記事にしていこうと思います。


・参考資料