2024-03-27

【aws-sdk-go-v2 & Datadog】LambdaからDatadogにトレースを送る方法

概要

lambdaでaws-sdk-go-v2を使って行う処理のトレースをDatadogに送る方法を記載する。

前提条件

Datadogのアカウントで、API Keyが発行済みであること
AWS Lambdaをコンテナイメージで動かせること

動作環境

AWS Lambda(コンテナイメージで動作）
Go 1.22

使用ライブラリ

github.com/aws/aws-sdk-go-v2
gopkg.in/DataDog/dd-trace-go.v1
github.com/DataDog/datadog-lambda-go

設定方法

Dockerfile

datadogのextenstionを入れる。

COPY --from=public.ecr.aws/datadog/lambda-extension:latest /opt/. /opt/

環境変数

API Keyはsecretsmanagerから取得しても良いし、環境変数に直接入れても良い。ここでは、secretsmanagerから取得する。 DD_API_KEY_SECRET_ARNという環境変数に、secretsのarnを入れる。

Lambdaから対象のsecretsへのアクセス権限が必要になるため、IAMポリシーに忘れずに入れる。

アプリケーションコード

awstrace.AppendMiddlewareという関数を使う。

cfg, err := config.LoadDefaultConfig(ctx, config.WithRegion("ap-northeast-1"))
if err != nil {
    return "", err
}
awstrace.AppendMiddleware(&cfg, awstrace.WithServiceName("{Datadog上に表示してほしいサービス名を入れる}"))
s3Client := s3.NewFromConfig(cfg)

このように初期化したS3のクライアントで何かしらの処理を行うと、その内容がDatadogのトレースとして送信されるようになる。

2024-02-04

【flutter freezedライブラリ】copyWithの引数にnullを渡す時の挙動に注意しよう

概要

flutterのfreezedライブラリを使ってモデルを定義し、コードを生成するとimmutableなオブジェクトを定義してくれる。このオブジェクトには、copyWithというメソッドが付いており、引数で与えられた属性値でオブジェクトを初期化して返してくれるのだが、1点気をつけるべき挙動がある。

具体的には、このcopyWithにnullを渡しても、属性値はnullになって返されるのではなく、copyWithを呼び出す前の状態のままになる、ということだ（2024年2月現在、freezedバージョン2.4.5）。

「copyWithの属性値がnullである」=「その属性値をnullにして初期化する」のではなく、「copyWithの属性値がnullである」=「その属性値は置き換えない」という挙動になっているのである。

具体例

例えば、以下のようにモデルを定義する。

@freezed
class SampleModel<T extends Object> with _$SampleModel<T> {
  const factory SampleModel({
    T? value,
  }) = _SampleModel<T>;
}

copyWithを使って、valueにnullを入れるとする。

SampleModel model = SampleModel(value: 10);
model = model.copyWith(value: null);
print(model.valule);

出力されるのはなんと 10 のままになる。

freezedで生成されたファイルを確認

freezedで生成されたファイル中の、copyWithメソッドの実装部分を見てみると、次のようになっている。

/// @nodoc
class _$SampleModelCopyWithImpl<T extends Object, $Res,
    $Val extends SampleModel<T>> implements $SampleModelCopyWith<T, $Res> {
  _$SampleModelCopyWithImpl(this._value, this._then);

  // ignore: unused_field
  final $Val _value;
  // ignore: unused_field
  final $Res Function($Val) _then;

  @pragma('vm:prefer-inline')
  @override
  $Res call({
    Object? value = null,
  }) {
    return _then(_value.copyWith(
      value: null == value
          ? _value.value
          : value // ignore: cast_nullable_to_non_nullable
              as T?,
    ) as $Val);
  }
}

valueがnullの場合、引数で与えられたvalueではなく、元の値（_value.value）を使っていることがわかる。

対処方法

この仕様を理解して、null値で置き換えたいときは、copyWithを使わずに初期化する。

SampleModel model = SampleModel(value: 10);
model = SampleModel(value: null);
print(model.value);

この方法だと属性値の多いクラスの記述がやたらと長くなってしまうし、チームで開発していると他メンバーが混乱しかねない。

値オブジェクトを定義し、「空である」という状態を定義する

@freezed
class SampleModel with _$SampleModel {
  const factory SampleModel({
    SampleValue value,
  }) = _SampleModel;
}

@freezed
class SampleValue<T extends Object> with _$SampleValue<T> {
  const factory SampleValue({
    T? value,
  }) = _SampleValue<T>;

  factory SampleValue.empty() => const SampleValue(value: null);

  bool isEmpty() {
    return value == null;
  }
}

この方法では、「空である」状態を表すことができる値オブジェクトのクラスを作り、nullを使わないようにする。個人的にはこちらのやり方が気に入っている。コード量はやや増えるものの、copyWithにnullを渡してハマるリスクを避けつつ、意図も明確に示せる。

参考リンク

github.com

2023-04-19

Go gin & Web Adapterで動作するLambdaからDatadogのトレースを送信する

Go AWS

概要

AWS Web Adapterというものがあり、ウェブAPIのコードをそのままにサーバーレス化することができる。

aws.amazon.com

これいいじゃんということで、GoのGinで動くウェブAPIをサーバーレス化したのだが、以前のECS Fargate環境では正常に動作していたDatadogのトレース送信が動かなくなった。

ハマった末にやり方が分かったので、Lambda Web Adapter & Gin の構成でウェブAPIを動かす際にDatadogのトレースを正常に送る方法をお伝えする。

Web Adapterを使っていない場合はどうやってトレースを送るのか？

まずは、Web Adapterを使っていない通常のLambdaの場合だとどのようにDatadogトレースを送るのか紹介する。

Dockerfileに以下のコードを書き、datadog extensionをコンテナイメージに含める。

COPY --from=public.ecr.aws/datadog/lambda-extension:latest /opt/. /opt/

datadog-lambda-goのライブラリをインストールし、

go get github.com/DataDog/datadog-lambda-go

ddlambda.WrapFunction を仕込む。

func main() {
  // Wrap your lambda handler
  lambda.Start(ddlambda.WrapFunction(myHandler, nil))
}

func myHandler(ctx context.Context, event MyEvent) (string, error) {
  // lambdaの処理をここに書く
}

このやり方は公式ドキュメントにも記載があるし、特に問題なく動かせると思う。（※ 2023年4月現在、英語でないとcontainer imageの場合のやり方が出ないので注意）

docs.datadoghq.com

Web Adapterを使っている場合はどうなるか？

通常、コンテナイメージのLambdaを使う場合、上記のように lambda.Start がエントリポイントとなる。しかし今回は、Web Adapterを使っているので、アプリケーションのコードはginを使った純粋なweb APIのコードになり、 lambda.Start をどこにも書いておらず、ddlambda.WrapFunctionを仕込むことができない。

ではどうやるのかというと、Datadogのトレーサーをスタートするときに、以下のようにオプションを入れると良い。

tracer.Start(
        tracer.WithEnv("環境名（stagingやproduction）を入れる"),
        tracer.WithService("Datadog上で扱われるサービス名を入れる"),
        tracer.WithLambdaMode(false), // trueにするとstdoutにtraceが出力され、datadog agentに送信されなくなる
        tracer.WithGlobalTag("_dd.origin", "lambda"),
    )
    defer tracer.Stop()

ECSからLambdaにginのAPIを移した際、当初はtracer.WithLambdaMode(false) と tracer.WithGlobalTag("_dd.origin", "lambda") を入れておらず、トレースがDatadogに送信されてこなかった。

コードのコメントにも書いた通り、tracer.WithLambdaMode がtrueだとトレースがstdoutに出力され、datadog agentに送信されなくなる。

実際にはLambdaで動かしているのに、WithLambdaModeをfalseにするのは奇妙だが、Datadogのライブラリの実装を追っていくと確かにそうしている。

github.com

まとめ

Web Adapterを使うことで簡単に既存のAPIをLambdaに載せれるし、ローカルでも簡単に動かしやすくなる。 Datadogとの連携がドキュメントにも書いておらず難しかったが、これで解決できたのでよかった。

2023-02-13

Lambda Web AdapterでGoのGinを動かすときに、DBコネクションの貼りすぎを防ぐ

AWS Go

概要

Lambda Web Adapterを使うと、それまで非サーバーレス環境下で動かしていたウェブAPIをLambda上で動かすことが可能になる。コンテナイメージさえあれば、web APIをサーバーレス化できるし、コードの共通化も容易になって便利だ。

ただし、RDSと接続するのならば、コネクションが増えすぎないように工夫する必要がある。対処方法として、RDS Proxyを使う方法はよく知られていて、導入はマストかと思う。

aws.amazon.com

その上で、RDS Proxyを使ったとしても、コネクション数が圧迫されることがある。アプリケーション内部で明示的にデータベースとの接続を切断しないと、接続がアイドル状態となってしまい、アイドルコネクションのタイムアウトを迎えるまでコネクション数を消費してしまうのだ。

今回は、Lambda Web Adapter & Go & Ginを使って動かす際に、アプリケーション内部で明示的にコネクションをクローズし、コネクション数を圧迫しないようにする工夫を紹介する。

解決策1: Graceful Shutdown時にコネクションをクローズする

Lambdaはシャットダウン時にSIGTERMのシグナルを受け取るので、これをトリガーにDBとのコネクションをクローズする。

https://docs.aws.amazon.com/lambda/latest/dg/runtimes-extensions-api.html より

サンプルコードは以下の通り。

r := gin.Default()
r.GET("/healthcheck", healthcheck)

srv := &http.Server{
    Addr:    ":8000",
    Handler: r,
}
go func() {
    if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
        panic(err)
    }
}()

// Wait for interrupt signal to gracefully shutdown the server with a timeout of 5 seconds.
quit := make(chan os.Signal, 1)
signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
<-quit

// DBとのコネクションを切断する
DbConnector.Close()

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := srv.Shutdown(ctx); err != nil {
    logging.Error(err, "Server Shutdown:")
}
// catching ctx.Done(). timeout of 5 seconds.
<-ctx.Done()

この方法の問題として、Lambdaがinvokeされなくなってからシャットダウンするまでの間もコネクションが張られ続けてしまう点がある。実際にAPIに対して同時接続数100件の負荷試験をかけてみると、RDSプロキシのコネクション数も同じ数くらいまで上がってしまった。

↓このようにリクエスト数に合わせてコネクション数が伸びてしまった。

解決策2: エンドポイントの処理が終了したらコネクションをクローズする（これで根本解決）

Ginのmiddlewareとして、DBのコネクションを閉じる処理を追加する。Ginのmiddlewareは、エンドポイントの処理の後に処理を挟み込むことができるので、これを利用する。

サンプルコードは以下の通り。

func SetDbConnClose(dbConn *sql.DB) func(*gin.Context) {
    return func(c *gin.Context) {
        c.Next()
        dbConn.Close()
    }
}


func main() {
    // DBとのコネクションを作成
    cfg := mysql.Config{
        User:   os.Getenv("DBUSER"),
        Passwd: os.Getenv("DBPASS"),
        Net:    "tcp",
        Addr:   "{RDS Proxyのエンドポイント}:3306",
        DBName: "sample",
    }
    dbConn, _ := sql.Open("mysql", cfg.FormatDSN())

    r := gin.Default()
    r.Use(SetDbConnClose())
    r.GET("/ping", func(c *gin.Context) {
        c.JSON(200, gin.H{
            "message": "pong",
        })
    })
    r.Run()
}

こうすることで、コネクション数が伸びてしまう現象をなくすことができた。

参考

Lambda Web Adapterとは？ aws.amazon.com

Lambdaのライフサイクル docs.aws.amazon.com

2023-02-03

ALBをトリガーにLambdaを起動する構成をTerraformで作ってみた

AWS インフラ

概要

HTTP APIをLambdaで動かすときは、AWS Apigateway & Lambdaの構成を取ることが多いが、今回はALBをトリガーとしてLambdaを動かす機会があったので、terraformでの作り方をまとめる。

LambdaをTerraformで作成する

まずは、Lambdaを作る。

resource "aws_lambda_function" "sample" {
  function_name = "sample_function"
  description   = "Awesome API"
  role          = aws_iam_role.metric_api.arn
  package_type  = "Image"
  image_uri     = "{コンテナイメージリポジトリのURI}"
  timeout       = 60
  memory_size   = 128
  vpc_config {
    security_group_ids = [{セキュリティグループのID}]
    subnet_ids         = [{サブネットのID}]
  }
}

ALBをTerraformで作成する

ALB、ターゲットグループ、リスナーグループを作成する。

resource "aws_alb" "alb" {
  name            = "sample-alb"
  internal        = false
  security_groups = [{セキュリティグループのID}]
  subnets         = [{サブネットのID}]

  enable_deletion_protection = false
  idle_timeout               = "60"

  access_logs {
    bucket  = "{アクセスログを置くS3バケット}"
    enabled = true
    prefix  = "alb/"
  }
}

resource "aws_alb_target_group" "tg" {
  name        = "sample_lambda_target_group"
  target_type = "lambda"
}

resource "aws_alb_listener" "https_listener" {
  load_balancer_arn = aws_alb.alb.arn
  port              = 443
  protocol          = "HTTPS"
  ssl_policy        = "ELBSecurityPolicy-2016-08"
  certificate_arn   = {ACMのARN}

  default_action {
    type             = "forward"
    target_group_arn = aws_alb_target_group.tg.arn
  }
}

ALBがLambdaをトリガーできるように構成する

ALBがLambdaをトリガーできるように設定する。

resource "aws_lambda_permission" "alb" {
  statement_id  = "AllowExecutionFromALB"
  action        = "lambda:InvokeFunction"
  function_name = aws_lambda_function.sample.function_name
  principal     = "elasticloadbalancing.amazonaws.com"
  source_arn    = aws_alb_target_group.tg.arn
}

最後に、ALBのターゲットグループとLambdaを紐付ける。

resource "aws_lb_target_group_attachment" "metric_api" {
  target_group_arn = aws_alb_target_group.tg.arn
  target_id        = aws_lambda_function.sample.arn
}

以上です。

2023-01-18

Redisへの大量データインサートには、パイプラインを使おう

Python

背景

Redisにダミーデータを投入するpython スクリプトを使っていた。データ件数が少ないうちはよかったのだが、多数のデータを入れるようになると、時間がかかりすぎて辛くなってきた。

パイプラインを使い、Redisとの通信ラウンドトリップを減らす

Redisのパイプラインを使うと、複数のRedisへのコマンド実行をまとめて一度に実行することができる。これにより、ラウンドトリップの回数を減らすことができ、パフォーマンスが向上する。

redis.io

サンプルコード

以下のスクリプトでは、時系列データを24時間分、5秒間隔で作成してRedisに入れている。

r = redis.Redis(host=HOST, password=PASSWORD, port=PORT, db=0)
unix_time_from = unix_time_now - 24 * 3600
for i, data in enumerate(data_list):
        unix_time_current = unix_time_from
        pipe = r.pipeline()
        while unix_time_now >= unix_time_current:
            pipe.zadd(data["id"], {data["value"]: unix_time_current})
            unix_time_current += 5
        pipe.execute()