From fa2cbb6f0cf9cc71e1b51ff62af25c59da0fd742 Mon Sep 17 00:00:00 2001 From: Diptanu Choudhury Date: Wed, 25 Oct 2017 23:51:53 -0700 Subject: [PATCH] Added the purge API on node endpoints --- command/agent/node_endpoint.go | 19 +++++++++ command/agent/node_endpoint_test.go | 65 +++++++++++++++++++++++++++++ nomad/node_endpoint.go | 15 ++++++- 3 files changed, 98 insertions(+), 1 deletion(-) diff --git a/command/agent/node_endpoint.go b/command/agent/node_endpoint.go index 8720503c5..5c173346c 100644 --- a/command/agent/node_endpoint.go +++ b/command/agent/node_endpoint.go @@ -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 +} diff --git a/command/agent/node_endpoint_test.go b/command/agent/node_endpoint_test.go index 391c10b9d..9f8a1edad 100644 --- a/command/agent/node_endpoint_test.go +++ b/command/agent/node_endpoint_test.go @@ -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) { diff --git a/nomad/node_endpoint.go b/nomad/node_endpoint.go index a8778216f..a62c68d1b 100644 --- a/nomad/node_endpoint.go +++ b/nomad/node_endpoint.go @@ -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)