// Package components defines utilities to spin up actual // beacon node and validator processes as needed by end to end tests. package components import ( "context" "fmt" "os" "os/exec" "path" "strconv" "strings" "syscall" "github.com/bazelbuild/rules_go/go/tools/bazel" "github.com/pkg/errors" cmdshared "github.com/prysmaticlabs/prysm/cmd" "github.com/prysmaticlabs/prysm/cmd/beacon-chain/flags" "github.com/prysmaticlabs/prysm/config/features" "github.com/prysmaticlabs/prysm/config/params" "github.com/prysmaticlabs/prysm/testing/endtoend/helpers" e2e "github.com/prysmaticlabs/prysm/testing/endtoend/params" e2etypes "github.com/prysmaticlabs/prysm/testing/endtoend/types" ) var _ e2etypes.ComponentRunner = (*BeaconNode)(nil) var _ e2etypes.ComponentRunner = (*BeaconNodeSet)(nil) var _ e2etypes.MultipleComponentRunners = (*BeaconNodeSet)(nil) var _ e2etypes.BeaconNodeSet = (*BeaconNodeSet)(nil) // BeaconNodeSet represents set of beacon nodes. type BeaconNodeSet struct { e2etypes.ComponentRunner config *e2etypes.E2EConfig nodes []e2etypes.ComponentRunner enr string ids []string started chan struct{} } // SetENR assigns ENR to the set of beacon nodes. func (s *BeaconNodeSet) SetENR(enr string) { s.enr = enr } // NewBeaconNodes creates and returns a set of beacon nodes. func NewBeaconNodes(config *e2etypes.E2EConfig) *BeaconNodeSet { return &BeaconNodeSet{ config: config, started: make(chan struct{}, 1), } } // Start starts all the beacon nodes in set. func (s *BeaconNodeSet) Start(ctx context.Context) error { if s.enr == "" { return errors.New("empty ENR") } // Create beacon nodes. nodes := make([]e2etypes.ComponentRunner, e2e.TestParams.BeaconNodeCount) for i := 0; i < e2e.TestParams.BeaconNodeCount; i++ { nodes[i] = NewBeaconNode(s.config, i, s.enr) } s.nodes = nodes // Wait for all nodes to finish their job (blocking). // Once nodes are ready passed in handler function will be called. return helpers.WaitOnNodes(ctx, nodes, func() { if s.config.UseFixedPeerIDs { for i := 0; i < len(nodes); i++ { s.ids = append(s.ids, nodes[i].(*BeaconNode).peerID) } s.config.PeerIDs = s.ids } // All nodes stated, close channel, so that all services waiting on a set, can proceed. close(s.started) }) } // Started checks whether beacon node set is started and all nodes are ready to be queried. func (s *BeaconNodeSet) Started() <-chan struct{} { return s.started } // Pause pauses the component and its underlying process. func (s *BeaconNodeSet) Pause() error { for _, n := range s.nodes { if err := n.Pause(); err != nil { return err } } return nil } // Resume resumes the component and its underlying process. func (s *BeaconNodeSet) Resume() error { for _, n := range s.nodes { if err := n.Resume(); err != nil { return err } } return nil } // Stop stops the component and its underlying process. func (s *BeaconNodeSet) Stop() error { for _, n := range s.nodes { if err := n.Stop(); err != nil { return err } } return nil } // PauseAtIndex pauses the component and its underlying process at the desired index. func (s *BeaconNodeSet) PauseAtIndex(i int) error { if i >= len(s.nodes) { return errors.Errorf("provided index exceeds slice size: %d >= %d", i, len(s.nodes)) } return s.nodes[i].Pause() } // ResumeAtIndex resumes the component and its underlying process at the desired index. func (s *BeaconNodeSet) ResumeAtIndex(i int) error { if i >= len(s.nodes) { return errors.Errorf("provided index exceeds slice size: %d >= %d", i, len(s.nodes)) } return s.nodes[i].Resume() } // StopAtIndex stops the component and its underlying process at the desired index. func (s *BeaconNodeSet) StopAtIndex(i int) error { if i >= len(s.nodes) { return errors.Errorf("provided index exceeds slice size: %d >= %d", i, len(s.nodes)) } return s.nodes[i].Stop() } // BeaconNode represents beacon node. type BeaconNode struct { e2etypes.ComponentRunner config *e2etypes.E2EConfig started chan struct{} index int enr string peerID string cmd *exec.Cmd } // NewBeaconNode creates and returns a beacon node. func NewBeaconNode(config *e2etypes.E2EConfig, index int, enr string) *BeaconNode { return &BeaconNode{ config: config, index: index, enr: enr, started: make(chan struct{}, 1), } } // Start starts a fresh beacon node, connecting to all passed in beacon nodes. func (node *BeaconNode) Start(ctx context.Context) error { binaryPath, found := bazel.FindBinary("cmd/beacon-chain", "beacon-chain") if !found { log.Info(binaryPath) return errors.New("beacon chain binary not found") } config, index, enr := node.config, node.index, node.enr stdOutFile, err := helpers.DeleteAndCreateFile(e2e.TestParams.LogPath, fmt.Sprintf(e2e.BeaconNodeLogFileName, index)) if err != nil { return err } expectedNumOfPeers := e2e.TestParams.BeaconNodeCount + e2e.TestParams.LighthouseBeaconNodeCount - 1 if node.config.TestSync { expectedNumOfPeers += 1 } jwtPath := path.Join(e2e.TestParams.TestPath, "eth1data/"+strconv.Itoa(node.index)+"/") if index == 0 { jwtPath = path.Join(e2e.TestParams.TestPath, "eth1data/miner/") } jwtPath = path.Join(jwtPath, "geth/jwtsecret") args := []string{ fmt.Sprintf("--%s=%s/eth2-beacon-node-%d", cmdshared.DataDirFlag.Name, e2e.TestParams.TestPath, index), fmt.Sprintf("--%s=%s", cmdshared.LogFileName.Name, stdOutFile.Name()), fmt.Sprintf("--%s=%s", flags.DepositContractFlag.Name, e2e.TestParams.ContractAddress.Hex()), fmt.Sprintf("--%s=%d", flags.RPCPort.Name, e2e.TestParams.Ports.PrysmBeaconNodeRPCPort+index), fmt.Sprintf("--%s=http://127.0.0.1:%d", flags.HTTPWeb3ProviderFlag.Name, e2e.TestParams.Ports.Eth1AuthRPCPort+index), fmt.Sprintf("--%s=%s", flags.ExecutionJWTSecretFlag.Name, jwtPath), fmt.Sprintf("--%s=%d", flags.MinSyncPeers.Name, 1), fmt.Sprintf("--%s=%d", cmdshared.P2PUDPPort.Name, e2e.TestParams.Ports.PrysmBeaconNodeUDPPort+index), fmt.Sprintf("--%s=%d", cmdshared.P2PTCPPort.Name, e2e.TestParams.Ports.PrysmBeaconNodeTCPPort+index), fmt.Sprintf("--%s=%d", cmdshared.P2PMaxPeers.Name, expectedNumOfPeers), fmt.Sprintf("--%s=%d", flags.MonitoringPortFlag.Name, e2e.TestParams.Ports.PrysmBeaconNodeMetricsPort+index), fmt.Sprintf("--%s=%d", flags.GRPCGatewayPort.Name, e2e.TestParams.Ports.PrysmBeaconNodeGatewayPort+index), fmt.Sprintf("--%s=%d", flags.ContractDeploymentBlock.Name, 0), fmt.Sprintf("--%s=%d", flags.MinPeersPerSubnet.Name, 0), fmt.Sprintf("--%s=%d", cmdshared.RPCMaxPageSizeFlag.Name, params.BeaconConfig().MinGenesisActiveValidatorCount), fmt.Sprintf("--%s=%s", cmdshared.BootstrapNode.Name, enr), fmt.Sprintf("--%s=%s", cmdshared.VerbosityFlag.Name, "debug"), "--" + cmdshared.ForceClearDB.Name, "--" + cmdshared.E2EConfigFlag.Name, "--" + cmdshared.AcceptTosFlag.Name, "--" + flags.EnableDebugRPCEndpoints.Name, } if config.UsePprof { args = append(args, "--pprof", fmt.Sprintf("--pprofport=%d", e2e.TestParams.Ports.PrysmBeaconNodePprofPort+index)) } // Only add in the feature flags if we either aren't performing a control test // on our features or the beacon index is a multiplier of 2 (idea is to split nodes // equally down the line with one group having feature flags and the other without // feature flags; this is to allow A-B testing on new features) if !config.TestFeature || index%2 == 0 { args = append(args, features.E2EBeaconChainFlags...) } args = append(args, config.BeaconFlags...) cmd := exec.CommandContext(ctx, binaryPath, args...) // #nosec G204 -- Safe // Write stdout and stderr to log files. stdout, err := os.Create(path.Join(e2e.TestParams.LogPath, fmt.Sprintf("beacon_node_%d_stdout.log", index))) if err != nil { return err } stderr, err := os.Create(path.Join(e2e.TestParams.LogPath, fmt.Sprintf("beacon_node_%d_stderr.log", index))) if err != nil { return err } defer func() { if err := stdout.Close(); err != nil { log.WithError(err).Error("Failed to close stdout file") } if err := stderr.Close(); err != nil { log.WithError(err).Error("Failed to close stderr file") } }() cmd.Stdout = stdout cmd.Stderr = stderr log.Infof("Starting beacon chain %d with flags: %s", index, strings.Join(args[2:], " ")) if err = cmd.Start(); err != nil { return fmt.Errorf("failed to start beacon node: %w", err) } if err = helpers.WaitForTextInFile(stdOutFile, "gRPC server listening on port"); err != nil { return fmt.Errorf("could not find multiaddr for node %d, this means the node had issues starting: %w", index, err) } if config.UseFixedPeerIDs { peerId, err := helpers.FindFollowingTextInFile(stdOutFile, "Running node with peer id of ") if err != nil { return fmt.Errorf("could not find peer id: %w", err) } node.peerID = peerId } // Mark node as ready. close(node.started) node.cmd = cmd return cmd.Wait() } // Started checks whether beacon node is started and ready to be queried. func (node *BeaconNode) Started() <-chan struct{} { return node.started } // Pause pauses the component and its underlying process. func (node *BeaconNode) Pause() error { return node.cmd.Process.Signal(syscall.SIGSTOP) } // Resume resumes the component and its underlying process. func (node *BeaconNode) Resume() error { return node.cmd.Process.Signal(syscall.SIGCONT) } // Stop stops the component and its underlying process. func (node *BeaconNode) Stop() error { return node.cmd.Process.Kill() }