基于 SQS ApproximateNumberOfMessagesVisible 自动扩展 Fargate 服务

Question

我想根据 SQS 队列的大小扩展我的 aws fargate 容器。看来我只能根据容器的 CPU 或内存使用情况进行扩展。有没有办法创建一个可以根据队列大小向外扩展和向内扩展的策略？是否有人能够根据其他 cloudwatch 指标进行扩展？

Answer 1

是的，您可以这样做。您必须使用步进扩展策略，并且您需要为您的 SQS 队列深度 (ApproximateNumberOfMessagesVisible) 创建一个警报。

转到 CloudWatch，创建一个新警报。我们将此警报称为 sqs-queue-depth-high，并在可见消息的大约数量为 1000 时触发它。

完成后，转到 ECS 到您要自动缩放的服务。单击服务的更新。添加扩展策略并选择 Step Tracking 品种。您会看到有一个选项可以创建新警报（仅允许您在 CPU 或 MemoryUtilization 之间进行选择）或使用现有警报。

在“使用现有警报”字段中键入 sqs-queue-depth-high 并按 Enter，您应该会看到一个绿色复选标记，让您知道名称有效（即警报存在） .您会看到新的下拉菜单，您现在可以在其中调整步数政策。

这适用于任何指标警报和 ECS 服务。如果您要尝试扩展此设置，例如针对多个环境，或者使其比 2 个步骤更复杂，请帮自己一个忙，使用 CloudFormation 或 Terraform 来帮助管理它。没有什么比不得不跨 10 项服务调整 5 步警报更糟糕的了。

Answer 2

AWS提供基于SQS队列的扩容解决方案：https://docs.aws.amazon.com/autoscaling/ec2/userguide/as-using-sqs-queue.html

主要思想

1. 使用以下公式创建 CloudWatch 自定义指标 sqs-backlog-per-task： sqs-backlog-per-task = sqs-messages-number / running-task-number。
2. 根据backlogPerInstance 指标创建目标跟踪扩展策略。

实现细节

自定义指标

在我的例子中，所有基础架构（Fargate、SQS 和其他资源）都在 CloudFormation 堆栈中进行了描述。因此，为了计算和记录自定义指标，我决定使用 AWS Lambda 函数，该函数也在 CloudFormation 堆栈中进行了描述，并与整个基础设施一起部署。

您可以在下面找到用于记录以下自定义指标的 AWS Lambda 函数的代码 sn-ps：

• sqs-backlog-per-task - 用于缩放
• running-task-number - 用于扩展优化和调试

CloudFormation 堆栈 (infrastructure.yml) 中 AWS SAM 语法中描述的 AWS Lambda 函数：

CustomMetricLoggerFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: custom-metric-logger
      Handler: custom-metric-logger.handler
      Runtime: nodejs8.10
      MemorySize: 128
      Timeout: 3
      Role: !GetAtt CustomMetricLoggerFunctionRole.Arn
      Environment:
        Variables:
          ECS_CLUSTER_NAME: !Ref Cluster
          ECS_SERVICE_NAME: !GetAtt Service.Name
          SQS_URL: !Ref Queue
      Events:
        Schedule:
          Type: Schedule
          Properties:
            Schedule: 'cron(0/1 * * * ? *)' # every one minute

用于计算和记录的 AWS Lambda Javascript 代码 (custom-metric-logger.js)：

var AWS = require('aws-sdk');

exports.handler = async () => {
  try {
    var sqsMessagesNumber = await getSqsMessagesNumber();
    var runningContainersNumber = await getRunningContainersNumber();

    var backlogPerInstance = sqsMessagesNumber;
    if (runningContainersNumber > 0) {
      backlogPerInstance = parseInt(sqsMessagesNumber / runningContainersNumber);
    }

    await putRunningTaskNumberMetricData(runningContainersNumber);
    await putSqsBacklogPerTaskMetricData(backlogPerInstance);

    return {
      statusCode: 200
    };
  } catch (err) {
    console.log(err);

    return {
      statusCode: 500
    };
  }
};

function getSqsMessagesNumber() {
  return new Promise((resolve, reject) => {
    var data = {
      QueueUrl: process.env.SQS_URL,
      AttributeNames: ['ApproximateNumberOfMessages']
    };

    var sqs = new AWS.SQS();
    sqs.getQueueAttributes(data, (err, data) => {
      if (err) {
        reject(err);
      } else {
        resolve(parseInt(data.Attributes.ApproximateNumberOfMessages));
      }
    });
  });
}

function getRunningContainersNumber() {
  return new Promise((resolve, reject) => {
    var data = {
      services: [
        process.env.ECS_SERVICE_NAME
      ],
      cluster: process.env.ECS_CLUSTER_NAME
    };

    var ecs = new AWS.ECS();
    ecs.describeServices(data, (err, data) => {
      if (err) {
        reject(err);
      } else {
        resolve(data.services[0].runningCount);
      }
    });
  });
}

function putRunningTaskNumberMetricData(value) {
  return new Promise((resolve, reject) => {
    var data = {
      MetricData: [{
        MetricName: 'running-task-number',
        Value: value,
        Unit: 'Count',
        Timestamp: new Date()
      }],
      Namespace: 'fargate-sqs-service'
    };

    var cloudwatch = new AWS.CloudWatch();
    cloudwatch.putMetricData(data, (err, data) => {
      if (err) {
        reject(err);
      } else {
        resolve(data);
      }
    });
  });
}

function putSqsBacklogPerTaskMetricData(value) {
  return new Promise((resolve, reject) => {
    var data = {
      MetricData: [{
        MetricName: 'sqs-backlog-per-task',
        Value: value,
        Unit: 'Count',
        Timestamp: new Date()
      }],
      Namespace: 'fargate-sqs-service'
    };

    var cloudwatch = new AWS.CloudWatch();
    cloudwatch.putMetricData(data, (err, data) => {
      if (err) {
        reject(err);
      } else {
        resolve(data);
      }
    });
  });
}

目标跟踪扩展策略

然后基于 sqs-backlog-per-task 指标，我在 Cloud Formation 模板中创建了目标跟踪扩展策略。

基于 sqs-backlog-per-task 指标 (infrastructure.yml) 的目标跟踪扩展策略：

ServiceScalingPolicy:
    Type: AWS::ApplicationAutoScaling::ScalingPolicy
    Properties:
      PolicyName: service-scaling-policy
      PolicyType: TargetTrackingScaling
      ScalingTargetId: !Ref ServiceScalableTarget
      TargetTrackingScalingPolicyConfiguration:
        ScaleInCooldown: 60
        ScaleOutCooldown: 60
        CustomizedMetricSpecification:
          Namespace: fargate-sqs-service
          MetricName: sqs-backlog-per-task
          Statistic: Average
          Unit: Count
        TargetValue: 2000

因此，AWS Application Auto Scaling 创建和管理触发扩展策略的 CloudWatch 警报，并根据指标和目标值计算扩展调整。扩展策略根据需要添加或删除容量，以使指标保持在或接近指定的目标值。除了将指标保持在目标值附近之外，目标跟踪扩展策略还会根据负载模式变化导致的指标变化进行调整。

Answer 3

我写了一篇关于这个主题的博客文章，包括一个 docker 容器来运行它。 文章可以在以下位置找到： https://allaboutaws.com/how-to-auto-scale-aws-ecs-containers-sqs-queue-metrics

预构建容器在 DockerHub 上可用： https://hub.docker.com/r/sh39sxn/ecs-autoscaling-sqs-metrics

这些文件在 GitHub 上可用： https://github.com/sh39sxn/ecs-autoscaling-sqs-metrics

希望对你有帮助。

Answer 4

更新到 2021 年（之前可能...）

对于那些需要它但在 CDK 中的人

一个示例用例：

// Create the vpc and cluster used by the queue processing service
const vpc = new ec2.Vpc(stack, 'Vpc', { maxAzs: 2 });
const cluster = new ecs.Cluster(stack, 'FargateCluster', { vpc });
const queue = new sqs.Queue(stack, 'ProcessingQueue', {
  QueueName: 'FargateEventQueue'
});

// Create the queue processing service
new QueueProcessingFargateService(stack, 'QueueProcessingFargateService', {
  cluster,
  image: ecs.ContainerImage.fromRegistry('amazon/amazon-ecs-sample'),
  desiredTaskCount: 2,
  maxScalingCapacity: 5,
  queue
});

来自：

https://github.com/aws/aws-cdk/blob/master/design/aws-ecs/aws-ecs-autoscaling-queue-worker.md