Added slots/max_slots to Copy Offload API

Signed-off-by: Blake Devcich <[email protected]>

controllers/datamovement_controller.go

@@ -405,6 +405,7 @@ func (r *DataMovementReconciler) Reconcile(ctx context.Context, req ctrl.Request
 			log.Error(err, "Data movement operation cancelled", "output", combinedOutBuf.String())
 			dm.Status.Status = nnfv1alpha1.DataMovementConditionReasonCancelled
 		} else if err != nil {
+			log.Error(err, "Data movement operation failed", "output", combinedOutBuf.String())
 			dm.Status.Status = nnfv1alpha1.DataMovementConditionReasonFailed
 			dm.Status.Message = fmt.Sprintf("%s: %s", err.Error(), combinedOutBuf.String())
 			resourceErr := dwsv1alpha2.NewResourceError("").WithError(err).WithUserMessage("data movement operation failed: %s", combinedOutBuf.String()).WithFatal()
@@ -471,6 +472,14 @@ func buildDMCommand(ctx context.Context, profile dmConfigProfile, hosts []string
 		slots := profile.Slots
 		maxSlots := profile.MaxSlots
 
+		// Allow the user to override the slots and max_slots in the hostfile.
+		if userConfig && dm.Spec.UserConfig.Slots != nil && *dm.Spec.UserConfig.Slots >= 0 {
+			slots = *dm.Spec.UserConfig.Slots
+		}
+		if userConfig && dm.Spec.UserConfig.MaxSlots != nil && *dm.Spec.UserConfig.MaxSlots >= 0 {
+			maxSlots = *dm.Spec.UserConfig.MaxSlots
+		}
+
 		hostfile, err = createMpiHostfile(dm.Name, hosts, slots, maxSlots)
 		if err != nil {
 			return nil, "", fmt.Errorf("error creating MPI hostfile: %v", err)
@@ -512,6 +521,7 @@ func buildDMCommand(ctx context.Context, profile dmConfigProfile, hosts []string
 
 // Create an MPI Hostfile given a list of hosts, slots, and maxSlots. A temporary directory is
 // created based on the DM Name. The hostfile is created inside of this directory.
+// A value of 0 for slots or maxSlots will not use it in the hostfile.
 func createMpiHostfile(dmName string, hosts []string, slots, maxSlots int) (string, error) {
 
 	tmpdir := filepath.Join("/tmp", dmName)

controllers/datamovement_controller_test.go

@@ -34,6 +34,7 @@ import (
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	. "github.com/onsi/gomega/gstruct"
+	"go.openly.dev/pointy"
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -748,6 +749,50 @@ var _ = Describe("Data Movement Test", func() {
 					Expect(strings.Join(cmd, " ")).Should(MatchRegexp(expectedCmdRegex))
 				})
 			})
+
+			When("slots/maxSlots are specified in the request", func() {
+				DescribeTable("it should use the user slots vs the profile",
+					func(numSlots *int) {
+						profileSlots, profileMaxSlots := 3, 8
+
+						profile := dmConfigProfile{
+							Command:  defaultCommand,
+							Slots:    profileSlots,
+							MaxSlots: profileMaxSlots,
+						}
+						dm.Spec.UserConfig = &nnfv1alpha1.NnfDataMovementConfig{
+							Slots:    numSlots,
+							MaxSlots: numSlots,
+						}
+						_, hostfilePath, err := buildDMCommand(context.TODO(), profile, hosts, &dm)
+						Expect(err).ToNot(HaveOccurred())
+						Expect(hostfilePath).ToNot(BeEmpty())
+						DeferCleanup(func() {
+							Expect(os.Remove(hostfilePath)).ToNot(HaveOccurred())
+						})
+
+						content, err := os.ReadFile(hostfilePath)
+						Expect(err).ToNot(HaveOccurred())
+						Expect(string(content)).ToNot(BeEmpty())
+
+						if numSlots == nil {
+							// if nil, use the profile's slots
+							Expect(string(content)).Should(MatchRegexp(fmt.Sprintf(" slots=%d", profileSlots)))
+							Expect(string(content)).Should(MatchRegexp(fmt.Sprintf(" max_slots=%d", profileMaxSlots)))
+						} else if *numSlots == 0 {
+							// if 0, then don't use slots at all
+							Expect(string(content)).ShouldNot(MatchRegexp(" slots"))
+							Expect(string(content)).ShouldNot(MatchRegexp(" max_slots"))
+						} else {
+							Expect(string(content)).Should(MatchRegexp(fmt.Sprintf(" slots=%d", *numSlots)))
+							Expect(string(content)).Should(MatchRegexp(fmt.Sprintf(" max_slots=%d", *numSlots)))
+						}
+					},
+					Entry("when non-zero", pointy.Int(17)),
+					Entry("when zero it should omit", pointy.Int(0)),
+					Entry("when nil it should use the profile", nil),
+				)
+			})
 		})
 	})
 })

daemons/compute/Readme.md

@@ -111,6 +111,19 @@ To build the generated code used for the Go and Python example clients, run `./p
 
 `client-go` and `client-py` will have updates to the generated files when the API changes.
 
+To run this script, you will need the following installed on your system:
+
+- [protoc](https://grpc.io/docs/protoc-installation/)
+- [protoc-gen-doc](https://github.com/pseudomuto/protoc-gen-doc#installation)
+- [grpc and grpc tools python modules](https://grpc.io/docs/languages/python/quickstart/)
+
+On OSX, this can be done easily via:
+
+```shell
+brew install protobuf protoc-gen-go protoc-gen-go-grpc
+pip3 install grpcio grpcio_tools
+```
+
 #### C and C++
 
 For the C and C++ clients, the clients must be built to generate the source code to support the API. Run the `Makefiles` in the `_client-c`, `client-cpp`, and `lib-cpp` directories to update the generated API files.

daemons/compute/api/datamovement.proto

@@ -106,6 +106,14 @@ message DataMovementCreateRequest {
 
     // If true, store stdout in DataMovementStatusResponse.Message when the command is successful. Failure output is always contained in the message.
     bool storeStdout = 7;
+
+    // The number of slots specified in the MPI hostfile. A value of 0 disables the use of slots in
+    // the hostfile. -1 will defer to the server side configuration.
+    int32 slots = 8;
+
+    // The number of max_slots specified in the MPI hostfile. A value of 0 disables the use of
+    // max_slots in the hostfile. -1 will defer to the server side configuration.
+    int32 maxSlots = 9;
 }
 
 // The Data Movement Create Response to indicate the status of of the Data Movement Request.