问用Go SDK检查AWS数据管道的状态
EN

Stack Overflow用户

提问于 2017-09-12 11:32:08

回答 1查看 548关注 0票数 1

情境:我有两个按需运行的数据管道.管道B在A管线完成之前无法运行。我试图在一个脚本/程序中自动运行这两个管道，但我不知道如何在Go中实现这一切。

我有一些Go代码可以激活数据管道：

func awsActivatePipeline(pipelineID, region string) (*datapipeline.ActivatePipelineOutput, error) {
    svc := datapipeline.New(session.New(&aws.Config{Region: aws.String(region)}))
    input := &datapipeline.ActivatePipelineInput{
        PipelineId: aws.String(pipelineID),
    }
    result, err := svc.ActivatePipeline(input)
    if err != nil {
        fmt.Println("error activating pipeline: ", err)
    }
    fmt.Println(result)
    return result, nil
}

激活后，我希望能够监视该管道，并确定它何时完成，以便能够运行第二个管道。类似于list-runs命令，但我不确定对应的Go函数是什么。

$ aws datapipeline list-runs --region us-west-2 --pipeline-id df-EXAMPLE
       Name                                                Scheduled Start      Status                 
       ID                                                  Started              Ended              
---------------------------------------------------------------------------------------------------
   1.  EC2ResourceObj                                      2017-09-12T17:49:55  FINISHED               
       @EC2ResourceObj_2017-09-12T17:49:55                 2017-09-12T17:49:58  2017-09-12T17:56:52

   2.  Installation                                        2017-09-12T17:49:55  FINISHED               
       @Installation_@ShellCommandActivityObj_2017-09-12T  2017-09-12T17:49:57  2017-09-12T17:54:09

   3.  S3OutputLocation                                    2017-09-12T17:49:55  FINISHED               
       @S3OutputLocation_2017-09-12T17:49:55               2017-09-12T17:49:58  2017-09-12T17:54:50

   4.  ShellCommandActivityObj                             2017-09-12T17:49:55  FINISHED               
       @ShellCommandActivityObj_2017-09-12T17:49:55        2017-09-12T17:49:57  2017-09-12T17:54:49

因此，一旦所有的动作被标记为‘完成’，我想激活我的第二个管道。做这件事最好的方法是什么？

amazon-web-services

amazon-data-pipeline

data-pipeline

校园云服务器 2H2G30元起

云+校园是针对学生用户推出的扶持计划，2核2G云服务器30元起，1核1G云数据库4元起，并享受免费体验6个月云开发

回答 1

Stack Overflow用户

回答已采纳

发布于 2018-12-11 09:40:15

如果其他人遇到这种情况，我就是这样解决这个问题的：

Golang调用来描述数据管道的对象/操作，如果所有对象都完成了，则返回true

    func awsDescribeObjects(pipelineID, region string, objects []string) bool {
        var r Object
        var s []string
        var f bool
        svc := datapipeline.New(session.New(&aws.Config{Region: aws.String(region)}))
        input := &datapipeline.DescribeObjectsInput{
            PipelineId: aws.String(pipelineID),
            ObjectIds:  aws.StringSlice(objects),
        }
        result, err := svc.DescribeObjects(input)
        if err != nil {
            fmt.Println("error describing pipeline objects: ", err)
            f = false
            return f
        }
        //fmt.Println("original result: ", result)
        result2 := re.ReplaceAllString(result.String(), `"$1"$2`) //add "" around keys
        result3 := re1.ReplaceAllString(result2, `$3$2`)          //remove key and string/ref value from fields struct
        result4 := strings.Replace(result3, "@", "", -1)          //remove @ from keys and values
        result5 := re2.ReplaceAllString(result4, `$1$3$5$7$9`)    //remove "" from timestamps
        result6 := re3.ReplaceAllString(result5, `$1,`)           // remove {} from fields struct
        json.Unmarshal([]byte(result6), &r)
        // fmt.Printf("R: %+v\n", r)
        p := r.PipelineObjects
        // fmt.Printf("P: %+v\n", p)
        for i := range p {
            for m := range p[i].Fields {
                fmt.Printf("%v STATUS: %v\n", p[i].Name, p[i].Fields[m].Status)
                s = append(s, p[i].Fields[m].Status)
                if p[i].Fields[m].Status != "FINISHED" {
                    f = false
                } else {
                    f = true
                }
            }
            // fmt.Println("bool: ", f)
        }
        return f
    }

我的主要围棋功能

    func main() {
        if *action == "describe" {
            obj := strings.Split(*object, ",")

            for i := 0; i <= 20; i++ {
                f := awsDescribeObjects(*pipeline, *region, obj)
                fmt.Printf("%v - Status Check %v - Finished?: %v\n", time.Now(), i, f)
                if f == true {
                    fmt.Println("FINISHED describing pipeline complete")
                    break
                }
                time.Sleep(5 * time.Minute)
                if i == 20 {
                    fmt.Println("TIME OUT - describe pipeline timed out, max time reached")
                    os.Exit(1)
                }
            }
        }
    }

具有go可执行文件的Shell脚本：

#PIPELINE 1
echo "Starting Pipeline 1..."
echo ./runpipeline.linux -region $REGION1 -pipeline-id $PIPELINEID1 -action activate
echo sleep 1m
echo ./runpipeline.linux -region $REGION1 -pipeline-id $PIPELINEID1 -action describe -object ShellCommandActivityObj
echo "Pipeline 1 complete"
#PIPELINE 2
echo "Starting Pipeline 2..."
echo ./runpipeline.linux -region $REGION2 -pipeline-id $PIPELINEID2 -action activate
echo sleep 1m
echo ./runpipeline.linux -region $REGION2 -pipeline-id $PIPELINEID2 -action describe -object ShellCommandActivityObj,CliActivity
echo "Pipeline 2 complete"
echo "FINISHED"