0

适用于 Go 的 AWS 开发工具包版本?

v2.0.0-preview.3

Go ( go version) 的版本?

go1.9.3 达尔文/amd64

你看到了什么问题?

我正在为 Go 中的 Step Functions 编写一个 Activity Worker。

什么时候:

  • 活动工作者正在运行,
  • 然后,我们从 SFN 控制台开始执行工作流

一切似乎都运行良好。

但是,当:

  • 活动工作者停止,
  • 然后,我们从 SFN 控制台开始执行工作流,
  • 然后工人重新启动,

工作人员似乎在轮询 SFN,但它不执行在停止期间启动的任务。如果我们此时开始一个新的工作流执行(当工作者运行时),工作者成功地执行新任务。在工作人员停止期间执行的工作流没有被工作人员拾取。

编辑:查看执行历史记录,我看到超时状态和以下事件日志:

在此处输入图像描述

重现步骤

在这里,如果我的 SFN 状态机:

{
  "Comment": "An example using a Task state.",
  "StartAt": "getGreeting",
  "Version": "1.0",
  "TimeoutSeconds": 300,
  "States":
  {
    "getGreeting": {
      "Type": "Task",
      "Resource": "arn:aws:states:ap-southeast-1:196709014601:activity:get-greeting",
      "End": true
    }
  }
}

这是我的 SFN 工人:

package main

import (
    "github.com/aws/aws-sdk-go-v2/aws"
    "github.com/aws/aws-sdk-go-v2/aws/endpoints"
    "github.com/aws/aws-sdk-go-v2/aws/external"
    "github.com/aws/aws-sdk-go-v2/service/sfn"
    "fmt"
    "encoding/json"
)

type Worker struct {
    svc             *sfn.SFN
    activityARN     string
}

type Task struct {
    input   *string
    token   *string
}

func New(activityARN string) *Worker {
    cfg, err := external.LoadDefaultAWSConfig()
    if err != nil {
        panic("unable to load SDK config, " + err.Error())
    }
    // Set the AWS Region that the service clients should use
    cfg.Region = endpoints.ApSoutheast1RegionID

    // Using the Config value, create the Step Functions client
    svc := sfn.New(cfg)

    w := &Worker{
        svc: svc,
        activityARN: activityARN,
    }
    return w
}

func (w *Worker) getTask() *Task {
    req := w.svc.GetActivityTaskRequest(&sfn.GetActivityTaskInput{
        ActivityArn: aws.String(w.activityARN),
    })
    res, err := req.Send()
    if err != nil { fmt.Println("failed to get tasks, "+err.Error()) }
    return &Task{
        input: res.Input,
        token: res.TaskToken,
    }
}

// Call SendTaskSuccess on success
func (w *Worker) handleSuccess(taskToken *string, json *string) error {
    req := w.svc.SendTaskSuccessRequest(&sfn.SendTaskSuccessInput{
        TaskToken: taskToken,
        Output: json, // JSON string
    })
    _, err := req.Send()
    if err != nil { fmt.Println("failed to send task success result, "+err.Error()) }
    return err
}

// Call SendTaskFailure on error
func (w *Worker) handleFailure(taskToken *string, err error) error {
    errorMessage := err.Error()
    req := w.svc.SendTaskFailureRequest(&sfn.SendTaskFailureInput{
        TaskToken: taskToken,
        Error: &errorMessage,
    })

    _, err = req.Send()
    if err != nil { fmt.Println("failed to send task failure result, "+err.Error()) }
    return err
}

func main() {
    activityARN := "arn:aws:states:ap-southeast-1:196709014601:activity:get-greeting"
    worker := New(activityARN)

    fmt.Println("Starting worker")
    for {
        // 1. Poll GetActivityTask API for tasks
        fmt.Println("Polling for tasks")
        task := worker.getTask()
        if task.token == nil { continue }

        // 2. Do some actual work
        fmt.Println("Working")
        result, err := work(task.input)

        // 3. Notify SFN on success and failure
        fmt.Println("Sending results")
        if err == nil {
            worker.handleSuccess(task.token, result)
        } else {
            worker.handleFailure(task.token, err)
        }
    }
}

// Handles marshalling and un-marshalling JSON
func work(jsonInput *string) (*string, error) {
    input := &GreetInput{}
    json.Unmarshal([]byte(*jsonInput), input)

    result, err := Greet(input) // Actual work
    if err != nil { return nil, err }

    outputBytes, _ := json.Marshal(result)
    output := string(outputBytes)
    return &output, nil
}

// Actual handler code
type GreetInput struct {
    Who string
}

type GreetOutput struct {
    Message string
}

func Greet(input *GreetInput) (*GreetOutput, error) {
    message := fmt.Sprintf("hello %s", input.Who)
    output := &GreetOutput {
        Message: message,
    }
    fmt.Println(message)
    return output, nil
}

跑步:

go build worker.go && ./worker
4

1 回答 1

1

根据您的更新,我认为工作人员没有正常停止(即在杀死工作人员时您没有等到GetActivityTask请求结束),因此 Step Functions 可能会响应(已经死亡的)工作人员。

所以工作流程如下:

  1. Worker 发送GetActivityTask请求并暂停(直到达到超时)。
  2. 工人在没有等待GetActivityTask结束的情况下被杀死。
  3. 创建新的执行。
  4. Step Functions 发现有些GetActivityTask仍然挂起 - 将新执行的任务发送给它。
  5. 但是工人已经死了,所以它不会收到那个任务。Step Functions 认为任务已交付,因此它会等到任务结束或超时。

要检查是否是这种情况,只需在杀死工作人员后稍等片刻(我不知道GetActivityTaskAWS SDK for Go 中的默认等待时间是多少 - 5 分钟应该可以完成这项工作),然后创建执行。如果新的执行按预期工作,那么您应该向工作人员添加优雅退出(等到GetActivityTask结束并最终处理任务)。

于 2018-03-19T19:35:10.277 回答