Fusion + Nats Complement

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

tools/enums.go

@@ -171,6 +171,7 @@ const (
 	PB_CONSIDERS
 	PB_ADMIRALTY_CONFIG
 	PB_MINIO_CONFIG
+	PB_PVC_CONFIG
 	PB_CLOSE_SEARCH
 	NONE
 )

tools/kubernetes.go

@@ -14,6 +14,7 @@ import (
 	v1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -598,6 +599,38 @@ func (k *KubernetesService) CreateSecret(context context.Context, minioId string
 	return nil
 }
 
+// CreatePVC creates a PersistentVolumeClaim in the given namespace.
+func (k *KubernetesService) CreatePVC(ctx context.Context, name, namespace, storageSize string) error {
+	pvc := &v1.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: namespace,
+		},
+		Spec: v1.PersistentVolumeClaimSpec{
+			AccessModes: []v1.PersistentVolumeAccessMode{v1.ReadWriteOnce},
+			Resources: v1.VolumeResourceRequirements{
+				Requests: v1.ResourceList{
+					v1.ResourceStorage: resource.MustParse(storageSize),
+				},
+			},
+		},
+	}
+	_, err := k.Set.CoreV1().PersistentVolumeClaims(namespace).Create(ctx, pvc, metav1.CreateOptions{})
+	if err != nil && !apierrors.IsAlreadyExists(err) {
+		return fmt.Errorf("CreatePVC %s/%s: %w", namespace, name, err)
+	}
+	return nil
+}
+
+// DeletePVC deletes a PersistentVolumeClaim from the given namespace.
+func (k *KubernetesService) DeletePVC(ctx context.Context, name, namespace string) error {
+	err := k.Set.CoreV1().PersistentVolumeClaims(namespace).Delete(ctx, name, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("DeletePVC %s/%s: %w", namespace, name, err)
+	}
+	return nil
+}
+
 // ============== ADMIRALTY ==============
 // Returns a concatenation of the peerId and namespace in order for
 // kubernetes ressources to have a unique name, under 63 characters

tools/nats_caller.go

@@ -29,7 +29,7 @@ type NATSMethod int
 var meths = []string{"remove execution", "create execution", "planner execution", "discovery",
 	"workflow event", "argo kube event", "create resource", "remove resource",
 	"propalgation event", "search event", "confirm event",
-	"considers event", "admiralty config event", "minio config event",
+	"considers event", "admiralty config event", "minio config event", "pvc config event",
 	"workflow started event", "workflow step done event", "workflow done event",
 }
 
@@ -53,6 +53,7 @@ const (
 	CONSIDERS_EVENT
 	ADMIRALTY_CONFIG_EVENT
 	MINIO_CONFIG_EVENT
+	PVC_CONFIG_EVENT
 
 	// Workflow lifecycle events emitted by oc-monitord.
 	// oc-scheduler listens to STARTED and DONE to maintain WorkflowExecution state.
@@ -70,7 +71,7 @@ func (n NATSMethod) String() string {
 func NameToMethod(name string) NATSMethod {
 	for _, v := range [...]NATSMethod{REMOVE_EXECUTION, CREATE_EXECUTION, PLANNER_EXECUTION, DISCOVERY, WORKFLOW_EVENT, ARGO_KUBE_EVENT,
 		CREATE_RESOURCE, REMOVE_RESOURCE, PROPALGATION_EVENT, SEARCH_EVENT, CONFIRM_EVENT,
-		CONSIDERS_EVENT, ADMIRALTY_CONFIG_EVENT, MINIO_CONFIG_EVENT,
+		CONSIDERS_EVENT, ADMIRALTY_CONFIG_EVENT, MINIO_CONFIG_EVENT, PVC_CONFIG_EVENT,
 		WORKFLOW_STARTED_EVENT, WORKFLOW_STEP_DONE_EVENT, WORKFLOW_DONE_EVENT} {
 		if strings.Contains(strings.ToLower(v.String()), strings.ToLower(name)) {
 			return v