「DogStatsD」を用いてDatadogでFargateコンテナ内のカスタムメトリクスを収集してみた

			
			FROM python:3.12-slim

# Set working directory
WORKDIR /app

# Copy requirements and install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY app.py .

# Expose port
EXPOSE 80

# Start application with uvicorn
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "80"]


		

			
			fastapi
uvicorn
datadog

		

			
			import os
import time
from typing import Optional
from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
from datadog import initialize, statsd
import uvicorn

options = {
    'statsd_host': os.getenv('DD_AGENT_HOST', '127.0.0.1'),
    'statsd_port': int(os.getenv('DD_DOGSTATSD_PORT', '8125'))
}

initialize(**options)

# 共通タグ
COMMON_TAGS = [
    f"environment:{os.getenv('DD_ENV', 'demo')}",
    f"service:{os.getenv('DD_SERVICE', 'demo-api')}"
]

def track_api_request(
    endpoint: str,
    method: str,
    status_code: int,
    response_time: float,
    request_size: int,
    response_size: int,
    extra_tags: Optional[list] = None
):
    tags = COMMON_TAGS + [
        f"endpoint:{endpoint}",
        f"method:{method}",
        f"status_code:{status_code}",
        f"status_family:{status_code // 100}xx"
    ]

    if extra_tags:
        tags.extend(extra_tags)

    # 1. COUNT: リクエスト総数
    statsd.increment('demo-api.requests.count', tags=tags)

    # 2. COUNT: エラー数
    if status_code >= 400:
        statsd.increment('demo-api.requests.errors', tags=tags)

    # 3. HISTOGRAM: レスポンスタイム
    statsd.histogram('demo-api.response.time_ms', response_time, tags=tags)

    # 4. DISTRIBUTION: リクエストサイズ
    statsd.distribution('demo-api.request.size_bytes', request_size, tags=tags)

    # 5. DISTRIBUTION: レスポンスサイズ
    statsd.distribution('demo-api.response.size_bytes', response_size, tags=tags)


app = FastAPI(
    title="Demo Fargate Datadog API",
    description="APIエンドポイント監視デモ（DogStatsDメトリクス送信）",
    version="1.0.0"
)

@app.middleware("http")
async def metrics_middleware(request: Request, call_next):
    # リクエスト開始時刻
    start_time = time.time()

    # リクエストサイズを取得
    request_size = int(request.headers.get("content-length", 0))

    try:
        # 次のミドルウェアまたはエンドポイント処理を実行
        response: Response = await call_next(request)

        # レスポンスサイズを取得
        response_size = int(response.headers.get("content-length", 0))

        # レスポンスタイムを計算（ミリ秒）
        response_time = (time.time() - start_time) * 1000

        # メトリクス送信
        track_api_request(
            endpoint=request.url.path,
            method=request.method,
            status_code=response.status_code,
            response_time=response_time,
            request_size=request_size,
            response_size=response_size
        )

        return response

    except Exception as e:
        # エラー発生時もメトリクス送信
        response_time = (time.time() - start_time) * 1000

        track_api_request(
            endpoint=request.url.path,
            method=request.method,
            status_code=500,
            response_time=response_time,
            request_size=request_size,
            response_size=0
        )

        # エラーレスポンスを返す
        return JSONResponse(
            status_code=500,
            content={"error": "Internal Server Error", "detail": str(e)}
        )


# ヘルスチェック用
@app.get("/")
async def health_check():
    return {"status": "healthy"}


# ユーザー一覧取得
@app.get("/api/v1/users")
async def get_users():
    time.sleep(0.05)
    return {
        "users": [
            {"id": 1, "name": "Alice"},
            {"id": 2, "name": "Bob"},
            {"id": 3, "name": "Charlie"}
        ]
    }

# エラーエンドポイント（エラー率テスト用）
@app.get("/api/v1/error")
async def error_endpoint():
    return JSONResponse(
        status_code=500,
        content={"error": "This is a test error"}
    )


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=80)


		

			
			import * as cdk from 'aws-cdk-lib';
import * as ec2 from 'aws-cdk-lib/aws-ec2';
import * as ecs from 'aws-cdk-lib/aws-ecs';
import * as elbv2 from 'aws-cdk-lib/aws-elasticloadbalancingv2';
import * as logs from 'aws-cdk-lib/aws-logs';
import { Platform } from 'aws-cdk-lib/aws-ecr-assets';
import { Construct } from 'constructs';
import { DatadogECSFargate } from 'datadog-cdk-constructs-v2';

export class DemoFargateDatadogSidecarStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    // VPCの作成
    const vpc = new ec2.Vpc(this, 'DemoVpc', {
      vpcName: 'demo-fargate-datadog-vpc',
      ipAddresses: ec2.IpAddresses.cidr('10.0.0.0/16'),
      maxAzs: 2,
      natGateways: 1,
      subnetConfiguration: [
        {
          cidrMask: 24,
          name: 'Public',
          subnetType: ec2.SubnetType.PUBLIC,
        },
        {
          cidrMask: 24,
          name: 'Private',
          subnetType: ec2.SubnetType.PRIVATE_WITH_EGRESS,
        },
      ],
    });

    // VPCにタグを追加
    cdk.Tags.of(vpc).add('Name', 'demo-fargate-datadog-vpc');

    // ALB用セキュリティグループの作成
    const albSecurityGroup = new ec2.SecurityGroup(this, 'ALBSecurityGroup', {
      securityGroupName: 'demo-fargate-datadog-alb-sg',
      description: 'Security group for ALB',
      vpc: vpc,
      allowAllOutbound: true,
    });

    // インターネットからのHTTPアクセスを許可
    albSecurityGroup.addIngressRule(
      ec2.Peer.anyIpv4(),
      ec2.Port.tcp(80),
      'Allow HTTP from internet'
    );

    // ECSサービス用セキュリティグループの作成
    const ecsSecurityGroup = new ec2.SecurityGroup(this, 'ECSSecurityGroup', {
      securityGroupName: 'demo-fargate-datadog-ecs-sg',
      description: 'Security group for ECS Fargate tasks',
      vpc: vpc,
      allowAllOutbound: true,
    });

    // ALBからのHTTPアクセスを許可
    ecsSecurityGroup.addIngressRule(
      albSecurityGroup,
      ec2.Port.tcp(80),
      'Allow HTTP from ALB'
    );

    // ECSクラスターの作成
    const cluster = new ecs.Cluster(this, 'DemoCluster', {
      clusterName: 'demo-fargate-datadog-cluster',
      vpc: vpc,
    });

    // ALBの作成
    const alb = new elbv2.ApplicationLoadBalancer(this, 'DemoALB', {
      loadBalancerName: 'demo-fargate-datadog-alb',
      vpc: vpc,
      internetFacing: true,
      vpcSubnets: {
        subnetType: ec2.SubnetType.PUBLIC,
      },
      securityGroup: albSecurityGroup,
    });

    // ターゲットグループの作成
    const targetGroup = new elbv2.ApplicationTargetGroup(this, 'DemoTargetGroup', {
      targetGroupName: 'demo-fargate-tg',
      vpc: vpc,
      port: 80,
      protocol: elbv2.ApplicationProtocol.HTTP,
      targetType: elbv2.TargetType.IP,
      healthCheck: {
        path: '/',
        interval: cdk.Duration.seconds(30),
        timeout: cdk.Duration.seconds(5),
        healthyThresholdCount: 2,
        unhealthyThresholdCount: 3,
      },
      deregistrationDelay: cdk.Duration.seconds(30),
    });

    // リスナーの作成
    alb.addListener('DemoListener', {
      port: 80,
      protocol: elbv2.ApplicationProtocol.HTTP,
      defaultTargetGroups: [targetGroup],
    });

    // CloudWatch Logsロググループの作成
    const logGroup = new logs.LogGroup(this, 'DemoLogGroup', {
      logGroupName: '/ecs/demo-fargate-datadog',
      removalPolicy: cdk.RemovalPolicy.DESTROY,
      retention: logs.RetentionDays.ONE_WEEK,
    });

    // Datadog ECS Fargateコンストラクトの初期化
    const datadogECS = new DatadogECSFargate({
      apiKeySecretArn: process.env.DD_API_KEY_SECRET_ARN,
      site: 'datadoghq.com',
      imageVersion: '7-rc-full',
      env: 'demo',
      service: 'demo-api',
      dogstatsd: { // カスタムメトリクス送信に必要
        isEnabled: true,
        isOriginDetectionEnabled: true,
        isSocketEnabled: true,
      },
    });

    // Datadogコンストラクトを使用したタスク定義の作成
    const taskDefinition = datadogECS.fargateTaskDefinition(this, 'DemoTaskDef', {
      family: 'demo-fargate-datadog-task',
      cpu: 512,
      memoryLimitMiB: 1024,
    });

    // メインアプリケーションコンテナ（FastAPI）の追加
    taskDefinition.addContainer('app', {
      containerName: 'app',
      image: ecs.ContainerImage.fromAsset('./app', {
        platform: Platform.LINUX_AMD64,
      }),
      logging: ecs.LogDrivers.awsLogs({
        streamPrefix: 'app',
        logGroup: logGroup,
      }),
      portMappings: [
        {
          containerPort: 80,
          protocol: ecs.Protocol.TCP,
        },
      ],
      essential: true,
    });

    // Fargateサービスの作成
    const service = new ecs.FargateService(this, 'DemoService', {
      serviceName: 'demo-fargate-datadog-service',
      cluster: cluster,
      taskDefinition: taskDefinition,
      desiredCount: 1,
      minHealthyPercent: 100,
      maxHealthyPercent: 200,
      enableExecuteCommand: true,
      vpcSubnets: {
        subnetType: ec2.SubnetType.PRIVATE_WITH_EGRESS,
      },
      assignPublicIp: false,
      securityGroups: [ecsSecurityGroup],
      healthCheckGracePeriod: cdk.Duration.seconds(60),
    });

    // サービスをターゲットグループに登録
    service.attachToApplicationTargetGroup(targetGroup);
  }
}


		

			
			// Datadog ECS Fargateコンストラクトの初期化
const datadogECS = new DatadogECSFargate({
    apiKeySecretArn: process.env.DD_API_KEY_SECRET_ARN,
    site: 'datadoghq.com',
    imageVersion: '7-rc-full',
    env: 'demo',
    service: 'demo-api',
    dogstatsd: { // カスタムメトリクス送信に必要
        isEnabled: true,
        isOriginDetectionEnabled: true,
        isSocketEnabled: true,
    },
});