mirror of
https://github.com/kemko/nomad.git
synced 2026-01-05 09:55:44 +03:00
Added the purge API on node endpoints
This commit is contained in:
@@ -42,6 +42,9 @@ func (s *HTTPServer) NodeSpecificRequest(resp http.ResponseWriter, req *http.Req
|
||||
case strings.HasSuffix(path, "/drain"):
|
||||
nodeName := strings.TrimSuffix(path, "/drain")
|
||||
return s.nodeToggleDrain(resp, req, nodeName)
|
||||
case strings.HasSuffix(path, "/purge"):
|
||||
nodeName := strings.TrimSuffix(path, "/purge")
|
||||
return s.nodePurge(resp, req, nodeName)
|
||||
default:
|
||||
return s.nodeQuery(resp, req, path)
|
||||
}
|
||||
@@ -142,3 +145,19 @@ func (s *HTTPServer) nodeQuery(resp http.ResponseWriter, req *http.Request,
|
||||
}
|
||||
return out.Node, nil
|
||||
}
|
||||
|
||||
func (s *HTTPServer) nodePurge(resp http.ResponseWriter, req *http.Request, nodeID string) (interface{}, error) {
|
||||
if req.Method != "POST" {
|
||||
return nil, CodedError(405, ErrInvalidMethod)
|
||||
}
|
||||
args := structs.NodeDeregisterRequest{
|
||||
NodeID: nodeID,
|
||||
}
|
||||
s.parseWriteRequest(req, &args.WriteRequest)
|
||||
var out structs.NodeUpdateResponse
|
||||
if err := s.agent.RPC("Node.Deregister", &args, &out); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
setIndex(resp, out.Index)
|
||||
return out, nil
|
||||
}
|
||||
|
||||
@@ -276,6 +276,71 @@ func TestHTTP_NodeDrain(t *testing.T) {
|
||||
})
|
||||
}
|
||||
|
||||
func TestHTTP_NodePurge(t *testing.T) {
|
||||
t.Parallel()
|
||||
httpTest(t, nil, func(s *TestAgent) {
|
||||
// Create the node
|
||||
node := mock.Node()
|
||||
args := structs.NodeRegisterRequest{
|
||||
Node: node,
|
||||
WriteRequest: structs.WriteRequest{Region: "global"},
|
||||
}
|
||||
var resp structs.NodeUpdateResponse
|
||||
if err := s.Agent.RPC("Node.Register", &args, &resp); err != nil {
|
||||
t.Fatalf("err: %v", err)
|
||||
}
|
||||
|
||||
// Add some allocations to the node
|
||||
state := s.Agent.server.State()
|
||||
alloc1 := mock.Alloc()
|
||||
alloc1.NodeID = node.ID
|
||||
if err := state.UpsertJobSummary(999, mock.JobSummary(alloc1.JobID)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
err := state.UpsertAllocs(1000, []*structs.Allocation{alloc1})
|
||||
if err != nil {
|
||||
t.Fatalf("err: %v", err)
|
||||
}
|
||||
|
||||
// Make the HTTP request to purge it
|
||||
req, err := http.NewRequest("POST", "/v1/node/"+node.ID+"/purge", nil)
|
||||
if err != nil {
|
||||
t.Fatalf("err: %v", err)
|
||||
}
|
||||
respW := httptest.NewRecorder()
|
||||
|
||||
// Make the request
|
||||
obj, err := s.Server.NodeSpecificRequest(respW, req)
|
||||
if err != nil {
|
||||
t.Fatalf("err: %v", err)
|
||||
}
|
||||
|
||||
// Check for the index
|
||||
if respW.HeaderMap.Get("X-Nomad-Index") == "" {
|
||||
t.Fatalf("missing index")
|
||||
}
|
||||
|
||||
// Check the response
|
||||
upd := obj.(structs.NodeUpdateResponse)
|
||||
if len(upd.EvalIDs) == 0 {
|
||||
t.Fatalf("bad: %v", upd)
|
||||
}
|
||||
|
||||
// Ensure that the node is not present anymore
|
||||
args1 := structs.NodeSpecificRequest{
|
||||
NodeID: node.ID,
|
||||
QueryOptions: structs.QueryOptions{Region: "global"},
|
||||
}
|
||||
var resp1 structs.SingleNodeResponse
|
||||
if err := s.Agent.RPC("Node.GetNode", &args1, &resp1); err != nil {
|
||||
t.Fatalf("err: %v", err)
|
||||
}
|
||||
if resp1.Node != nil {
|
||||
t.Fatalf("node still exists after purging: %#v", resp1.Node)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestHTTP_NodeQuery(t *testing.T) {
|
||||
t.Parallel()
|
||||
httpTest(t, nil, func(s *TestAgent) {
|
||||
|
||||
@@ -213,6 +213,20 @@ func (n *Node) Deregister(args *structs.NodeDeregisterRequest, reply *structs.No
|
||||
if args.NodeID == "" {
|
||||
return fmt.Errorf("missing node ID for client deregistration")
|
||||
}
|
||||
// Look for the node
|
||||
snap, err := n.srv.fsm.State().Snapshot()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
ws := memdb.NewWatchSet()
|
||||
node, err := snap.NodeByID(ws, args.NodeID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if node == nil {
|
||||
return fmt.Errorf("node not found")
|
||||
}
|
||||
|
||||
// Commit this update via Raft
|
||||
_, index, err := n.srv.raftApply(structs.NodeDeregisterRequestType, args)
|
||||
@@ -232,7 +246,6 @@ func (n *Node) Deregister(args *structs.NodeDeregisterRequest, reply *structs.No
|
||||
}
|
||||
|
||||
// Determine if there are any Vault accessors on the node
|
||||
ws := memdb.NewWatchSet()
|
||||
accessors, err := n.srv.State().VaultAccessorsByNode(ws, args.NodeID)
|
||||
if err != nil {
|
||||
n.srv.logger.Printf("[ERR] nomad.client: looking up accessors for node %q failed: %v", args.NodeID, err)
|
||||
|
||||
Reference in New Issue
Block a user