mirror of
https://github.com/kemko/nomad.git
synced 2026-01-01 16:05:42 +03:00
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
This commit is contained in:
@@ -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
|
||||
}
|
||||
|
||||
|
||||
143
command/agent/host_volume_endpoint.go
Normal file
143
command/agent/host_volume_endpoint.go
Normal file
@@ -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
|
||||
}
|
||||
101
command/agent/host_volume_endpoint_test.go
Normal file
101
command/agent/host_volume_endpoint_test.go
Normal file
@@ -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)
|
||||
})
|
||||
}
|
||||
@@ -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))
|
||||
|
||||
Reference in New Issue
Block a user