implement ScenarioOperation CRUD logic

Makefile

@@ -32,6 +32,7 @@ format:
 # run test on all modules
 test:
 	cd simulator/ && make test
+	cd scenario/ && make test
 
 .PHONY: mod-download
 mod-download:
@@ -53,6 +54,11 @@ docker_build_server:
 docker_build_scheduler:
 	docker $(BUILD) -f simulator/cmd/scheduler/Dockerfile -t simulator-scheduler simulator
 
+.PHONY: docker_build_scenario
+docker_build_scenario:
+	docker $(BUILD) -f scenario/Dockerfile -t scenario-controller scenario
+	docker $(BUILD) -f scenario/bootstrap/Dockerfile -t bootstrap scenario/bootstrap
+
 .PHONY: docker_build_front
 docker_build_front:
 	docker $(BUILD) -t simulator-frontend ./web/
@@ -65,6 +71,10 @@ docker_up:
 docker_up_local:
 	docker compose -f compose.yml -f compose.local.yml up -d
 
+.PHONY: docker_up_scenario
+docker_up_scenario:
+	docker compose -f compose.yml -f compose.local.yml --profile scenario up -d
+
 .PHONY: docker_build_and_up
 docker_build_and_up: docker_build docker_up_local
 
@@ -75,3 +85,7 @@ docker_down:
 .PHONY: docker_down_local
 docker_down_local:
 	docker compose -f compose.yml -f compose.local.yml down --volumes
+
+.PHONY: docker_down_scenario
+docker_down_scenario:
+	docker compose -f compose.yml -f compose.local.yml --profile scenario down --volumes

compose.local.yml

@@ -32,9 +32,50 @@ services:
       - simulator-internal-network
     profiles:
       - externalImportEnabled
+
+  scenario-controller:
+    image: scenario-controller
+    profiles: ["scenario"]
+    command:
+      - "--kubeconfig=/config/kubeconfig.yaml"
+      - "--webhook-cert-path=/tmp/k8s-webhook-server/serving-certs"
+    volumes:
+      - ./simulator/cmd/scheduler/kubeconfig.yaml:/config/kubeconfig.yaml:ro
+      - scenario-webhook-certs:/tmp/k8s-webhook-server/serving-certs:ro
+    ports:
+      - "9443:9443"
+    depends_on:
+      bootstrap:
+        condition: service_healthy
+    networks:
+      - simulator-internal-network
+
+  # Bootstrap container: prepares CRDs and generates webhook certificates
+  bootstrap:
+    image: bootstrap
+    profiles: ["scenario"]
+    depends_on:
+      - simulator-cluster
+    volumes:
+      - ./simulator/cmd/scheduler/kubeconfig.yaml:/config/kubeconfig.yaml:ro
+      - ./scenario/config/webhook:/tmp/webhook:ro
+      - ./scenario/config/crd:/manifests/crd:ro
+      - scenario-webhook-certs:/manifests/webhook/certs
+      - scenario-webhook-work:/manifests/webhook
+    networks:
+      - simulator-internal-network
+    healthcheck:
+      test: ["CMD", "sh", "-c", "test -f /manifests/webhook/certs/tls.crt"]
+      interval: 5s
+      timeout: 3s
+      retries: 5
+      start_period: 5s
+
 networks:
   simulator-internal-network:
     driver: bridge
 volumes:
   simulator-etcd-data:
+  scenario-webhook-certs:
+  scenario-webhook-work:
   conf:

scenario/PROJECT

@@ -17,4 +17,7 @@ resources:
   kind: Scenario
   path: sigs.k8s.io/kube-scheduler-simulator/scenario/api/v1alpha1
   version: v1alpha1
+  webhooks:
+    validation: true
+    webhookVersion: v1
 version: "3"

scenario/api/v1alpha1/run_suite_test.go

@@ -0,0 +1,57 @@
+package v1alpha1
+
+import (
+	"context"
+	"path/filepath"
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/rest"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/envtest"
+	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var (
+	cfg     *rest.Config
+	k8sCli  client.Client
+	testEnv *envtest.Environment
+	ctx     context.Context
+	cancel  context.CancelFunc
+)
+
+func TestRun(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "ScenarioOperation Run() API Suite")
+}
+
+var _ = BeforeSuite(func() {
+	ctrl.SetLogger(zap.New(zap.WriteTo(GinkgoWriter), zap.UseDevMode(true)))
+	ctx, cancel = context.WithCancel(context.TODO())
+
+	testEnv = &envtest.Environment{
+		CRDDirectoryPaths: []string{
+			filepath.Join("..", "..", "..", "config", "crd", "bases"),
+		},
+	}
+	var err error
+	cfg, err = testEnv.Start()
+	Expect(err).NotTo(HaveOccurred())
+
+	scheme := runtime.NewScheme()
+	Expect(corev1.AddToScheme(scheme)).To(Succeed())
+	Expect(AddToScheme(scheme)).To(Succeed())
+
+	k8sCli, err = client.New(cfg, client.Options{Scheme: scheme})
+	Expect(err).NotTo(HaveOccurred())
+})
+
+var _ = AfterSuite(func() {
+	cancel()
+	Expect(testEnv.Stop()).To(Succeed())
+})

scenario/api/v1alpha1/scenario_operation.go

@@ -0,0 +1,117 @@
+package v1alpha1
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"golang.org/x/xerrors"
+	runtime "k8s.io/apimachinery/pkg/runtime"
+	runtimeschema "k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/discovery"
+	memory "k8s.io/client-go/discovery/cached"
+	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/restmapper"
+)
+
+func (s *ScenarioOperation) Run(ctx context.Context, cfg *rest.Config) (bool, error) {
+	switch {
+	case s.Create != nil:
+		ope := s.Create
+		gvk := ope.Object.GetObjectKind().GroupVersionKind()
+		client, err := buildClient(gvk, cfg)
+		if err != nil {
+			return true, xerrors.Errorf("build client failed for id: %s error: %w", s.ID, err)
+		}
+		_, err = client.Namespace(ope.Object.GetNamespace()).Create(ctx, ope.Object, ope.CreateOptions)
+		if err != nil {
+			return true, xerrors.Errorf("run create operation: id: %s error: %w", s.ID, err)
+		}
+	case s.Patch != nil:
+		ope := s.Patch
+		gvk := ope.TypeMeta.GroupVersionKind()
+		client, err := buildClient(gvk, cfg)
+		if err != nil {
+			return true, xerrors.Errorf("build client failed for id: %s error: %w", s.ID, err)
+		}
+		_, err = client.Namespace(ope.ObjectMeta.Namespace).Patch(ctx, ope.ObjectMeta.Name, ope.PatchType, []byte(ope.Patch), ope.PatchOptions)
+		if err != nil {
+			return true, xerrors.Errorf("run patch operation: id: %s error: %w", s.ID, err)
+		}
+	case s.Delete != nil:
+		ope := s.Delete
+		gvk := ope.TypeMeta.GroupVersionKind()
+		client, err := buildClient(gvk, cfg)
+		if err != nil {
+			return true, xerrors.Errorf("build client failed for id: %s error: %w", s.ID, err)
+		}
+		err = client.Namespace(ope.ObjectMeta.Namespace).Delete(ctx, ope.ObjectMeta.Name, ope.DeleteOptions)
+		if err != nil {
+			return true, xerrors.Errorf("run delete operation: id: %s error: %w", s.ID, err)
+		}
+	case s.Done != nil:
+		return true, nil
+	default:
+		return true, ErrUnknownOperation
+	}
+
+	return false, nil
+}
+
+// ValidateCreate implements webhook.Validator so a webhook will be registered for the type.
+func (s *ScenarioOperation) ValidateCreate() error {
+	return s.validateOperations()
+}
+
+// ValidateUpdate implements webhook.Validator so a webhook will be registered for the type.
+func (s *ScenarioOperation) ValidateUpdate(old runtime.Object) error {
+	return s.validateOperations()
+}
+
+// ValidateDelete implements webhook.Validator so a webhook will be registered for the type.
+func (s *ScenarioOperation) ValidateDelete() error {
+	return nil
+}
+
+// validateOperations checks that exactly one operation is set.
+func (s *ScenarioOperation) validateOperations() error {
+	var count int
+	if s.Create != nil {
+		count++
+	}
+	if s.Patch != nil {
+		count++
+	}
+	if s.Delete != nil {
+		count++
+	}
+	if s.Done != nil {
+		count++
+	}
+	if count != 1 {
+		return fmt.Errorf("exactly one operation type must be specified, but found %d", count)
+	}
+	return nil
+}
+
+var ErrUnknownOperation = errors.New("unknown operation")
+
+func buildClient(gvk runtimeschema.GroupVersionKind, cfg *rest.Config) (dynamic.NamespaceableResourceInterface, error) {
+	cli, err := dynamic.NewForConfig(cfg)
+	if err != nil {
+		return nil, xerrors.Errorf("build dynamic client: %w", err)
+	}
+
+	dc, err := discovery.NewDiscoveryClientForConfig(cfg)
+	if err != nil {
+		return nil, xerrors.Errorf("build discovery client: %w", err)
+	}
+	mapper := restmapper.NewDeferredDiscoveryRESTMapper(memory.NewMemCacheClient(dc))
+	mapping, err := mapper.RESTMapping(gvk.GroupKind(), gvk.Version)
+	if err != nil {
+		return nil, xerrors.Errorf("build mapping from RESTMapper: %w", err)
+	}
+
+	return cli.Resource(mapping.Resource), nil
+}

scenario/api/v1alpha1/scenario_operation_run_test.go

@@ -0,0 +1,122 @@
+package v1alpha1
+
+import (
+	"context"
+
+	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/types"
+	"k8s.io/apimachinery/pkg/util/uuid"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+const ns = "default"
+
+func podToUnstructured(p *corev1.Pod) *unstructured.Unstructured {
+	objMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(p)
+	Expect(err).NotTo(HaveOccurred())
+	return &unstructured.Unstructured{Object: objMap}
+}
+
+var _ = Describe("ScenarioOperation.Run()", func() {
+
+	It("executes CreateOperation", func() {
+		pod := &corev1.Pod{
+			TypeMeta:   metav1.TypeMeta{APIVersion: "v1", Kind: "Pod"},
+			ObjectMeta: metav1.ObjectMeta{Name: "run-create-pod", Namespace: ns},
+			Spec: corev1.PodSpec{Containers: []corev1.Container{
+				{Name: "pause", Image: "registry.k8s.io/pause:3.9"},
+			}},
+		}
+
+		op := &ScenarioOperation{
+			ID: "create-" + string(uuid.NewUUID()),
+			Create: &CreateOperation{
+				Object: podToUnstructured(pod),
+			},
+		}
+
+		done, err := op.Run(context.TODO(), cfg)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(done).To(BeFalse())
+
+		Eventually(func() error {
+			return k8sCli.Get(context.TODO(),
+				types.NamespacedName{Name: "run-create-pod", Namespace: ns},
+				&corev1.Pod{})
+		}).Should(Succeed())
+	})
+
+	It("executes PatchOperation", func() {
+		base := &corev1.Pod{
+			TypeMeta:   metav1.TypeMeta{APIVersion: "v1", Kind: "Pod"},
+			ObjectMeta: metav1.ObjectMeta{Name: "run-patch-pod", Namespace: ns},
+			Spec: corev1.PodSpec{Containers: []corev1.Container{
+				{Name: "pause", Image: "registry.k8s.io/pause:3.9"},
+			}},
+		}
+		Expect(k8sCli.Create(context.TODO(), base)).To(Succeed())
+
+		op := &ScenarioOperation{
+			ID: "patch-" + string(uuid.NewUUID()),
+			Patch: &PatchOperation{
+				TypeMeta:   metav1.TypeMeta{APIVersion: "v1", Kind: "Pod"},
+				ObjectMeta: metav1.ObjectMeta{Name: "run-patch-pod", Namespace: ns},
+				PatchType:  types.MergePatchType,
+				Patch:      `{"metadata":{"labels":{"patched":"true"}}}`,
+			},
+		}
+
+		_, err := op.Run(context.TODO(), cfg)
+		Expect(err).NotTo(HaveOccurred())
+
+		Eventually(func() map[string]string {
+			tmp := &corev1.Pod{}
+			_ = k8sCli.Get(context.TODO(),
+				types.NamespacedName{Name: "run-patch-pod", Namespace: ns}, tmp)
+			return tmp.Labels
+		}).Should(HaveKeyWithValue("patched", "true"))
+	})
+
+	It("executes DeleteOperation", func() {
+		p := &corev1.Pod{
+			TypeMeta:   metav1.TypeMeta{APIVersion: "v1", Kind: "Pod"},
+			ObjectMeta: metav1.ObjectMeta{Name: "run-delete-pod", Namespace: ns},
+			Spec: corev1.PodSpec{Containers: []corev1.Container{
+				{Name: "pause", Image: "registry.k8s.io/pause:3.9"},
+			}},
+		}
+		Expect(k8sCli.Create(context.TODO(), p)).To(Succeed())
+
+		op := &ScenarioOperation{
+			ID: "delete-" + string(uuid.NewUUID()),
+			Delete: &DeleteOperation{
+				TypeMeta:   metav1.TypeMeta{APIVersion: "v1", Kind: "Pod"},
+				ObjectMeta: metav1.ObjectMeta{Name: "run-delete-pod", Namespace: ns},
+			},
+		}
+
+		_, err := op.Run(context.TODO(), cfg)
+		Expect(err).NotTo(HaveOccurred())
+
+		Eventually(func() error {
+			return k8sCli.Get(context.TODO(),
+				types.NamespacedName{Name: "run-delete-pod", Namespace: ns},
+				&corev1.Pod{})
+		}).Should(MatchError(ContainSubstring("not found")))
+	})
+
+	It("executes DoneOperation", func() {
+		op := &ScenarioOperation{
+			ID:   "done-" + string(uuid.NewUUID()),
+			Done: &DoneOperation{},
+		}
+		done, err := op.Run(context.TODO(), cfg)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(done).To(BeTrue())
+	})
+})