dynamic host volumes: HTTP API endpoint (#24380)

This changeset implements the HTTP API endpoints for Dynamic Host Volumes.

The `GET /v1/volumes` endpoint is shared between CSI and DHV with a query
parameter for the type. In the interest of getting some working handlers
available for use in development (and minimizing the size of the diff to
review), this changeset doesn't do any sort of refactoring of how the existing
List Volumes CSI endpoint works. That will come in a later PR, as will the
corresponding `api` package updates we need to support the CLI.

Ref: https://hashicorp.atlassian.net/browse/NET-11549

command/agent/csi_endpoint.go

@@ -27,7 +27,13 @@ func (s *HTTPServer) CSIVolumesRequest(resp http.ResponseWriter, req *http.Reque
 	if !ok {
 		return []*structs.CSIVolListStub{}, nil
 	}
-	if qtype[0] != "csi" {
+	// TODO(1.10.0): move handling of GET /v1/volumes/ out so that we're not
+	// co-mingling the call for listing host volume here
+	switch qtype[0] {
+	case "host":
+		return s.HostVolumesListRequest(resp, req)
+	case "csi":
+	default:
 		return nil, nil
 	}
 

command/agent/host_volume_endpoint.go

@@ -0,0 +1,143 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package agent
+
+import (
+	"net/http"
+	"strings"
+
+	"github.com/hashicorp/nomad/nomad/structs"
+)
+
+func (s *HTTPServer) HostVolumesListRequest(resp http.ResponseWriter, req *http.Request) (any, error) {
+	args := structs.HostVolumeListRequest{}
+	if s.parse(resp, req, &args.Region, &args.QueryOptions) {
+		return nil, nil
+	}
+
+	query := req.URL.Query()
+	args.Prefix = query.Get("prefix")
+	args.NodePool = query.Get("node_pool")
+	args.NodeID = query.Get("node_id")
+
+	var out structs.HostVolumeListResponse
+	if err := s.agent.RPC("HostVolume.List", &args, &out); err != nil {
+		return nil, err
+	}
+
+	setMeta(resp, &out.QueryMeta)
+	return out.Volumes, nil
+}
+
+// HostVolumeSpecificRequest dispatches GET and PUT
+func (s *HTTPServer) HostVolumeSpecificRequest(resp http.ResponseWriter, req *http.Request) (any, error) {
+	// Tokenize the suffix of the path to get the volume id, tolerating a
+	// present or missing trailing slash
+	reqSuffix := strings.TrimPrefix(req.URL.Path, "/v1/volume/host/")
+	tokens := strings.FieldsFunc(reqSuffix, func(c rune) bool { return c == '/' })
+
+	if len(tokens) == 0 {
+		return nil, CodedError(404, resourceNotFoundErr)
+	}
+
+	switch req.Method {
+
+	// PUT /v1/volume/host/create
+	// POST /v1/volume/host/create
+	// PUT /v1/volume/host/register
+	// POST /v1/volume/host/register
+	case http.MethodPut, http.MethodPost:
+		switch tokens[0] {
+		case "create", "":
+			return s.hostVolumeCreate(resp, req)
+		case "register":
+			return s.hostVolumeRegister(resp, req)
+		default:
+			return nil, CodedError(404, resourceNotFoundErr)
+		}
+
+	// DELETE /v1/volume/host/:id
+	case http.MethodDelete:
+		return s.hostVolumeDelete(tokens[0], resp, req)
+
+	// GET /v1/volume/host/:id
+	case http.MethodGet:
+		return s.hostVolumeGet(tokens[0], resp, req)
+	}
+
+	return nil, CodedError(404, resourceNotFoundErr)
+}
+
+func (s *HTTPServer) hostVolumeGet(id string, resp http.ResponseWriter, req *http.Request) (any, error) {
+	args := structs.HostVolumeGetRequest{
+		ID: id,
+	}
+	if s.parse(resp, req, &args.Region, &args.QueryOptions) {
+		return nil, nil
+	}
+
+	var out structs.HostVolumeGetResponse
+	if err := s.agent.RPC("HostVolume.Get", &args, &out); err != nil {
+		return nil, err
+	}
+
+	setMeta(resp, &out.QueryMeta)
+	if out.Volume == nil {
+		return nil, CodedError(404, "volume not found")
+	}
+
+	return out.Volume, nil
+}
+
+func (s *HTTPServer) hostVolumeRegister(resp http.ResponseWriter, req *http.Request) (any, error) {
+
+	args := structs.HostVolumeRegisterRequest{}
+	if err := decodeBody(req, &args); err != nil {
+		return err, CodedError(400, err.Error())
+	}
+	s.parseWriteRequest(req, &args.WriteRequest)
+
+	var out structs.HostVolumeRegisterResponse
+	if err := s.agent.RPC("HostVolume.Register", &args, &out); err != nil {
+		return nil, err
+	}
+
+	setIndex(resp, out.Index)
+
+	return &out, nil
+}
+
+func (s *HTTPServer) hostVolumeCreate(resp http.ResponseWriter, req *http.Request) (any, error) {
+
+	args := structs.HostVolumeCreateRequest{}
+	if err := decodeBody(req, &args); err != nil {
+		return err, CodedError(400, err.Error())
+	}
+	s.parseWriteRequest(req, &args.WriteRequest)
+
+	var out structs.HostVolumeCreateResponse
+	if err := s.agent.RPC("HostVolume.Create", &args, &out); err != nil {
+		return nil, err
+	}
+
+	setIndex(resp, out.Index)
+
+	return &out, nil
+}
+
+func (s *HTTPServer) hostVolumeDelete(id string, resp http.ResponseWriter, req *http.Request) (any, error) {
+	// HTTP API only supports deleting a single ID because of compatibility with
+	// the existing HTTP routes for CSI
+	args := structs.HostVolumeDeleteRequest{VolumeIDs: []string{id}}
+	s.parseWriteRequest(req, &args.WriteRequest)
+
+	var out structs.HostVolumeDeleteResponse
+	if err := s.agent.RPC("HostVolume.Delete", &args, &out); err != nil {
+		return nil, err
+	}
+
+	setIndex(resp, out.Index)
+
+	return nil, nil
+}

command/agent/host_volume_endpoint_test.go

@@ -0,0 +1,101 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package agent
+
+import (
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"testing"
+
+	"github.com/hashicorp/nomad/nomad/mock"
+	"github.com/hashicorp/nomad/nomad/structs"
+	"github.com/shoenig/test/must"
+)
+
+func TestHostVolumeEndpoint_CRUD(t *testing.T) {
+	httpTest(t, nil, func(s *TestAgent) {
+
+		// Create a volume on the test node
+
+		vol := mock.HostVolumeRequest()
+		reqBody := struct {
+			Volumes []*structs.HostVolume
+		}{Volumes: []*structs.HostVolume{vol}}
+		buf := encodeReq(reqBody)
+		req, err := http.NewRequest(http.MethodPut, "/v1/volume/host/create", buf)
+		must.NoError(t, err)
+		respW := httptest.NewRecorder()
+
+		// Make the request and verify we got a valid volume back
+
+		obj, err := s.Server.HostVolumeSpecificRequest(respW, req)
+		must.NoError(t, err)
+		must.NotNil(t, obj)
+		resp := obj.(*structs.HostVolumeCreateResponse)
+		must.Len(t, 1, resp.Volumes)
+		must.Eq(t, vol.Name, resp.Volumes[0].Name)
+		must.Eq(t, s.client.NodeID(), resp.Volumes[0].NodeID)
+		must.NotEq(t, "", respW.Result().Header.Get("X-Nomad-Index"))
+
+		volID := resp.Volumes[0].ID
+
+		// Verify volume was created
+
+		path, err := url.JoinPath("/v1/volume/host/", volID)
+		must.NoError(t, err)
+		req, err = http.NewRequest(http.MethodGet, path, nil)
+		must.NoError(t, err)
+		obj, err = s.Server.HostVolumeSpecificRequest(respW, req)
+		must.NoError(t, err)
+		must.NotNil(t, obj)
+		respVol := obj.(*structs.HostVolume)
+		must.Eq(t, s.client.NodeID(), respVol.NodeID)
+
+		// Update the volume (note: this doesn't update the volume on the client)
+
+		vol = respVol.Copy()
+		vol.Parameters = map[string]string{"bar": "foo"} // swaps key and value
+		reqBody = struct {
+			Volumes []*structs.HostVolume
+		}{Volumes: []*structs.HostVolume{vol}}
+		buf = encodeReq(reqBody)
+		req, err = http.NewRequest(http.MethodPut, "/v1/volume/host/register", buf)
+		must.NoError(t, err)
+		obj, err = s.Server.HostVolumeSpecificRequest(respW, req)
+		must.NoError(t, err)
+		must.NotNil(t, obj)
+		regResp := obj.(*structs.HostVolumeRegisterResponse)
+		must.Len(t, 1, regResp.Volumes)
+		must.Eq(t, map[string]string{"bar": "foo"}, regResp.Volumes[0].Parameters)
+
+		// Verify volume was updated
+
+		path = fmt.Sprintf("/v1/volumes?type=host&node_id=%s", s.client.NodeID())
+		req, err = http.NewRequest(http.MethodGet, path, nil)
+		must.NoError(t, err)
+		obj, err = s.Server.HostVolumesListRequest(respW, req)
+		must.NoError(t, err)
+		vols := obj.([]*structs.HostVolumeStub)
+		must.Len(t, 1, vols)
+
+		// Delete the volume
+
+		req, err = http.NewRequest(http.MethodDelete, fmt.Sprintf("/v1/volume/host/%s", volID), nil)
+		must.NoError(t, err)
+		_, err = s.Server.HostVolumeSpecificRequest(respW, req)
+		must.NoError(t, err)
+
+		// Verify volume was deleted
+
+		path, err = url.JoinPath("/v1/volume/host/", volID)
+		must.NoError(t, err)
+		req, err = http.NewRequest(http.MethodGet, path, nil)
+		must.NoError(t, err)
+		obj, err = s.Server.HostVolumeSpecificRequest(respW, req)
+		must.EqError(t, err, "volume not found")
+		must.Nil(t, obj)
+	})
+}

command/agent/http.go

@@ -410,6 +410,7 @@ func (s *HTTPServer) registerHandlers(enableDebug bool) {
 	s.mux.HandleFunc("/v1/volume/csi/", s.wrap(s.CSIVolumeSpecificRequest))
 	s.mux.HandleFunc("/v1/plugins", s.wrap(s.CSIPluginsRequest))
 	s.mux.HandleFunc("/v1/plugin/csi/", s.wrap(s.CSIPluginSpecificRequest))
+	s.mux.HandleFunc("/v1/volume/host/", s.wrap(s.HostVolumeSpecificRequest))
 
 	s.mux.HandleFunc("/v1/acl/policies", s.wrap(s.ACLPoliciesRequest))
 	s.mux.HandleFunc("/v1/acl/policy/", s.wrap(s.ACLPolicySpecificRequest))