Add RPC modifications to include the lock work

command/agent/variable_endpoint.go

@@ -4,6 +4,7 @@
 package agent
 
 import (
+	"errors"
 	"fmt"
 	"net/http"
 	"net/url"
@@ -13,10 +14,11 @@ import (
 	"github.com/hashicorp/nomad/nomad/structs"
 )
 
-const (
-	renewLockQueryParam   = "renew"
-	acquireLockQueryParam = "acquire"
-	releaseLockQueryParam = "release"
+var (
+	renewLockQueryParam = "lock-renew"
+
+	acquireLockQueryParam = string(structs.VarOpLockAcquire)
+	releaseLockQueryParam = string(structs.VarOpLockRelease)
 )
 
 func (s *HTTPServer) VariablesListRequest(resp http.ResponseWriter, req *http.Request) (interface{}, error) {
@@ -52,67 +54,83 @@ func (s *HTTPServer) VariableSpecificRequest(resp http.ResponseWriter, req *http
 	case http.MethodGet:
 		return s.variableQuery(resp, req, path)
 	case http.MethodPut, http.MethodPost:
+		urlParams := req.URL.Query()
+		lockOperation, err := getLockOperation(urlParams)
+		if err != nil {
+			return nil, CodedError(http.StatusBadRequest, err.Error())
+		}
 
-		queryParams := req.URL.Query()
-		_, renewLock := queryParams[renewLockQueryParam]
-		_, acquireLock := queryParams[acquireLockQueryParam]
-		_, releaseLock := queryParams[releaseLockQueryParam]
+		cq := req.URL.Query().Get("cas")
+
+		if cq != "" && lockOperation != "" {
+			return nil, CodedError(http.StatusBadRequest, "CAS can't be used with lock operations")
+		}
 
-		if renewLock || acquireLock || releaseLock {
-			if !isOneAndOnlyOneSet(renewLock, acquireLock, releaseLock) {
-				return nil, CodedError(http.StatusBadRequest, "multiple lock operations")
-			}
-			return s.variableLockOperation(resp, req, queryParams)
+		if lockOperation == "" {
+			return s.variableUpsert(resp, req, path)
 		}
 
-		return s.variableUpsert(resp, req, path)
+		if lockOperation == renewLockQueryParam {
+			return s.variableLockRenew(resp, req, path)
+		}
+
+		return s.variableLockOperation(resp, req, path, lockOperation)
+
 	case http.MethodDelete:
 		return s.variableDelete(resp, req, path)
 	default:
 		return nil, CodedError(http.StatusBadRequest, ErrInvalidMethod)
 	}
 }
 
-func (s *HTTPServer) variableLockOperation(resp http.ResponseWriter, req *http.Request,
-	operation url.Values) (interface{}, error) {
+func (s *HTTPServer) variableLockRenew(resp http.ResponseWriter, req *http.Request, path string) (interface{}, error) {
 
 	// Parse the Variable
 	var Variable structs.VariableDecrypted
 	if err := decodeBody(req, &Variable); err != nil {
 		return nil, CodedError(http.StatusBadRequest, err.Error())
 	}
 
-	if operation[renewLockQueryParam][0] == renewLockQueryParam {
-		args := structs.VariablesRenewLockRequest{
-			Path: Variable.Path,
+	args := structs.VariablesRenewLockRequest{
+		Path:   path,
+		LockID: Variable.LockID(),
+	}
 
-			LockID: Variable.Lock.ID,
-		}
+	s.parseWriteRequest(req, &args.WriteRequest)
 
-		s.parseWriteRequest(req, &args.WriteRequest)
+	var out structs.VariablesRenewLockResponse
+	if err := s.agent.RPC(structs.VariablesRenewLockRPCMethod, &args, &out); err != nil {
+		return nil, err
+	}
 
-		var out structs.VariablesRenewLockResponse
-		if err := s.agent.RPC(structs.VariablesRenewLockRPCMethod, &args, &out); err != nil {
-			return nil, err
-		}
+	return out.VarMeta, nil
+}
+
+func (s *HTTPServer) variableLockOperation(resp http.ResponseWriter, req *http.Request,
+	path, operation string) (interface{}, error) {
 
-		return out.VarMeta, nil
+	// Parse the Variable
+	var Variable structs.VariableDecrypted
+	if err := decodeBody(req, &Variable); err != nil {
+		return nil, CodedError(http.StatusBadRequest, err.Error())
 	}
 
 	// At this point, the operation can be either acquire or release, and they are
 	// both handled by the VariablesApplyRPCMethod.
 	args := structs.VariablesApplyRequest{
-		Op:  structs.VarOpSet,
+		Op:  structs.VarOp(operation),
 		Var: &Variable,
 	}
 
+	Variable.Path = path
+
 	s.parseWriteRequest(req, &args.WriteRequest)
 
 	var out structs.VariablesApplyResponse
 	err := s.agent.RPC(structs.VariablesApplyRPCMethod, &args, &out)
 	defer setIndex(resp, out.WriteMeta.Index)
 	if err != nil {
-		return nil, CodedError(http.StatusInternalServerError, err.Error())
+		return nil, err
 	}
 
 	if out.Conflict != nil {
@@ -153,6 +171,7 @@ func (s *HTTPServer) variableUpsert(resp http.ResponseWriter, req *http.Request,
 	if err := decodeBody(req, &Variable); err != nil {
 		return nil, CodedError(http.StatusBadRequest, err.Error())
 	}
+
 	if len(Variable.Items) == 0 {
 		return nil, CodedError(http.StatusBadRequest, "variable missing required Items object")
 	}
@@ -263,3 +282,30 @@ func parseCAS(req *http.Request) (bool, uint64, error) {
 func isOneAndOnlyOneSet(a, b, c bool) bool {
 	return (a || b || c) && !a != !b != !c != !(a && b && c)
 }
+
+// getLockOperation returns the lock operation to be performed in case there is
+// one. It returns error if more than one is set.
+func getLockOperation(queryParams url.Values) (string, error) {
+	_, renewLock := queryParams[renewLockQueryParam]
+	_, acquireLock := queryParams[acquireLockQueryParam]
+	_, releaseLock := queryParams[releaseLockQueryParam]
+
+	if !renewLock && !acquireLock && !releaseLock {
+		return "", nil
+	}
+
+	if !isOneAndOnlyOneSet(renewLock, acquireLock, releaseLock) {
+		return "", errors.New("multiple lock operations")
+	}
+
+	switch {
+	case renewLock:
+		return renewLockQueryParam, nil
+	case acquireLock:
+		return acquireLockQueryParam, nil
+	case releaseLock:
+		return releaseLockQueryParam, nil
+	default:
+		return "", errors.New("unspecified lock operation")
+	}
+}

command/agent/variable_endpoint_test.go

@@ -294,7 +294,8 @@ func TestHTTP_Variables(t *testing.T) {
 				require.Equal(t, &svU, out)
 			}
 		})
-		t.Run("update-cas", func(t *testing.T) {
+
+		t.Run("update_cas", func(t *testing.T) {
 			sv := mock.Variable()
 			require.NoError(t, rpcWriteSV(s, sv, sv))
 
@@ -375,7 +376,164 @@ func TestHTTP_Variables(t *testing.T) {
 				require.Equal(t, fmt.Sprint(svU.Items), fmt.Sprint(out.Items))
 			}
 		})
-		rpcResetSV(s)
+
+		t.Run("error_cas_and_acquire_lock", func(t *testing.T) {
+			svLA := sv1.Copy()
+			svLA.Items["new"] = "new"
+
+			// break the request body
+			badBuf := encodeBrokenReq(&svLA)
+
+			req, err := http.NewRequest("PUT", "/v1/var/"+sv1.Path+"?cas=1&"+acquireLockQueryParam, badBuf)
+			require.NoError(t, err)
+			respW := httptest.NewRecorder()
+
+			// Make the request
+			obj, err := s.Server.VariableSpecificRequest(respW, req)
+			require.EqualError(t, err, "CAS can't be used with lock operations")
+
+			var cErr HTTPCodedError
+			require.ErrorAs(t, err, &cErr)
+			require.Equal(t, http.StatusBadRequest, cErr.Code())
+			require.Nil(t, obj)
+		})
+		t.Run("error_parse_acquire_lock", func(t *testing.T) {
+			svLA := sv1.Copy()
+			svLA.Items["new"] = "new"
+
+			// break the request body
+			badBuf := encodeBrokenReq(&svLA)
+
+			req, err := http.NewRequest("PUT", "/v1/var/"+sv1.Path+"?"+acquireLockQueryParam, badBuf)
+			require.NoError(t, err)
+			respW := httptest.NewRecorder()
+
+			// Make the request
+			obj, err := s.Server.VariableSpecificRequest(respW, req)
+			require.EqualError(t, err, "unexpected EOF")
+			var cErr HTTPCodedError
+			require.ErrorAs(t, err, &cErr)
+			require.Equal(t, http.StatusBadRequest, cErr.Code())
+			require.Nil(t, obj)
+		})
+		t.Run("error_rpc_acquire_lock", func(t *testing.T) {
+			svLA := sv1.Copy()
+			svLA.Items["new"] = "new"
+
+			// test broken rpc error
+			buf := encodeReq(&svLA)
+			req, err := http.NewRequest("PUT", "/v1/var/"+sv1.Path+"?region=bad&"+acquireLockQueryParam, buf)
+			require.NoError(t, err)
+			respW := httptest.NewRecorder()
+
+			// Make the request
+			obj, err := s.Server.VariableSpecificRequest(respW, req)
+			require.EqualError(t, err, "No path to region")
+			require.Nil(t, obj)
+		})
+
+		t.Run("acquire_lock", func(t *testing.T) {
+			svLA := sv1
+
+			svLA.Items["new"] = "new"
+			// Make the HTTP request
+			buf := encodeReq(&svLA)
+			req, err := http.NewRequest("PUT", "/v1/var/"+svLA.Path+"?"+acquireLockQueryParam, buf)
+			require.NoError(t, err)
+			respW := httptest.NewRecorder()
+
+			obj, err := s.Server.VariableSpecificRequest(respW, req)
+			require.NoError(t, err)
+
+			// Test the returned object and rehydrate to a VariableDecrypted
+			require.NotNil(t, obj)
+			out, ok := obj.(*structs.VariableDecrypted)
+			require.True(t, ok, "Unable to convert obj to VariableDecrypted")
+
+			// Check for the index
+			require.NotZero(t, respW.HeaderMap.Get("X-Nomad-Index"))
+			require.Equal(t, fmt.Sprint(out.ModifyIndex), respW.HeaderMap.Get("X-Nomad-Index"))
+
+			// Check for the lock
+			require.NotNil(t, out.VariableMetadata.Lock)
+			require.NotEmpty(t, out.LockID())
+
+			// Check that written varible does not equal the input to rule out input mutation
+			require.NotEqual(t, &svLA.Items, out.Items)
+
+			// Update the lock information for the following tests
+			sv1.VariableMetadata = out.VariableMetadata
+		})
+
+		t.Run("error_rpc_renew_lock", func(t *testing.T) {
+			svRL := sv1.Copy()
+
+			// test broken rpc error
+			buf := encodeReq(&svRL)
+			req, err := http.NewRequest("PUT", "/v1/var/"+sv1.Path+"?region=bad&"+renewLockQueryParam, buf)
+			require.NoError(t, err)
+			respW := httptest.NewRecorder()
+
+			// Make the request
+			obj, err := s.Server.VariableSpecificRequest(respW, req)
+			require.EqualError(t, err, "No path to region")
+			require.Nil(t, obj)
+		})
+
+		t.Run("renew_lock", func(t *testing.T) {
+			svRL := sv1.Copy()
+
+			// Make the HTTP request
+			buf := encodeReq(&svRL)
+			req, err := http.NewRequest("PUT", "/v1/var/"+svRL.Path+"?"+renewLockQueryParam, buf)
+			require.NoError(t, err)
+			respW := httptest.NewRecorder()
+
+			obj, err := s.Server.VariableSpecificRequest(respW, req)
+			require.NoError(t, err)
+
+			// Test the returned object and rehydrate to a VariableDecrypted
+			require.NotNil(t, obj)
+			out, ok := obj.(*structs.VariableMetadata)
+			require.True(t, ok, "Unable to convert obj to VariableDecrypted")
+
+			// Check for the lock
+			require.NotNil(t, out.Lock)
+			require.Equal(t, sv1.LockID(), out.Lock.ID)
+		})
+
+		t.Run("release_lock", func(t *testing.T) {
+			svLR := sv1
+
+			svLR.Items["new"] = "new"
+			// Make the HTTP request
+			buf := encodeReq(&svLR)
+			req, err := http.NewRequest("PUT", "/v1/var/"+svLR.Path+"?"+releaseLockQueryParam, buf)
+			require.NoError(t, err)
+			respW := httptest.NewRecorder()
+
+			obj, err := s.Server.VariableSpecificRequest(respW, req)
+			require.NoError(t, err)
+
+			// Test the returned object and rehydrate to a VariableDecrypted
+			require.NotNil(t, obj)
+			out, ok := obj.(*structs.VariableDecrypted)
+			require.True(t, ok, "Unable to convert obj to VariableDecrypted")
+
+			// Check for the index
+			require.NotZero(t, respW.HeaderMap.Get("X-Nomad-Index"))
+			require.Equal(t, fmt.Sprint(out.ModifyIndex), respW.HeaderMap.Get("X-Nomad-Index"))
+
+			// Check for the lock
+			require.Nil(t, out.VariableMetadata.Lock)
+			require.Empty(t, out.LockID())
+
+			// Check that written variable is equal the input
+			require.Equal(t, sv1.Items, out.Items)
+
+			// Remove the lock information from the mock variable for the following tests
+			sv1.VariableMetadata = out.VariableMetadata
+		})
 
 		t.Run("error_rpc_delete", func(t *testing.T) {
 			sv1 := mock.Variable()