Adding tests

api/v1/coherence_types.go

@@ -2952,6 +2952,12 @@ type CoherenceJob struct {
 	// +optional
 	Completions *int32 `json:"completions,omitempty"`
 
+	// SyncCompletions is a flag to indicate that the Operator should always set the
+	// Completions value to be the same as the Replicas value.
+	// When a Job is then scaled, the Completions value will also be changed.
+	// +optional
+	SyncCompletionsToReplicas *bool `json:"syncCompletionsToReplicas,omitempty"`
+
 	// Specifies the policy of handling failed pods. In particular, it allows to
 	// specify the set of actions and conditions which need to be
 	// satisfied to take the associated action.
@@ -3018,19 +3024,32 @@ type CoherenceJob struct {
 }
 
 // UpdateJob updates a JobSpec from the fields in this CoherenceJob
-func (in *CoherenceJob) UpdateJob(spec *batchv1.JobSpec) {
+func (in *CoherenceJob) UpdateJob(spec *batchv1.JobSpec, c *CoherenceResourceSpec) {
 	if in == nil {
 		return
 	}
 
-	spec.Completions = in.Completions
+	if in.IsSyncCompletions() {
+		spec.Completions = pointer.Int32(c.GetReplicas())
+	} else {
+		spec.Completions = in.Completions
+	}
+
 	spec.PodFailurePolicy = in.PodFailurePolicy
 	spec.BackoffLimit = in.BackoffLimit
 	spec.TTLSecondsAfterFinished = in.TTLSecondsAfterFinished
 	spec.CompletionMode = in.CompletionMode
 	spec.Suspend = in.Suspend
 }
 
+// IsSyncCompletions returns true if Completions should always match Parallelism
+func (in *CoherenceJob) IsSyncCompletions() bool {
+	if in == nil {
+		return false
+	}
+	return in.SyncCompletionsToReplicas != nil && *in.SyncCompletionsToReplicas
+}
+
 // ----- helper methods -----------------------------------------------------
 
 // Int32PtrToStringWithDefault converts an int32 pointer to a string using the default if the pointer is nil.

api/v1/coherence_webhook.go

@@ -12,6 +12,7 @@ import (
 	"github.com/oracle/coherence-operator/pkg/operator"
 	appsv1 "k8s.io/api/apps/v1"
 	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
 	apiequality "k8s.io/apimachinery/pkg/api/equality"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/intstr"
@@ -46,6 +47,14 @@ var _ webhook.Defaulter = &Coherence{}
 
 // Default implements webhook.Defaulter so a webhook will be registered for the type
 func (in *Coherence) Default() {
+	if in.IsRunAsJob() {
+		in.SetJobDefaults()
+	}
+	in.SetCommonDefaults()
+}
+
+// SetCommonDefaults sets defaults common to both a Job and StatefulSet
+func (in *Coherence) SetCommonDefaults() {
 	logger := webhookLogger.WithValues("namespace", in.Namespace, "name", in.Name)
 	if in.Status.Phase == "" {
 		logger.Info("Setting defaults for new resource")
@@ -63,7 +72,7 @@ func (in *Coherence) Default() {
 
 		// set the default replicas if not present
 		if in.Spec.Replicas == nil {
-			in.Spec.SetReplicas(3)
+			in.Spec.SetReplicas(in.Spec.GetReplicas())
 		}
 
 		// set the default Coherence local port and local port adjust if not present
@@ -106,6 +115,25 @@ func (in *Coherence) Default() {
 	}
 }
 
+// SetJobDefaults sets defaults for Jobs
+func (in *Coherence) SetJobDefaults() {
+	coherenceSpec := in.Spec.Coherence
+	if in.Spec.Coherence == nil {
+		coherenceSpec = &CoherenceSpec{}
+		in.Spec.Coherence = coherenceSpec
+	}
+
+	// default to storage disabled to false
+	if coherenceSpec.StorageEnabled == nil {
+		coherenceSpec.StorageEnabled = pointer.Bool(false)
+	}
+
+	// default the restart policy to never
+	if in.Spec.RestartPolicy == nil {
+		in.Spec.RestartPolicy = in.Spec.RestartPolicyPointer(corev1.RestartPolicyNever)
+	}
+}
+
 func (in *Coherence) AddAnnotation(key, value string) {
 	if in != nil {
 		if in.Annotations == nil {
@@ -134,6 +162,12 @@ func (in *Coherence) ValidateCreate() error {
 		return err
 	}
 	err = in.validateNodePorts()
+	if in.IsRunAsJob() {
+		errorList := in.validateJob()
+		if len(errorList) > 0 {
+			err = fmt.Errorf("rejecting update as it would have resulted in an invalid Job: %v", errorList)
+		}
+	}
 	return err
 }
 
@@ -163,17 +197,20 @@ func (in *Coherence) ValidateUpdate(previous runtime.Object) error {
 	}
 
 	var errorList field.ErrorList
+	var deploymentType string
 
 	if in.IsRunAsJob() {
 		errorList = in.validateJob()
+		deploymentType = "Job"
 	} else {
+		deploymentType = "StatefulSet"
 		sts := in.Spec.CreateStatefulSet(in)
 		stsOld := prev.Spec.CreateStatefulSet(prev)
 		errorList = ValidateStatefulSetUpdate(&sts, &stsOld)
 	}
 
 	if len(errorList) > 0 {
-		return fmt.Errorf("rejecting update as it would have resulted in an invalid statefuleset: %v", errorList)
+		return fmt.Errorf("rejecting update as it would have resulted in an invalid %s: %v", deploymentType, errorList)
 	}
 
 	return nil

api/v1/coherenceresource_types.go

@@ -10,6 +10,7 @@ import (
 	"fmt"
 	"golang.org/x/mod/semver"
 	appsv1 "k8s.io/api/apps/v1"
+	batchv1 "k8s.io/api/batch/v1"
 	coreV1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -401,6 +402,51 @@ func (in *CoherenceResourceStatus) Update(deployment *Coherence, sts *appsv1.Sta
 	return updated
 }
 
+// UpdateFromJob the status based on the condition of the Job status.
+func (in *CoherenceResourceStatus) UpdateFromJob(deployment *Coherence, jobStatus *batchv1.JobStatus) bool {
+	// ensure that there is an Initialized condition
+	updated := in.ensureInitialized(deployment)
+
+	if jobStatus != nil {
+		count := jobStatus.Active + jobStatus.Succeeded
+		// update CurrentReplicas from Job if required
+		if in.CurrentReplicas != count {
+			in.CurrentReplicas = count
+			updated = true
+		}
+
+		// update ReadyReplicas from Job if required
+		if jobStatus.Ready != nil && in.ReadyReplicas != *jobStatus.Ready {
+			in.ReadyReplicas = *jobStatus.Ready
+			updated = true
+		}
+
+		if in.Phase != ConditionTypeReady && in.Replicas == in.ReadyReplicas && in.Replicas == in.CurrentReplicas {
+			updated = in.setPhase(ConditionTypeReady)
+		}
+	} else {
+		// update CurrentReplicas to zero
+		if in.CurrentReplicas != 0 {
+			in.CurrentReplicas = 0
+			updated = true
+		}
+		// update ReadyReplicas to zero
+		if in.ReadyReplicas != 0 {
+			in.ReadyReplicas = 0
+			updated = true
+		}
+	}
+
+	if deployment.Spec.GetReplicas() == 0 {
+		// scaled to zero
+		if in.Phase != ConditionTypeStopped {
+			updated = in.setPhase(ConditionTypeStopped)
+		}
+	}
+
+	return updated
+}
+
 // set a status phase.
 func (in *CoherenceResourceStatus) setPhase(phase ConditionType) bool {
 	if in.Phase == phase {

api/v1/coherenceresourcespec_types.go

@@ -418,6 +418,9 @@ func (in *CoherenceResourceSpec) GetReplicas() int32 {
 		return 0
 	}
 	if in.Replicas == nil {
+		if in.IsRunAsJob() {
+			return DefaultJobReplicas
+		}
 		return DefaultReplicas
 	}
 	return *in.Replicas
@@ -435,6 +438,23 @@ func (in *CoherenceResourceSpec) IsRunAsJob() bool {
 	return in != nil && in.RunAsJob != nil && *in.RunAsJob
 }
 
+// GetRestartPolicy returns the name of the application image to use
+func (in *CoherenceResourceSpec) GetRestartPolicy() *corev1.RestartPolicy {
+	if in == nil {
+		return nil
+	}
+	if in.IsRunAsJob() {
+		if in.RestartPolicy == nil {
+			return in.RestartPolicyPointer(corev1.RestartPolicyNever)
+		}
+	}
+	return in.RestartPolicy
+}
+
+func (in *CoherenceResourceSpec) RestartPolicyPointer(policy corev1.RestartPolicy) *corev1.RestartPolicy {
+	return &policy
+}
+
 // GetCoherenceImage returns the name of the application image to use
 func (in *CoherenceResourceSpec) GetCoherenceImage() *string {
 	if in != nil {
@@ -846,15 +866,12 @@ func (in *CoherenceResourceSpec) CreateJob(deployment *Coherence) batchv1.Job {
 
 	job.Spec = batchv1.JobSpec{
 		Parallelism: &replicas,
-		Selector: &metav1.LabelSelector{
-			MatchLabels: in.CreatePodSelectorLabels(deployment),
-		},
-		Template: podTemplate,
+		Template:    podTemplate,
 	}
 
 	job.Spec.ActiveDeadlineSeconds = in.ActiveDeadlineSeconds
 
-	in.JobSpec.UpdateJob(&job.Spec)
+	in.JobSpec.UpdateJob(&job.Spec, in)
 
 	return job
 }
@@ -927,6 +944,8 @@ func (in *CoherenceResourceSpec) CreatePodTemplateSpec(deployment *Coherence) co
 
 	if in.RestartPolicy != nil {
 		podTemplate.Spec.RestartPolicy = *in.RestartPolicy
+	} else if in.IsRunAsJob() {
+		podTemplate.Spec.RestartPolicy = corev1.RestartPolicyNever
 	}
 
 	// Add any ConfigMap Volumes

api/v1/constants.go

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2020, 2022, Oracle and/or its affiliates.
+ * Copyright (c) 2020, 2023, Oracle and/or its affiliates.
  * Licensed under the Universal Permissive License v 1.0 as shown at
  * http://oss.oracle.com/licenses/upl.
  */
@@ -11,6 +11,8 @@ import "github.com/oracle/coherence-operator/pkg/operator"
 const (
 	// DefaultReplicas is the default number of replicas that will be created for a deployment if no value is specified in the spec
 	DefaultReplicas int32 = 3
+	// DefaultJobReplicas is the default number of replicas that will be created for a Job deployment if no value is specified in the spec
+	DefaultJobReplicas int32 = 1
 	// WKAServiceNameSuffix is the suffix appended to a deployment name to give the WKA service name
 	WKAServiceNameSuffix = "-wka"
 	// HeadlessServiceNameSuffix is the suffix appended to a deployment name to give the StatefulSet headless-service name

api/v1/create_job_test.go

@@ -21,6 +21,91 @@ func TestCreateJobFromMinimalRoleSpec(t *testing.T) {
 	deployment := createTestDeployment(spec)
 	// Create expected Job
 	expected := createMinimalExpectedJob(deployment)
-	// assert that the StatefulSet is as expected
+	// assert that the Job is as expected
+	assertJobCreation(t, deployment, expected)
+}
+
+func TestCreateJobWithReplicas(t *testing.T) {
+	// Create minimal spec spec
+	spec := coh.CoherenceResourceSpec{
+		Replicas: pointer.Int32(19),
+		RunAsJob: pointer.Bool(true),
+	}
+
+	// Create the test deployment
+	deployment := createTestDeployment(spec)
+
+	// Create expected Job
+	expected := createMinimalExpectedJob(deployment)
+	expected.Spec.Parallelism = pointer.Int32(19)
+
+	// assert that the Job is as expected
+	assertJobCreation(t, deployment, expected)
+}
+
+func TestCreateJobWithReplicasAndCompletions(t *testing.T) {
+	// Create minimal spec spec
+	spec := coh.CoherenceResourceSpec{
+		Replicas: pointer.Int32(19),
+		RunAsJob: pointer.Bool(true),
+		JobSpec: &coh.CoherenceJob{
+			Completions: pointer.Int32(21),
+		},
+	}
+
+	// Create the test deployment
+	deployment := createTestDeployment(spec)
+
+	// Create expected Job
+	expected := createMinimalExpectedJob(deployment)
+	expected.Spec.Parallelism = pointer.Int32(19)
+	expected.Spec.Completions = pointer.Int32(21)
+
+	// assert that the Job is as expected
+	assertJobCreation(t, deployment, expected)
+}
+
+func TestCreateJobWithReplicasAndSyncedCompletions(t *testing.T) {
+	// Create minimal spec spec
+	spec := coh.CoherenceResourceSpec{
+		Replicas: pointer.Int32(19),
+		RunAsJob: pointer.Bool(true),
+		JobSpec: &coh.CoherenceJob{
+			SyncCompletionsToReplicas: pointer.Bool(true),
+		},
+	}
+
+	// Create the test deployment
+	deployment := createTestDeployment(spec)
+
+	// Create expected Job
+	expected := createMinimalExpectedJob(deployment)
+	expected.Spec.Parallelism = pointer.Int32(19)
+	expected.Spec.Completions = pointer.Int32(19)
+
+	// assert that the Job is as expected
+	assertJobCreation(t, deployment, expected)
+}
+
+func TestCreateJobWithReplicasAndSyncedCompletionsOverride(t *testing.T) {
+	// Create minimal spec spec
+	spec := coh.CoherenceResourceSpec{
+		Replicas: pointer.Int32(19),
+		RunAsJob: pointer.Bool(true),
+		JobSpec: &coh.CoherenceJob{
+			Completions:               pointer.Int32(21),
+			SyncCompletionsToReplicas: pointer.Bool(true),
+		},
+	}
+
+	// Create the test deployment
+	deployment := createTestDeployment(spec)
+
+	// Create expected Job
+	expected := createMinimalExpectedJob(deployment)
+	expected.Spec.Parallelism = pointer.Int32(19)
+	expected.Spec.Completions = pointer.Int32(19)
+
+	// assert that the Job is as expected
 	assertJobCreation(t, deployment, expected)
 }