Added profile to Copy Offload API + fix nnf-dm logging

`DataMovementCreateRequest`can now specify a server-side DM profile via the `profile` parameter. If empty, it will use the `default` profile. nnf-dm was not performing `ctrl.SetLogger`, so there was no logging happening in sysconfig. I assume this is due to the updates to controller runtime. This has been updated to match what the DM controllers are doing. Additionally, fixed an issue with the API incorrectly reporting invalid DM requests as Pending. Signed-off-by: Blake Devcich <[email protected]>
NearNodeFlash · Nov 22, 2023 · ec0020c · ec0020c
1 parent 88dbb1b
commit ec0020c
Show file tree

Hide file tree

Showing 10 changed files with 233 additions and 184 deletions.
diff --git a/daemons/compute/api/datamovement.proto b/daemons/compute/api/datamovement.proto
@@ -114,6 +114,12 @@ message DataMovementCreateRequest {
     // 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 name of the data movement configuration profile to use. The above parameters (e.g. slots,
+    // logStdout) will override the settings defined by the profile. This profile must exist on the
+    // server otherwise the data movement operation will be invalid. Empty will default to the
+    // default profile.
+    string profile = 10;
 }
 
 // The Data Movement Create Response to indicate the status of of the Data Movement Request.

diff --git a/daemons/compute/client-go/api/datamovement.pb.go b/daemons/compute/client-go/api/datamovement.pb.go
diff --git a/daemons/compute/client-go/api/datamovement_grpc.pb.go b/daemons/compute/client-go/api/datamovement_grpc.pb.go
diff --git a/daemons/compute/client-go/main.go b/daemons/compute/client-go/main.go
@@ -22,6 +22,7 @@ package main
 import (
 	"context"
 	"flag"
+	"fmt"
 	"log"
 	"os"
 	"strings"
@@ -40,7 +41,7 @@ func main() {
 	namespace := flag.String("namespace", os.Getenv("DW_WORKFLOW_NAMESPACE"), "parent workflow namespace")
 	source := flag.String("source", "", "source file or directory")
 	destination := flag.String("destination", "", "destination file or directory")
-	dryrun := flag.Bool("dryrun", false, "perfrom dry run of operation")
+	dryrun := flag.Bool("dryrun", false, "perform dry run of operation")
 	skipDelete := flag.Bool("skip-delete", false, "skip deleting the resource after completion")
 	socket := flag.String("socket", "/var/run/nnf-dm.sock", "socket address")
 	maxWaitTime := flag.Int64("max-wait-time", 0, "maximum time to wait for status completion, in seconds.")
@@ -51,6 +52,7 @@ func main() {
 	storeStdout := flag.Bool("store-stdout", false, "store stdout in status message on successful dm")
 	slots := flag.Int("slots", -1, "slots to use in mpirun hostfile. -1 defers to system config, 0 omits from hostfile")
 	maxSlots := flag.Int("max-slots", -1, "max_slots to use in mpirun hostfile. -1 defers to system config, 0 omits from hostfile")
+	profile := flag.String("profile", "", "which data movement profile to use on the server, empty defaults to default profile")
 
 	flag.Parse()
 
@@ -99,7 +101,10 @@ func main() {
 			defer wg.Done()
 
 			log.Printf("Creating request %d of %d...", i+1, *count)
-			createResponse, err := createRequest(ctx, c, *workflow, *namespace, *source, *destination, *dryrun, *dcpOptions, *logStdout, *storeStdout, *slots, *maxSlots)
+			createResponse, err := createRequest(ctx, c, *workflow, *namespace,
+				*source, *destination, *dryrun, *dcpOptions,
+				*logStdout, *storeStdout, *slots, *maxSlots,
+				*profile)
 			if err != nil {
 				log.Fatalf("could not create data movement request: %v", err)
 			}
@@ -145,11 +150,12 @@ func main() {
 					log.Fatalf("failed to get data movement status: %v", err)
 				}
 
-				if statusResponse.GetStatus() == pb.DataMovementStatusResponse_FAILED {
-					log.Fatalf("data movement status failed: %+v", statusResponse)
+				if statusResponse.GetStatus() == pb.DataMovementStatusResponse_FAILED ||
+					statusResponse.GetStatus() == pb.DataMovementStatusResponse_INVALID {
+					log.Fatalf("Data movement failed: %s", statusString(statusResponse))
 				}
 
-				log.Printf("Data movement status: %+v", statusResponse)
+				log.Printf("Data movement status: %s", statusString(statusResponse))
 				if statusResponse.GetState() == pb.DataMovementStatusResponse_COMPLETED {
 					break
 				}
@@ -212,7 +218,7 @@ func versionRequest(ctx context.Context, client pb.DataMoverClient) (*pb.DataMov
 
 func createRequest(ctx context.Context, client pb.DataMoverClient, workflow, namespace,
 	source, destination string, dryrun bool, dcpOptions string, logStdout, storeStdout bool,
-	slots, maxSlots int) (*pb.DataMovementCreateResponse, error) {
+	slots, maxSlots int, profile string) (*pb.DataMovementCreateResponse, error) {
 
 	rsp, err := client.Create(ctx, &pb.DataMovementCreateRequest{
 		Workflow: &pb.DataMovementWorkflow{
@@ -227,6 +233,7 @@ func createRequest(ctx context.Context, client pb.DataMoverClient, workflow, nam
 		StoreStdout: storeStdout,
 		Slots:       int32(slots),
 		MaxSlots:    int32(maxSlots),
+		Profile:     profile,
 	})
 
 	if err != nil {
@@ -253,6 +260,10 @@ func getStatus(ctx context.Context, client pb.DataMoverClient, workflow string,
 	return rsp, nil
 }
 
+func statusString(status *pb.DataMovementStatusResponse) string {
+	return fmt.Sprintf("state: %s, status: %s, message: %s, commandStatus: %+v", status.State, status.Status, status.Message, status.CommandStatus)
+}
+
 func listRequests(ctx context.Context, client pb.DataMoverClient, workflow string, namespace string) (*pb.DataMovementListResponse, error) {
 	rsp, err := client.List(ctx, &pb.DataMovementListRequest{
 		Workflow: &pb.DataMovementWorkflow{

diff --git a/daemons/compute/client-py/datamovement_pb2.py b/daemons/compute/client-py/datamovement_pb2.py
diff --git a/daemons/compute/copy-offload-api.html b/daemons/compute/copy-offload-api.html
@@ -461,6 +461,16 @@ <h3 id="datamovement.DataMovementCreateRequest">DataMovementCreateRequest</h3>
 max_slots in the hostfile. -1 will defer to the server side configuration. </p></td>
                 </tr>
 
+                <tr>
+                  <td>profile</td>
+                  <td><a href="#string">string</a></td>
+                  <td></td>
+                  <td><p>The name of the data movement configuration profile to use. The above parameters (e.g. slots,
+logStdout) will override the settings defined by the profile. This profile must exist on the
+server otherwise the data movement operation will be invalid. Empty will default to the
+default profile. </p></td>
+                </tr>
+
             </tbody>
           </table>
 

diff --git a/daemons/compute/lib-cpp/client.cc b/daemons/compute/lib-cpp/client.cc
@@ -202,7 +202,7 @@ std::vector<std::string> VersionResponse::apiversions() {
     return apiVersions;
 }
 
-CreateRequest::CreateRequest(std::string source, std::string destination, bool dryrun, std::string dcpOptions, bool logStdout, bool storeStdout, int slots, int maxSlots) {
+CreateRequest::CreateRequest(std::string source, std::string destination, bool dryrun, std::string dcpOptions, bool logStdout, bool storeStdout, int slots, int maxSlots, std::string profile) {
     auto request = new datamovement::DataMovementCreateRequest();
 
     request->set_source(source);
@@ -213,6 +213,7 @@ CreateRequest::CreateRequest(std::string source, std::string destination, bool d
     request->set_storestdout(storeStdout);
     request->set_slots(slots);
     request->set_maxslots(maxSlots);
+    request->set_profile(profile);
 
     data_ = static_cast<void *>(request);
 }