Repository: iter8-tools/iter8
Pull request #1195 (2022-03-24 19:09:23)

Readiness Task
Migration of readiness task from v0.7 to v0.9.  Uses the k8s dynamic client (replacing https://github.com/iter8-tools/iter8/pull/1191).  Related issue is https://github.com/iter8-tools/iter8/issues/1190.

kalantar: Perhaps name of task should be singular `k8s-object-ready` instead of plural (`objects`).

Patch:
diff --git a/base/experiment.go b/base/experiment.go
index b6caabe6..ca224a15 100644
--- a/base/experiment.go
+++ b/base/experiment.go
@@ -162,6 +162,10 @@ func (s *ExperimentSpec) UnmarshalJSON(data []byte) error {
 		} else {
 			// this is some other task
 			switch *t.Task {
+			case ReadinessTaskName:
+				rt := &readinessTask{}
+				json.Unmarshal(tBytes, rt)
+				tsk = rt
 			case CollectHTTPTaskName:
 				cht := &collectHTTPTask{}
 				err := json.Unmarshal(tBytes, cht)
diff --git a/base/kubedriver.go b/base/kubedriver.go
new file mode 100644
index 00000000..7fd917db
--- /dev/null
+++ b/base/kubedriver.go
@@ -0,0 +1,57 @@
+package base
+
+import (
+	"errors"
+
+	// Import to initialize client auth plugins.
+	_ "k8s.io/client-go/plugin/pkg/client/auth"
+
+	"github.com/iter8-tools/iter8/base/log"
+
+	"helm.sh/helm/v3/pkg/cli"
+
+	"k8s.io/client-go/dynamic"
+)
+
+var (
+	kd = NewKubeDriver(cli.New())
+)
+
+// KubeDriver embeds Kube configuration, and
+// enables interaction with a Kubernetes cluster through Kube APIs
+type KubeDriver struct {
+	// EnvSettings provides generic Kubernetes options
+	*cli.EnvSettings
+	// dynamicClient enables unstructured interaction with a Kubernetes cluster
+	dynamicClient dynamic.Interface
+}
+
+// NewKubeDriver creates and returns a new KubeDriver
+func NewKubeDriver(s *cli.EnvSettings) *KubeDriver {
+	kd := &KubeDriver{
+		EnvSettings:   s,
+		dynamicClient: nil,
+	}
+	return kd
+}
+
+// initKube initializes the Kubernetes clientset
+func (kd *KubeDriver) initKube() (err error) {
+	if kd.dynamicClient == nil {
+		// get REST config
+		restConfig, err := kd.EnvSettings.RESTClientGetter().ToRESTConfig()
+		if err != nil {
+			e := errors.New("unable to get Kubernetes REST config")
+			log.Logger.WithStackTrace(err.Error()).Error(e)
+			return e
+		}
+		kd.dynamicClient, err = dynamic.NewForConfig(restConfig)
+		if err != nil {
+			e := errors.New("unable to get Kubernetes dynamic client")
+			log.Logger.WithStackTrace(err.Error()).Error(e)
+			return e
+		}
+	}
+
+	return nil
+}
diff --git a/base/readiness.go b/base/readiness.go
new file mode 100644
index 00000000..bd4f1b77
--- /dev/null
+++ b/base/readiness.go
@@ -0,0 +1,204 @@
+package base
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	log "github.com/iter8-tools/iter8/base/log"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/util/retry"
+)
+
+const (
+	// ReadinessTaskName is the task name
+	ReadinessTaskName = "k8s-object-ready"
+
+	// defaultTimeout is default timeout for readiness command
+	defaultTimeout = "10s"
+)
+
+// ReadinessInputs identifies the K8s object to test for existence and
+// the (optional) condition that should be tested (succeeds if true).
+type readinessInputs struct {
+	// Group of the object. Optional. If unspecified it will be defaulted to ""
+	Group string `json:"group,omitempty" yaml:"group,omitempty"`
+	// Version of the object. Optional. If unspecified it will be defaulted to ""
+	Version string `json:"version,omitempty" yaml:"version,omitempty"`
+	// Resource type of the object. Required.
+	Resource string `json:"resource" yaml:"resource"`
+	// Namespace of the object. Optional. If left unspecified, this will be defaulted to the namespace of the experiment
+	Namespace *string `json:"namespace,omitempty" yaml:"namespace,omitempty"`
+	// Name of the object
+	Name string `json:"name" yaml:"name"`
+	// Condition is label of condition to check for value of "True"
+	Condition *string `json:"condition" yaml:"condition"`
+	// Timeout is maximum time spent trying to find object and check condition
+	Timeout *string `json:"timeout" yaml:"timeout"`
+}
+
+// ReadinessTask checks existence and readiness of specified resources
+type readinessTask struct {
+	TaskMeta
+	With readinessInputs `json:"with" yaml:"with"`
+}
+
+// initializeDefaults sets default values for the readiness task
+func (t *readinessTask) initializeDefaults() {
+	if t.With.Timeout == nil {
+		t.With.Timeout = StringPointer(defaultTimeout)
+	}
+
+	kd.initKube()
+	// set Namespace (from context) if not already set
+	if t.With.Namespace == nil {
+		t.With.Namespace = StringPointer(kd.Namespace())
+	}
+}
+
+// validateInputs validates task inputs
+// at present all validation at initialization
+func (t *readinessTask) validateInputs() error {
+	return nil
+}
+
+// run executes the task
+func (t *readinessTask) run(exp *Experiment) error {
+	// validation
+	err := t.validateInputs()
+	if err != nil {
+		return err
+	}
+
+	// initialization
+	t.initializeDefaults()
+
+	// parse timeout
+	timeout, err := time.ParseDuration(*t.With.Timeout)
+	if err != nil {
+		e := errors.New("invalid format for timeout")
+		log.Logger.WithStackTrace(err.Error()).Error(e)
+		return e
+	}
+
+	// get rest config
+	restConfig, err := kd.EnvSettings.RESTClientGetter().ToRESTConfig()
+	if err != nil {
+		e := errors.New("unable to get Kubernetes REST config")
+		log.Logger.WithStackTrace(err.Error()).Error(e)
+		return e
+	}
+
+	// do the work: check for object and condition
+	// repeat until time out
+	interval := 1 * time.Second
+	err = retry.OnError(
+		wait.Backoff{
+			Steps:    int(timeout / interval),
+			Cap:      timeout,
+			Duration: interval,
+			Factor:   1.0,
+			Jitter:   0.1,
+		},
+		func(err error) bool {
+			log.Logger.Error(err)
+			return true
+		}, // retry on all failures
+		func() error {
+			return checkObjectExistsAndConditionTrue(t, restConfig)
+		},
+	)
+	return err
+}
+
+// checkObjectExistsAndConditionTrue determines if the object exists
+// if so, it further checks if the requested condition is "True"
+func checkObjectExistsAndConditionTrue(t *readinessTask, restCfg *rest.Config) error {
+	log.Logger.Trace("looking for resource (", t.With.Group, "/", t.With.Version, ") ", t.With.Resource, ": ", t.With.Name, " in namespace ", *t.With.Namespace)
+
+	obj, err := kd.dynamicClient.Resource(gvr(&t.With)).Namespace(*t.With.Namespace).Get(context.Background(), t.With.Name, metav1.GetOptions{})
+	if err != nil {
+		return err
+	}
+
+	// if no condition to check was specified, we can return now
+	if t.With.Condition == nil {
+		return nil
+	}
+
+	// otherwise, find the condition and check that it is "True"
+	log.Logger.Trace("looking for condition: ", *t.With.Condition)
+
+	cs, err := getConditionStatus(obj, *t.With.Condition)
+	if err != nil {
+		return err
+	}
+	if strings.EqualFold(*cs, string(corev1.ConditionTrue)) {
+		return nil
+	}
+	return errors.New("condition status not True")
+}
+
+func gvr(objRef *readinessInputs) schema.GroupVersionResource {
+	return schema.GroupVersionResource{
+		Group:    objRef.Group,
+		Version:  objRef.Version,
+		Resource: objRef.Resource,
+	}
+}
+
+func getConditionStatus(obj *unstructured.Unstructured, conditionType string) (*string, error) {
+	if obj == nil {
+		return nil, errors.New("no object")
+	}
+
+	resultJson, err := obj.MarshalJSON()
+	if err != nil {
+		return nil, err
+	}
+
+	resultObj := make(map[string]interface{})
+	err = json.Unmarshal(resultJson, &resultObj)
+	if err != nil {
+		return nil, err
+	}
+
+	// get object status
+	objStatusInterface, ok := resultObj["status"]
+	if !ok {
+		return nil, errors.New("object does not contain a status")
+	}
+	objStatus := objStatusInterface.(map[string]interface{})
+
+	conditionsInterface, ok := objStatus["conditions"]
+	if !ok {
+		return nil, errors.New("object status does not contain conditions")
+	}
+	conditions := conditionsInterface.([]interface{})
+	for _, conditionInterface := range conditions {
+		condition := conditionInterface.(map[string]interface{})
+		cTypeInterface, ok := condition["type"]
+		if !ok {
+			return nil, errors.New("condition does not have a type")
+		}
+		cType := cTypeInterface.(string)
+		if strings.EqualFold(cType, conditionType) {
+			conditionStatusInterface, ok := condition["status"]
+			if !ok {
+				return nil, fmt.Errorf("condition %s does not have a value", cType)
+			}
+			conditionStatus := conditionStatusInterface.(string)
+			return StringPointer(conditionStatus), nil
+		}
+	}
+	return nil, errors.New("expected condition not found")
+}
diff --git a/driver/kubedriver.go b/driver/kubedriver.go
index 64130706..8dfcce81 100644
--- a/driver/kubedriver.go
+++ b/driver/kubedriver.go
@@ -122,14 +122,14 @@ func (driver *KubeDriver) initRevision() error {
 }
 
 // Init initializes the KubeDriver
-func (driver *KubeDriver) Init() error {
-	if err := driver.initKube(); err != nil {
+func (kd *KubeDriver) Init() error {
+	if err := kd.initKube(); err != nil {
 		return err
 	}
-	if err := driver.initHelm(); err != nil {
+	if err := kd.initHelm(); err != nil {
 		return err
 	}
-	if err := driver.initRevision(); err != nil {
+	if err := kd.initRevision(); err != nil {
 		return err
 	}
 	return nil


Test patch:
diff --git a/base/readiness_test.go b/base/readiness_test.go
new file mode 100644
index 00000000..9f477903
--- /dev/null
+++ b/base/readiness_test.go
@@ -0,0 +1,175 @@
+package base
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"helm.sh/helm/v3/pkg/cli"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+// TestNoObject tests that task fails if the object is not present
+func TestNoObject(t *testing.T) {
+	ns, nm := "default", "test-pod"
+	pod := newPod(ns, nm).withCondition("Ready", "True").build()
+	rTask := newReadinessTask("non-existant-pod").withVersion("v1").withResource("pods").withNamespace(ns).withCondition("Ready").build()
+	runTaskTest(t, rTask, false, ns, pod)
+}
+
+// TestWithoutCondition tests the task succeeds when there are no conditions on the object
+// It should be successful
+// Also validates parsing of timeout
+// Also validates setting of default namespace
+func TestWithoutConditions(t *testing.T) {
+	ns, nm := "default", "test-pod"
+	pod := newPod(ns, nm).build()
+	rTask := newReadinessTask(nm).withVersion("v1").withResource("pods").withTimeout("20s").build()
+	runTaskTest(t, rTask, true, ns, pod)
+}
+
+// TestWithCondition tests that the task succeeds when the condition is present and True
+func TestWithCondition(t *testing.T) {
+	ns, nm := "default", "test-pod"
+	pod := newPod(ns, nm).withCondition("Ready", "True").build()
+	rTask := newReadinessTask(nm).withVersion("v1").withResource("pods").withNamespace(ns).withCondition("Ready").build()
+	runTaskTest(t, rTask, true, ns, pod)
+}
+
+// TestWithFalseCondition tests that the task fails when the condition is present and not True
+func TestWithFalseCondition(t *testing.T) {
+	ns, nm := "default", "test-pod"
+	pod := newPod(ns, nm).withCondition("Ready", "False").build()
+	rTask := newReadinessTask(nm).withVersion("v1").withResource("pods").withNamespace(ns).withCondition("Ready").build()
+	runTaskTest(t, rTask, false, ns, pod)
+}
+
+// TestConditionNotPresent tests that the task fails when the condition is not present (but others are)
+func TestConditionNotPresent(t *testing.T) {
+	ns, nm := "default", "test-pod"
+	pod := newPod(ns, nm).build()
+	rTask := newReadinessTask(nm).withVersion("v1").withResource("pods").withNamespace(ns).withCondition("NotPresent").build()
+	runTaskTest(t, rTask, false, ns, pod)
+}
+
+// TestInvalidTimeout tests that the task fails when the specified timeout is invalid (not parseable)
+func TestInvalidTimeout(t *testing.T) {
+	ns, nm := "default", "test-pod"
+	pod := newPod(ns, nm).withCondition("Ready", "True").build()
+	rTask := newReadinessTask(nm).withVersion("v1").withResource("pods").withNamespace(ns).withTimeout("timeout").build()
+	runTaskTest(t, rTask, false, ns, pod)
+}
+
+// UTILITY METHODS for all tests
+
+// runTaskTest creates fake cluster with pod and runs rTask
+func runTaskTest(t *testing.T, rTask *readinessTask, success bool, ns string, pod *unstructured.Unstructured) {
+	*kd = *NewFakeKubeDriver(cli.New())
+	rs := schema.GroupVersionResource{Group: "", Version: "v1", Resource: "pods"}
+	_, err := kd.dynamicClient.Resource(rs).Namespace(ns).Create(context.Background(), pod, metav1.CreateOptions{})
+	assert.NoError(t, err, "get failed")
+
+	err = rTask.run(&Experiment{
+		Tasks:  []Task{rTask},
+		Result: &ExperimentResult{},
+	})
+	if success {
+		assert.NoError(t, err)
+	} else {
+		assert.Error(t, err)
+	}
+}
+
+type podBuilder corev1.Pod
+
+func newPod(ns string, nm string) *podBuilder {
+	pod := &corev1.Pod{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: "v1",
+			Kind:       "Pod",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      nm,
+			Namespace: ns,
+		},
+	}
+	return (*podBuilder)(pod)
+}
+
+func (p *podBuilder) build() *unstructured.Unstructured {
+	o, err := runtime.DefaultUnstructuredConverter.ToUnstructured((*corev1.Pod)(p))
+	if err != nil {
+		return nil
+	}
+	return &unstructured.Unstructured{Object: o}
+}
+
+func (p *podBuilder) withCondition(typ string, value string) *podBuilder {
+	c := corev1.PodCondition{Type: (corev1.PodConditionType(typ))}
+	switch strings.ToLower(value) {
+	case "true":
+		c.Status = corev1.ConditionTrue
+	case "false":
+		c.Status = corev1.ConditionFalse
+	default:
+		c.Status = corev1.ConditionUnknown
+	}
+	p.Status.Conditions = append(p.Status.Conditions, c)
+
+	return p
+}
+
+type readinessTaskBuilder readinessTask
+
+func newReadinessTask(name string) *readinessTaskBuilder {
+	rTask := &readinessTask{
+		TaskMeta: TaskMeta{
+			Task: StringPointer(ReadinessTaskName),
+		},
+		With: readinessInputs{
+			Name: name,
+		},
+	}
+
+	return (*readinessTaskBuilder)(rTask)
+}
+
+// func (t *readinessTaskBuilder) withGroup(group string) *readinessTaskBuilder {
+// 	t.With.Group = group
+// 	return t
+// }
+func (t *readinessTaskBuilder) withVersion(version string) *readinessTaskBuilder {
+	t.With.Version = version
+	return t
+}
+
+func (t *readinessTaskBuilder) withResource(resource string) *readinessTaskBuilder {
+	t.With.Resource = resource
+	return t
+}
+
+func (t *readinessTaskBuilder) withNamespace(ns string) *readinessTaskBuilder {
+	t.With.Namespace = StringPointer(ns)
+	return t
+}
+
+func (t *readinessTaskBuilder) withTimeout(timeout string) *readinessTaskBuilder {
+	t.With.Timeout = StringPointer(timeout)
+	return t
+}
+
+func (t *readinessTaskBuilder) withCondition(condition string) *readinessTaskBuilder {
+	t.With.Condition = &condition
+	return t
+}
+
+func (t *readinessTaskBuilder) build() *readinessTask {
+	return (*readinessTask)(t)
+}
diff --git a/base/test_helpers_driver.go b/base/test_helpers_driver.go
new file mode 100644
index 00000000..aca690c6
--- /dev/null
+++ b/base/test_helpers_driver.go
@@ -0,0 +1,28 @@
+package base
+
+import (
+	"helm.sh/helm/v3/pkg/cli"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	dynamicfake "k8s.io/client-go/dynamic/fake"
+)
+
+// initKubeFake initialize the Kube clientset with a fake
+func initKubeFake(kd *KubeDriver, objects ...runtime.Object) {
+	kd.dynamicClient = dynamicfake.NewSimpleDynamicClient(runtime.NewScheme())
+}
+
+// initFake initializes fake Kubernetes and Helm clients
+func initFake(kd *KubeDriver, objects ...runtime.Object) error {
+	initKubeFake(kd, objects...)
+	return nil
+}
+
+// NewFakeKubeDriver creates and returns a new KubeDriver with fake clients
+func NewFakeKubeDriver(s *cli.EnvSettings, objects ...runtime.Object) *KubeDriver {
+	kd := &KubeDriver{
+		EnvSettings: s,
+	}
+	initFake(kd, objects...)
+	return kd
+}
diff --git a/driver/test_helpers.go b/driver/test_helpers.go
index 51da1cc6..6242411d 100644
--- a/driver/test_helpers.go
+++ b/driver/test_helpers.go
@@ -18,7 +18,7 @@ import (
 )
 
 // initKubeFake initialize the Kube clientset with a fake
-func (kd *KubeDriver) initKubeFake(objects ...runtime.Object) {
+func initKubeFake(kd *KubeDriver, objects ...runtime.Object) {
 	// secretDataReactor sets the secret.Data field based on the values from secret.StringData
 	// Credit: this function is adapted from https://github.com/creydr/go-k8s-utils
 	var secretDataReactor = func(action ktesting.Action) (bool, runtime.Object, error) {
@@ -44,7 +44,7 @@ func (kd *KubeDriver) initKubeFake(objects ...runtime.Object) {
 // initHelmFake initializes the Helm config with a fake
 // Credit: this function is adapted from helm
 // https://github.com/helm/helm/blob/e9abdc5efe11cdc23576c20c97011d452201cd92/pkg/action/action_test.go#L37
-func (kd *KubeDriver) initHelmFake() {
+func initHelmFake(kd *KubeDriver) {
 	registryClient, err := registry.NewClient()
 	if err != nil {
 		log.Logger.Error(err)
@@ -61,9 +61,9 @@ func (kd *KubeDriver) initHelmFake() {
 }
 
 // initFake initializes fake Kubernetes and Helm clients
-func (driver *KubeDriver) initFake(objects ...runtime.Object) error {
-	driver.initKubeFake(objects...)
-	driver.initHelmFake()
+func initFake(kd *KubeDriver, objects ...runtime.Object) error {
+	initKubeFake(kd, objects...)
+	initHelmFake(kd)
 	return nil
 }
 
@@ -73,6 +73,6 @@ func NewFakeKubeDriver(s *cli.EnvSettings, objects ...runtime.Object) *KubeDrive
 		EnvSettings: s,
 		Group:       DefaultExperimentGroup,
 	}
-	kd.initFake(objects...)
+	initFake(kd, objects...)
 	return kd
 }