feat: kubernetes pvc support (#539)

Allow creating PVC, so the kube jobs can use PVCs as a cache across multiple jobs.

drivers/cmd/kubernetes/main.go

@@ -1,4 +1,4 @@
-// Copyright 2022 PayPal Inc.
+// Copyright 2023 PayPal Inc.
 
 // This Source Code Form is subject to the terms of the MIT License.
 // If a copy of the MIT License was not distributed with this file,
@@ -26,7 +26,6 @@ import (
 	"github.com/op/go-logging"
 	appsv1 "k8s.io/api/apps/v1"
 	batchv1 "k8s.io/api/batch/v1"
-	batchv1beta1 "k8s.io/api/batch/v1beta1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -79,6 +78,8 @@ func main() {
 	driver.Commands["waitForJob"] = waitForJob
 	driver.Commands["getJobLog"] = getJobLog
 	driver.Commands["deleteJob"] = deleteJob
+	driver.Commands["createPVC"] = createPVC
+	driver.Commands["deletePVC"] = deletePVC
 	driver.Reload = func(*dipper.Message) {}
 	driver.Run()
 }
@@ -370,10 +371,10 @@ func constructJob(m *dipper.Message, namespace string, client *kubernetes.Client
 
 		switch {
 		case errors.IsNotFound(err):
-			v1beta1client := client.BatchV1beta1().CronJobs(cronJobNamespace)
+			v1beta1client := client.BatchV1().CronJobs(cronJobNamespace)
 			ctx2, cancel2 := context.WithTimeout(context.Background(), driver.APITimeout*time.Second)
 			defer cancel2()
-			cronJobv1beta1 := dipper.Must(v1beta1client.Get(ctx2, cronJobName, metav1.GetOptions{})).(*batchv1beta1.CronJob)
+			cronJobv1beta1 := dipper.Must(v1beta1client.Get(ctx2, cronJobName, metav1.GetOptions{})).(*batchv1.CronJob)
 
 			job.Spec = cronJobv1beta1.Spec.JobTemplate.Spec
 		case err != nil:

drivers/cmd/kubernetes/pv.go

@@ -0,0 +1,69 @@
+// Copyright 2023 PayPal Inc.
+
+// This Source Code Form is subject to the terms of the MIT License.
+// If a copy of the MIT License was not distributed with this file,
+// you can obtain one at https://mit-license.org/.
+
+// Package honeydipper is an event-driven, rule based orchestration platform tailor towards
+// DevOps and system engineering workflows.
+package main
+
+import (
+	"github.com/ghodss/yaml"
+	"github.com/honeydipper/honeydipper/pkg/dipper"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func createPVC(m *dipper.Message) {
+	k8sclient := prepareKubeConfig(m)
+	nameSpace, ok := dipper.GetMapDataStr(m.Payload, "namespace")
+	if !ok {
+		nameSpace = DefaultNamespace
+	}
+
+	pvc := corev1.PersistentVolumeClaim{}
+	source := dipper.MustGetMapData(m.Payload, "pvc")
+	buf, err := yaml.Marshal(source)
+	if err != nil {
+		log.Panicf("[%s] unable to marshal pvc manifest %+v", driver.Service, err)
+	}
+	err = yaml.Unmarshal(buf, &pvc)
+	if err != nil {
+		log.Panicf("[%s] invalid pvc manifest %+v", driver.Service, err)
+	}
+
+	client := k8sclient.CoreV1().PersistentVolumeClaims(nameSpace)
+	ctx, cancel := driver.GetContext()
+	defer cancel()
+	pvcResult, e := client.Create(ctx, &pvc, metav1.CreateOptions{})
+	if e != nil {
+		log.Panicf("[%s] failed to create pvc %+v", driver.Service, e)
+	}
+
+	m.Reply <- dipper.Message{
+		Payload: map[string]interface{}{
+			"metadata": pvcResult.ObjectMeta,
+			"status":   pvcResult.Status,
+		},
+	}
+}
+
+func deletePVC(m *dipper.Message) {
+	k8sclient := prepareKubeConfig(m)
+	nameSpace, ok := dipper.GetMapDataStr(m.Payload, "namespace")
+	if !ok {
+		nameSpace = DefaultNamespace
+	}
+	pvcName := dipper.MustGetMapDataStr(m.Payload, "pvc")
+
+	client := k8sclient.CoreV1().PersistentVolumeClaims(nameSpace)
+	ctx, cancel := driver.GetContext()
+	defer cancel()
+	e := client.Delete(ctx, pvcName, metav1.DeleteOptions{})
+	if e != nil {
+		log.Panicf("[%s] failed to delete pvc %+v", driver.Service, e)
+	}
+
+	m.Reply <- dipper.Message{}
+}