Hive 源码分析

func main() {
    ...
    // 从文件目录查询可用的 clients and simulators.
    inv, err := libhive.LoadInventory(".")
    ...
    // 使用命令行中指定的 --sim 参数正则匹配生成要执行的 simulator
    simList, err := inv.MatchSimulators(*simPattern)
    ...
    // Create the docker backends.
    dockerConfig := &libdocker.Config{
        Inventory:   inv,
        PullEnabled: *dockerPull,
    }
    ...
    // builder 用于 hive runner 构建镜像
    // cb(containerBackend) 管理 container 创建、启动、删除，
    // 以及 docker 中创建、删除网络，在网络中添加、删除 container
    builder, cb, err := libdocker.Connect(*dockerEndpoint, dockerConfig)
    ...

    // env 是启动 container(simulator 以及其他测试依赖 container) 相关配置参数
    env := libhive.SimEnv{
        LogDir:             *testResultsRoot,
        SimLogLevel:        *simLogLevel,
        SimTestPattern:     *simTestPattern,
        SimParallelism:     *simParallelism,
        SimDurationLimit:   *simTimeLimit,
        ClientStartTimeout: *clientTimeout,
    }
    // runner 用于执行 simulation 的对象实例
    runner := libhive.NewRunner(inv, builder, cb)

    // 获取命令行 --client 指定的客户端列表
    clientList := splitAndTrim(*clients, ",")

    // 构建执行 simulation 需要相关镜像： hiveproxy、clients、simulator
    if err := runner.Build(ctx, clientList, simList); err != nil {
        fatal(err)
    }
    ...

    var failCount int
    // 执行命令行 --sim 指定的 simulator（可以通过正则表达式匹配多个）
    // 可以看到这里传入了启动 container 相关的 env 参数
    for _, sim := range simList {
        result, err := runner.Run(ctx, sim, env)
        if err != nil {
            ...
        }
    }
}

func (r *Runner) Build(ctx context.Context, clientList, simList []string) error {
    // hive proxy image
    if err := r.container.Build(ctx, r.builder); err != nil {
        return err
    }
    // clients images
    if err := r.buildClients(ctx, clientList); err != nil {
        return err
    }
    // simulator images
    return r.buildSimulators(ctx, simList)
}

// run runs one simulation.
func (r *Runner) run(ctx context.Context, sim string, env SimEnv) (SimResult, error) {
    log15.Info(fmt.Sprintf("running simulation: %s", sim))

    // 命令行定义的 clients
    clientDefs := make(map[string]*ClientDefinition)
    ...
    // Start the simulation API.
    tm := NewTestManager(env, r.container, clientDefs)
    ...
    log15.Debug("starting simulator API server")
    // 启动 hiveproxy container, 将 simulation API 转发到外部的 hive 控制器
    server, err := r.container.ServeAPI(ctx, tm.API())

    // Create the simulator container.
    opts := ContainerOptions{
        Env: map[string]string{
            "HIVE_SIMULATOR":    "http://" + server.Addr().String(), // hiveproxy ip:port
            "HIVE_PARALLELISM":  strconv.Itoa(env.SimParallelism),
            "HIVE_LOGLEVEL":     strconv.Itoa(env.SimLogLevel),
            "HIVE_TEST_PATTERN": env.SimTestPattern,
        },
    }
    containerID, err := r.container.CreateContainer(ctx, r.simImages[sim], opts)
    ...
    // 启动 simulator container，执行定义的相关测试
    sc, err := r.container.StartContainer(ctx, containerID, opts)
    // 等待 simulator finish，搜集结果返回
    // Count the results.
    var result SimResult
    for _, suite := range tm.Results() {
        ...
        for _, test := range suite.TestCases {
            ...
        }
    }
    return result, err
}

// newSimulationAPI creates handlers for the simulation API.
func newSimulationAPI(b ContainerBackend, env SimEnv, tm *TestManager) http.Handler {
    api := &simAPI{backend: b, env: env, tm: tm}

    // API routes.
    router := mux.NewRouter()
    router.HandleFunc("/clients", api.getClientTypes).Methods("GET")
    router.HandleFunc("/testsuite/{suite}/test/{test}/node/{node}/exec", api.execInClient).Methods("POST")
    router.HandleFunc("/testsuite/{suite}/test/{test}/node/{node}", api.getNodeStatus).Methods("GET")
    router.HandleFunc("/testsuite/{suite}/test/{test}/node", api.startClient).Methods("POST")
    router.HandleFunc("/testsuite/{suite}/test/{test}/node/{node}", api.stopClient).Methods("DELETE")
    router.HandleFunc("/testsuite/{suite}/test", api.startTest).Methods("POST")
    // post because the delete http verb does not always support a message body
    router.HandleFunc("/testsuite/{suite}/test/{test}", api.endTest).Methods("POST")
    router.HandleFunc("/testsuite", api.startSuite).Methods("POST")
    router.HandleFunc("/testsuite/{suite}", api.endSuite).Methods("DELETE")
    router.HandleFunc("/testsuite/{suite}/network/{network}", api.networkCreate).Methods("POST")
    router.HandleFunc("/testsuite/{suite}/network/{network}", api.networkRemove).Methods("DELETE")
    router.HandleFunc("/testsuite/{suite}/network/{network}/{node}", api.networkIPGet).Methods("GET")
    router.HandleFunc("/testsuite/{suite}/network/{network}/{node}", api.networkConnect).Methods("POST")
    router.HandleFunc("/testsuite/{suite}/network/{network}/{node}", api.networkDisconnect).Methods("DELETE")
    return router
}

// ServeAPI starts the API server.
func (cb *ContainerBackend) ServeAPI(ctx context.Context, h http.Handler) (libhive.APIServer, error) {
    inR, inW := io.Pipe()
    outR, outW := io.Pipe()

    // hive 通过 Input 与 Output 与 hiveporxy 进行交流。
    opts := libhive.ContainerOptions{Output: outW, Input: inR}
    // 创建 hiveproxy container
    id, err := cb.CreateContainer(ctx, hiveproxyTag, opts)
    ...
    // Launch the proxy server before starting the container.
    var (
        proxy     *hiveproxy.Proxy
        proxyErrC = make(chan error, 1)
    )
    go func() {
        var err error
        // 后台运行，接收来自 hiveproxy 的请求
        proxy, err = hiveproxy.RunBackend(outR, inW, h)
    }()

    // 启动 hiveproxy container
    info, err := cb.StartContainer(ctx, id, opts)
    cb.proxy = proxy

    srv := &proxyContainer{
        cb:              cb,
        containerID:     id,
        containerIP:     net.ParseIP(info.IP),
        containerWait:   info.Wait,
        containerStdin:  inR,
        containerStdout: outW,
        proxy:           proxy,
    }

    // Register proxy in ContainerBackend, so it can be used for CheckLive.
    cb.proxy = proxy
    log15.Info("hiveproxy started", "container", id[:12], "addr", srv.Addr())
    return srv, nil
}

func main() {
    discv4 := hivesim.Suite{
        Name:        "discv4",
        Description: "This suite runs Discovery v4 protocol tests.",
        Tests: []hivesim.AnyTest{
            hivesim.ClientTestSpec{
                ...
            },
        },
    }

    eth := hivesim.Suite{
        Name:        "eth",
        Description: "This suite tests a client's ability to accurately respond to basic eth protocol messages.",
        Tests: []hivesim.AnyTest{
            hivesim.ClientTestSpec{
                ...
            },
        },
    }

    snap := hivesim.Suite{
        ...
    }

    hivesim.MustRun(hivesim.New(), discv4, eth, snap)
}

// RunSuite runs all tests in a suite.
func RunSuite(host *Simulation, suite Suite) error {
    ...
    suiteID, err := host.StartSuite(suite.Name, suite.Description, "")
    if err != nil {
        return err
    }
    defer host.EndSuite(suiteID)

    for _, test := range suite.Tests {
        // 调用了所有 test case 的 runTest 方法
        if err := test.runTest(host, suiteID, &suite); err != nil {
            return err
        }
    }
    return nil
}

func (spec TestSpec) runTest(host *Simulation, suiteID SuiteID, suite *Suite) error {
    test := testSpec{
        suiteID:   suiteID,
        suite:     suite,
        name:      spec.Name,
        desc:      spec.Description,
        alwaysRun: spec.AlwaysRun,
    }
    return runTest(host, test, spec.Run)
}

func (spec ClientTestSpec) runTest(host *Simulation, suiteID SuiteID, suite *Suite) error {
    clients, err := host.ClientTypes()
    if err != nil {
        return err
    }
    for _, clientDef := range clients {
        // 根据 test case 中指定的 role 过滤命令行中指定的 client，然后对其执行所有的 test case.
        // 如果 test case 没有指定 role，对所有的额 client 执行测试。
        // 'role' is an optional filter, so eth1 tests, beacon node tests,
        // validator tests, etc. can all live in harmony.
        if spec.Role != "" && !clientDef.HasRole(spec.Role) {
            continue
        }
        test := testSpec{
            suiteID:   suiteID,
            suite:     suite,
            name:      clientTestName(spec.Name, clientDef.Name),
            desc:      spec.Description,
            alwaysRun: spec.AlwaysRun,
        }
        // runTest 是一个封装函数
        err := runTest(host, test, func(t *T) {
            // 执行测试前启动 client
            client := t.StartClient(clientDef.Name, spec.Parameters, WithStaticFiles(spec.Files))
            // 调用自定义的测试逻辑
            spec.Run(t, client)
        })
        if err != nil {
            return err
        }
    }
    return nil
}

func runTest(host *Simulation, test testSpec, runit func(t *T)) error {
    if !test.alwaysRun && !host.m.match(test.suite.Name, test.name) {
        fmt.Fprintf(os.Stderr, "skipping test %q because it doesn't match test pattern %s\n", test.name, host.m.pattern)
        return nil
    }

    // Register test on simulation server and initialize the T.
    t := &T{
        Sim:     host,
        SuiteID: test.suiteID,
        suite:   test.suite,
    }
    // 调用 hiveproxy.StartTest 上报
    testID, err := host.StartTest(test.suiteID, test.name, test.desc)
    if err != nil {
        return err
    }
    t.TestID = testID
    t.result.Pass = true
    defer func() {
        t.mu.Lock()
        defer t.mu.Unlock()
        // 调用 hiveproxy.EndTest 上报
        host.EndTest(test.suiteID, testID, t.result)
    }()

    // Run the test function.
    done := make(chan struct{})
    go func() {
        defer func() {
            if err := recover(); err != nil {
                buf := make([]byte, 4096)
                i := runtime.Stack(buf, false)
                t.Logf("panic: %v\n\n%s", err, buf[:i])
                t.Fail()
            }
            close(done)
        }()
        // 自己编写的的测试函数
        runit(t)
    }()
    <-done
    return nil
}