基于SQS的近似数量的消息自动缩放Fargate服务可见

Question

我想根据SQS队列的大小来扩展AWS Fargate容器。看来我只能根据容器的CPU或内存使用量进行扩展。有没有一种方法可以创建根据队列大小进行横向扩展和纵向扩展的策略？有人能够根据其他cloudwatch指标进行扩展吗？

Answer 1

是的，您可以这样做。您必须使用逐步扩展策略，并且已经为您的SQS队列深度（roximateNumberOfMessagesVisible）创建了警报。

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

完成后，转到ECS到您要自动扩展的服务。单击该服务的更新。添加缩放策略，然后选择“逐步跟踪”选项。您会看到有一个创建新警报的选项（只能让您在CPU或MemoryUtilization之间选择），或使用现有警报。

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

这适用于任何度量标准警报和ECS服务。如果您打算尝试扩展此设置，例如对于多个环境，或者使其变得比两个步骤都要复杂，请帮个忙，并使用CloudFormation或Terraform来帮助管理它。没有比调整10个服务的5步警报更糟糕的了。

Answer 2

更新至 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

Answer 3

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

大意

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

实施细节

自定义指标

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

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

• 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模板中创建了Target Tracking Scaling策略。

基于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警报，这些警报触发了扩展策略并根据指标和目标值计算了扩展调整。缩放策略会根据需要添加或删除容量，以将指标保持在指定的目标值或接近指定的目标值。除了使度量接近目标值之外，目标跟踪缩放策略还根据负载模式的变化来调整度量的变化。