Add jobs flag to spec on pgupgrade to determine number of pgupgrade jobs

internal/controller/pgupgrade/jobs.go

@@ -43,6 +43,12 @@ func upgradeCommand(upgrade *v1beta1.PGUpgrade, fetchKeyCommand string) []string
 		initdb += ` --encryption-key-command "` + fetchKeyCommand + `"`
 	}
 
+	// Determine the number of jobs for pg_upgrade
+	var jobsFlag string
+	if upgrade.Spec.Jobs > 0 {
+		jobsFlag = fmt.Sprintf(" --jobs=%d", upgrade.Spec.Jobs)
+	}
+
 	args := []string{oldVersion, newVersion}
 	script := strings.Join([]string{
 		`declare -r data_volume='/pgdata' old_version="$1" new_version="$2"`,
@@ -104,7 +110,7 @@ func upgradeCommand(upgrade *v1beta1.PGUpgrade, fetchKeyCommand string) []string
 		`echo -e "\nStep 6: Running pg_upgrade...\n"`,
 		`time /usr/pgsql-"${new_version}"/bin/pg_upgrade --old-bindir /usr/pgsql-"${old_version}"/bin \`,
 		`--new-bindir /usr/pgsql-"${new_version}"/bin --old-datadir /pgdata/pg"${old_version}" \`,
-		`--new-datadir /pgdata/pg"${new_version}" --link`,
+		`--new-datadir /pgdata/pg"${new_version}" --link` + jobsFlag, // Add jobs flag here if it's defined
 
 		// Since we have cleared the Patroni cluster step by removing the EndPoints, we copy patroni.dynamic.json
 		// from the old data dir to help retain PostgreSQL parameters you had set before.

internal/controller/pgupgrade/jobs_test.go

@@ -155,6 +155,141 @@ status: {}
 		`/usr/pgsql-"${new_version}"/bin/initdb -k -D /pgdata/pg"${new_version}" --encryption-key-command "echo testKey"`))
 }
 
+func TestGenerateUpgradeJobWithJobs(t *testing.T) {
+	ctx := context.Background()
+	reconciler := &PGUpgradeReconciler{}
+
+	upgrade := &v1beta1.PGUpgrade{}
+	upgrade.Namespace = "ns1"
+	upgrade.Name = "pgu2"
+	upgrade.UID = "uid3"
+	upgrade.Spec.Image = initialize.Pointer("img4")
+	upgrade.Spec.PostgresClusterName = "pg5"
+	upgrade.Spec.FromPostgresVersion = 19
+	upgrade.Spec.ToPostgresVersion = 25
+	upgrade.Spec.Resources.Requests = corev1.ResourceList{
+		corev1.ResourceCPU: resource.MustParse("3.14"),
+	}
+	upgrade.Spec.Jobs = 1
+
+	startup := &appsv1.StatefulSet{}
+	startup.Spec.Template.Spec = corev1.PodSpec{
+		Containers: []corev1.Container{{
+			Name: ContainerDatabase,
+
+			SecurityContext: &corev1.SecurityContext{Privileged: new(bool)},
+			VolumeMounts: []corev1.VolumeMount{
+				{Name: "vm1", MountPath: "/mnt/some/such"},
+			},
+		}},
+		Volumes: []corev1.Volume{
+			{
+				Name: "vol2",
+				VolumeSource: corev1.VolumeSource{
+					HostPath: new(corev1.HostPathVolumeSource),
+				},
+			},
+		},
+	}
+
+	job := reconciler.generateUpgradeJob(ctx, upgrade, startup, "")
+	assert.Assert(t, cmp.MarshalMatches(job, `
+apiVersion: batch/v1
+kind: Job
+metadata:
+  annotations:
+    kubectl.kubernetes.io/default-container: database
+  creationTimestamp: null
+  labels:
+    postgres-operator.crunchydata.com/cluster: pg5
+    postgres-operator.crunchydata.com/pgupgrade: pgu2
+    postgres-operator.crunchydata.com/role: pgupgrade
+    postgres-operator.crunchydata.com/version: "25"
+  name: pgu2-pgdata
+  namespace: ns1
+  ownerReferences:
+  - apiVersion: postgres-operator.crunchydata.com/v1beta1
+    blockOwnerDeletion: true
+    controller: true
+    kind: PGUpgrade
+    name: pgu2
+    uid: uid3
+spec:
+  backoffLimit: 0
+  template:
+    metadata:
+      annotations:
+        kubectl.kubernetes.io/default-container: database
+      creationTimestamp: null
+      labels:
+        postgres-operator.crunchydata.com/cluster: pg5
+        postgres-operator.crunchydata.com/pgupgrade: pgu2
+        postgres-operator.crunchydata.com/role: pgupgrade
+        postgres-operator.crunchydata.com/version: "25"
+    spec:
+      containers:
+      - command:
+        - bash
+        - -ceu
+        - --
+        - |-
+          declare -r data_volume='/pgdata' old_version="$1" new_version="$2"
+          printf 'Performing PostgreSQL upgrade from version "%s" to "%s" ...\n\n' "$@"
+          gid=$(id -G); NSS_WRAPPER_GROUP=$(mktemp)
+          (sed "/^postgres:x:/ d; /^[^:]*:x:${gid%% *}:/ d" /etc/group
+          echo "postgres:x:${gid%% *}:") > "${NSS_WRAPPER_GROUP}"
+          uid=$(id -u); NSS_WRAPPER_PASSWD=$(mktemp)
+          (sed "/^postgres:x:/ d; /^[^:]*:x:${uid}:/ d" /etc/passwd
+          echo "postgres:x:${uid}:${gid%% *}::${data_volume}:") > "${NSS_WRAPPER_PASSWD}"
+          export LD_PRELOAD='libnss_wrapper.so' NSS_WRAPPER_GROUP NSS_WRAPPER_PASSWD
+          cd /pgdata || exit
+          echo -e "Step 1: Making new pgdata directory...\n"
+          mkdir /pgdata/pg"${new_version}"
+          echo -e "Step 2: Initializing new pgdata directory...\n"
+          /usr/pgsql-"${new_version}"/bin/initdb -k -D /pgdata/pg"${new_version}"
+          echo -e "\nStep 3: Setting the expected permissions on the old pgdata directory...\n"
+          chmod 700 /pgdata/pg"${old_version}"
+          echo -e "Step 4: Copying shared_preload_libraries setting to new postgresql.conf file...\n"
+          echo "shared_preload_libraries = '$(/usr/pgsql-"""${old_version}"""/bin/postgres -D \
+          /pgdata/pg"""${old_version}""" -C shared_preload_libraries)'" >> /pgdata/pg"${new_version}"/postgresql.conf
+          echo -e "Step 5: Running pg_upgrade check...\n"
+          time /usr/pgsql-"${new_version}"/bin/pg_upgrade --old-bindir /usr/pgsql-"${old_version}"/bin \
+          --new-bindir /usr/pgsql-"${new_version}"/bin --old-datadir /pgdata/pg"${old_version}"\
+           --new-datadir /pgdata/pg"${new_version}" --link --check
+          echo -e "\nStep 6: Running pg_upgrade...\n"
+          time /usr/pgsql-"${new_version}"/bin/pg_upgrade --old-bindir /usr/pgsql-"${old_version}"/bin \
+          --new-bindir /usr/pgsql-"${new_version}"/bin --old-datadir /pgdata/pg"${old_version}" \
+          --new-datadir /pgdata/pg"${new_version}" --link --jobs=1
+          echo -e "\nStep 7: Copying patroni.dynamic.json...\n"
+          cp /pgdata/pg"${old_version}"/patroni.dynamic.json /pgdata/pg"${new_version}"
+          echo -e "\npg_upgrade Job Complete!"
+        - upgrade
+        - "19"
+        - "25"
+        image: img4
+        name: database
+        resources:
+          requests:
+            cpu: 3140m
+        securityContext:
+          privileged: false
+        volumeMounts:
+        - mountPath: /mnt/some/such
+          name: vm1
+      restartPolicy: Never
+      volumes:
+      - hostPath:
+          path: ""
+        name: vol2
+status: {}
+	`))
+
+	tdeJob := reconciler.generateUpgradeJob(ctx, upgrade, startup, "echo testKey")
+	b, _ := yaml.Marshal(tdeJob)
+	assert.Assert(t, strings.Contains(string(b),
+		`/usr/pgsql-"${new_version}"/bin/initdb -k -D /pgdata/pg"${new_version}" --encryption-key-command "echo testKey"`))
+}
+
 func TestGenerateRemoveDataJob(t *testing.T) {
 	ctx := context.Background()
 	reconciler := &PGUpgradeReconciler{}

pkg/apis/postgres-operator.crunchydata.com/v1beta1/pgupgrade_types.go

@@ -68,6 +68,11 @@ type PGUpgradeSpec struct {
 	// +optional
 	ToPostgresImage string `json:"toPostgresImage,omitempty"`
 
+	// Jobs specifies the number of parallel jobs pg_upgrade should use.
+	// If set to 0 or not provided, pg_upgrade will use its default behavior.
+	//TODO: Should we include kubebuilder:validation:Minimum/Maximum?
+	Jobs int `json:"jobs,omitempty"`
+
 	// Resource requirements for the PGUpgrade container.
 	// +optional
 	Resources corev1.ResourceRequirements `json:"resources,omitempty"`