后端任务调度工作流自动化云原生MLOps微服务【免费下载链接】flyteDynamic, resilient AI orchestration. Coordinate data, models, and compute as you build AI workflows.项目地址https://gitcode.com/gh_mirrors/fl/flyte点击查看免费下载Flyte CoPilot 是 Flyte 面向任意容器场景的关键基础设施它通过一个理解 FlyteIDL 元数据格式的flyte-copilot二进制把 Flyte 的输入输出协议翻译成普通的本地文件系统操作从而让完全不了解 Flyte SDK 的容器也能作为 Flyte 任务运行。本文以 flytecopilot/README.md 为骨架结合仓库内 CLI 实现、数据层源码与 flyteidl2/core/tasks.proto 协议定义讲透它的两种运行模式、容器生命周期编排思路、全部命令行参数以及底层数据搬运原理。核心思想把 Flyte 协议翻译成文件系统协议传统 Flyte 任务容器内通常需要安装 flytekitPython等 SDK 才能与 Flyte 平台交换输入输出。CoPilot 的目标是消除这一依赖让任意镜像、任意语言写的容器只需要读写本地文件系统就能参与 Flyte 任务。从 flyteidl2/core/tasks.proto 中DataLoadingConfig的注释可以确认这一设计意图This configuration allows executing raw containers in Flyte using the Flyte CoPilot system. Flyte CoPilot, eliminates the needs of flytekit or sdk inside the container. Any inputs required by the users container are side-loaded in the input_path. Any outputs generated by the user container - within output_path are automatically uploaded.也就是说CoPilot 负责两件事输入侧加载side-load把远程存储中的输入元数据Flyte Metadata Format即core.LiteralMap和实际数据下载到容器可见的本地路径输出侧上传auto-upload主容器结束后把约定路径下生成的输出文件尤其是元数据outputs.pb上传回远程存储。这一切由同一个二进制flyte-copilot的两种模式完成。从 flytecopilot/main.go 看二进制入口会构造根命令NewDataCommand()并挂载download与sidecar两个子命令见 flytecopilot/cmd/root.go根命令的定位描述是flytedata is a simple go binary that can be used to retrieve and upload data from/to remote stow store to local disk与 README 中2 种模式的说法一致。两种运行模式总览模式CLI 子命令部署形态职责Downloaderflyte-copilot downloadK8s init container下载元数据与数据到共享卷Sidecarflyte-copilot sidecarK8s sidecar 容器监控主容器生命周期并上传输出需要说明的是README 以 Downloader 指代该模式而源码中 cobra 子命令的实际名称是download见 flytecopilot/cmd/download.go二者是同一功能的两种叫法。Mode: Downloader —— 在 init container 中预置输入基本用法$ flyte-copilot download在 Kubernetes 中download通常作为init container运行并将下载卷shared volume挂载到主容器init container 先于主容器启动并完成数据下载下载完成后主容器启动时输入数据已全部就位在本地路径上这保证了主容器永远看不到半成品输入。从 flytecopilot/cmd/download.go 的实现看Download()会依次校验to-output-prefix、to-local-dir、from-remote三个必填参数再校验format、download-mode、file-input-layout的合法性最终调用data.NewDownloader(...).DownloadInputs(...)完成递归下载。完整命令行参数参数短选项默认值说明--from-remote-f无必填远程存储中输入元数据inputs.pb的路径/键--to-output-prefix无无必填输出元数据前缀主要用于写error.pb--to-local-dir-o无必填本地下载目录--format-mJSON元数据编码格式可选JSON/YAML/PROTO--download-mode-dDOWNLOAD_EAGER下载模式见下文 IOStrategy--file-input-layout无DIRECTFile/list[File] 输入在磁盘上的布局--timeout-t1h下载允许的最大时长--input-interface-i无base64 编码的core.VariableMap输入接口 protobuf所有取值校验都直接对照flyteidl2/core/tasks.proto中的枚举实现见 flytecopilot/cmd/download.go取值不合法时会返回带全部可选值的明确报错例如incorrect input download format specified, given [xxx], possible values [JSON YAML PROTO]。下载后的目录布局从DataLoadingConfig.input_path的注释flyteidl2/core/tasks.proto可以归纳出输入目录的标准形态例如输入接口为(x: int, y: blob, z: multipart_blob)、输入路径为/var/flyte/inputs/var/flyte/inputs/ ├── inputs.json # 元数据摘要文件依 format 可为 .pb/.json/.yaml ├── inputs.pb # 始终写入的 protobuf 版 LiteralMap ├── x # 整数 x 的字符串形式 ├── y # Blob y 的二进制内容 └── z/... # multipart blob z 是一个目录内部为各 part从 flytecopilot/data/download.go 的DownloadInputs()可以看到下载器总是写出inputs.pb当formatJSON时额外写出inputs.json当formatYAML时额外写出inputs.yaml远程引用在元数据中会被替换为本地文件系统路径handleBlob返回toPath并改写 scalar 的 URI。Mode: Sidecar —— 守护主容器生命周期并回收输出五步目标README 明确了 sidecar 模式的核心目标流程识别主容器identify the main container等待主容器启动wait for the main container to start up等待主容器退出wait for the main container to exit把数据尤其是元数据复制到远程存储copy the data to remote store退出exit$ flyte-copilot sidecar从 flytecopilot/cmd/sidecar.go 的Sidecar()看流程被实现为uploader() 错误兜底正常路径下等待容器退出后执行RecursiveUpload若上传过程中出现RawContainerError主容器以 Flyte 错误文档形式失败则直接上传该错误文档其他失败则统一上传OutputUploadFailed错误文档。完整命令行参数参数短选项默认值说明--to-output-prefix-o无必填输出元数据在 stow store 中的远程前缀--to-raw-output-x无原始输出数据的远程前缀沙箱目录--from-local-dir-f无主容器输出所在的本地目录--format-mJSON原始/结构化类型的输出元数据格式--upload-mode-uUPLOAD_ON_EXIT上传时机见下文 IOStrategy--meta-output-name无outputs.pb成功执行时输出元数据文件的键名--timeout-t1h上传允许的最大时长旧名--start-timeout已废弃但保留兼容--interface-i无base64 编码的core.TypedInterface声明输出接口--start-watcher-type无signal等待容器启动的 watcher 类型--exit-watcher-type无signal等待容器退出的 watcher 类型输出接口驱动的上传Sidecar 上传并非盲目拷贝整个目录而是严格按TypedInterface的输出变量逐个处理flytecopilot/cmd/sidecar.go未提供--interface或输出接口为空时直接按 Void 输出处理、立即退出每个输出变量按其类型分派简单类型SimpleType走handleSimpleTypeBlob 类型走handleBlobType简单类型文件有1024 字节的上限校验maxPrimitiveSize见 flytecopilot/data/upload.goBlob 类型的目录会递归遍历并并发上传每个 part上传完成后把所有core.Literal聚合为LiteralMap并以 protobuf 写入outputs.pb--meta-output-name可改名。值得注意的是源码中定义了三个约定文件名常量flytecopilot/cmd/sidecar.goconst ( StartFile _START SuccessFile _SUCCESS ErrorFile _ERROR )其中_ERROR是主容器输出目录中的错误文件Uploader 在开始上传前会先检查它若内容可反序列化为core.ErrorDocument且包含 code/message则抛出RawContainerError让平台拿到结构化错误否则按普通文本错误处理见 flytecopilot/data/upload.go 与RecursiveUpload开头部分。如何识别主容器与感知退出三种方案的演进README 中保留了Raw notes原始设计笔记对比了三种识别主容器、等待其退出的方案这对理解 watcher 抽象很有价值方案 1轮询 Kube APIpoll Kubeapi. Works perfectly fine, but too much load on kubeapi通过持续调用 Kubernetes API 查询 Pod 状态。功能上可行但会给 API Server 带来过大负载被否决。方案 2约定_SUCCESS文件协议Main container will exit and write a _SUCCESS file to a known location主容器正常退出时向约定位置写一个_SUCCESS文件sidecar 据此判定完成。缺点也很明显发生 OOM 或随机退出时不会写该文件uploader 会被卡住。README 提出的缓解思路是引入超时或在主容器异常退出时由 sidecar 直接杀掉 Pod。方案 3共享进程命名空间最终采用Use shared process namespace. This allows all pids in a pod to share the namespace. Thus pids can see each other.利用 Pod 的shareProcessNamespace让 sidecar 与主容器共享 PID 空间。README 记录了两个待解问题及解法如何识别主容器容器 ID 无法提前得知容器名到 PID 的映射不可行一种思路是调用 Kube API 获取 Pod 信息找到容器 ID另一种更轻量的思路是轮询/proc/pid/cgroup文件——该文件包含容器 ID可以建立盲的容器 ID → PID 映射如何等待主容器启动同样可以借助 Kube API 获取容器 ID 后结合 cgroup 映射解决一旦拿到主容器等待其退出、复制数据这两步已实现。落地实现Watcher 抽象当前仓库把等待启动/等待退出抽象为containerwatcher.Watcher接口flytecopilot/cmd/containerwatcher/iface.gotype Watcher interface { WaitToStart(ctx context.Context) error WaitToExit(ctx context.Context) error }内置两种实现--start-watcher-type/--exit-watcher-type可选signal默认依赖 Kubernetes 1.28 的 sidecar container 原生特性sidecar 监听SIGTERM收到信号即认为主容器退出flytecopilot/cmd/containerwatcher/signal_watcher.go。WaitToStart对 signal watcher 而言是 no-opnoop调试用占位实现直接假定容器已启动/已退出flytecopilot/cmd/containerwatcher/noop_watcher.go。协议层支撑IOStrategy 与 DataLoadingConfigCoPilot 的何时下载/何时上传策略由 flyteidl2/core/tasks.proto 中的IOStrategy定义这也是 downloader/sidecar 各参数取值的权威来源枚举取值语义DownloadModeDOWNLOAD_EAGER默认主容器启动前全部下载完成DownloadModeDOWNLOAD_STREAM流式下载写 End-Of-Stream 标记表示全部就绪DownloadModeDO_NOT_DOWNLOAD大对象offloaded不下载UploadModeUPLOAD_ON_EXIT默认主容器退出后统一上传UploadModeUPLOAD_EAGER数据出现即上传UploadModeDO_NOT_UPLOAD只写引用不上传数据DataLoadingConfigflyteidl2/core/tasks.proto则定义了数据加载的完整开关与路径约定enabled总开关未设置则不启用数据加载input_path/output_path输入下载目录 / 输出上传目录从根开始的绝对路径formatJSON/YAML/PROTO三种元数据编码——有 protobuf 定义时推荐PROTO无定义时JSON/YAML更易读io_strategy下载/上传时机策略file_input_layoutDIRECT默认File 直接落到input_path/varlist[File] 落到input_path/var/index丢弃原始文件名与扩展名/NAMED_DIR每个 File 放入独立目录并保留扩展名list[File] 元素为input_path/var/indexext便于依赖扩展名识别格式的工具按 glob 消费。NAMED_DIR布局在下载器中的实现可参见 flytecopilot/data/download.golist[File] 保留原始 basename重名时回退为索引前缀与RecursiveDownload中单 File 的目录化处理同文件 L525-L531。数据层实现亮点data包是 CoPilot 的数据搬运核心flytecopilot/data/common.go 的注释将其定位为目前只有两个工具downloader 与 uploader几个值得关注的实现细节多部分 Blob 并发下载handleBlob对MULTIPART类型先通过store.List分页每批 100 项递归列出全部 part再以 goroutine WaitGroup 并发下载并用 Mutex 保护目录创建与计数统计任何文件下载失败或读写流未正常关闭都会汇总为明确错误flytecopilot/data/download.goHTTP 直下载对http/https协议的引用走DownloadFileFromHTTP带 context 取消的 GET 请求其余走DownloadFileFromStorage先 Head 确认存在再 ReadRaw兼容 S3/GCS 等 stow 后端flytecopilot/data/utils.goOffloaded 大对象处理RecursiveDownload在下载每个变量前检查literal.GetOffloadedMetadata()若存在则先从远程读取真正的字面量内容再继续下载flytecopilot/data/download.go简单类型按原生格式落盘handlePrimitive支持 string、bool、integer、float、datetimeRFC3339Nano、duration 等原生类型序列化写入flytecopilot/data/download.go。错误处理闭环CoPilot 的错误处理设计保证了主容器失败也能被平台感知主容器写入_ERROR文件内容可为结构化core.ErrorDocument或纯文本Uploader 上传前先检查该文件结构化错误 →RawContainerError由 sidecar 原样上传错误文档kind/origin 由调用方设置见 flytecopilot/cmd/root.go平台据此判断是否可重试非结构化错误 → 包装为User Error: 内容上传downloader/sidecar 自身的失败如下载失败则通过UploadError(ctx, InputDownloadFailed / OutputUploadFailed, ...)写入error.pb错误文件名由根命令的--err-output-name控制默认error.pb。限制与注意事项简单类型输出文件上限 1024 字节超出会报错flytecopilot/data/upload.goUploader 目前仅支持LiteralType_Blob与LiteralType_Simple两类输出类型其余类型返回 currently CoPilot uploader does not support ... system errorflytecopilot/data/upload.go重试尚未在data层自动处理见 flytecopilot/data/common.go 的 TODO 注释DataLoadingConfig的注释明确指出该能力仅支持 KubernetesThis is supported only on K8s at the moment。源码阅读地图若想深入理解推荐按以下路径阅读协议定义flyteidl2/core/tasks.protoIOStrategy / DataLoadingConfig / FileInputLayoutCLI 入口与根命令flytecopilot/main.go、flytecopilot/cmd/root.go下载器命令与实现flytecopilot/cmd/download.go、flytecopilot/data/download.go上传器命令与实现flytecopilot/cmd/sidecar.go、flytecopilot/data/upload.go容器生命周期 watcherflytecopilot/cmd/containerwatcher/iface.go、flytecopilot/cmd/containerwatcher/signal_watcher.go总结而言Flyte CoPilot 的优雅之处在于它不要求容器理解任何 Flyte 专有协议只要求容器读写本地文件系统——下载器把远程输入翻译成本地文件sidecar 把本地输出翻译回远程元数据中间的主容器则完全保持原生。赞分享后端任务调度工作流自动化云原生MLOps微服务【免费下载链接】flyteDynamic, resilient AI orchestration. Coordinate data, models, and compute as you build AI workflows.项目地址https://gitcode.com/gh_mirrors/fl/flyte点击查看免费下载相关推荐Flyte项目原生调度器架构深度解析Flyte项目原生调度器架构深度解析 概述云原生工作流编排的核心引擎 Flyte Propeller螺旋桨是Flyte项目的核心调度引擎作为Kubern后端任务调度工作流自动化云原生MLOps微服务Distrobox 实战指南在任意 Linux 主机上运行任意发行版容器Distrobox 实战指南在任意 Linux 主机上运行任意发行版容器 Distrobox 是一个基于 podman 、 docker 或 lilipod开发工具CLI探索downkyicore用专业工具轻松提取B站音频的实战指南探索downkyicore用专业工具轻松提取B站音频的实战指南 downkyicore作为一款功能强大的哔哩哔哩视频下载工具不仅支持8K、HDR、杜比视界等上一篇23种设计模式在Android开源项目中的终极实践指南从理论到AOSP源码分析下一篇Go测试完全指南Go Practical Tips中的表格驱动测试与并行测试技巧创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考