通过 SQS 向 Elastic Beanstalk 工作线程环境发布消息

Question

我在 Elastic beanstalk 上有一个带有 Web 服务器和工作环境的 docker 应用程序。

工作环境当前通过 cron 运行计划作业。 我正在尝试将服务器连接到工作人员以实现以下目标：

1. 客户端向服务器发送请求（/trigger_job）
2. 服务器通过向 SQS 队列 (/perform_job) 发送 JSON 消息来将作业卸载给工作人员
3. Worker 通过读取来自 SQS 的消息来执行作业

我无法找到有关 JSON 消息应是什么样子的文档。 官方文档提到了一些 HTTP 标头。但没有提到标头来指定工作环境中所需的端点。

# server.py
from bottle import post, HTTPResponse


@post('/trigger_job')
def trigger_worker_job():
    # should send a JSON message to sqs to trigger the '/perform_job'
    # Need help with what the JSON message looks like
    return HTTPResponse(status=200, body={'Msg': 'Sent message'})

# worker.py
from bottle import post, HTTPResponse


@post('/perform_job')
def perform_job():
    # job is performed in the worker environment

    return HTTPResponse(status=200, body={'Msg': 'Success'})

Answer 1

在 Python 中，您可以从 python 示例应用程序中了解这一点，您可以在此 aws 文档步骤 4：部署新应用程序版本。

您可以在beanstalk工作环境控制台中配置SQS端点。配置 > 工作线程 > 选择工作线程队列

# for example: 
environ['HTTP_X_AWS_SQSD_TASKNAME']
environ['HTTP_X_AWS_SQSD_SCHEDULED_AT']
logger.info("environ X-Aws-Sqsd-Queue %s" % environ['HTTP_X_AWS_SQSD_QUEUE'])

# regarding your message attribute. For example, the attribute name is Email, 
# you can extract it via environ['HTTP_X_AWS_SQSD_ATTR_EMAIL']). 
# Make sure that the attribute name is all capital.  
logger.info("environ X-Aws-Sqsd-Attr Email %s" % environ['HTTP_X_AWS_SQSD_ATTR_EMAIL'])

消息将包含图像中的以下信息。您可以阅读有关 aws AWS Elastic Beanstalk 工作线程环境

的更多信息

Answer 2

我从死信队列中一些失败的 cron 作业中发现，有以下三个属性：

• "beanstalk.sqsd.path"

• "beanstalk.sqsd.task_name"

• "beanstalk.sqsd.scheduled_time"

这与 cron 作业上设置的属性类似。

您可以在工作环境的 SQS 队列中手动创建一条新消息，并设置这些属性以匹配您希望执行的 cron 作业，尽管这很乏味，而且我更喜欢通过代码来执行此操作。

我在 python 中使用

boto3

框架，但希望在其他语言中也是类似的过程。

def send_elastic_beanstalk_message(name, path, message):
    client = boto3.client('sqs')
    return client.send_message(
        QueueUrl=url, 
        MessageAttributes={
            "beanstalk.sqsd.path": {
                "DataType": "String",
                "StringValue": path
            },
            "beanstalk.sqsd.task_name": {
                "DataType": "String",
                "StringValue": name
            },
            "beanstalk.sqsd.scheduled_time": {
                "DataType": "String",
                "StringValue": datetime.now(timezone.utc).strftime(
                    "%Y-%m-%d %H:%M:%S UTC"
                )
            }
        }
    )

这将在 SQS 队列中创建一条新消息，该消息将由工作守护进程解析并触发 cron 作业。

不幸的是，解析消息时似乎不包含消息正文，因此触发作业是我能够解决的唯一操作。