Use Flink API to retrieve savepoint name

cmd/cli/flink/savepoint.go

@@ -7,10 +7,7 @@ import (
 	"io/ioutil"
 )
 
-type createSavepointRequest struct {
-	TargetDirectory string `json:"target-directory"`
-	CancelJob       bool   `json:"cancel-job"`
-}
+type createSavepointRequest map[string]interface{}
 
 // CreateSavepointResponse represents the response body
 // used by the create savepoint API
@@ -19,11 +16,13 @@ type CreateSavepointResponse struct {
 }
 
 // CreateSavepoint creates a savepoint for a job specified by job ID
-func (c FlinkRestClient) CreateSavepoint(jobID string, savepointPath string) (CreateSavepointResponse, error) {
-	req := createSavepointRequest{
-		TargetDirectory: savepointPath,
-		CancelJob:       false,
+func (c FlinkRestClient) CreateSavepoint(jobID string, savepointDir string) (CreateSavepointResponse, error) {
+	req := make(map[string]interface{})
+	if len(savepointDir) > 0 {
+		// The target-directory attribute has to be omitted if the cluster-default savepoint directory should be used
+		req["target-directory"] = savepointDir
 	}
+	req["cancel-job"] = true
 
 	reqBody := new(bytes.Buffer)
 	json.NewEncoder(reqBody).Encode(req)
@@ -63,6 +62,12 @@ type SavepointCreationStatus struct {
 // used by the savepoint monitoring API
 type MonitorSavepointCreationResponse struct {
 	Status SavepointCreationStatus `json:"status"`
+	Operation SavepointCreationOperation `json:"operation"`
+}
+
+type SavepointCreationOperation struct {
+	Location string `json:"location"`
+	FailureCause string `json:"failure-cause"`
 }
 
 // MonitorSavepointCreation allows for monitoring the status of a savepoint creation

cmd/cli/flink/savepoint_test.go

@@ -12,7 +12,7 @@ import (
  * Create Savepoint
  */
 func TestCreateSavepointReturnsAnErrorWhenTheStatusIsNot202(t *testing.T) {
-	server := createTestServerWithBodyCheck(t, "/jobs/1/savepoints", `{"target-directory":"/data/flink","cancel-job":false}`, http.StatusOK, "{}")
+	server := createTestServerWithBodyCheck(t, "/jobs/1/savepoints", `{"cancel-job":true,"target-directory":"/data/flink"}`, http.StatusOK, "{}")
 	defer server.Close()
 
 	api := FlinkRestClient{server.URL, retryablehttp.NewClient()}
@@ -22,7 +22,7 @@ func TestCreateSavepointReturnsAnErrorWhenTheStatusIsNot202(t *testing.T) {
 }
 
 func TestCreateSavepointReturnsAnErrorWhenItCannotDeserializeTheResponseAsJSON(t *testing.T) {
-	server := createTestServerWithBodyCheck(t, "/jobs/1/savepoints", `{"target-directory":"/data/flink","cancel-job":false}`, http.StatusAccepted, `{"jobs: []}`)
+	server := createTestServerWithBodyCheck(t, "/jobs/1/savepoints", `{"cancel-job":true,"target-directory":"/data/flink"}`, http.StatusAccepted, `{"jobs: []}`)
 	defer server.Close()
 
 	api := FlinkRestClient{server.URL, retryablehttp.NewClient()}
@@ -32,7 +32,7 @@ func TestCreateSavepointReturnsAnErrorWhenItCannotDeserializeTheResponseAsJSON(t
 }
 
 func TestCreateSavepointCorrectlyReturnsARequestID(t *testing.T) {
-	server := createTestServerWithBodyCheck(t, "/jobs/1/savepoints", `{"target-directory":"/data/flink","cancel-job":false}`, http.StatusAccepted, `{"request-id": "1"}`)
+	server := createTestServerWithBodyCheck(t, "/jobs/1/savepoints", `{"cancel-job":true,"target-directory":"/data/flink"}`, http.StatusAccepted, `{"request-id": "1"}`)
 	defer server.Close()
 
 	api := FlinkRestClient{server.URL, retryablehttp.NewClient()}

cmd/cli/main.go

@@ -77,14 +77,7 @@ func DeployAction(c *cli.Context) error {
 		deploy.ProgramArgs = programArgs
 	}
 
-	savepointDir := c.String("savepoint-dir")
 	savepointPath := c.String("savepoint-path")
-	if len(savepointDir) > 0 && len(savepointPath) > 0 {
-		return cli.NewExitError("both flags 'savepoint-dir' and 'savepoint-path' specified, only one allowed", -1)
-	}
-	if len(savepointDir) > 0 {
-		deploy.SavepointDir = savepointDir
-	}
 	if len(savepointPath) > 0 {
 		deploy.SavepointPath = savepointPath
 	}
@@ -152,7 +145,7 @@ func UpdateAction(c *cli.Context) error {
 	if len(savepointDir) != 0 {
 		update.SavepointDir = savepointDir
 	} else {
-		return cli.NewExitError("unspecified flag 'savepoint-dir'", -1)
+		update.SavepointDir = ""
 	}
 
 	update.AllowNonRestoredState = c.Bool("allow-non-restored-state")

cmd/cli/main_test.go

@@ -90,20 +90,6 @@ func TestDeployActionShouldThrowAnErrorWhenBothTheLocalFilenameAndRemoteFilename
 	assert.EqualError(t, err, "both flags 'file-name' and 'remote-file-name' specified, only one allowed")
 }
 
-func TestDeployActionShouldThrowAnErrorWhenBothTheSavepointDirAndSavepointPathArgumentsAreSet(t *testing.T) {
-	operator = TestOperator{}
-
-	app := cli.App{}
-	set := flag.FlagSet{}
-	set.String("file-name", "file.jar", "")
-	set.String("savepoint-dir", "/data/flink", "")
-	set.String("savepoint-path", "/data/flink/savepoint-abc", "")
-	context := cli.NewContext(&app, &set, nil)
-	err := DeployAction(context)
-
-	assert.EqualError(t, err, "both flags 'savepoint-dir' and 'savepoint-path' specified, only one allowed")
-}
-
 func TestDeployActionShouldThrowAnErrorWhenTheCommandFails(t *testing.T) {
 	mockedDeployError = errors.New("failed")
 	operator = TestOperator{}

cmd/cli/operations/deploy.go

@@ -2,7 +2,6 @@ package operations
 
 import (
 	"errors"
-	"fmt"
 	"log"
 	"strings"
 )
@@ -16,7 +15,6 @@ type Deploy struct {
 	EntryClass            string
 	Parallelism           int
 	ProgramArgs           string
-	SavepointDir          string
 	SavepointPath         string
 	AllowNonRestoredState bool
 }
@@ -30,26 +28,7 @@ func (o RealOperator) extractJarIDFromFilename(filename string) string {
 func (o RealOperator) Deploy(d Deploy) error {
 	log.Println("Starting deploy")
 
-	if len(d.SavepointDir) > 0 && len(d.SavepointPath) > 0 {
-		return errors.New("both properties 'SavepointDir' and 'SavepointPath' are specified")
-	}
-
-	if len(d.SavepointDir) > 0 {
-		log.Printf("Using savepoint directory to retrieve the latest savepoint: %v", d.SavepointDir)
-
-		latestSavepoint, err := o.retrieveLatestSavepoint(d.SavepointDir)
-		if err != nil {
-			return fmt.Errorf("retrieving the latest savepoint failed: %v", err)
-		}
-
-		if len(latestSavepoint) != 0 {
-			d.SavepointPath = latestSavepoint
-		}
-	}
-
-	if len(d.SavepointPath) > 0 {
-		log.Printf("Using savepoint for deployment: %v", d.SavepointPath)
-	}
+	log.Printf("Using savepoint for deployment: %v", d.SavepointPath)
 
 	if d.AllowNonRestoredState == true {
 		log.Printf("Allowing non restorable state")

cmd/cli/operations/deploy_test.go

@@ -6,8 +6,6 @@ import (
 	"testing"
 
 	"github.com/ing-bank/flink-deployer/cmd/cli/flink"
-	"github.com/spf13/afero"
-
 	"github.com/stretchr/testify/assert"
 )
 
@@ -25,21 +23,10 @@ func TestExtractJarIDFromFilenameShouldReturnThePartAfterTheLastSlash(t *testing
 /*
  * Deploy
  */
-func TestDeployShouldReturnAnErrorWhenBothTheSavepointDirAndSavepointPathAreSet(t *testing.T) {
-	operator := RealOperator{}
-
-	err := operator.Deploy(Deploy{
-		SavepointDir:  "/data/flink",
-		SavepointPath: "/data/flink/savepoint-abc",
-	})
-
-	assert.EqualError(t, err, "both properties 'SavepointDir' and 'SavepointPath' are specified")
-}
-
 func TestDeployShouldReturnAnErrorWhenNeitherTheLocalOrRemoteFileNameAreSet(t *testing.T) {
 	operator := RealOperator{}
 
-	err := operator.Deploy(Deploy{})
+	err := operator.Deploy(Deploy{SavepointPath: "dummy"})
 
 	assert.EqualError(t, err, "both properties 'RemoteFilename' and 'LocalFilename' are unspecified")
 }
@@ -61,32 +48,6 @@ func TestDeployShouldReturnAnErrorWhenTheJarUploadFails(t *testing.T) {
 	assert.EqualError(t, err, "failed")
 }
 
-func TestDeployShouldReturnAnErrorWhenTheLatestSavepointCannotBeRetrieved(t *testing.T) {
-	filesystem := afero.NewMemMapFs()
-	filesystem.Mkdir("/data/flink/", 0755)
-
-	mockedUploadJarResponse = flink.UploadJarResponse{
-		Filename: "/data/flink/sample.jar",
-		Status:   "success",
-	}
-	mockedUploadJarError = nil
-
-	operator := RealOperator{
-		Filesystem: filesystem,
-		FlinkRestAPI: TestFlinkRestClient{
-			BaseURL: "http://localhost",
-			Client:  &http.Client{},
-		},
-	}
-
-	err := operator.Deploy(Deploy{
-		LocalFilename: "testdata/sample.jar",
-		SavepointDir:  "/data/flink",
-	})
-
-	assert.EqualError(t, err, "retrieving the latest savepoint failed: No savepoints present in directory: /data/flink")
-}
-
 func TestDeployShouldReturnAnErrorWhenTheJarRunFails(t *testing.T) {
 	mockedUploadJarResponse = flink.UploadJarResponse{
 		Filename: "/data/flink/sample.jar",