aim-proxy/call.go

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
}