时间戳

首页 > 解决方案 > 是否可以获得 SQS 中的消息总数？

kubernetes - 是否可以获得 SQS 中的消息总数？

问题描述

我看到有 2 个单独的指标ApproximateNumberOfMessagesVisible和 ApproximateNumberOfMessagesNotVisible.

使用可见消息数会导致处理 pod 在从队列中获取消息后立即触发终止，因为它们不再可见。如果我使用不可见的消息数量，它将不会扩大。

我正在尝试使用水平 pod 自动缩放器和 SQS 的外部指标来扩展 kubernetes 服务。这是模板外部指标：

apiVersion: metrics.aws/v1alpha1
kind: ExternalMetric
metadata:
  name: metric-name
spec:
  name: metric-name
  queries:
    - id: metric_name
      metricStat:
        metric:
          namespace: "AWS/SQS"
          metricName: "ApproximateNumberOfMessagesVisible"
          dimensions:
            - name: QueueName
              value: "queue_name"
        period: 60
        stat: Average
        unit: Count
      returnData: true

这是 HPA 模板：

kind: HorizontalPodAutoscaler
apiVersion: autoscaling/v2beta1
metadata:
  name: hpa-name
spec:
  scaleTargetRef:
    apiVersion: apps/v1beta1
    kind: Deployment
    name: deployment-name
  minReplicas: 1
  maxReplicas: 50
  metrics:
  - type: External
    external:
      metricName: metric-name
      targetAverageValue: 1

如果我可以定义另一个自定义指标，它是这两个指标的总和，问题将得到解决，我还能如何解决这个问题？

标签： kubernetesamazon-sqsamazon-ekshorizontalpodautoscaler

解决方案

我们使用 lambda 来获取两个指标并发布一个自定义指标，该指标是正在进行和等待的消息的总和，并使用 cloudwatch 事件以您想要的任何频率触发此 lambda，https: //console.aws.amazon.com/cloudwatch /home?region=us-east-1#rules:action=create

以下是供参考的 lambda 代码：

const AWS = require('aws-sdk');
const cloudwatch = new AWS.CloudWatch({region: ''});  // fill region here
const sqs = new AWS.SQS();

const SQS_URL = ''; // fill queue url here

async function getSqsMetric(queueUrl) {
    var params = {
      QueueUrl: queueUrl,
      AttributeNames: ['All']
    };
    return new Promise((res, rej) => {
        sqs.getQueueAttributes(params, function(err, data) {
            if (err) rej(err);
            else res(data);
        });
    })
    
}

function buildMetric(numMessages) {
  return {
    Namespace: 'yourcompany-custom-metrics',
    MetricData: [{
      MetricName: 'mymetric',
      Dimensions: [{
          Name: 'env',
          Value: 'prod'
      }],
      Timestamp: new Date(),
      Unit: 'Count',
      Value: numMessages
    }]
  }
}

async function pushMetrics(metrics) {
  await new Promise((res) => cloudwatch.putMetricData(metrics, (err, data) => {
    if (err) {
      console.log('err', err, err.stack); // an error occurred
      res(err);
    } else {
      console.log('response', data);           // successful response
      res(data);
    }
  }));
}

exports.handler = async (event) => {
    console.log('Started');
    const sqsMetrics = await getSqsMetric(SQS_URL).catch(console.error);
    var queueSize = null;
    if (sqsMetrics) {
        console.log('Got sqsMetrics', sqsMetrics);
        if (sqsMetrics.Attributes) {
          queueSize = parseInt(sqsMetrics.Attributes.ApproximateNumberOfMessages) + parseInt(sqsMetrics.Attributes.ApproximateNumberOfMessagesNotVisible);
          console.log('Pushing', queueSize);
          await pushMetrics(buildMetric(queueSize))
        } 
    } else {
        console.log('Failed fetching sqsMetrics');
    }
    const response = {
        statusCode: 200,
        body: JSON.stringify('Pushed ' + queueSize),
    };
    return response;
};

推荐阅读