feat(k6): add support for k8s loading/unloading of Seldon CRs (#5563)

This adds initial k6 load test support for k8s via the `USE_KUBE_CONTROL_PLANE` environment
variable:

- models/pipelines/experiments in components/model.js are modified to also
return CR yamls
- updates functions in components/utils.js (including `setupBase(...)` and
`teardownBase(...)` to use xk6-kubernetes, when configured
- all scenarios already using `setupBase(...)` should work unchanged

**Which issue(s) this PR fixes:**
- INFRA-949 (internal issue) Extend existing k6 scenarios to use xk6-kubernetes

tests/k6/Makefile

@@ -94,5 +94,5 @@ create-secret:
 
 xk6-install:
 	# Install xk6
-	go install github.com/grafana/xk6/cmd/xk6@latest
+	go install go.k6.io/xk6/cmd/xk6@latest
 	xk6 build --with github.com/grafana/xk6-kubernetes

tests/k6/components/k8s.js

@@ -0,0 +1,104 @@
+import { Kubernetes } from "k6/x/kubernetes";
+import { getConfig } from '../components/settings.js'
+import {
+  awaitStatus,
+  awaitPipelineStatus,
+  awaitExperimentStart,
+  awaitExperimentStop
+} from '../components/scheduler.js';
+import { seldonObjectType } from '../components/seldon.js'
+
+const kubeclient = new Kubernetes();
+const namespace = getConfig().namespace;
+var schedulerClient = null;
+
+export function connectScheduler(schedulerCl) {
+  schedulerClient = schedulerCl
+}
+
+export function disconnectScheduler() {
+  schedulerClient = null
+}
+
+function seldonObjExists(kind, name, ns) {
+    // This is ugly, but xk6-kubernetes kubeclient.get(...) throws an exception if the
+    // underlying k8s CR doesn't exist.
+
+    // The alternative here would be to list all objects of the given kind from the namespace
+    // and see if the one with the specified name exists among them. However, that would end
+    // up being considerably slower, and we don't want to do it on every single
+    // model/pipeline/experiment load or unload.
+    try {
+        kubeclient.get(kind.description, name, ns)
+        return true
+    } catch(error) {
+        return false
+    }
+}
+
+export function loadModel(modelName, data, awaitReady=true) {
+    // TODO: Update existing model with new CR definition.
+    // At the moment, if an object with the same name exists, it will not be
+    // re-loaded with different settings. This is because we get a k8s apply
+    // conflict caused by a FieldManager being set on `.spec.memory`
+    if(!seldonObjExists(seldonObjectType.MODEL, modelName, namespace)) {
+        kubeclient.apply(data)
+        let created = kubeclient.get(seldonObjectType.MODEL.description, modelName, namespace)
+        if ('uid' in created.metadata) {
+            if (awaitReady && schedulerClient != null) {
+                awaitStatus(modelName, "ModelAvailable")
+            }
+        }
+    }
+}
+
+export function unloadModel(modelName, awaitReady=true) {
+    if(seldonObjExists(seldonObjectType.MODEL, modelName, namespace)) {
+        kubeclient.delete(seldonObjectType.MODEL.description, modelName, namespace)
+        if (awaitReady && schedulerClient != null) {
+            awaitStatus(modelName, "ModelTerminated")
+        }
+    }
+}
+
+export function loadPipeline(pipelineName, data, awaitReady=true) {
+    if(!seldonObjExists(seldonObjectType.PIPELINE, pipelineName, namespace)) {
+        kubeclient.apply(data)
+        let created = kubeclient.get(seldonObjectType.PIPELINE.description, pipelineName, namespace)
+        if ('uid' in created.metadata) {
+            if (awaitReady && schedulerClient != null) {
+                awaitStatus(pipelineName, "PipelineReady")
+            }
+        }
+    }
+}
+
+export function unloadPipeline(pipelineName, awaitReady = true) {
+    if(seldonObjExists(seldonObjectType.PIPELINE, pipelineName, namespace)) {
+        kubeclient.delete(seldonObjectType.PIPELINE.description, pipelineName, namespace)
+        if (awaitReady && schedulerClient != null) {
+            awaitStatus(pipelineName, "PipelineTerminated")
+        }
+    }
+}
+
+export function loadExperiment(experimentName, data, awaitReady=true) {
+    if(!seldonObjExists(seldonObjectType.EXPERIMENT, experimentName, namespace)) {
+        kubeclient.apply(data)
+        let created = kubeclient.get(seldonObjectType.EXPERIMENT.description, experimentName, namespace)
+        if ('uid' in created.metadata) {
+            if (awaitReady && schedulerClient != null) {
+                awaitExperimentStart(experimentName)
+            }
+        }
+    }
+}
+
+export function unloadExperiment(experimentName, awaitReady=true) {
+    if(seldonObjExists(seldonObjectType.EXPERIMENT, experimentName, namespace)) {
+        kubeclient.delete(seldonObjExists.EXPERIMENT.description, experimentName, namespace)
+        if (awaitReady && schedulerClient != null) {
+            awaitExperimentStop(experimentName)
+        }
+    }
+}