FATE Board 执行流程探索

2024-08-30 10:36
文章标签 board 流程 探索 执行 fate

本文主要是介绍FATE Board 执行流程探索,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

背景介绍

FATE Board 是 FATE 提供的一个工程,用于给 FATE 提供可视化能力,方便在联邦学习训练中实时查看执行状态,更好地定位执行中遇到的问题。

查看 FATE 架构可以看到 FATE Board 是建立在 MySQL 和 FATE Flow Server 的基础上的,看起来数据来源是来自于这两者。FATE Flow Server 在之前的文章 中已经介绍过,FATE 中隐私计算的主要调度流程都是实现在这个服务中。

请添加图片描述

FATE Board 代码仓库地址 https://github.com/FederatedAI/FATE-Board, 本文的探索基于 v1.11.1,后续版本可能有所不同

FATE Board 实现探索

FATE Board 工程中包含前端与后端的实现,前端是基于 Vue 实现的,后端则是基于 Java 实现。本文在探索时主要基于两个场景串联了一下完整的流程,分别是主页面的 job 列表页,以及 job 日志详情,通过查看完整的调用链路,对 FATE Board 建立基础的认识。

Job 列表页

请添加图片描述

通过 Chrome 调试模式查看对应的请求,即可比较容易发现获取 job 列表数据对应的请求为 /job/query/page/new , 通过对应的接口路径全局搜索可以发现后端的实现为 src/main/java/com/webank/ai/fate/board/controller/JobManagerController.java 中的 queryPagedJob() 方法,对应的代码实现如下:

public PageBean<Map<String, Object>> queryPagedJobs(PagedJobQO pagedJobQO) {String jobId = pagedJobQO.getJobId();FlowJobQO flowJobQO = new FlowJobQO();if (jobId != null && 0 != jobId.trim().length()) {flowJobQO.setJob_id(pagedJobQO.getJobId());}// 构造请求参数 ...// 实际获取数据Map<String, Object> jobMap = getJobMap(flowJobQO);// ... 冗长的业务处理
}

可以看到的真正的数据获取部分基本就是直接调用 getJobMap() ,对应的实现如下所示:

private Map<String, Object> getJobMap(Object query) {result = flowFeign.post(Dict.URL_JOB_QUERY, JSON.toJSONString(query));// ... 冗长的结果转换
}

实际的获取是通过一次 HTTP 请求获取到,对应的请求地址为 /v1/job/list/job,看情况应该是调用 FATE Flow Server 获取的,在 FATE-Flow 中看到的对应的接口,处于路径 FATE-Flow/python/fate_flow/apps/job_app.py 中的 list_job(),实际的实现就是一次简单的数据库查询,不再进一步展开。

Job 日志

请添加图片描述

通过 chrome 调试模式看到实际获取 Job 日志是通过 websocket 获取的,请求的地址为 /log/new/202307260855242117390/host/8889/default,目前来看日志的获取和 job 列表的获取存在一些差异

依旧利用请求地址搜索对应的代码实现,可以确认后端对应的实现路径为 src/main/java/com/webank/ai/fate/board/websocket/LogWebSocketController.java 中的 LogWebSocketController 类实现,对于 websocket 的服务端,消息处理都是在 onMessage 实现的,我们可以看到对应的代码实现如下:

@OnMessage
public void onMessage(String message,Session session,@PathParam("jobId") String jobId,@PathParam("role") String role,@PathParam("partyId") String partyId,@PathParam("componentId") String componentId) throws Exception {synchronized (session) {LogQuery logQuery = JSON.parseObject(message, LogQuery.class);// 根据类型主要包含 logSize 和 logCat,其中 logSize 用于获取日志行数,logCat 获取日志内容if (logQuery.getType().equals(LogTypeEnum.LOG_SIZE.boardValue)) {logSize(session, jobId, role, partyId, componentId, logQuery);} else {logCat(session, jobId, role, partyId, componentId, logQuery);}}
}

可以看到的通过路径获取 jobId, role, partyId, componentId 的参数,然后调用 logSize()logCat() 执行实际的处理,我们主要关注日志内容的获取,可以看到 logCat() 对应的实现如下所示:

private void logCat(Session session, String jobId, String role, String partyId, String componentId, LogQuery logQuery) {// 构造请求FlowLogCatReq flowLogCatReq = new FlowLogCatReq();flowLogCatReq.setJob_id(jobId);flowLogCatReq.setLog_type(Dict.logTypeMap.get(logQuery.getType()));flowLogCatReq.setRole(role);flowLogCatReq.setParty_id(Integer.valueOf(partyId));flowLogCatReq.setComponent_name(componentId);flowLogCatReq.setInstance_id(logQuery.getInstanceId());flowLogCatReq.setBegin(logQuery.getBegin());flowLogCatReq.setEnd(logQuery.getEnd());// 实际获取数据FlowResponse<List<FlowLogCatResp>> resultFlow = flowLogFeign.logCat(flowLogCatReq);// 构造响应数据LogContentResponse logContentResponse = new LogContentResponse();logContentResponse.setType(logQuery.getType());logContentResponse.setData(resultFlow.getData().stream().map(LogContentResponse.LogContent::fromFlowContent).collect(Collectors.toList()));try {session.getBasicRemote().sendText(JSON.toJSONString(logContentResponse));} catch (IOException e) {e.printStackTrace();logger.error("websocket send error: {}", logContentResponse);}
}

根据最核心的数据获取是调用 flowLogFeign.logCat() ,对应的实现:

@FeignClient(url = RouteTargeter.URL_PLACE_HOLDER + "/v1/log", name = "flowLogFeign", configuration = FeignRequestInterceptor.class)
public interface FlowLogFeign {// 构造 http 请求@RequestMapping(value = "/cat", method = RequestMethod.POST)FlowResponse<List<FlowLogCatResp>> logCat(FlowLogCatReq request);@RequestMapping(value = "/size", method = RequestMethod.POST)FlowResponse<FlowLogSizeResp> logSize(FlowLogSizeReq request);
}

最后兜了一圈,看起来还是转换了一次网络请求,看起来还是发送给了 FATE Flow Server,追踪 FATE-Flow 工程中的对应实现,可以看到对应的网络请求位于 FATE-Flow/python/fate_flow/apps/log_app.py 路径下,具体的实现位于 FATE-Flow/python/fate_flow/utils/log_sharing_utils.py 中的 cat_log() 方法中,实现如下:

def cat_log(self, begin, end):line_list = []log_path = self.get_log_file_path()if begin and end:cmd = f"cat {log_path} | tail -n +{begin}| head -n {end-begin+1}"elif begin:cmd = f"cat {log_path} | tail -n +{begin}"elif end:cmd = f"cat {log_path} | head -n {end}"else:cmd = f"cat {log_path}"lines = self.execute(cmd)if lines:line_list = []line_num = begin if begin else 1for line in lines.split("\n"):line = replace_ip(line)line_list.append({"line_num": line_num, "content": line})line_num += 1return line_list

可以看到最终就是调用系统的 cat 命令,最终文件对应的内容,整体实现简单直接。

总结

通过对 FATE-Board 两个请求的调用链路的跟踪,可以对 FATE-Board 工程有了一些了解,看起来 FATE-Board 是建立在 FATE-Flow 基础上的一个简单可视化,使用的能力基本都是通过 FATE-Flow 提供,而 FATE-Board 仅仅提供必要的数据包装与前端的展示呈现,过程简单清晰。后续如果希望了解 FATE-Board 对应的可视化的能力范围,直接查看 FATE-Flow 对应提供的接口即可

这篇关于FATE Board 执行流程探索的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1120576

相关文章

Spring定时任务只执行一次的原因分析与解决方案

《Spring定时任务只执行一次的原因分析与解决方案》在使用Spring的@Scheduled定时任务时,你是否遇到过任务只执行一次,后续不再触发的情况?这种情况可能由多种原因导致,如未启用调度、线程... 目录1. 问题背景2. Spring定时任务的基本用法3. 为什么定时任务只执行一次?3.1 未启用

Spring AI ectorStore的使用流程

《SpringAIectorStore的使用流程》SpringAI中的VectorStore是一种用于存储和检索高维向量数据的数据库或存储解决方案,它在AI应用中发挥着至关重要的作用,本文给大家介... 目录一、VectorStore的基本概念二、VectorStore的核心接口三、VectorStore的

python之流程控制语句match-case详解

《python之流程控制语句match-case详解》:本文主要介绍python之流程控制语句match-case使用,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐... 目录match-case 语法详解与实战一、基础值匹配(类似 switch-case)二、数据结构解构匹

在VSCode中本地运行DeepSeek的流程步骤

《在VSCode中本地运行DeepSeek的流程步骤》本文详细介绍了如何在本地VSCode中安装和配置Ollama和CodeGPT,以使用DeepSeek进行AI编码辅助,无需依赖云服务,需要的朋友可... 目录步骤 1:在 VSCode 中安装 Ollama 和 CodeGPT安装Ollama下载Olla

linux环境openssl、openssh升级流程

《linux环境openssl、openssh升级流程》该文章详细介绍了在Ubuntu22.04系统上升级OpenSSL和OpenSSH的方法,首先,升级OpenSSL的步骤包括下载最新版本、安装编译... 目录一.升级openssl1.官网下载最新版openssl2.安装编译环境3.下载后解压安装4.备份

C#集成DeepSeek模型实现AI私有化的流程步骤(本地部署与API调用教程)

《C#集成DeepSeek模型实现AI私有化的流程步骤(本地部署与API调用教程)》本文主要介绍了C#集成DeepSeek模型实现AI私有化的方法,包括搭建基础环境,如安装Ollama和下载DeepS... 目录前言搭建基础环境1、安装 Ollama2、下载 DeepSeek R1 模型客户端 ChatBo

grom设置全局日志实现执行并打印sql语句

《grom设置全局日志实现执行并打印sql语句》本文主要介绍了grom设置全局日志实现执行并打印sql语句,包括设置日志级别、实现自定义Logger接口以及如何使用GORM的默认logger,通过这些... 目录gorm中的自定义日志gorm中日志的其他操作日志级别Debug自定义 Loggergorm中的

JavaScript中的reduce方法执行过程、使用场景及进阶用法

《JavaScript中的reduce方法执行过程、使用场景及进阶用法》:本文主要介绍JavaScript中的reduce方法执行过程、使用场景及进阶用法的相关资料,reduce是JavaScri... 目录1. 什么是reduce2. reduce语法2.1 语法2.2 参数说明3. reduce执行过程

Linux流媒体服务器部署流程

《Linux流媒体服务器部署流程》文章详细介绍了流媒体服务器的部署步骤,包括更新系统、安装依赖组件、编译安装Nginx和RTMP模块、配置Nginx和FFmpeg,以及测试流媒体服务器的搭建... 目录流媒体服务器部署部署安装1.更新系统2.安装依赖组件3.解压4.编译安装(添加RTMP和openssl模块

pip install jupyterlab失败的原因问题及探索

《pipinstalljupyterlab失败的原因问题及探索》在学习Yolo模型时,尝试安装JupyterLab但遇到错误,错误提示缺少Rust和Cargo编译环境,因为pywinpty包需要它... 目录背景问题解决方案总结背景最近在学习Yolo模型,然后其中要下载jupyter(有点LSVmu像一个