Keda podspec (#2021)

* Add podspec to MQT

Signed-off-by: Harsh Thakur <harshthakur9030@gmail.com>

* Add preupgrade check

Signed-off-by: Harsh Thakur <harshthakur9030@gmail.com>

* Use zap logger

Signed-off-by: Harsh Thakur <harshthakur9030@gmail.com>

Co-authored-by: Sanket Sudake <sanketsudake@gmail.com>

cmd/preupgradechecks/preupgradechecks.go

@@ -48,6 +48,7 @@ type (
 const (
 	maxRetries  = 5
 	FunctionCRD = "functions.fission.io"
+	MqtCRD      = "messagequeuetriggers.fission.io"
 )
 
 func makePreUpgradeTaskClient(logger *zap.Logger, fnPodNs, envBuilderNs string) (*PreUpgradeTaskClient, error) {
@@ -79,16 +80,36 @@ func (client *PreUpgradeTaskClient) GetFunctionCRD() *v1.CustomResourceDefinitio
 	return nil
 }
 
+// GetMqtCRD checks if MQT CRD is present on the cluster and returns it. It returns nil if not found
+func (client *PreUpgradeTaskClient) GetMqtCRD() *v1.CustomResourceDefinition {
+	crd, err := client.apiExtClient.ApiextensionsV1().CustomResourceDefinitions().Get(context.TODO(), MqtCRD, metav1.GetOptions{})
+	if err != nil {
+		client.logger.Error("Could not find MQT CRD", zap.Error(err))
+		return nil
+	}
+	return crd
+}
+
 // LatestSchemaApplied ensures that the end user has applied the latest CRDs generated to the cluster.
 // For future reference: whenever a new field is added, we need to check for that field's existence in this function
 func (client *PreUpgradeTaskClient) LatestSchemaApplied() error {
 	client.logger.Info("Checking if user has applied the latest CRDs")
-	crd := client.GetFunctionCRD()
+	funcCRD := client.GetFunctionCRD()
-	if crd == nil {
+	if funcCRD == nil {
 		return errors.New("Could not get the Function CRD")
 	}
 	// Any new field added in Function spec can be checked here provided the substring matches the description in CRD Validation of the field
-	if !strings.Contains(crd.Spec.String(), "RequestsPerPod") || !strings.Contains(crd.Spec.String(), "OnceOnly") {
+	if !strings.Contains(funcCRD.Spec.String(), "RequestsPerPod") || !strings.Contains(funcCRD.Spec.String(), "OnceOnly") {
+		return errors.New("Apply the newer CRDs before upgrading")
+	}
+
+	mqtCRD := client.GetMqtCRD()
+	if mqtCRD == nil {
+		return errors.New("Could not get the MQT CRD")
+	}
+
+	// Any new field added in MQT spec can be checked here provided the substring matches the description in CRD Validation of the field
+	if !strings.Contains(mqtCRD.Spec.String(), "PodSpec") {
 		return errors.New("Apply the newer CRDs before upgrading")
 	}
 	return nil

crds/v1/fission.io_httptriggers.yaml

@@ -51,7 +51,6 @@ spec:
                     description: 'Type indicates whether this function reference is by name or selector. For now, the only supported reference type is by "name".  Future reference types:   * Function by label or annotation   * Branch or tag of a versioned function   * A "rolling upgrade" from one version of a function to another Available value: - name - function-weights'
                     type: string
                 required:
-                - functionweights
                 - name
                 - type
                 type: object

crds/v1/fission.io_kuberneteswatchtriggers.yaml

@@ -48,7 +48,6 @@ spec:
                     description: 'Type indicates whether this function reference is by name or selector. For now, the only supported reference type is by "name".  Future reference types:   * Function by label or annotation   * Branch or tag of a versioned function   * A "rolling upgrade" from one version of a function to another Available value: - name - function-weights'
                     type: string
                 required:
-                - functionweights
                 - name
                 - type
                 type: object

crds/v1/fission.io_timetriggers.yaml

@@ -51,7 +51,6 @@ spec:
                     description: 'Type indicates whether this function reference is by name or selector. For now, the only supported reference type is by "name".  Future reference types:   * Function by label or annotation   * Branch or tag of a versioned function   * A "rolling upgrade" from one version of a function to another Available value: - name - function-weights'
                     type: string
                 required:
-                - functionweights
                 - name
                 - type
                 type: object

go.sum

@@ -1108,7 +1108,6 @@ gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLks
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f h1:BLraFXnmrev5lT+xlilqcH8XK9/i0At2xKjWk4p6zsU=
 gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/cheggaaa/pb.v1 v1.0.25/go.mod h1:V/YB90LKu/1FcN3WVnfiiE5oMCibMjukxqG/qStrOgw=

pkg/apis/core/v1/types.go

@@ -490,6 +490,7 @@ type (
 		// Function Reference by weight. this map contains function name as key and its weight
 		// as the value. This is for canary upgrade purpose.
 		// +nullable
+		// +optional
 		FunctionWeights map[string]int `json:"functionweights"`
 	}
 
@@ -768,6 +769,16 @@ type (
 		// Kind of Message Queue Trigger to be created, by default its fission
 		// +optional
 		MqtKind string `json:"mqtkind,omitempty"`
+
+		// (Optional) Podspec allows modification of deployed runtime pod with Kubernetes PodSpec
+		// The merging logic is briefly described below and detailed MergePodSpec function
+		// - Volumes mounts and env variables for function and fetcher container are appended
+		// - All additional containers and init containers are appended
+		// - Volume definitions are appended
+		// - Lists such as tolerations, ImagePullSecrets, HostAliases are appended
+		// - Structs are merged and variables from pod spec take precedence
+		// +optional
+		PodSpec *apiv1.PodSpec `json:"podspec,omitempty"`
 	}
 
 	// TimeTriggerSpec invokes the specific function at a time or

pkg/apis/core/v1/zz_generated.deepcopy.go

@@ -737,6 +737,11 @@ func (in *MessageQueueTriggerSpec) DeepCopyInto(out *MessageQueueTriggerSpec) {
 			(*out)[key] = val
 		}
 	}
+	if in.PodSpec != nil {
+		in, out := &in.PodSpec, &out.PodSpec
+		*out = new(corev1.PodSpec)
+		(*in).DeepCopyInto(*out)
+	}
 	return
 }
 

pkg/mqtrigger/scalermanager.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"reflect"
 	"regexp"
 	"strconv"
 	"strings"
@@ -23,6 +24,7 @@ import (
 
 	fv1 "github.com/fission/fission/pkg/apis/core/v1"
 	"github.com/fission/fission/pkg/crd"
+	"github.com/fission/fission/pkg/executor/util"
 	"github.com/fission/fission/pkg/utils"
 )
 
@@ -265,7 +267,12 @@ func checkAndUpdateTriggerFields(mqt, newMqt *fv1.MessageQueueTrigger) bool {
 		updated = true
 	}
 	if len(newMqt.Spec.FunctionReference.Name) > 0 && newMqt.Spec.FunctionReference.Name != mqt.Spec.FunctionReference.Name {
-		newMqt.Spec.FunctionReference.Name = mqt.Spec.FunctionReference.Name
+		mqt.Spec.FunctionReference.Name = newMqt.Spec.FunctionReference.Name
+		updated = true
+	}
+
+	if !reflect.DeepEqual(newMqt.Spec.PodSpec, mqt.Spec.PodSpec) {
+		mqt.Spec.PodSpec = newMqt.Spec.PodSpec
 		updated = true
 	}
 
@@ -395,6 +402,21 @@ func getDeploymentSpec(mqt *fv1.MessageQueueTrigger, routerURL string, kubeClien
 	image := os.Getenv(strings.ToUpper(imageName))
 	imagePullPolicy := utils.GetImagePullPolicy(os.Getenv("CONNECTOR_IMAGE_PULL_POLICY"))
 
+	podSpec := &apiv1.PodSpec{
+		Containers: []apiv1.Container{
+			{
+				Name:            mqt.ObjectMeta.Name,
+				Image:           image,
+				ImagePullPolicy: imagePullPolicy,
+				Env:             envVars,
+			},
+		},
+	}
+	podSpec, err = util.MergePodSpec(podSpec, mqt.Spec.PodSpec)
+	if err != nil {
+		return nil, err
+	}
+
 	blockOwnerDeletion := true
 	return &appsv1.Deployment{
 		ObjectMeta: metav1.ObjectMeta{
@@ -424,16 +446,7 @@ func getDeploymentSpec(mqt *fv1.MessageQueueTrigger, routerURL string, kubeClien
 						"app": mqt.ObjectMeta.Name,
 					},
 				},
-				Spec: apiv1.PodSpec{
+				Spec: *podSpec,
-					Containers: []apiv1.Container{
-						{
-							Name:            mqt.ObjectMeta.Name,
-							Image:           image,
-							ImagePullPolicy: imagePullPolicy,
-							Env:             envVars,
-						},
-					},
-				},
 			},
 		},
 	}, nil