Merge pull request #28 from crytic/additional-faults

Additional faults. Addresses #26 #16 #3

Author: bsamuels453

README.md

@@ -16,7 +16,7 @@
 3. Install chaos-mesh
    1. `kubectl create ns chaos-mesh`
    2. `helm repo add chaos-mesh https://charts.chaos-mesh.org`
-   3. `helm install chaos-mesh chaos-mesh/chaos-mesh -n=chaos-mesh --version 2.6.1 --set chaosDaemon.runtime=containerd --set chaosDaemon.socketPath=/run/containerd/containerd.sock --set dashboard.securityMode=false`
+   3. `helm install chaos-mesh chaos-mesh/chaos-mesh -n=chaos-mesh --version 2.6.1 --set chaosDaemon.runtime=containerd --set chaosDaemon.socketPath=/run/containerd/containerd.sock --set dashboard.securityMode=false --set bpfki.create=true`
    4. To access chaos dashboard, use `kubectl --namespace chaos-mesh port-forward svc/chaos-dashboard 2333`
 4. Install kurtosis locally.
 5. Run `kurtosis cluster set cloud`

cmd/attacknet/main.go

@@ -49,6 +49,7 @@ func main() {
 		err = pkg.StartTestSuite(ctx, cfg)
 		if err != nil {
 			log.Fatal(err)
+			os.Exit(1)
 		}
 	}
 }

pkg/chaos-mesh/session.go

@@ -6,6 +6,7 @@ import (
 	api "github.com/chaos-mesh/chaos-mesh/api/v1alpha1"
 	"github.com/kurtosis-tech/stacktrace"
 	log "github.com/sirupsen/logrus"
+	v1 "k8s.io/api/core/v1"
 	"reflect"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"time"
@@ -23,38 +24,73 @@ const (
 	Error      FaultStatus = "Error"
 )
 
+var FaultHasNoDurationErr = fmt.Errorf("this fault has no expected duration")
+
 // succeeded (inject worked, now back to normal)
 // failure?
 // time out?
 
 type FaultSession struct {
-	client              *ChaosClient
-	faultKind           *api.ChaosKind
-	faultSpec           map[string]interface{}
-	Name                string
-	podsFailingRecovery map[string]*api.Record
-	TestStartTime       time.Time
-	TestDuration        *time.Duration
-	TestEndTime         time.Time
+	client                   *ChaosClient
+	faultKind                *api.ChaosKind
+	faultType                string
+	faultAction              string
+	faultSpec                map[string]interface{}
+	Name                     string
+	podsFailingRecovery      map[string]*api.Record
+	checkedForMissingPods    bool
+	podsExpectedMissing      int
+	TestStartTime            time.Time
+	TestDuration             *time.Duration
+	TestEndTime              *time.Time
+	TargetSelectionCompleted bool
 }
 
 func NewFaultSession(ctx context.Context, client *ChaosClient, faultKind *api.ChaosKind, faultSpec map[string]interface{}, name string) (*FaultSession, error) {
 	now := time.Now()
 
+	faultKindStr, ok := faultSpec["kind"].(string)
+	if !ok {
+		return nil, stacktrace.NewError("failed to decode faultSpec.kind to string: %s", faultSpec["kind"])
+	}
+
+	spec, ok := faultSpec["spec"].(map[string]interface{})
+	if !ok {
+		return nil, stacktrace.NewError("failed to decode faultSpec.spec to map[string]interface{}")
+	}
+
+	faultAction, ok := spec["action"].(string)
+	if !ok {
+		return nil, stacktrace.NewError("failed to decode faultSpec.spec.action to string: %s", spec["action"])
+	}
+
 	partial := &FaultSession{
-		client:              client,
-		faultKind:           faultKind,
-		faultSpec:           faultSpec,
-		Name:                name,
-		podsFailingRecovery: map[string]*api.Record{},
-		TestStartTime:       now,
+		client:                   client,
+		faultKind:                faultKind,
+		faultType:                faultKindStr,
+		faultSpec:                spec,
+		faultAction:              faultAction,
+		Name:                     name,
+		podsFailingRecovery:      map[string]*api.Record{},
+		TestStartTime:            now,
+		podsExpectedMissing:      0,
+		checkedForMissingPods:    false,
+		TargetSelectionCompleted: false,
 	}
 	duration, err := partial.getDuration(ctx)
 	if err != nil {
-		return nil, err
+		if err == FaultHasNoDurationErr {
+			partial.TestDuration = nil
+			partial.TestEndTime = nil
+		} else {
+			return nil, err
+		}
+	} else {
+		partial.TestDuration = duration
+		endTime := now.Add(*duration)
+		partial.TestEndTime = &endTime
 	}
-	partial.TestDuration = duration
-	partial.TestEndTime = now.Add(*duration)
+
 	return partial, nil
 }
 
@@ -72,12 +108,39 @@ func (f *FaultSession) getKubeResource(ctx context.Context) (client.Object, erro
 	return resource, nil
 }
 
-func (f *FaultSession) getDetailedStatus(ctx context.Context) ([]*api.Record, error) {
+func (f *FaultSession) checkTargetSelectionCompleted(resource client.Object) error {
+	if f.TargetSelectionCompleted {
+		return nil
+	}
+	conditionsVal := reflect.ValueOf(resource).Elem().FieldByName("Status").FieldByName("ChaosStatus").FieldByName("Conditions")
+	conditions, ok := conditionsVal.Interface().([]api.ChaosCondition)
+	if !ok || conditions == nil {
+		return stacktrace.NewError("Unable to decode status.chaosstatus.conditions")
+	}
+	for _, condition := range conditions {
+		if condition.Type != api.ConditionSelected {
+			continue
+		}
+		if condition.Status == v1.ConditionTrue {
+			log.Info("chaos-mesh has identified pods to inject into")
+			f.TargetSelectionCompleted = true
+		}
+		break
+	}
+	return nil
+}
+
+func (f *FaultSession) getFaultRecords(ctx context.Context) ([]*api.Record, error) {
 	resource, err := f.getKubeResource(ctx)
 	if err != nil {
 		return nil, err
 	}
 
+	err = f.checkTargetSelectionCompleted(resource)
+	if err != nil {
+		return nil, err
+	}
+
 	// Feel free to figure out a better way to do this. These fields are part of every Chaos status struct we support,
 	// but since they don't implement a common interface containing the status fields, there's no clean or simple way
 	// to extract the values in Go. One alternate option may be to serialize to json, then deserialize into an object
@@ -117,18 +180,51 @@ func (f *FaultSession) checkForFailedRecovery(record *api.Record) (bool, []strin
 	return true, distinctMessages
 }
 
+/*
+Determines whether the fault will leave some pods in a terminated state, and how many pods will be impacted.
+This must be run after the fault manifest has been applied and the handler webhook has run.
+*/
+func (f *FaultSession) checkForMissingPods(records []*api.Record) error {
+	if !f.checkedForMissingPods {
+		f.checkedForMissingPods = true
+		// we expect missing pods when the fault is pod kill.
+
+		podsInjected := countInjectedPods(records)
+		log.Infof("Chaos-mesh has identified %d pods matching the targeting criteria", podsInjected)
+		if f.faultType == "PodChaos" && f.faultAction == "pod-kill" {
+			f.podsExpectedMissing = podsInjected
+			log.Infof("We're expecting %d pods to be terminated from the selected fault", f.podsExpectedMissing)
+		}
+	}
+	return nil
+}
+
+func countInjectedPods(records []*api.Record) int {
+	podsInjected := 0
+	for _, record := range records {
+		if record.Phase == "Injected" {
+			podsInjected += 1
+		}
+	}
+	return podsInjected
+}
+
 // todo: we need a better way of monitoring fault injection status. There's a ton of statefulness represented in
 // chaos-mesh that we're glancing over. Situations such as a pod crashing during a fault may produce unexpected behavior
 // in this code as it currently stands.
 func (f *FaultSession) GetStatus(ctx context.Context) (FaultStatus, error) {
-	records, err := f.getDetailedStatus(ctx)
+	records, err := f.getFaultRecords(ctx)
 	if err != nil {
 		return Error, err
 	}
 
 	if records == nil {
 		return Starting, nil
 	}
+	err = f.checkForMissingPods(records)
+	if err != nil {
+		return Error, err
+	}
 
 	podsInjectedAndRecovered := 0
 	podsInjectedNotRecovered := 0
@@ -152,15 +248,13 @@ func (f *FaultSession) GetStatus(ctx context.Context) (FaultStatus, error) {
 		}
 	}
 
-	// todo: check if unrecovered pods are failing to recover ^^ up here PodRecord.Events[-1].Operation = "Recover", Type="Failed". Emit Message
-
 	if podsNotInjected > 0 {
 		return Starting, nil
 	}
-	if podsInjectedNotRecovered > 0 && podsInjectedAndRecovered == 0 {
+	if podsInjectedNotRecovered-f.podsExpectedMissing > 0 && podsInjectedAndRecovered == 0 {
 		return InProgress, nil
 	}
-	if podsInjectedAndRecovered+len(f.podsFailingRecovery) == len(records) {
+	if podsInjectedAndRecovered+len(f.podsFailingRecovery)+f.podsExpectedMissing == len(records) {
 		return Completed, nil
 	}
 	if podsInjectedNotRecovered > 0 && podsInjectedAndRecovered > 0 {
@@ -182,6 +276,10 @@ func (f *FaultSession) getDuration(ctx context.Context) (*time.Duration, error)
 	if !ok {
 		return nil, stacktrace.NewError("unable to cast durationVal to string")
 	}
+	if durationStr == nil {
+		return nil, FaultHasNoDurationErr
+	}
+
 	duration, err := time.ParseDuration(*durationStr)
 	if err != nil {
 		return nil, err

pkg/grafana.go

@@ -14,6 +14,7 @@ type GrafanaTunnel struct {
 	Client                   *grafanaSdk.Client
 	portForwardStopCh        chan struct{}
 	allowPostFaultInspection bool
+	cleanedUp                bool
 }
 
 func CreateGrafanaClient(ctx context.Context, namespace string, config AttacknetConfig) (*GrafanaTunnel, error) {
@@ -42,14 +43,16 @@ func CreateGrafanaClient(ctx context.Context, namespace string, config Attacknet
 		return nil, stacktrace.Propagate(err, "unable to create Grafana client")
 	}
 
-	return &GrafanaTunnel{client, stopCh, config.AllowPostFaultInspection}, nil
+	return &GrafanaTunnel{client, stopCh, config.AllowPostFaultInspection, false}, nil
 }
 
-func (t *GrafanaTunnel) Cleanup() {
-	if t.allowPostFaultInspection {
-		log.Info("Attacknet has completed, but since allowPostFaultInspection is set to true, the program will continue to run to facilitate the Grafana port-forward connection.")
-		log.Info("Press enter to terminate the port-forward connection.")
-		_, _ = fmt.Scanln()
+func (t *GrafanaTunnel) Cleanup(skipInspection bool) {
+	if !t.cleanedUp {
+		if t.allowPostFaultInspection && !skipInspection {
+			log.Info("Press enter to terminate the port-forward connection.")
+			_, _ = fmt.Scanln()
+		}
+		close(t.portForwardStopCh)
+		t.cleanedUp = true
 	}
-	close(t.portForwardStopCh)
 }

pkg/kubernetes/port-forward.go

@@ -4,12 +4,12 @@ import (
 	"errors"
 	"fmt"
 	"github.com/kurtosis-tech/stacktrace"
+	log "github.com/sirupsen/logrus"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/portforward"
 	"k8s.io/client-go/transport/spdy"
 	"net/http"
 	"net/url"
-	"os"
 	"time"
 )
 
@@ -31,11 +31,16 @@ func StartPortForwarding(pod, namespace string, port uint16, kubeConfig *rest.Co
 
 	stopCh = make(chan struct{}, 1)
 	readyCh := make(chan struct{}, 1)
-	portForward, err := portforward.New(dialer, []string{portFwd}, stopCh, readyCh, os.Stdout, os.Stderr)
+	logger := log.New()
+
+	errLogger := CreatePrefixWriter("[port-forward] ", logger.WriterLevel(log.ErrorLevel))
+	stdLogger := CreatePrefixWriter("[port-forward] ", logger.WriterLevel(log.InfoLevel))
+
+	portForward, err := portforward.New(dialer, []string{portFwd}, stopCh, readyCh, stdLogger, errLogger)
 	if err != nil {
 		return nil, stacktrace.Propagate(err, "unable to create port forward dialer")
 	}
-	fmt.Print("Starting port-forward to grafana pod")
+	log.Info("Starting port-forward to grafana pod")
 
 	go func() {
 		if err = portForward.ForwardPorts(); err != nil {
@@ -45,7 +50,7 @@ func StartPortForwarding(pod, namespace string, port uint16, kubeConfig *rest.Co
 
 	select {
 	case <-readyCh:
-		fmt.Print("Port-forward established.")
+		log.Info("Port-forward established.")
 	case <-time.After(time.Minute):
 		return nil, errors.New("timed out after waiting to establish port forward")
 	}

pkg/kubernetes/utils.go

@@ -0,0 +1,16 @@
+package kubernetes
+
+import "io"
+
+type LogPrefixWriter struct {
+	prefix string
+	writer io.Writer
+}
+
+func CreatePrefixWriter(prefix string, writer io.Writer) *LogPrefixWriter {
+	return &LogPrefixWriter{prefix, writer}
+}
+
+func (pw *LogPrefixWriter) Write(p []byte) (n int, err error) {
+	return pw.writer.Write(append([]byte(pw.prefix), p...))
+}

pkg/kurtosis.go

@@ -139,24 +139,24 @@ func StartNetwork(ctx context.Context, enclaveCtx *EnclaveContextWrapper, harnes
 		if progress != nil {
 			progressMsgs := progress.CurrentStepInfo
 			for i := progressIndex; i < len(progressMsgs); i++ {
-				log.Infof("Kurtosis: %s", progressMsgs[i])
+				log.Infof("[Kurtosis] %s", progressMsgs[i])
 			}
 			progressIndex = len(progressMsgs)
 		}
 
 		info := t.GetInfo()
 		if info != nil {
-			log.Infof("Kurtosis: %s", info.InfoMessage)
+			log.Infof("[Kurtosis] %s", info.InfoMessage)
 		}
 
 		warn := t.GetWarning()
 		if warn != nil {
-			log.Warnf("Kurtosis: %s", warn.WarningMessage)
+			log.Warnf("[Kurtosis] %s", warn.WarningMessage)
 		}
 
 		e := t.GetError()
 		if e != nil {
-			log.Errorf("Kurtosis: %s", e.String())
+			log.Errorf("[Kurtosis] %s", e.String())
 			return stacktrace.Propagate(errors.New("kurtosis deployment failed during execution"), "%s", e.String())
 		}
 
@@ -166,17 +166,17 @@ func StartNetwork(ctx context.Context, enclaveCtx *EnclaveContextWrapper, harnes
 
 		insRes := t.GetInstructionResult()
 		if insRes != nil {
-			log.Infof("Kurtosis: %s", insRes.SerializedInstructionResult)
+			log.Infof("[Kurtosis] %s", insRes.SerializedInstructionResult)
 		}
 
 		finishRes := t.GetRunFinishedEvent()
 		if finishRes != nil {
-			log.Infof("Kurtosis: %s", finishRes.GetSerializedOutput())
+			log.Infof("[Kurtosis] %s", finishRes.GetSerializedOutput())
 			if finishRes.IsRunSuccessful {
-				log.Info("Kurtosis: Devnet genesis successful. Passing back to Attacknet")
+				log.Info("[Kurtosis] Devnet genesis successful. Passing back to Attacknet")
 				return nil
 			} else {
-				log.Error("Kurtosis: There was an error during genesis.")
+				log.Error("[Kurtosis] There was an error during genesis.")
 				return stacktrace.Propagate(errors.New("kurtosis deployment failed"), "%s", finishRes.GetSerializedOutput())
 			}
 		}