Cromwell 工作流引擎支持

开通批量计算服务

要使用批量计算服务，请根据官方文档里面的指导开通批量计算和其依赖的相关服务，如OSS等。

下载 Cromwell

Cromwell 官方下载

开通 ECS 作为 Cromwell server

当前批量计算提供了 Cromwell server 的 ECS 镜像，用户可以用此镜像开通一台 ECS 作为 server。镜像中提供了 Cromwell 官网要求的基本配置和常用软件。在此镜像中，Cromwell 的工作目录位于/home/cromwell，上一步下载的 Crowwell jar 包可以放置在 /home/cromwell/cromwell 目录下。

配置文件

Cromwell 运行的配置文件，包括：

• Cromwell 公共配置。
• 批量计算相关配置，包含了批量计算作为后端需要的存储、计算等资源配置。

关于配置参数的详细介绍请参考 Cromwell 官方文档。如下是一个批量计算配置文件的例子 bcs.conf:

include required(classpath("application"))


database {
  profile = "slick.jdbc.MySQLProfile$"
  db {
    driver = "com.mysql.jdbc.Driver"
    url = "jdbc:mysql://localhost/db_cromwell?rewriteBatchedStatements=true&useSSL=false&allowPublicKeyRetrieval=true"
    user = "user_cromwell"
    #Your mysql password
    password = ""
    connectionTimeout = 5000
  }
}

workflow-options {
    workflow-log-dir = "/home/cromwell/cromwell/logs/"
}

call-caching {
  # Allows re-use of existing results for jobs you've already run
  # (default: false)
  enabled = false

  # Whether to invalidate a cache result forever if we cannot reuse them. Disable this if you expect some cache copies
  # to fail for external reasons which should not invalidate the cache (e.g. auth differences between users):
  # (default: true)
  invalidate-bad-cache-results = true
}

docker {
  hash-lookup {
    enabled = false
    # Set this to match your available quota against the Google Container Engine API
    #gcr-api-queries-per-100-seconds = 1000

    # Time in minutes before an entry expires from the docker hashes cache and needs to be fetched again
    #cache-entry-ttl = "20 minutes"

    # Maximum number of elements to be kept in the cache. If the limit is reached, old elements will be removed from the cache
    #cache-size = 200

    # How should docker hashes be looked up. Possible values are "local" and "remote"
    # "local": Lookup hashes on the local docker daemon using the cli
    # "remote": Lookup hashes on docker hub and gcr
    method = "remote"
    #method = "local"
    alibabacloudcr {
      num-threads = 5
      #aliyun CR credentials
      auth {
    #endpoint = "cr.cn-shanghai.aliyuncs.com"
        access-id = ""
        access-key = ""
      }
    }
  }
}

engine {
  filesystems {
    oss {
      auth {
        endpoint = "oss-cn-shanghai.aliyuncs.com"
        access-id = ""
        access-key = ""
      }
    }
  }
}

backend {
  default = "BCS"

  providers {
    BCS {
      actor-factory = "cromwell.backend.impl.bcs.BcsBackendLifecycleActorFactory"
      config {
        root = "oss://your-bucket/cromwell_dir"
        region = "cn-shanghai"
        access-id = ""
        access-key = ""

        filesystems {
          oss {
            auth {
              endpoint = "oss-cn-shanghai.aliyuncs.com"
              access-id = ""
              access-key = ""
            }

        caching {
              # When a cache hit is found, the following duplication strategy will be followed to use the cached outputs
              # Possible values: "copy", "reference". Defaults to "copy"
              # "copy": Copy the output files
              # "reference": DO NOT copy the output files but point to the original output files instead.
              #              Will still make sure than all the original output files exist and are accessible before
              #              going forward with the cache hit.
              duplication-strategy = "reference"
            }
          }
        }

        default-runtime-attributes {
          failOnStderr: false
          continueOnReturnCode: 0
          autoReleaseJob: false
          cluster: "OnDemand ecs.sn1.medium img-ubuntu-vpc"
          #cluster: cls-6kihku8blloidu3s1t0006
          vpc: "192.168.0.0/16"
        } 
      }
    }
  }
}

如果使用前面章节中的镜像开通 ECS 作为 Cromwell server，配置文件位于 /home/cromwell/cromwell/bcs_sample.conf，只需要填写自己的配置即可使用 Cromwell。

运行模式

Cromwell支持两种模式：

• run 模式
• server 模式

关于两种模式的详细描述，请参考 Cromwell 官网文档。下面重点介绍这两种模式下如何使用批量计算。

run模式

run模式适用于本地运行一个单独的 WDL 文件描述的工作流，命令行如下：java -Dconfig.file=bcs.conf -jar cromwell.jar run echo.wdl --inputs echo.inputs

• WDL 文件：描述详细的工作流。工作流中每个 task 对应批量计算的一个作业（Job）。
• inputs文件：是 WDL 中定义的工作流的输入信息inputs 文件是用来描述 WDL 文件中定义的工作流及其 task 的输入文件。如下所示：

  {
    "workflow_name.task_name.input1": "xxxxxx"
  }

运行成功后，WDL 文件中描述的工作流中的一个 task 会作为批量计算的一个作业（Job）来提交。此时登录批量计算的控制台就可以看到当前的 Job 状态。

当 workflow 中所有的 task 对应的作业运行完成后，工作流运行完成。

server 模式

启动 server

相比 run 模式一次运行只能处理一个 WDL 文件，server 模式可以并行处理多个 WDL 文件。关于 server 模式的更多信息，请参考 Cromwell 官方文档。可以采用如下命令行启动 server：java -Dconfig.file=bsc.conf -jar cromwell.jar serverserver 启动成功后，就可以接收来自 client 的工作流处理请求。下面分别介绍如何使用 API 和 CLI 的方式向 server 提交工作流。

使用 API 提交工作流

server 启动后，可以通过浏览器访问 Cromwell Server，比如 Server 的 IP 为39.105.xxx.yyy，则在浏览器中输入http://39.105.xxx.yyy:8000，通过如下图所示的界面提交任务：更多API接口及用法，请参考 Cromwell 官网文档。

使用 CLI 提交工作流[推荐]

除了可以使用 API 提交工作流以外，Cromwell 官方还提供了一个开源的 CLI 命令行工具 widder。可以使用如下的命令提交一个工作流：

python widdler.py run echo.wdl echo.inputs -o bcs_workflow_tag:tagxxx -S localhost

其中-o key:value是用于设置option，批量计算提供了 bcs_workflow_tag:tagxxx 选项，用于配置作业输出目录的tag（下一节查看运行结果中会介绍）。

如果使用前面章节中的镜像开通 ECS 作为 Cromwell server，镜像中已经安装了 widdler，位于 /home/cromwell/widdler。可以使用如下的命令提交工作流：

widdler run echo.wdl echo.inputs -o bcs_workflow_tag:tagxxx -S localhost

更多命令用法可使用widdler -h命令查看，或参考官方文档。

查看运行结果

工作流运行结束后，输出结果被上传到了配置文件或 WDL 中定义的 OSS 路径下。在OSS路径上面的目录结构如下：

如上图所示，在配置文件中的config.root目录下有如下输出目录：

• 第一层：workflowname 工作流的名称
• 第二层：通过上一节中 CLI 命令的-o设置的目录tag
• 第三层：workflow id，每次运行会生成一个
• 第四层：workflow 中每个 task 的运行输出，比如上图中的 workflow 15e45adf-6dc7-4727-850c-89545faf81b0 有两个 task，每个task对应的目录命名是call-taskname，目录中包含三部分内容：
  • 批量计算的日志，包括 bcs-stdout 和 bcs-stderr
  • 当前 task 的输出，比如图中的 output1/output2 等
  • 当前 task 执行的 stdout 和 stderr

使用建议

在使用过程中，关于 BCS 的配置，有如下的建议供参考：

使用集群

批量计算提供了两种使用集群的方式：

• 自动集群
• 固定集群

自动集群

在config配置文件中指定默认的资源类型、实例类型以及镜像类型，在提交批量计算 Job 时就会使用这些配置自动创建集群，比如：

default-runtime-attributes {
        cluster : "OnDemand ecs.sn1ne.large img-ubuntu-vpc"
      }

如果在某些 workflow 中不使用默认集群配置，也可以通过inputs文件中指定 workflow 中某个 task 的对应的批量计算的集群配置（将 cluster_config 作为 task 的一个输入），比如：

{
      "workflow_name.task_name.cluster_config": "OnDemand ecs.sn2ne.8xlarge img-ubuntu-vpc"
}

然后在 task 中重新设置运行配置：

task task_demo {
    String cluster_config

    runtime {
        cluster: cluster_config
  }
}

就会覆盖默认配置，使用新的配置信息创建集群。

固定集群

使用自动集群时，需要创建新集群，会有一个等待集群的时间。如果对于启动时间有要求，或者有了大量的作业提交，可以考虑使用固定集群。比如：

default-runtime-attributes {
        cluster : "cls-xxxxxxxxxx"
      }

注意：使用固定集群时，如果使用完毕，请及时释放集群，否则集群中的实例会持续收费。

Cromwell Server 配置建议

• 大压力作业时，建议使用较高配置的机器作为 Cromwell Server，比如ecs.sn1ne.8xlarge等32核64GB的机器。
• 大压力作业时，修改 Cromwell Server 的最大打开文件数。比如在ubuntu下可以通过修改/etc/security/limits.conf文件，比如修改最大文件数为100万：

  root soft nofile 1000000
  root hard nofile 1000000
  * soft nofile 1000000
  * hard nofile 1000000

• 确认 Cromwell Server 有配置数据库，防止作业信息丢失。
• 设置 bcs.conf 里面的并发作业数，比如 system.max-concurrent-workflows = 1000

开通批量计算相关配额

如果有大压力场景，可能需要联系批量计算服务开通对应的配额，比如：

• 一个用户所有作业的数量（包括完成的、运行的、等待的等多种状态下）；
• 同时运行的作业的集群的数量（包括固定集群和自动集群）；

使用 NAS

使用 NAS 时要注意以下几点：

• NAS 必须在 VPC 内使用，要求添加挂载点时，必须指定 VPC；
• 所以要求在 runtime 中必须包含：
  • VPC 信息
  • mounts 信息

下面的例子可供参考：

runtime {
    cluster: cluster_config
    mounts: "nas://1f****04-xkv88.cn-beijing.nas.aliyuncs.com:/ /mnt/ true"
    vpc: "192.168.0.0/16 vpc-2zexxxxxxxx1hxirm"
  }

如果是有多个目录需要mount，可采用下面的方式

runtime {
    mounts: "nas://1f****04-xkv88.cn-beijing.nas.aliyuncs.com:/ /mnt1/ true, nas://1f****04-xkv99.cn-beijing.nas.aliyuncs.com:/ /mnt2/ true"
}

即两组mount之间用逗号隔开。

高级特性支持

Glob

Cromwell 支持使用 glob 来指定工作流中多个文件作为 task 的输出，比如：

task globber {
  command <<<
    for i in `seq 1 5`
    do
      mkdir out-$i
      echo globbing is my number $i best hobby  out-$i/$i.txt
    done
  >>>
  output {
    Array[File] outFiles = glob("out-*/*.txt")
  }
}

workflow test {
  call globber
}

当 task 执行结束时，通过 glob 指定的多个文件会作为输出，上传到 OSS 上。

Call Caching

Call Caching 是 Cromwell 提供的高级特性，如果检测到工作流中某个 task（对应一个批量计算的 job）和之前已经执行过的某个 task 具有相同的输入和运行时等条件，则不需要再执行，直接取之前的运行结果，这样可以为客户节省时间和费用。一个常见的场景是如果一个工作流有 n 个 task，当执行到中间某一个 task 时由于某些原因失败了，排除了错误之后，再次提交这个工作流运行后，Cromwell 判断如果满足条件，则已经完成的几个 task 不需要重新执行，只需要从出错的 task 开始继续运行。

配置 Call Caching

要在 BCS 后端情况下使用 Call Caching 特性，需要如下配置项：

database {
  profile = "slick.jdbc.MySQLProfile$"
  db {
    driver = "com.mysql.jdbc.Driver"
    url = "jdbc:mysql://localhost/db_cromwell?rewriteBatchedStatements=true&useSSL=false"
    user = "user_cromwell"
    password = "xxxxx"
    connectionTimeout = 5000
  }
}

call-caching {
  # Allows re-use of existing results for jobs you have already run
  # (default: false)
  enabled = true

  # Whether to invalidate a cache result forever if we cannot reuse them. Disable this if you expect some cache copies
  # to fail for external reasons which should not invalidate the cache (e.g. auth differences between users):
  # (default: true)
  invalidate-bad-cache-results = true
}

docker {
  hash-lookup {

    enabled = true

    # How should docker hashes be looked up. Possible values are local and remote
    # local: Lookup hashes on the local docker daemon using the cli
    # remote: Lookup hashes on alibab cloud Container Registry
    method = remote
    alibabacloudcr {
      num-threads = 10
      auth {
        access-id = "xxxx"
        access-key = "yyyy"
      }
    }
  }
}

engine {
  filesystems {
    oss {
      auth {
        endpoint = "oss-cn-shanghai.aliyuncs.com"
        access-id = "xxxx"
        access-key = "yyyy"
      }
    }
  }
}

backend {
  default = "BCS"

  providers {
    BCS {
      actor-factory = "cromwell.backend.impl.bcs.BcsBackendLifecycleActorFactory"
      config {

        #其他配置省略

        filesystems {
          oss {
            auth {
              endpoint = "oss-cn-shanghai.aliyuncs.com"
              access-id = "xxxx"
              access-key = "yyyy"
            }
            caching {
              # When a cache hit is found, the following duplication strategy will be followed to use the cached outputs
              # Possible values: copy, reference. Defaults to copy
              # copy: Copy the output files
              # reference: DO NOT copy the output files but point to the original output files instead.
              #              Will still make sure than all the original output files exist and are accessible before
              #              going forward with the cache hit.
              duplication-strategy = "reference"
            }
          }
        }

        default-runtime-attributes {
          failOnStderr: false
          continueOnReturnCode: 0
          cluster: "OnDemand ecs.sn1.medium img-ubuntu-vpc"
          vpc: "192.168.0.0/16"
        }
      }
    }
  }
}

• database 配置：Cromwell 将 workflow 的执行元数据存储在数据库中，所以需要添加数据库配置，详细情况参考Cromwell 官网指导。
• call-caching 配置：Call Caching 的开关配置等；
• docker.hash-lookup 配置： 设置 Hash 查找开关及阿里云 CR 等信息，用于查找镜像的 Hash 值。
• backend.providers.BCS.config.filesystems.oss.caching 配置：设置 Call Caching命中后，使用原来输出的方式，批量计算在这里支持 reference 模式，不需要拷贝原有的结果，节省时间和成本。

命中条件

使用批量计算作为后端时，Cromwell 通过如下条件判断一个 task 是否需要重新执行：

如果一个 task 的上述参数未发生改变，Cromwell 会判定为不需要执行的 task，直接获取上次执行的结果，并继续工作流的执行。