forked from ci4s/aim-proxy
增加get-batch调用
This commit is contained in:
parent
12951636fb
commit
16c1fb7b7b
|
|
@ -0,0 +1,103 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"log"
|
||||
"net/http"
|
||||
"sync"
|
||||
"time"
|
||||
"io/ioutil"
|
||||
"fmt"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
"github.com/go-chi/chi/v5/middleware"
|
||||
"github.com/go-chi/render"
|
||||
)
|
||||
|
||||
type Run struct {
|
||||
Hash string `json:"hash"`
|
||||
}
|
||||
|
||||
type RunInfo struct {
|
||||
RunHash string `json:"run_hash"`
|
||||
Data map[string]interface{} `json:"data"`
|
||||
// other fields from the info API response
|
||||
}
|
||||
|
||||
type IndexedRun struct {
|
||||
Index int `json:"index"`
|
||||
Run
|
||||
}
|
||||
|
||||
// var host = "http://my-aim-ui-service.aim:43800"
|
||||
var host = "http://172.20.32.181:30039"
|
||||
func main() {
|
||||
r := chi.NewRouter()
|
||||
r.Use(middleware.Logger)
|
||||
r.Post("/api/fetch_run_info", fetchRunInfo)
|
||||
|
||||
log.Fatal(http.ListenAndServe(":7124", r))
|
||||
}
|
||||
|
||||
func fetchRunInfo(w http.ResponseWriter, r *http.Request) {
|
||||
var indexedRuns []IndexedRun
|
||||
body, err := ioutil.ReadAll(r.Body)
|
||||
if err != nil {
|
||||
fmt.Println("read body error:", err)
|
||||
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
r.Body.Close()
|
||||
fmt.Println("request body:", string(body))
|
||||
if err := json.Unmarshal(body, &indexedRuns); err != nil {
|
||||
fmt.Println("unmarshal error:", err)
|
||||
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
results := fetchRunDetails(indexedRuns)
|
||||
|
||||
render.JSON(w, r, results)
|
||||
}
|
||||
|
||||
func fetchRunDetails(indexedRuns []IndexedRun) []map[string]interface{} {
|
||||
results := make([]map[string]interface{}, len(indexedRuns))
|
||||
var wg sync.WaitGroup
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
|
||||
defer cancel()
|
||||
concurrencyLimit := 20
|
||||
sem := make(chan struct{}, concurrencyLimit)
|
||||
|
||||
for _, indexedRun := range indexedRuns {
|
||||
wg.Add(1)
|
||||
sem <- struct{}{} // 向通道发送一个值以占用一个槽
|
||||
go func(indexedRun IndexedRun) {
|
||||
defer wg.Done()
|
||||
defer func() { <-sem }() // 从通道中读取一个值以释放一个槽
|
||||
url := host + "/api/runs/" + indexedRun.Hash + "/info"
|
||||
req, _ := http.NewRequestWithContext(ctx, "GET", url, nil)
|
||||
fmt.Println("fetching details url for run: ", url)
|
||||
client := &http.Client{}
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
log.Printf("Failed to fetch details for run %s: %v", indexedRun.Hash, err)
|
||||
return
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
|
||||
var data map[string]interface{}
|
||||
if err := json.NewDecoder(resp.Body).Decode(&data); err != nil {
|
||||
log.Printf("Failed to decode response for run %s: %v", indexedRun.Hash, err)
|
||||
return
|
||||
}
|
||||
data["run_hash"] = indexedRun.Hash
|
||||
fmt.Println("run details:", data)
|
||||
results[indexedRun.Index] = data
|
||||
}(indexedRun)
|
||||
}
|
||||
fmt.Println("waiting for all requests to complete...")
|
||||
wg.Wait()
|
||||
fmt.Println("all requests completed, returning results:", results)
|
||||
return results
|
||||
}
|
||||
|
|
@ -25,26 +25,34 @@ def print_run_details(run):
|
|||
df = run.dataframe(include_props=False)
|
||||
print("all params: ", df.to_dict(orient='records'))
|
||||
|
||||
props = run.dataframe(include_props=True,include_params=False)
|
||||
print("props: ", props.to_dict())
|
||||
# 假设 Run 对象有以下属性
|
||||
print(f"Run ID: {run.hash}")
|
||||
print(f"Experiment: {run.experiment}")
|
||||
print(f"Description: {run.description}")
|
||||
print(f"Created at: {run.creation_time}")
|
||||
print(f"Tags: {run.tags}")
|
||||
# props = run.dataframe(include_props=True,include_params=False)
|
||||
# print("props: ", props.to_dict())
|
||||
# # 假设 Run 对象有以下属性
|
||||
# print(f"Run ID: {run.hash}")
|
||||
# print(f"Experiment: {run.experiment}")
|
||||
# print(f"Description: {run.description}")
|
||||
# print(f"Created at: {run.creation_time}")
|
||||
# print(f"Tags: {run.tags}")
|
||||
|
||||
# params = repo.collect_params_info()
|
||||
# print("collectparams: ", params)
|
||||
|
||||
seq_type = repo.available_sequence_types()
|
||||
print("available sequence types: ", seq_type)
|
||||
seq =run.collect_sequence_info("metric", skip_last_value=False)
|
||||
print("collect_sequence_info: ", seq)
|
||||
|
||||
for metric_name, metric_context, metric_run in run.iter_metrics_info():
|
||||
print(f"Metric: {metric_name}, Context: {metric_context}, Run: {metric_run}")
|
||||
# 通过metric_name, metric_context, metric_run获取metric的具体值
|
||||
metric_value = run.get_metric(metric_name, metric_context)
|
||||
print(f"{metric_name}: {metric_value}")
|
||||
|
||||
for metric in run.metrics():
|
||||
result = metric.values.values_list()
|
||||
last = metric.values.last()
|
||||
print(f"{metric.name} result is :{result} ")
|
||||
print(f"{metric.name}-{metric.context}: {last}")
|
||||
#
|
||||
# for metric in run.metrics():
|
||||
# result = metric.values.values_list()
|
||||
# last = metric.values.last()
|
||||
# print(f"{metric.name} result is :{result} ")
|
||||
# print(f"{metric.name}-{metric.context}: {last}")
|
||||
|
||||
#tt = run.collect_sequence_info(sequence_types="{'context': {'subset': 'train'}, 'name': 'loss')}")
|
||||
#print("collect info is ", tt)
|
||||
|
|
@ -84,7 +92,7 @@ def print_run_details(run):
|
|||
# # 查询指定的 runs
|
||||
# hashes_to_find = ["581c75204d134704b3c9ac27", "3dfdf9b3a4b0453790c8c7d0", "34fd862f49f64d9c8c12aa4a"]
|
||||
|
||||
query_res = repo.query_runs(query='run.experiment=="experiment-88888"', paginated=True).iter_runs()
|
||||
query_res = repo.query_runs(query='run.hash=="07cd1e6130c94e76bd48c528"', paginated=True).iter_runs()
|
||||
|
||||
|
||||
#runs = repo.query_runs(query='', report_mode=QueryReportMode.PROGRESS_TUPLE)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,37 @@
|
|||
#!/bin/bash
|
||||
|
||||
# 要调用的URL
|
||||
url="http://172.20.32.181:30059/api/runs/07cd1e6130c94e76bd48c528/info"
|
||||
|
||||
# 定义一个函数，用于发出HTTP请求
|
||||
fetch_url() {
|
||||
curl -s -o /dev/null -w "%{http_code}" "$url"
|
||||
}
|
||||
|
||||
# 总共调用次数
|
||||
total_requests=60
|
||||
|
||||
# 每次并发数
|
||||
concurrent_requests=10
|
||||
|
||||
# 计算需要的批次数
|
||||
batches=$((total_requests / concurrent_requests))
|
||||
|
||||
#记录开始时间
|
||||
start_time=$(date +%s)
|
||||
# 执行请求
|
||||
for ((i=0; i<batches; i++)); do
|
||||
for ((j=0; j<concurrent_requests; j++)); do
|
||||
fetch_url &
|
||||
done
|
||||
wait
|
||||
done
|
||||
|
||||
# 记录结束时间
|
||||
end_time=$(date +%s)
|
||||
|
||||
# 计算总共耗时
|
||||
total_time=$((end_time - start_time))
|
||||
|
||||
# 输出结果
|
||||
echo "Total time: $total_time seconds"
|
||||
|
|
@ -0,0 +1,57 @@
|
|||
from fastapi import FastAPI, HTTPException
|
||||
from aim.sdk.repo import Repo
|
||||
import httpx
|
||||
from datetime import datetime
|
||||
import uvicorn
|
||||
import traceback
|
||||
|
||||
repo_path = 'aim://172.20.32.181:30038'
|
||||
#repo_path = 'aim://my-aim-server-service.aim:53800'
|
||||
# host = 'http://my-aim-ui-service.aim:43800'
|
||||
go_service_host = 'http://172.20.32.186:7124' # Go服务的地址
|
||||
|
||||
# 打开 Aim repository
|
||||
repo = Repo(path=repo_path)
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
@app.get("/api/runs/search/run")
|
||||
async def search_runs(query: str = None, paginated: bool = True, offset: str = None, limit: int = None):
|
||||
try:
|
||||
print("query is ", query)
|
||||
print("paginated is ", paginated)
|
||||
print("offset is ", offset)
|
||||
print("limit is ", limit)
|
||||
query_start_time = datetime.now().timestamp()
|
||||
query_res = repo.query_runs(query=query, paginated=paginated, offset=offset).iter_runs()
|
||||
runs = [item.run for item in query_res]
|
||||
query_end_time = datetime.now().timestamp()
|
||||
query_duration = query_end_time - query_start_time
|
||||
print("query time is ", query_duration)
|
||||
|
||||
# 根据run.creation_time 降序排序
|
||||
runs = sorted(runs, key=lambda x: x.creation_time, reverse=True)
|
||||
if paginated and limit:
|
||||
runs = runs[:limit]
|
||||
|
||||
indexed_runs = [{"index": i, "hash": run.hash, "creation_time": run.creation_time} for i, run in enumerate(runs)]
|
||||
print("indexed_runs:", indexed_runs)
|
||||
# 将indexed_runs传递给Go服务以获取详细信息
|
||||
http_start_time = datetime.now().timestamp()
|
||||
async with httpx.AsyncClient(timeout=20) as client:
|
||||
response = await client.post(f"{go_service_host}/api/fetch_run_info", json=indexed_runs)
|
||||
response.raise_for_status()
|
||||
results = response.json()
|
||||
print("result len is ", len(results))
|
||||
http_end_time = datetime.now().timestamp()
|
||||
http_duration = http_end_time - http_start_time
|
||||
print("http time is ", http_duration)
|
||||
|
||||
return results
|
||||
|
||||
except Exception as e:
|
||||
traceback.print_exc()
|
||||
raise HTTPException(status_code=500, detail=str(e))
|
||||
|
||||
if __name__ == "__main__":
|
||||
uvicorn.run(app, host="0.0.0.0", port=7123)
|
||||
|
|
@ -0,0 +1,91 @@
|
|||
from fastapi import FastAPI, HTTPException
|
||||
from aim.sdk.repo import Repo
|
||||
import httpx
|
||||
import traceback
|
||||
import uvicorn
|
||||
from datetime import datetime
|
||||
import asyncio
|
||||
|
||||
# AIM repository路径和目标服务的主机地址
|
||||
repo_path = 'aim://172.20.32.181:30058'
|
||||
host = 'http://172.20.32.181:30059'
|
||||
|
||||
# 打开 AIM repository
|
||||
repo = Repo(path=repo_path)
|
||||
|
||||
# 创建 FastAPI 应用实例
|
||||
app = FastAPI()
|
||||
|
||||
# 定义异步函数，用于获取每个 run 的详细信息
|
||||
async def fetch_run_info(run, results):
|
||||
try:
|
||||
# 构建获取信息的 URL
|
||||
url = f"{host}/api/runs/{run.hash}/info"
|
||||
# 使用 httpx.AsyncClient 发起异步 GET 请求
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.get(url)
|
||||
response.raise_for_status()
|
||||
info = response.json()
|
||||
info['run_hash'] = run.hash # 将 run_hash 添加到返回的信息中
|
||||
results.append(info) # 将信息添加到结果列表中
|
||||
except Exception as e:
|
||||
traceback.print_exc() # 打印异常堆栈信息
|
||||
raise HTTPException(status_code=500, detail=str(e))
|
||||
|
||||
# 定义路由 /api/runs/search/run，接收查询参数并发起调用
|
||||
@app.get("/api/runs/search/run")
|
||||
async def search_runs(query: str = None, paginated: bool = True, offset: str = None, limit: int = None):
|
||||
try:
|
||||
# 记录查询开始时间
|
||||
query_start_time = datetime.now().timestamp()
|
||||
# 使用 repo.query_runs 查询符合条件的 runs，并获取迭代器
|
||||
query_res = repo.query_runs(query=query, paginated=paginated, offset=offset).iter_runs()
|
||||
# 将查询结果中的 run 对象提取到列表中
|
||||
runs = [item.run for item in query_res]
|
||||
# 记录查询结束时间和持续时间
|
||||
query_end_time = datetime.now().timestamp()
|
||||
query_duration = query_end_time - query_start_time
|
||||
print("query time is ", query_duration)
|
||||
|
||||
# 根据 run.creation_time 降序排序 runs 列表
|
||||
runs = sorted(runs, key=lambda x: x.creation_time, reverse=True)
|
||||
|
||||
results = [] # 初始化结果列表
|
||||
|
||||
# 记录 HTTP 请求开始时间
|
||||
http_start_time = datetime.now().timestamp()
|
||||
|
||||
tasks = [] # 初始化任务列表
|
||||
for run in runs:
|
||||
# 创建异步任务，并添加到任务列表
|
||||
task = fetch_run_info(run, results)
|
||||
tasks.append(task)
|
||||
# 当任务数达到 10 时，使用 asyncio.gather 并发执行这些任务
|
||||
if len(tasks) >= 8:
|
||||
await asyncio.gather(*tasks)
|
||||
tasks = [] # 清空任务列表，以便下一批任务
|
||||
|
||||
# 等待剩余的任务完成
|
||||
if tasks:
|
||||
await asyncio.gather(*tasks)
|
||||
|
||||
# 记录 HTTP 请求结束时间和持续时间
|
||||
http_end_time = datetime.now().timestamp()
|
||||
http_duration = http_end_time - http_start_time
|
||||
print("http time is ", http_duration)
|
||||
|
||||
# 如果是分页模式，并且有限制结果数量，则截取结果列表
|
||||
if paginated:
|
||||
results = results[:limit] if limit else results
|
||||
|
||||
# 返回最终结果
|
||||
print("results len is ", len(results))
|
||||
return results
|
||||
|
||||
except Exception as e:
|
||||
traceback.print_exc() # 打印异常堆栈信息
|
||||
raise HTTPException(status_code=500, detail=str(e))
|
||||
|
||||
# 如果是主程序入口，则运行 FastAPI 应用
|
||||
if __name__ == "__main__":
|
||||
uvicorn.run(app, host="0.0.0.0", port=7123)
|
||||
|
|
@ -0,0 +1,89 @@
|
|||
from fastapi import FastAPI, HTTPException
|
||||
from aim.sdk.repo import Repo
|
||||
import httpx
|
||||
import traceback
|
||||
from datetime import datetime
|
||||
import uvicorn
|
||||
|
||||
repo_path = 'aim://my-aim-server-service.aim:53800'
|
||||
#repo_path = 'aim://172.20.32.181:30058'
|
||||
host = 'http://my-aim-ui-service.aim:43800'
|
||||
#host = 'http://172.20.32.181:30059'
|
||||
repo = Repo(path=repo_path)
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
|
||||
async def fetch_run_info(client, host, run_hash):
|
||||
try:
|
||||
url = f"{host}/api/runs/{run_hash}/info"
|
||||
response = await client.get(url)
|
||||
response.raise_for_status()
|
||||
info = response.json()
|
||||
return info
|
||||
except httpx.HTTPStatusError as e:
|
||||
raise HTTPException(status_code=e.response.status_code, detail=str(e))
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=str(e))
|
||||
|
||||
|
||||
async def fetch_run_metrics(client, host, run_hash, data):
|
||||
try:
|
||||
url = f"{host}/api/runs/{run_hash}/metric/get-batch"
|
||||
response = await client.post(url, json=data)
|
||||
response.raise_for_status()
|
||||
metrics = response.json()
|
||||
return metrics
|
||||
except httpx.HTTPStatusError as e:
|
||||
raise HTTPException(status_code=e.response.status_code, detail=str(e))
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=str(e))
|
||||
|
||||
|
||||
def extract_latest_metrics(metrics, name, context):
|
||||
for metric in metrics:
|
||||
if metric["name"] == name and metric["context"] == context:
|
||||
return metric["values"][-1] if metric["values"] else None
|
||||
return None
|
||||
|
||||
|
||||
@app.get("/api/runs/search/run")
|
||||
async def search_runs(query: str = None, paginated: bool = True, offset: str = None, limit: int = None):
|
||||
try:
|
||||
query_start_time = datetime.now().timestamp()
|
||||
query_res = repo.query_runs(query=query, paginated=paginated, offset=offset).iter_runs()
|
||||
runs = [item.run for item in query_res]
|
||||
query_end_time = datetime.now().timestamp()
|
||||
query_duration = query_end_time - query_start_time
|
||||
print("query time is ", query_duration)
|
||||
|
||||
runs = sorted(runs, key=lambda x: x.creation_time, reverse=True)
|
||||
if paginated and limit:
|
||||
runs = runs[:limit]
|
||||
|
||||
results = []
|
||||
async with httpx.AsyncClient() as client:
|
||||
for run in runs:
|
||||
run_info = await fetch_run_info(client, host, run.hash)
|
||||
data = run_info["traces"]["metric"]
|
||||
run_metrics = await fetch_run_metrics(client, host, run.hash, data)
|
||||
for metric in run_metrics:
|
||||
metric["last_value"] = metric["values"][-1] if metric["values"] else None
|
||||
|
||||
run_info["traces"]["metric"] = run_metrics
|
||||
run_info["run_hash"] = run.hash
|
||||
results.append(run_info)
|
||||
|
||||
http_end_time = datetime.now().timestamp()
|
||||
http_duration = http_end_time - query_end_time
|
||||
print("http time is ", http_duration)
|
||||
|
||||
return results
|
||||
|
||||
except Exception as e:
|
||||
traceback.print_exc()
|
||||
raise HTTPException(status_code=500, detail=str(e))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
uvicorn.run(app, host="0.0.0.0", port=7123)
|
||||
Loading…
Reference in New Issue