package helpers import ( "errors" "fmt" "io" "net/http" "os" "os/exec" "path/filepath" "runtime" "strings" "time" ) const ( NodeTypeMaster = "master" NodeTypeSlave = "slave" DefaultCNPort = 3100 DefaultOVPort = 3101 DefaultSessionSecret = "user-migration-e2e-session-secret" DefaultCNDBFileName = "cn.db" DefaultOVDBFileName = "ov.db" DefaultCNLogFileName = "cn.log" DefaultOVLogFileName = "ov.log" DefaultCNSessionName = "cn_session" DefaultOVSessionName = "ov_session" DefaultSyncAPIKey = "e2e-sync-key" DefaultRootUsername = "root" DefaultRootPassword = "Password123!" defaultNodeReadyTimout = 45 * time.Second ) type NodeProcess struct { Name string Port int BaseURL string DBPath string SessionName string NodeType string LogPath string cmd *exec.Cmd logFile *os.File waitCh chan error } type Cluster struct { RepoRoot string RunDir string BinaryPath string CNDBPath string OVDBPath string CN *NodeProcess OV *NodeProcess } type Environment struct { RepoRoot string ArtifactRoot string RunDir string BinaryPath string CNDBPath string OVDBPath string Cluster *Cluster CNClient *APIClient OVClient *APIClient } func BuildBinary(repoRoot, artifactRoot string) (string, error) { binDir := filepath.Join(artifactRoot, "bin") if err := os.MkdirAll(binDir, 0o755); err != nil { return "", fmt.Errorf("create binary dir: %w", err) } binaryName := "new-api-e2e" if runtime.GOOS == "windows" { binaryName += ".exe" } binaryPath := filepath.Join(binDir, binaryName) cmd := exec.Command("go", "build", "-o", binaryPath, ".") cmd.Dir = repoRoot output, err := cmd.CombinedOutput() if err != nil { return "", fmt.Errorf("build e2e binary: %w\n%s", err, string(output)) } return binaryPath, nil } func PrepareEnvironment(repoRoot, artifactRoot, binaryPath, scenarioName string) (*Environment, error) { runDir := filepath.Join(artifactRoot, "runs", sanitizeName(scenarioName)) if err := os.RemoveAll(runDir); err != nil { return nil, fmt.Errorf("reset run dir: %w", err) } if err := os.MkdirAll(runDir, 0o755); err != nil { return nil, fmt.Errorf("create run dir: %w", err) } cluster := NewCluster(repoRoot, runDir, binaryPath) if err := cluster.Start(NodeTypeMaster, NodeTypeMaster); err != nil { _ = cluster.Stop() return nil, err } cnClient, err := NewAPIClient(cluster.CN.BaseURL) if err != nil { _ = cluster.Stop() return nil, err } ovClient, err := NewAPIClient(cluster.OV.BaseURL) if err != nil { _ = cluster.Stop() return nil, err } if err := SetupRoot(cnClient, DefaultRootUsername, DefaultRootPassword); err != nil { _ = cluster.Stop() return nil, fmt.Errorf("setup CN root: %w", err) } if err := SetupRoot(ovClient, DefaultRootUsername, DefaultRootPassword); err != nil { _ = cluster.Stop() return nil, fmt.Errorf("setup OV root: %w", err) } if err := cluster.Stop(); err != nil { return nil, fmt.Errorf("stop bootstrap cluster: %w", err) } if err := SeedCNData(cluster.CNDBPath); err != nil { return nil, fmt.Errorf("seed CN data: %w", err) } if err := SeedOVData(cluster.OVDBPath); err != nil { return nil, fmt.Errorf("seed OV data: %w", err) } if err := cluster.Start(NodeTypeMaster, NodeTypeSlave); err != nil { _ = cluster.Stop() return nil, err } cnClient, err = NewAPIClient(cluster.CN.BaseURL) if err != nil { _ = cluster.Stop() return nil, err } ovClient, err = NewAPIClient(cluster.OV.BaseURL) if err != nil { _ = cluster.Stop() return nil, err } if err := Login(cnClient, DefaultRootUsername, DefaultRootPassword); err != nil { _ = cluster.Stop() return nil, fmt.Errorf("login CN root: %w", err) } if err := Login(ovClient, DefaultRootUsername, DefaultRootPassword); err != nil { _ = cluster.Stop() return nil, fmt.Errorf("login OV root: %w", err) } if err := ConfigureCNRegionSync(cnClient, cluster.CN.BaseURL, cluster.OV.BaseURL, DefaultSyncAPIKey); err != nil { _ = cluster.Stop() return nil, fmt.Errorf("configure CN region sync: %w", err) } if err := ConfigureOVRegionSync(ovClient, cluster.CN.BaseURL, DefaultSyncAPIKey); err != nil { _ = cluster.Stop() return nil, fmt.Errorf("configure OV region sync: %w", err) } if err := WaitOptionApplied(cluster.OV.BaseURL, DefaultSyncAPIKey, 5, 20*time.Second); err != nil { _ = cluster.Stop() return nil, fmt.Errorf("wait OV migration API ready: %w", err) } return &Environment{ RepoRoot: repoRoot, ArtifactRoot: artifactRoot, RunDir: runDir, BinaryPath: binaryPath, CNDBPath: cluster.CNDBPath, OVDBPath: cluster.OVDBPath, Cluster: cluster, CNClient: cnClient, OVClient: ovClient, }, nil } func (e *Environment) Close() error { if e == nil || e.Cluster == nil { return nil } return e.Cluster.Stop() } func NewCluster(repoRoot, runDir, binaryPath string) *Cluster { return &Cluster{ RepoRoot: repoRoot, RunDir: runDir, BinaryPath: binaryPath, CNDBPath: filepath.Join(runDir, DefaultCNDBFileName), OVDBPath: filepath.Join(runDir, DefaultOVDBFileName), } } func (c *Cluster) Start(cnNodeType, ovNodeType string) error { if err := os.MkdirAll(c.RunDir, 0o755); err != nil { return fmt.Errorf("create run dir: %w", err) } cn, err := c.startNode("cn", DefaultCNPort, c.CNDBPath, DefaultCNSessionName, cnNodeType, DefaultCNLogFileName) if err != nil { return err } c.CN = cn ov, err := c.startNode("ov", DefaultOVPort, c.OVDBPath, DefaultOVSessionName, ovNodeType, DefaultOVLogFileName) if err != nil { _ = c.CN.Stop() c.CN = nil return err } c.OV = ov if err := c.CN.WaitReady(defaultNodeReadyTimout); err != nil { _ = c.Stop() return err } if err := c.OV.WaitReady(defaultNodeReadyTimout); err != nil { _ = c.Stop() return err } return nil } func (c *Cluster) Stop() error { var errs []string if c.OV != nil { if err := c.OV.Stop(); err != nil { errs = append(errs, err.Error()) } c.OV = nil } if c.CN != nil { if err := c.CN.Stop(); err != nil { errs = append(errs, err.Error()) } c.CN = nil } if len(errs) == 0 { return nil } return errors.New(strings.Join(errs, "; ")) } func (c *Cluster) startNode(name string, port int, dbPath, sessionName, nodeType, logFileName string) (*NodeProcess, error) { logPath := filepath.Join(c.RunDir, logFileName) logFile, err := os.OpenFile(logPath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644) if err != nil { return nil, fmt.Errorf("open %s log file: %w", name, err) } systemLogDir := filepath.Join(c.RunDir, "system-logs", name) if err := os.MkdirAll(systemLogDir, 0o755); err != nil { _ = logFile.Close() return nil, fmt.Errorf("create %s system log dir: %w", name, err) } cmd := exec.Command(c.BinaryPath, "--log-dir", systemLogDir) cmd.Dir = c.RepoRoot cmd.Env = mergeEnv(os.Environ(), map[string]string{ "PORT": fmt.Sprintf("%d", port), "SQLITE_PATH": dbPath, "SESSION_SECRET": DefaultSessionSecret, "SESSION_NAME": sessionName, "SESSION_SECURE": "false", "SESSION_SAMESITE": "strict", "GLOBAL_API_RATE_LIMIT_ENABLE": "false", "GLOBAL_WEB_RATE_LIMIT_ENABLE": "false", "CRITICAL_RATE_LIMIT_ENABLE": "false", "GIN_MODE": "release", "NODE_TYPE": nodeType, }) cmd.Stdout = logFile cmd.Stderr = logFile waitCh := make(chan error, 1) if err := cmd.Start(); err != nil { _ = logFile.Close() return nil, fmt.Errorf("start %s node: %w", name, err) } go func() { waitCh <- cmd.Wait() }() return &NodeProcess{ Name: name, Port: port, BaseURL: fmt.Sprintf("http://127.0.0.1:%d", port), DBPath: dbPath, SessionName: sessionName, NodeType: nodeType, LogPath: logPath, cmd: cmd, logFile: logFile, waitCh: waitCh, }, nil } func (n *NodeProcess) WaitReady(timeout time.Duration) error { client := &http.Client{Timeout: 1 * time.Second} deadline := time.Now().Add(timeout) for time.Now().Before(deadline) { select { case err := <-n.waitCh: if err == nil { return fmt.Errorf("%s node exited before ready; see %s", n.Name, n.LogPath) } return fmt.Errorf("%s node exited before ready: %w; see %s", n.Name, err, n.LogPath) default: } resp, err := client.Get(n.BaseURL + "/api/status") if err == nil { _, _ = io.Copy(io.Discard, resp.Body) resp.Body.Close() if resp.StatusCode == http.StatusOK { return nil } } time.Sleep(250 * time.Millisecond) } return fmt.Errorf("%s node did not become ready within %s; see %s", n.Name, timeout, n.LogPath) } func (n *NodeProcess) Stop() error { if n == nil { return nil } var errs []string killRequested := false if n.cmd != nil && n.cmd.Process != nil { if err := n.cmd.Process.Kill(); err != nil && !errors.Is(err, os.ErrProcessDone) { errs = append(errs, fmt.Sprintf("kill %s node: %v", n.Name, err)) } else { killRequested = true } select { case err := <-n.waitCh: if err != nil && !killRequested && !strings.Contains(strings.ToLower(err.Error()), "signal: killed") { errs = append(errs, fmt.Sprintf("wait %s node: %v", n.Name, err)) } case <-time.After(5 * time.Second): errs = append(errs, fmt.Sprintf("wait %s node timeout", n.Name)) } } if n.logFile != nil { if err := n.logFile.Close(); err != nil { errs = append(errs, fmt.Sprintf("close %s log: %v", n.Name, err)) } n.logFile = nil } n.cmd = nil if len(errs) == 0 { return nil } return errors.New(strings.Join(errs, "; ")) } func sanitizeName(name string) string { replacer := strings.NewReplacer("/", "_", "\\", "_", ":", "_", " ", "_") return replacer.Replace(name) } func mergeEnv(base []string, overrides map[string]string) []string { filtered := make([]string, 0, len(base)+len(overrides)) for _, entry := range base { parts := strings.SplitN(entry, "=", 2) if len(parts) != 2 { continue } if _, ok := overrides[parts[0]]; ok { continue } filtered = append(filtered, entry) } for key, value := range overrides { filtered = append(filtered, key+"="+value) } return filtered }