Функция AWS Step - Подождите, пока событие

Ответ 1

Невозможно предложить простое решение, только несколько направлений для изучения.

Во-первых, функции Step имеют особый способ обработки долговременной фоновой работы: действия. https://docs.aws.amazon.com/step-functions/latest/dg/concepts-activities.html это в основном очередь.

Если вы хотите 100% -ный сервер без сервера, это будет сложным или уродливым.

• либо, как вы сказали, создать новую функцию шага для каждого файла
• или, цикл опроса S3 в конечной машине с использованием пользовательского кода ошибки и предложения Retry

Если вы можете выделить экземпляр "1/8 микро" для рабочего фона, он не изящный, но простой и может быть реализован с мгновенной реакцией. Низкое требование к оборудованию подсказывает, что мы будем использовать машину только для синхронизации.

Определите действие StepFunction, названное, например, video-duration. Определите очередь SQS для мгновенной реакции или опроса S3 для результатов продолжительности.

Псевдокод функции состояния:

{
  StartAt: ffprobe
  ffprobe: {
    Type: Task
    Resource: arn:...lambda:launch-ffprobe
    Next: wait-duration
  }
  wait-duration: {
    Type: Task
    Resource: arn...activity:video-duration
    End: true
  }
}

Фоновый псевдоним пользователя:

statemap = dict/map filename to result

thread1:
  loop:
    taskToken, input = SF.GetActivityTask('video-duration')  # long poll
    sync(key=input.filename, waiter=taskToken)
thread2:
  loop:
    msg = SQS.ReceiveMessage(...)  # or poll S3
    sync(key=msg.filename, duration=msg.result)

function sync(key, waiter, duration):
  state = statemap[key]
  if waiter:
    state.waiter = waiter
  if duration:
    state.duration = duration
  if state.waiter and state.duration:
    SF.SendTaskSuccess(state.waiter, state.duration)

Псевдокод триггера S3:

if filename is video:
  SF.StartExecution(...)
else if filename is duration:
  content = S3.GetObject(filename)
  SQS.SendMessage(queue, content)

Ответ 2

Когда вы отправляете запрос на перегрузку, сохраните taskToken для шага в s3 с помощью предсказуемого ключа на основе загруженного файла. Например, если файл мультимедиа находится в 's3://my-media-bucket/foobar/media-001.mp3', вы можете сделать файл JSON, содержащий маркер задачи текущего шага, и сохранить его с тем же ключом в другом ковше, например 's3://ffprobe-tasks/foobar/media-001.mp3.json'. В конце шага, который отправляет носитель в transloadit, не назовите успех или неудачу на шаге - оставьте его в рабочем состоянии.

Затем, когда вы получите уведомление s3 о том, что результат transloadit готов, вы можете определить ключ s3 для получения маркера задачи ('s3://ffprobe-tasks/foobar/media-001.mp3'), загрузить JSON (и удалить это от s3) и отправить успех для этой задачи. Функция шага будет продолжать следующее состояние в выполнении.

Ответ 3

Ну, я бы вдохновил себя на https://aws.amazon.com/blogs/compute/implementing-serverless-manual-approval-steps-in-aws-step-functions-and-amazon-api-gateway/

Вы можете заменить шлюз API в этом функцией AWS Lambda, вызванной, например, событием S3 (Документация: http://docs.aws.amazon.com/lambda/latest/dg/with-s3.html). Просто убедитесь, что ваша задача имеет соответствующий тайм-аут.

Ответ 4

Я также перехожу к этой проблеме, когда я попытался объединить SFN для упорядочивания рабочих заданий AWS. описанные выше методы проблематичны, так как вы должны передать taskToken, так что вам нужно, с помощью лямбда внутри машины состояния, опросить TaskToken из очереди и передать его на S3 или где-нибудь, чтобы другая лямбда сообщила статус активности,

Проблема в том, что при опросе taskToken вы не можете знать, принадлежит ли это вашему экземпляру state-machine. вы можете получить токен на другом экземпляре одной и той же sate-машины. лично, я думаю, было бы здорово, если AWS поддержит эту функциональность, которую они легко могут сделать...

Ответ 5

Обычно вы хотите инициировать асинхронную задачу как действие функции Step. Ключевое слово здесь - инициация - другими словами, как только ваша деятельность имеет ожидающее действие, это при запуске вашего асинхронного действия. Причина этого в том, что вам нужен токен задачи, связанный с ожидающим действием, - тогда, пока ваше "будущее" может каким-то образом включить этот токен (например, вы можете установить его как ссылку или идентификатор запроса), тогда вы можете "завершить", активность с успехом или неудачей с использованием вызова SendTaskSuccess или SendTaskFailure.

Существует два подхода к инициированию задачи:

1. Опрос на новую деятельность. Вы должны настроить запланированное событие CloudWatch, чтобы вызвать вызов GetActivityTask каждые n минут.

2. Пожарайте новую задачу "инициатор" параллельно с вашей деятельностью в функции шага. Этот инициатор выполняет те же действия, что и # 1, и вызывает вызов GetActivityTask, единственное различие заключается в немедленном срабатывании и не требует механизма опроса. Вызов GetActivityTask блокируется до тех пор, пока не станет доступной новая задача, поэтому нет проблем с условиями гонки. Обратите внимание, что есть шанс, что вы можете получить активность от другого выполнения, поэтому этому инициатору нужно только учитывать ввод активности, а не вход, который сам получает инициатор.

Вот что выглядит # 2 в функции Step:

И пример базового кода, связанный с задачей InitiateManualApprovalActivity:

import boto3
import time

client = boto3.client('stepfunctions')
activity = "arn:aws:states:us-east-1:123456789012:activity:ManualStep"

def lambda_handler(event, context):
    print(event)
    # This will block until an activity task becomes available
    task = client.get_activity_task(activityArn=activity, workerName="test")
    print(task)
    # Perform your task here
    # In this example we continue on in the same function,
    # but the continuation could be a separate event, 
    # just as long as you can retrieve the task token
    time.sleep(60)
    response = client.send_task_success(taskToken=task['taskToken'], output=task['input'])
    print(response)
    return "done"

Ответ 6

Шаговые функции AWS теперь поддерживают асинхронные обратные вызовы для длительных шагов как первоклассные.

Это похоже на ответ @mixja выше, но упрощено. SendTaskSuccess состояние в вашем рабочем процессе может напрямую вызывать Lambda, SNS, SQS или ECS и ожидать вызова SendTaskSuccess.

Есть хороший пример, документированный для SQS, где пошаговая функция отправляет сообщение и приостанавливает выполнение рабочего процесса, пока что-нибудь не предоставит обратный вызов. Лямбда будет эквивалентна (при условии, что основная обработка, такая как transloadit происходит вне самой Лямбды)

Ваше определение шаговой функции будет выглядеть

"Invoke transloadit": {
  "Type": "Task",
  "Resource": "arn:aws:states:::lambda:invoke.waitForTaskToken",
  "Parameters": {
    "FunctionName": "InvokeTransloadit",
    "Payload": {
        "some_other_param": "...",
        "token.$": "$$.Task.Token"
     }
  },
  "Next": "NEXT_STATE"
}

Тогда в вашей лямбде вы бы сделали что-то вроде

def lambda_handler(event, context):
    token = event['token']

    # invoke transloadit via SSM, ECS, passing token along

затем в вашем основном долгосрочном процессе вы должны aws stepfunctions send-task-success --task-token $token обратный вызов с токеном вроде aws stepfunctions send-task-success --task-token $token из сценария оболочки /CLI или аналогичным с вызовами API.