fix determinism

Author: bsamuels453

pkg/exploration/explore.go

@@ -12,16 +12,18 @@ import (
 	"attacknet/cmd/pkg/types"
 	"context"
 	"fmt"
+	"github.com/kurtosis-tech/stacktrace"
 	log "github.com/sirupsen/logrus"
 	"golang.org/x/exp/rand"
+	"net/http"
 	"os"
 	"os/signal"
 	"syscall"
 	"time"
 )
 
 const WaitBetweenTestsSecs = 60
-const Seed = 555
+const Seed = 557
 
 func getRandomAttackSize() suite.AttackSize {
 	//return suite.AttackOne
@@ -51,7 +53,7 @@ func buildRandomLatencyTest(targetDescription string, targetSelectors []*suite.C
 	minDelayMilliSeconds := 10
 	maxDelayMilliSeconds := 1000
 	minDurationSeconds := 10
-	maxDurationSeconds := 1000
+	maxDurationSeconds := 300
 	minJitterMilliseconds := 10
 	maxJitterMilliseconds := 1000
 	minCorrelation := 0
@@ -75,6 +77,38 @@ func buildRandomLatencyTest(targetDescription string, targetSelectors []*suite.C
 	)
 }
 
+func buildRandomClockSkewTest(targetDescription string, targetSelectors []*suite.ChaosTargetSelector) (*types.SuiteTest, error) {
+	minDelaySeconds := -600
+	maxDelaySeconds := 600
+	minDurationSeconds := 10
+	maxDurationSeconds := 300
+
+	grace := time.Second * 300
+	delay := fmt.Sprintf("%ds", rand.Intn(maxDelaySeconds-minDelaySeconds)+minDelaySeconds)
+	duration := fmt.Sprintf("%ds", rand.Intn(maxDurationSeconds-minDurationSeconds)+minDurationSeconds)
+
+	description := fmt.Sprintf("Apply %s clock skew for %s against %d targets. %s", delay, duration, len(targetSelectors), targetDescription)
+	log.Info(description)
+	return suite.ComposeNodeClockSkewTest(
+		description,
+		targetSelectors,
+		delay,
+		duration,
+		&grace,
+	)
+}
+
+func buildRandomTest(targetDescription string, targetSelectors []*suite.ChaosTargetSelector) (*types.SuiteTest, error) {
+	testId := rand.Intn(2)
+	if testId == 0 {
+		return buildRandomLatencyTest(targetDescription, targetSelectors)
+	}
+	if testId == 1 {
+		return buildRandomClockSkewTest(targetDescription, targetSelectors)
+	}
+	return nil, stacktrace.NewError("unknown test id")
+}
+
 func pickRandomClient(config *plan.PlannerConfig) (string, bool) {
 	//return "reth", true
 	isExec := rand.Intn(2)
@@ -104,6 +138,11 @@ func StartExploration(config *plan.PlannerConfig) error {
 	}
 	testableNodes := nodes[1:]
 
+	for _, n := range nodes {
+		log.Infof("%s", suite.ConvertToNodeIdTag(len(nodes), n, "execution"))
+		log.Infof("%s", suite.ConvertToNodeIdTag(len(nodes), n, "consensus"))
+	}
+
 	// dedupe from runtime?
 	kubeClient, err := kubernetes.CreateKubeClient(config.KubernetesNamespace)
 	if err != nil {
@@ -118,22 +157,29 @@ func StartExploration(config *plan.PlannerConfig) error {
 	}
 
 	var testArtifacts []*artifacts.TestArtifact
-	var done = make(chan bool, 1)
-	sigs := make(chan os.Signal, 1)
+	var done = make(chan bool, 2)
+	sigs := make(chan os.Signal, 2)
 	signal.Notify(sigs, syscall.SIGINT)
 	go func() {
 		sig := <-sigs
 		fmt.Println()
 		fmt.Println(sig, "Signal received. Ending after next test is completed.")
 		done <- true // Signal that we're done
-	}()
 
+	}()
+	killall := false
 	for {
+		loc := time.FixedZone("GMT", 0)
+		log.Infof("Start loop. GMT time: %s", time.Now().In(loc).Format(http.TimeFormat))
 		select {
 		case <-done:
 			fmt.Println("Writing test artifacts")
 			return cleanup(testArtifacts)
 		default:
+			if killall {
+				fmt.Println("Writing test artifacts")
+				return cleanup(testArtifacts)
+			}
 			clientUnderTest, isExec := pickRandomClient(config)
 			targetSpec := getTargetSpec()
 			attackSize := getRandomAttackSize()
@@ -149,14 +195,25 @@ func StartExploration(config *plan.PlannerConfig) error {
 				continue
 			}
 
+			for _, selector := range targetSelectors {
+				for _, s := range selector.Selector {
+					msg := "Hitting "
+					for _, target := range s.Values {
+						msg = fmt.Sprintf("%s %s,", msg, target)
+					}
+					log.Info(msg)
+				}
+			}
+			log.Infof("time: %d", time.Now().Unix())
+
 			var targetingDescription string
 			if targetSpec == suite.TargetMatchingNode {
 				targetingDescription = fmt.Sprintf("Impacting the full node of targeted %s clients. Injecting into %s of the matching targets.", clientUnderTest, attackSize)
 			} else {
 				targetingDescription = fmt.Sprintf("Impacting the client of targeted %s clients. Injecting into %s of the matching targets.", clientUnderTest, attackSize)
 			}
 
-			test, err := buildRandomLatencyTest(
+			test, err := buildRandomTest(
 				targetingDescription,
 				targetSelectors,
 			)
@@ -169,12 +226,13 @@ func StartExploration(config *plan.PlannerConfig) error {
 			err = executor.RunTestPlan(ctx)
 			if err != nil {
 				log.Errorf("Error while running test")
-				return err
+				fmt.Println("Writing test artifacts")
+				return cleanup(testArtifacts)
 			} else {
 				log.Infof("Test steps completed.")
 			}
 
-			log.Info("Starting health checks")
+			log.Infof("Starting health checks at %s", time.Now().In(loc).Format(http.TimeFormat))
 			podsUnderTest, err := executor.GetPodsUnderTest()
 			if err != nil {
 				return err
@@ -186,6 +244,9 @@ func StartExploration(config *plan.PlannerConfig) error {
 			}
 			results, err := hc.RunChecks(ctx)
 			if err != nil {
+
+				fmt.Println("Writing test artifacts")
+				err := cleanup(testArtifacts)
 				return err
 			}
 			testArtifact := artifacts.BuildTestArtifact(results, podsUnderTest, *test)

pkg/health/checker.go

@@ -53,7 +53,7 @@ func (hc *CheckOrchestrator) RunChecks(ctx context.Context) (*types.HealthCheckR
 		}
 
 		if time.Now().After(latestAllowable) {
-			log.Warn("Grace period elapsed and a health check is still failing")
+			log.Warnf("Grace period elapsed and a health check is still failing. Time: %d", time.Now().Unix())
 			return results, nil
 		} else {
 			log.Warn("Health checks failed but still in grace period")

pkg/health/ethereum/beacon_rpc.go

@@ -119,21 +119,32 @@ func (e *EthNetworkChecker) dialToBeaconClients(ctx context.Context) ([]*BeaconC
 }
 
 func dialBeaconRpcClient(ctx context.Context, session *kubernetes.PortForwardsSession) (*BeaconClientRpc, error) {
-	httpClient, err := http.New(ctx,
-		http.WithAddress(fmt.Sprintf("http://localhost:%d", session.LocalPort)),
-		http.WithLogLevel(zerolog.WarnLevel),
-	)
-	if err != nil {
-		return nil, stacktrace.Propagate(err, "err while dialing RPC for %s", session.Pod.GetName())
-	}
-	provider, isProvider := httpClient.(eth2client.BeaconBlockHeadersProvider)
-	if !isProvider {
-		return nil, stacktrace.NewError("unable to cast http client to beacon rpc provider for %s", session.Pod.GetName())
+	// 3 attempts
+	retryCount := 8
+	for i := 0; i <= retryCount; i++ {
+		httpClient, err := http.New(ctx,
+			http.WithAddress(fmt.Sprintf("http://localhost:%d", session.LocalPort)),
+			http.WithLogLevel(zerolog.WarnLevel),
+		)
+		if err != nil {
+			if i == retryCount {
+				return nil, stacktrace.Propagate(err, "err while dialing RPC for %s", session.Pod.GetName())
+			} else {
+				time.Sleep(1 * time.Second)
+				continue
+			}
+
+		}
+		provider, isProvider := httpClient.(eth2client.BeaconBlockHeadersProvider)
+		if !isProvider {
+			return nil, stacktrace.NewError("unable to cast http client to beacon rpc provider for %s", session.Pod.GetName())
+		}
+		return &BeaconClientRpc{
+			session: session,
+			client:  provider,
+		}, nil
 	}
-	return &BeaconClientRpc{
-		session: session,
-		client:  provider,
-	}, nil
+	return nil, stacktrace.NewError("unreachable beacon rpc")
 }
 
 func (c *BeaconClientRpc) Close() {
@@ -158,7 +169,14 @@ func (c *BeaconClientRpc) GetLatestBlockBy(ctx context.Context, blockType string
 				}
 			}
 		}
-		return nil, stacktrace.Propagate(err, "Unable to query for blockType %s with client for %s", blockType, c.session.Pod.GetName())
+		// chock it up to a failure we need to retry
+		choice := &ClientForkChoice{
+			Pod:         c.session.Pod,
+			BlockNumber: 0,
+			BlockHash:   "N/A",
+		}
+		return choice, nil
+		//return nil, stacktrace.Propagate(err, "Unable to query for blockType %s with client for %s", blockType, c.session.Pod.GetName())
 	}
 
 	slot := uint64(result.Data.Header.Message.Slot)

pkg/health/ethereum/consensus.go

@@ -4,6 +4,7 @@ import (
 	"attacknet/cmd/pkg/kubernetes"
 	"context"
 	log "github.com/sirupsen/logrus"
+	"time"
 )
 
 // var UnableToReachLatestConsensusError = fmt.Errorf("there are nodes that disagree on the latest block")
@@ -101,7 +102,7 @@ func reportConsensusDataToLogger(consensusType string,
 
 	log.Infof("Consensus %s block height: %d", consensusType, consensusBlockNum[0].BlockNumber)
 	if len(wrongBlockNum) > 0 {
-		log.Warnf("Some nodes are out of consensus for block type '%s'", consensusType)
+		log.Warnf("Some nodes are out of consensus for block type '%s'. Time: %d", consensusType, time.Now().Unix())
 		for _, n := range wrongBlockNum {
 			log.Warnf("---> Node: %s %s BlockHeight: %d BlockHash: %s", n.Pod.GetName(), consensusType, n.BlockNumber, n.BlockHash)
 		}

pkg/kubernetes/port_forward.go

@@ -114,25 +114,37 @@ func (c *KubeClient) StartPortForwarding(pod string, localPort, remotePort int,
 	portForwardIssueCh := make(chan error, 1)
 	defer close(portForwardIssueCh)
 
-	retries := 3
-	go func(retriesRem int) {
+	retriesRem := 5
+	/*go func(retriesRem int) {
 		for retriesRem > 0 {
-			if err = portForward.ForwardPorts(); err == nil {
+			if err = portForward.ForwardPorts(); err != nil {
 				retriesRem--
+				time.Sleep(200 * time.Millisecond)
 				if retriesRem == 0 {
 					portForwardIssueCh <- stacktrace.Propagate(err, "unable to start port forward session")
 				}
 			}
 		}
-	}(retries)
+	}(retries)*/
+
+	for retriesRem > 0 {
+		if err = portForward.ForwardPorts(); err != nil {
+			retriesRem--
+			time.Sleep(200 * time.Millisecond)
+		} else {
+			return stopCh, nil
+		}
+	}
+	//portForwardIssueCh <- stacktrace.Propagate(err, "unable to start port forward session")
+	return nil, errors.New("Failed to establish port forward")
 
-	select {
+	/*select {
 	case <-readyCh:
 		log.Debugf("Port-forward established to pod/%s:%d", pod, remotePort)
 	case <-time.After(time.Minute):
 		return nil, errors.New("timed out after waiting to establish port forward")
 	case err = <-portForwardIssueCh:
 		return nil, err
 	}
-	return stopCh, nil
+	return stopCh, nil*/
 }

pkg/plan/network/consensus.go

@@ -11,7 +11,7 @@ const defaultValCpu = 500
 const defaultClMem = 1536
 const defaultValMem = 512
 
-func composeConsensusTesterNetwork(nodeMultiplier int, consensusClient string, execClientMap, consClientMap map[string]ClientVersion) ([]*Node, error) {
+func composeConsensusTesterNetwork(nodeMultiplier int, consensusClient string, execClientList []ClientVersion, consClientMap map[string]ClientVersion) ([]*Node, error) {
 
 	// make sure consensusClient actually exists
 	clientUnderTest, ok := consClientMap[consensusClient]
@@ -21,11 +21,11 @@ func composeConsensusTesterNetwork(nodeMultiplier int, consensusClient string, e
 
 	// start from 2 because bootnode is index 1
 	index := 2
-	nodes, err := composeNodesForClTesting(nodeMultiplier, index, clientUnderTest, execClientMap)
+	nodes, err := composeNodesForClTesting(nodeMultiplier, index, clientUnderTest, execClientList)
 	return nodes, err
 }
 
-func composeNodesForClTesting(nodeMultiplier, index int, consensusClient ClientVersion, execClients map[string]ClientVersion) ([]*Node, error) {
+func composeNodesForClTesting(nodeMultiplier, index int, consensusClient ClientVersion, execClients []ClientVersion) ([]*Node, error) {
 	var nodes []*Node
 
 	for _, execClient := range execClients {

pkg/plan/network/execution.go

@@ -4,10 +4,10 @@ import (
 	"github.com/kurtosis-tech/stacktrace"
 )
 
-const defaultElCpu = 512
-const defaultElMem = 512
+const defaultElCpu = 768
+const defaultElMem = 1024
 
-func composeExecTesterNetwork(nodeMultiplier int, execClient string, execClientMap, consClientMap map[string]ClientVersion) ([]*Node, error) {
+func composeExecTesterNetwork(nodeMultiplier int, execClient string, consClientList []ClientVersion, execClientMap map[string]ClientVersion) ([]*Node, error) {
 
 	// make sure execClient actually exists
 	clientUnderTest, ok := execClientMap[execClient]
@@ -17,14 +17,14 @@ func composeExecTesterNetwork(nodeMultiplier int, execClient string, execClientM
 
 	// start from 2 because bootnode is index 1
 	index := 2
-	nodes, err := composeNodesForElTesting(nodeMultiplier, index, clientUnderTest, consClientMap)
+	nodes, err := composeNodesForElTesting(nodeMultiplier, index, clientUnderTest, consClientList)
 	return nodes, err
 }
 
-func composeNodesForElTesting(nodeMultiplier, index int, execClient ClientVersion, consensusClients map[string]ClientVersion) ([]*Node, error) {
+func composeNodesForElTesting(nodeMultiplier, index int, execClient ClientVersion, consClientList []ClientVersion) ([]*Node, error) {
 	var nodes []*Node
 
-	for _, consensusClient := range consensusClients {
+	for _, consensusClient := range consClientList {
 		for i := 0; i < nodeMultiplier; i++ {
 			node := buildNode(index, execClient, consensusClient)
 			nodes = append(nodes, node)

pkg/plan/network/network_builder.go

@@ -65,9 +65,9 @@ func ComposeNetworkTopology(topology Topology, clientUnderTest string, execClien
 	// assume already checked clientUnderTest is a member of consClients or execClients
 	var nodesToTest []*Node
 	if isExecutionClient {
-		nodesToTest, err = composeExecTesterNetwork(nodeMultiplier, clientUnderTest, execClientMap, consClientMap)
+		nodesToTest, err = composeExecTesterNetwork(nodeMultiplier, clientUnderTest, consClients, execClientMap)
 	} else {
-		nodesToTest, err = composeConsensusTesterNetwork(nodeMultiplier, clientUnderTest, execClientMap, consClientMap)
+		nodesToTest, err = composeConsensusTesterNetwork(nodeMultiplier, clientUnderTest, execClients, consClientMap)
 	}
 	if err != nil {
 		return nil, err
@@ -78,6 +78,7 @@ func ComposeNetworkTopology(topology Topology, clientUnderTest string, execClien
 	extraNodes, err := composeNodesToSatisfyTargetPercent(
 		topology.TargetsAsPercentOfNetwork,
 		len(nodes)-1,
+		nodes[len(nodes)-1].Index+1,
 		clientUnderTest,
 		execClients,
 		consClients,
@@ -89,7 +90,7 @@ func ComposeNetworkTopology(topology Topology, clientUnderTest string, execClien
 	return nodes, nil
 }
 
-func composeNodesToSatisfyTargetPercent(percentTarget float32, targetedNodeCount int, clientUnderTest string, execClients, consClients []ClientVersion) ([]*Node, error) {
+func composeNodesToSatisfyTargetPercent(percentTarget float32, targetedNodeCount int, startIndex int, clientUnderTest string, execClients, consClients []ClientVersion) ([]*Node, error) {
 	// percent target is unconfigured
 	if percentTarget == 0 {
 		return []*Node{}, nil
@@ -100,8 +101,7 @@ func composeNodesToSatisfyTargetPercent(percentTarget float32, targetedNodeCount
 		return nil, err
 	}
 
-	startNodeIndex := targetedNodeCount + 2
-	nodes, err := pickExtraNodeClients(startNodeIndex, nodesToAdd, clientUnderTest, execClients, consClients)
+	nodes, err := pickExtraNodeClients(startIndex, nodesToAdd, clientUnderTest, execClients, consClients)
 	return nodes, err
 }