Merge pull request #2305 from hashicorp/f-operator

Add nomad operator command for interacting with Raft configuration

api/operator.go

@@ -0,0 +1,81 @@
+package api
+
+// Operator can be used to perform low-level operator tasks for Nomad.
+type Operator struct {
+	c *Client
+}
+
+// Operator returns a handle to the operator endpoints.
+func (c *Client) Operator() *Operator {
+	return &Operator{c}
+}
+
+// RaftServer has information about a server in the Raft configuration.
+type RaftServer struct {
+	// ID is the unique ID for the server. These are currently the same
+	// as the address, but they will be changed to a real GUID in a future
+	// release of Nomad.
+	ID string
+
+	// Node is the node name of the server, as known by Nomad, or this
+	// will be set to "(unknown)" otherwise.
+	Node string
+
+	// Address is the IP:port of the server, used for Raft communications.
+	Address string
+
+	// Leader is true if this server is the current cluster leader.
+	Leader bool
+
+	// Voter is true if this server has a vote in the cluster. This might
+	// be false if the server is staging and still coming online, or if
+	// it's a non-voting server, which will be added in a future release of
+	// Nomad.
+	Voter bool
+}
+
+// RaftConfigration is returned when querying for the current Raft configuration.
+type RaftConfiguration struct {
+	// Servers has the list of servers in the Raft configuration.
+	Servers []*RaftServer
+
+	// Index has the Raft index of this configuration.
+	Index uint64
+}
+
+// RaftGetConfiguration is used to query the current Raft peer set.
+func (op *Operator) RaftGetConfiguration(q *QueryOptions) (*RaftConfiguration, error) {
+	r := op.c.newRequest("GET", "/v1/operator/raft/configuration")
+	r.setQueryOptions(q)
+	_, resp, err := requireOK(op.c.doRequest(r))
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	var out RaftConfiguration
+	if err := decodeBody(resp, &out); err != nil {
+		return nil, err
+	}
+	return &out, nil
+}
+
+// RaftRemovePeerByAddress is used to kick a stale peer (one that it in the Raft
+// quorum but no longer known to Serf or the catalog) by address in the form of
+// "IP:port".
+func (op *Operator) RaftRemovePeerByAddress(address string, q *WriteOptions) error {
+	r := op.c.newRequest("DELETE", "/v1/operator/raft/peer")
+	r.setWriteOptions(q)
+
+	// TODO (alexdadgar) Currently we made address a query parameter. Once
+	// IDs are in place this will be DELETE /v1/operator/raft/peer/<id>.
+	r.params.Set("address", string(address))
+
+	_, resp, err := requireOK(op.c.doRequest(r))
+	if err != nil {
+		return err
+	}
+
+	resp.Body.Close()
+	return nil
+}

api/operator_test.go

@@ -0,0 +1,36 @@
+package api
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestOperator_RaftGetConfiguration(t *testing.T) {
+	c, s := makeClient(t, nil, nil)
+	defer s.Stop()
+
+	operator := c.Operator()
+	out, err := operator.RaftGetConfiguration(nil)
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	if len(out.Servers) != 1 ||
+		!out.Servers[0].Leader ||
+		!out.Servers[0].Voter {
+		t.Fatalf("bad: %v", out)
+	}
+}
+
+func TestOperator_RaftRemovePeerByAddress(t *testing.T) {
+	c, s := makeClient(t, nil, nil)
+	defer s.Stop()
+
+	// If we get this error, it proves we sent the address all the way
+	// through.
+	operator := c.Operator()
+	err := operator.RaftRemovePeerByAddress("nope", nil)
+	if err == nil || !strings.Contains(err.Error(),
+		"address \"nope\" was not found in the Raft configuration") {
+		t.Fatalf("err: %v", err)
+	}
+}

command/agent/http.go

@@ -170,6 +170,8 @@ func (s *HTTPServer) registerHandlers(enableDebug bool) {
 	s.mux.HandleFunc("/v1/status/leader", s.wrap(s.StatusLeaderRequest))
 	s.mux.HandleFunc("/v1/status/peers", s.wrap(s.StatusPeersRequest))
 
+	s.mux.HandleFunc("/v1/operator/", s.wrap(s.OperatorRequest))
+
 	s.mux.HandleFunc("/v1/system/gc", s.wrap(s.GarbageCollectRequest))
 	s.mux.HandleFunc("/v1/system/reconcile/summaries", s.wrap(s.ReconcileJobSummaries))
 

command/agent/operator_endpoint.go

@@ -0,0 +1,69 @@
+package agent
+
+import (
+	"net/http"
+	"strings"
+
+	"github.com/hashicorp/nomad/nomad/structs"
+	"github.com/hashicorp/raft"
+)
+
+func (s *HTTPServer) OperatorRequest(resp http.ResponseWriter, req *http.Request) (interface{}, error) {
+	path := strings.TrimPrefix(req.URL.Path, "/v1/operator/raft/")
+	switch {
+	case strings.HasPrefix(path, "configuration"):
+		return s.OperatorRaftConfiguration(resp, req)
+	case strings.HasPrefix(path, "peer"):
+		return s.OperatorRaftPeer(resp, req)
+	default:
+		return nil, CodedError(404, ErrInvalidMethod)
+	}
+}
+
+// OperatorRaftConfiguration is used to inspect the current Raft configuration.
+// This supports the stale query mode in case the cluster doesn't have a leader.
+func (s *HTTPServer) OperatorRaftConfiguration(resp http.ResponseWriter, req *http.Request) (interface{}, error) {
+	if req.Method != "GET" {
+		resp.WriteHeader(http.StatusMethodNotAllowed)
+		return nil, nil
+	}
+
+	var args structs.GenericRequest
+	if done := s.parse(resp, req, &args.Region, &args.QueryOptions); done {
+		return nil, nil
+	}
+
+	var reply structs.RaftConfigurationResponse
+	if err := s.agent.RPC("Operator.RaftGetConfiguration", &args, &reply); err != nil {
+		return nil, err
+	}
+
+	return reply, nil
+}
+
+// OperatorRaftPeer supports actions on Raft peers. Currently we only support
+// removing peers by address.
+func (s *HTTPServer) OperatorRaftPeer(resp http.ResponseWriter, req *http.Request) (interface{}, error) {
+	if req.Method != "DELETE" {
+		resp.WriteHeader(http.StatusMethodNotAllowed)
+		return nil, nil
+	}
+
+	var args structs.RaftPeerByAddressRequest
+	s.parseRegion(req, &args.Region)
+
+	params := req.URL.Query()
+	if _, ok := params["address"]; ok {
+		args.Address = raft.ServerAddress(params.Get("address"))
+	} else {
+		resp.WriteHeader(http.StatusBadRequest)
+		resp.Write([]byte("Must specify ?address with IP:port of peer to remove"))
+		return nil, nil
+	}
+
+	var reply struct{}
+	if err := s.agent.RPC("Operator.RaftRemovePeerByAddress", &args, &reply); err != nil {
+		return nil, err
+	}
+	return nil, nil
+}

command/agent/operator_endpoint_test.go

@@ -0,0 +1,58 @@
+package agent
+
+import (
+	"bytes"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/hashicorp/nomad/nomad/structs"
+)
+
+func TestHTTP_OperatorRaftConfiguration(t *testing.T) {
+	httpTest(t, nil, func(s *TestServer) {
+		body := bytes.NewBuffer(nil)
+		req, err := http.NewRequest("GET", "/v1/operator/raft/configuration", body)
+		if err != nil {
+			t.Fatalf("err: %v", err)
+		}
+
+		resp := httptest.NewRecorder()
+		obj, err := s.Server.OperatorRaftConfiguration(resp, req)
+		if err != nil {
+			t.Fatalf("err: %v", err)
+		}
+		if resp.Code != 200 {
+			t.Fatalf("bad code: %d", resp.Code)
+		}
+		out, ok := obj.(structs.RaftConfigurationResponse)
+		if !ok {
+			t.Fatalf("unexpected: %T", obj)
+		}
+		if len(out.Servers) != 1 ||
+			!out.Servers[0].Leader ||
+			!out.Servers[0].Voter {
+			t.Fatalf("bad: %v", out)
+		}
+	})
+}
+
+func TestHTTP_OperatorRaftPeer(t *testing.T) {
+	httpTest(t, nil, func(s *TestServer) {
+		body := bytes.NewBuffer(nil)
+		req, err := http.NewRequest("DELETE", "/v1/operator/raft/peer?address=nope", body)
+		if err != nil {
+			t.Fatalf("err: %v", err)
+		}
+
+		// If we get this error, it proves we sent the address all the
+		// way through.
+		resp := httptest.NewRecorder()
+		_, err = s.Server.OperatorRaftPeer(resp, req)
+		if err == nil || !strings.Contains(err.Error(),
+			"address \"nope\" was not found in the Raft configuration") {
+			t.Fatalf("err: %v", err)
+		}
+	})
+}

command/job_dispatch.go

@@ -33,11 +33,11 @@ General Options:
 Dispatch Options:
 
   -meta <key>=<value>
-	Meta takes a key/value pair seperated by "=". The metadata key will be
-	merged into the job's metadata. The job may define a default value for the
-	key which is overriden when dispatching. The flag can be provided more than
-	once to inject multiple metadata key/value pairs. Arbitrary keys are not
-	allowed. The parameterized job must allow the key to be merged.
+    Meta takes a key/value pair seperated by "=". The metadata key will be
+    merged into the job's metadata. The job may define a default value for the
+    key which is overriden when dispatching. The flag can be provided more than
+    once to inject multiple metadata key/value pairs. Arbitrary keys are not
+    allowed. The parameterized job must allow the key to be merged.
     
   -detach
     Return immediately instead of entering monitor mode. After job dispatch,

command/operator.go

@@ -0,0 +1,32 @@
+package command
+
+import (
+	"strings"
+
+	"github.com/mitchellh/cli"
+)
+
+type OperatorCommand struct {
+	Meta
+}
+
+func (f *OperatorCommand) Help() string {
+	helpText := `
+Usage: nomad operator <subcommand> [options]
+
+  Provides cluster-level tools for Nomad operators, such as interacting with
+  the Raft subsystem. NOTE: Use this command with extreme caution, as improper
+  use could lead to a Nomad outage and even loss of data.
+
+  Run nomad operator <subcommand> with no arguments for help on that subcommand.
+`
+	return strings.TrimSpace(helpText)
+}
+
+func (f *OperatorCommand) Synopsis() string {
+	return "Provides cluster-level tools for Nomad operators"
+}
+
+func (f *OperatorCommand) Run(args []string) int {
+	return cli.RunResultHelp
+}

command/operator_raft.go

@@ -0,0 +1,30 @@
+package command
+
+import (
+	"strings"
+
+	"github.com/mitchellh/cli"
+)
+
+type OperatorRaftCommand struct {
+	Meta
+}
+
+func (c *OperatorRaftCommand) Help() string {
+	helpText := `
+Usage: nomad operator raft <subcommand> [options]
+
+The Raft operator command is used to interact with Nomad's Raft subsystem. The
+command can be used to verify Raft peers or in rare cases to recover quorum by
+removing invalid peers.
+`
+	return strings.TrimSpace(helpText)
+}
+
+func (c *OperatorRaftCommand) Synopsis() string {
+	return "Provides access to the Raft subsystem"
+}
+
+func (c *OperatorRaftCommand) Run(args []string) int {
+	return cli.RunResultHelp
+}