open-consul/command/agent/http.go

545 lines
17 KiB
Go
Raw Normal View History

2013-12-23 19:38:51 +00:00
package agent
import (
"crypto/tls"
2013-12-23 19:38:51 +00:00
"encoding/json"
"fmt"
2013-12-23 19:38:51 +00:00
"io"
"log"
"net"
"net/http"
"net/http/pprof"
"net/url"
"os"
"strconv"
2014-08-22 19:59:47 +00:00
"strings"
2013-12-23 22:26:34 +00:00
"time"
2014-08-22 19:59:47 +00:00
"github.com/hashicorp/consul/consul/structs"
"github.com/hashicorp/consul/tlsutil"
2014-08-22 19:59:47 +00:00
"github.com/mitchellh/mapstructure"
2013-12-23 19:38:51 +00:00
)
2015-02-06 22:10:01 +00:00
var (
// scadaHTTPAddr is the address associated with the
// HTTPServer. When populating an ACL token for a request,
// this is checked to switch between the ACLToken and
// AtlasACLToken
scadaHTTPAddr = "SCADA"
)
2013-12-23 19:38:51 +00:00
// HTTPServer is used to wrap an Agent and expose various API's
// in a RESTful manner
type HTTPServer struct {
agent *Agent
mux *http.ServeMux
listener net.Listener
logger *log.Logger
2014-04-23 19:40:59 +00:00
uiDir string
addr string
2013-12-23 19:38:51 +00:00
}
// NewHTTPServers starts new HTTP servers to provide an interface to
2013-12-23 19:38:51 +00:00
// the agent.
func NewHTTPServers(agent *Agent, config *Config, logOutput io.Writer) ([]*HTTPServer, error) {
var servers []*HTTPServer
2013-12-23 19:38:51 +00:00
if config.Ports.HTTPS > 0 {
httpAddr, err := config.ClientListener(config.Addresses.HTTPS, config.Ports.HTTPS)
if err != nil {
return nil, err
}
tlsConf := &tlsutil.Config{
VerifyIncoming: config.VerifyIncoming,
VerifyOutgoing: config.VerifyOutgoing,
CAFile: config.CAFile,
CertFile: config.CertFile,
KeyFile: config.KeyFile,
NodeName: config.NodeName,
ServerName: config.ServerName}
tlsConfig, err := tlsConf.IncomingTLSConfig()
if err != nil {
return nil, err
}
ln, err := net.Listen(httpAddr.Network(), httpAddr.String())
if err != nil {
return nil, fmt.Errorf("Failed to get Listen on %s: %v", httpAddr.String(), err)
}
list := tls.NewListener(tcpKeepAliveListener{ln.(*net.TCPListener)}, tlsConfig)
// Create the mux
mux := http.NewServeMux()
// Create the server
srv := &HTTPServer{
agent: agent,
mux: mux,
listener: list,
logger: log.New(logOutput, "", log.LstdFlags),
uiDir: config.UiDir,
addr: httpAddr.String(),
}
srv.registerHandlers(config.EnableDebug)
// Start the server
go http.Serve(list, mux)
2014-11-19 19:51:25 +00:00
servers = append(servers, srv)
2013-12-23 19:38:51 +00:00
}
if config.Ports.HTTP > 0 {
httpAddr, err := config.ClientListener(config.Addresses.HTTP, config.Ports.HTTP)
if err != nil {
return nil, fmt.Errorf("Failed to get ClientListener address:port: %v", err)
}
// Error if we are trying to bind a domain socket to an existing path
socketPath, isSocket := unixSocketAddr(config.Addresses.HTTP)
if isSocket {
if _, err := os.Stat(socketPath); !os.IsNotExist(err) {
agent.logger.Printf("[WARN] agent: Replacing socket %q", socketPath)
}
if err := os.Remove(socketPath); err != nil && !os.IsNotExist(err) {
return nil, fmt.Errorf("error removing socket file: %s", err)
2015-01-16 01:24:15 +00:00
}
}
ln, err := net.Listen(httpAddr.Network(), httpAddr.String())
if err != nil {
return nil, fmt.Errorf("Failed to get Listen on %s: %v", httpAddr.String(), err)
}
var list net.Listener
if isSocket {
2015-01-24 01:57:04 +00:00
// Set up ownership/permission bits on the socket file
if err := setFilePermissions(socketPath, config.UnixSockets); err != nil {
return nil, fmt.Errorf("Failed setting up HTTP socket: %s", err)
}
2015-01-24 01:57:04 +00:00
list = ln
} else {
list = tcpKeepAliveListener{ln.(*net.TCPListener)}
}
// Create the mux
mux := http.NewServeMux()
// Create the server
srv := &HTTPServer{
agent: agent,
mux: mux,
listener: list,
logger: log.New(logOutput, "", log.LstdFlags),
uiDir: config.UiDir,
addr: httpAddr.String(),
}
srv.registerHandlers(config.EnableDebug)
// Start the server
go http.Serve(list, mux)
2014-11-19 19:51:25 +00:00
servers = append(servers, srv)
2013-12-23 19:38:51 +00:00
}
return servers, nil
}
2015-02-05 02:17:45 +00:00
// newScadaHttp creates a new HTTP server wrapping the SCADA
// listener such that HTTP calls can be sent from the brokers.
func newScadaHttp(agent *Agent, list net.Listener) *HTTPServer {
// Create the mux
mux := http.NewServeMux()
// Create the server
srv := &HTTPServer{
agent: agent,
mux: mux,
listener: list,
logger: agent.logger,
addr: scadaHTTPAddr,
2015-02-05 02:17:45 +00:00
}
srv.registerHandlers(false) // Never allow debug for SCADA
2015-02-05 02:17:45 +00:00
// Start the server
go http.Serve(list, mux)
return srv
}
// tcpKeepAliveListener sets TCP keep-alive timeouts on accepted
// connections. It's used by NewHttpServer so
// dead TCP connections eventually go away.
type tcpKeepAliveListener struct {
*net.TCPListener
}
func (ln tcpKeepAliveListener) Accept() (c net.Conn, err error) {
tc, err := ln.AcceptTCP()
if err != nil {
return
}
tc.SetKeepAlive(true)
tc.SetKeepAlivePeriod(30 * time.Second)
return tc, nil
2013-12-23 19:38:51 +00:00
}
// Shutdown is used to shutdown the HTTP server
func (s *HTTPServer) Shutdown() {
if s != nil {
2015-02-06 01:29:55 +00:00
s.logger.Printf("[DEBUG] http: Shutting down http server (%v)", s.addr)
s.listener.Close()
}
2013-12-23 19:38:51 +00:00
}
// registerHandlers is used to attach our handlers to the mux
func (s *HTTPServer) registerHandlers(enableDebug bool) {
2013-12-25 01:09:51 +00:00
s.mux.HandleFunc("/", s.Index)
2013-12-23 19:38:51 +00:00
s.mux.HandleFunc("/v1/status/leader", s.wrap(s.StatusLeader))
s.mux.HandleFunc("/v1/status/peers", s.wrap(s.StatusPeers))
2013-12-23 22:26:34 +00:00
s.mux.HandleFunc("/v1/catalog/register", s.wrap(s.CatalogRegister))
s.mux.HandleFunc("/v1/catalog/deregister", s.wrap(s.CatalogDeregister))
2013-12-23 22:26:34 +00:00
s.mux.HandleFunc("/v1/catalog/datacenters", s.wrap(s.CatalogDatacenters))
s.mux.HandleFunc("/v1/catalog/nodes", s.wrap(s.CatalogNodes))
s.mux.HandleFunc("/v1/catalog/services", s.wrap(s.CatalogServices))
s.mux.HandleFunc("/v1/catalog/service/", s.wrap(s.CatalogServiceNodes))
s.mux.HandleFunc("/v1/catalog/node/", s.wrap(s.CatalogNodeServices))
2014-01-04 01:15:51 +00:00
if !s.agent.config.DisableCoordinates {
s.mux.HandleFunc("/v1/coordinate/datacenters", s.wrap(s.CoordinateDatacenters))
s.mux.HandleFunc("/v1/coordinate/nodes", s.wrap(s.CoordinateNodes))
} else {
s.mux.HandleFunc("/v1/coordinate/datacenters", s.wrap(coordinateDisabled))
s.mux.HandleFunc("/v1/coordinate/nodes", s.wrap(coordinateDisabled))
}
s.mux.HandleFunc("/v1/health/node/", s.wrap(s.HealthNodeChecks))
s.mux.HandleFunc("/v1/health/checks/", s.wrap(s.HealthServiceChecks))
s.mux.HandleFunc("/v1/health/state/", s.wrap(s.HealthChecksInState))
s.mux.HandleFunc("/v1/health/service/", s.wrap(s.HealthServiceNodes))
2014-01-10 23:13:37 +00:00
s.mux.HandleFunc("/v1/agent/self", s.wrap(s.AgentSelf))
s.mux.HandleFunc("/v1/agent/maintenance", s.wrap(s.AgentNodeMaintenance))
2014-01-04 01:15:51 +00:00
s.mux.HandleFunc("/v1/agent/services", s.wrap(s.AgentServices))
s.mux.HandleFunc("/v1/agent/checks", s.wrap(s.AgentChecks))
2014-01-04 01:15:51 +00:00
s.mux.HandleFunc("/v1/agent/members", s.wrap(s.AgentMembers))
s.mux.HandleFunc("/v1/agent/join/", s.wrap(s.AgentJoin))
s.mux.HandleFunc("/v1/agent/force-leave/", s.wrap(s.AgentForceLeave))
s.mux.HandleFunc("/v1/agent/check/register", s.wrap(s.AgentRegisterCheck))
s.mux.HandleFunc("/v1/agent/check/deregister/", s.wrap(s.AgentDeregisterCheck))
s.mux.HandleFunc("/v1/agent/check/pass/", s.wrap(s.AgentCheckPass))
s.mux.HandleFunc("/v1/agent/check/warn/", s.wrap(s.AgentCheckWarn))
s.mux.HandleFunc("/v1/agent/check/fail/", s.wrap(s.AgentCheckFail))
s.mux.HandleFunc("/v1/agent/check/update/", s.wrap(s.AgentCheckUpdate))
s.mux.HandleFunc("/v1/agent/service/register", s.wrap(s.AgentRegisterService))
2014-04-18 17:54:18 +00:00
s.mux.HandleFunc("/v1/agent/service/deregister/", s.wrap(s.AgentDeregisterService))
s.mux.HandleFunc("/v1/agent/service/maintenance/", s.wrap(s.AgentServiceMaintenance))
2014-08-28 20:42:07 +00:00
s.mux.HandleFunc("/v1/event/fire/", s.wrap(s.EventFire))
s.mux.HandleFunc("/v1/event/list", s.wrap(s.EventList))
2014-04-01 00:12:10 +00:00
s.mux.HandleFunc("/v1/kv/", s.wrap(s.KVSEndpoint))
s.mux.HandleFunc("/v1/kv-txn", s.wrap(s.KVSTxn))
2014-04-01 00:12:10 +00:00
2014-05-16 22:49:17 +00:00
s.mux.HandleFunc("/v1/session/create", s.wrap(s.SessionCreate))
s.mux.HandleFunc("/v1/session/destroy/", s.wrap(s.SessionDestroy))
s.mux.HandleFunc("/v1/session/renew/", s.wrap(s.SessionRenew))
2014-05-16 22:49:17 +00:00
s.mux.HandleFunc("/v1/session/info/", s.wrap(s.SessionGet))
s.mux.HandleFunc("/v1/session/node/", s.wrap(s.SessionsForNode))
s.mux.HandleFunc("/v1/session/list", s.wrap(s.SessionList))
if s.agent.config.ACLDatacenter != "" {
s.mux.HandleFunc("/v1/acl/create", s.wrap(s.ACLCreate))
s.mux.HandleFunc("/v1/acl/update", s.wrap(s.ACLUpdate))
2014-08-18 19:05:01 +00:00
s.mux.HandleFunc("/v1/acl/destroy/", s.wrap(s.ACLDestroy))
s.mux.HandleFunc("/v1/acl/info/", s.wrap(s.ACLGet))
s.mux.HandleFunc("/v1/acl/clone/", s.wrap(s.ACLClone))
s.mux.HandleFunc("/v1/acl/list", s.wrap(s.ACLList))
} else {
s.mux.HandleFunc("/v1/acl/create", s.wrap(aclDisabled))
s.mux.HandleFunc("/v1/acl/update", s.wrap(aclDisabled))
2014-08-18 19:05:01 +00:00
s.mux.HandleFunc("/v1/acl/destroy/", s.wrap(aclDisabled))
s.mux.HandleFunc("/v1/acl/info/", s.wrap(aclDisabled))
s.mux.HandleFunc("/v1/acl/clone/", s.wrap(aclDisabled))
s.mux.HandleFunc("/v1/acl/list", s.wrap(aclDisabled))
}
2014-08-06 00:50:36 +00:00
s.mux.HandleFunc("/v1/query", s.wrap(s.PreparedQueryGeneral))
s.mux.HandleFunc("/v1/query/", s.wrap(s.PreparedQuerySpecific))
if enableDebug {
s.mux.HandleFunc("/debug/pprof/", pprof.Index)
s.mux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
s.mux.HandleFunc("/debug/pprof/profile", pprof.Profile)
s.mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
}
2014-04-23 19:57:06 +00:00
// Use the custom UI dir if provided.
2014-04-23 19:57:06 +00:00
if s.uiDir != "" {
2014-04-24 18:10:23 +00:00
s.mux.Handle("/ui/", http.StripPrefix("/ui/", http.FileServer(http.Dir(s.uiDir))))
2015-12-22 17:30:19 +00:00
} else if s.agent.config.EnableUi {
s.mux.Handle("/ui/", http.StripPrefix("/ui/", http.FileServer(assetFS())))
2015-02-11 21:25:04 +00:00
}
2014-04-28 21:52:30 +00:00
2015-09-02 01:28:32 +00:00
// API's are under /internal/ui/ to avoid conflict
s.mux.HandleFunc("/v1/internal/ui/nodes", s.wrap(s.UINodes))
s.mux.HandleFunc("/v1/internal/ui/node/", s.wrap(s.UINodeInfo))
s.mux.HandleFunc("/v1/internal/ui/services", s.wrap(s.UIServices))
2013-12-23 19:38:51 +00:00
}
// wrap is used to wrap functions to make them more convenient
2013-12-24 00:20:51 +00:00
func (s *HTTPServer) wrap(handler func(resp http.ResponseWriter, req *http.Request) (interface{}, error)) func(resp http.ResponseWriter, req *http.Request) {
2013-12-23 19:38:51 +00:00
f := func(resp http.ResponseWriter, req *http.Request) {
setHeaders(resp, s.agent.config.HTTPAPIResponseHeaders)
// Obfuscate any tokens from appearing in the logs
formVals, err := url.ParseQuery(req.URL.RawQuery)
if err != nil {
2016-01-05 16:49:40 +00:00
s.logger.Printf("[ERR] http: Failed to decode query: %s from=%s", err, req.RemoteAddr)
2016-02-05 22:06:42 +00:00
resp.WriteHeader(http.StatusInternalServerError) // 500
return
}
logURL := req.URL.String()
if tokens, ok := formVals["token"]; ok {
for _, token := range tokens {
if token == "" {
logURL += "<hidden>"
continue
}
logURL = strings.Replace(logURL, token, "<hidden>", -1)
}
}
// TODO (slackpad) We may want to consider redacting prepared
// query names/IDs here since they are proxies for tokens. But,
// knowing one only gives you read access to service listings
// which is pretty trivial, so it's probably not worth the code
// complexity and overhead of filtering them out. You can't
// recover the token it's a proxy for with just the query info;
// you'd need the actual token (or a management token) to read
// that back.
2013-12-23 19:38:51 +00:00
// Invoke the handler
2013-12-23 22:26:34 +00:00
start := time.Now()
defer func() {
2016-01-05 16:49:40 +00:00
s.logger.Printf("[DEBUG] http: Request %s %v (%v) from=%s", req.Method, logURL, time.Now().Sub(start), req.RemoteAddr)
2013-12-23 22:26:34 +00:00
}()
2013-12-24 00:20:51 +00:00
obj, err := handler(resp, req)
2013-12-23 19:38:51 +00:00
// Check for an error
HAS_ERR:
if err != nil {
2016-01-05 16:49:40 +00:00
s.logger.Printf("[ERR] http: Request %s %v, error: %v from=%s", req.Method, logURL, err, req.RemoteAddr)
2016-02-05 22:06:42 +00:00
code := http.StatusInternalServerError // 500
2014-08-22 19:59:47 +00:00
errMsg := err.Error()
if strings.Contains(errMsg, "Permission denied") || strings.Contains(errMsg, "ACL not found") {
2016-02-05 22:06:42 +00:00
code = http.StatusForbidden // 403
2014-08-22 19:59:47 +00:00
}
resp.WriteHeader(code)
2013-12-23 19:38:51 +00:00
resp.Write([]byte(err.Error()))
return
}
2013-12-24 00:20:51 +00:00
if obj != nil {
var buf []byte
buf, err = s.marshalJSON(req, obj)
if err != nil {
2013-12-24 00:20:51 +00:00
goto HAS_ERR
}
resp.Header().Set("Content-Type", "application/json")
resp.Write(buf)
2013-12-23 19:38:51 +00:00
}
}
return f
}
2013-12-24 00:20:51 +00:00
// marshalJSON marshals the object into JSON, respecting the user's pretty-ness
// configuration.
func (s *HTTPServer) marshalJSON(req *http.Request, obj interface{}) ([]byte, error) {
if _, ok := req.URL.Query()["pretty"]; ok {
buf, err := json.MarshalIndent(obj, "", " ")
return buf, err
}
buf, err := json.Marshal(obj)
return buf, err
}
// Returns true if the UI is enabled.
func (s *HTTPServer) IsUIEnabled() bool {
return s.uiDir != "" || s.agent.config.EnableUi
}
2013-12-25 01:09:51 +00:00
// Renders a simple index page
func (s *HTTPServer) Index(resp http.ResponseWriter, req *http.Request) {
2014-04-27 19:10:38 +00:00
// Check if this is a non-index path
if req.URL.Path != "/" {
2016-02-05 22:06:42 +00:00
resp.WriteHeader(http.StatusNotFound) // 404
2014-04-27 19:10:38 +00:00
return
2013-12-25 01:09:51 +00:00
}
2014-04-27 19:10:38 +00:00
// Give them something helpful if there's no UI so they at least know
// what this server is.
if !s.IsUIEnabled() {
2014-04-27 19:10:38 +00:00
resp.Write([]byte("Consul Agent"))
return
}
// Redirect to the UI endpoint
2016-02-05 22:06:42 +00:00
http.Redirect(resp, req, "/ui/", http.StatusMovedPermanently) // 301
2013-12-25 01:09:51 +00:00
}
2013-12-24 00:20:51 +00:00
// decodeBody is used to decode a JSON request body
func decodeBody(req *http.Request, out interface{}, cb func(interface{}) error) error {
var raw interface{}
2013-12-24 00:20:51 +00:00
dec := json.NewDecoder(req.Body)
if err := dec.Decode(&raw); err != nil {
return err
}
// Invoke the callback prior to decode
if cb != nil {
if err := cb(raw); err != nil {
return err
}
}
return mapstructure.Decode(raw, out)
2013-12-24 00:20:51 +00:00
}
// setIndex is used to set the index response header
func setIndex(resp http.ResponseWriter, index uint64) {
resp.Header().Set("X-Consul-Index", strconv.FormatUint(index, 10))
}
// setKnownLeader is used to set the known leader header
func setKnownLeader(resp http.ResponseWriter, known bool) {
s := "true"
if !known {
s = "false"
}
resp.Header().Set("X-Consul-KnownLeader", s)
}
// setLastContact is used to set the last contact header
func setLastContact(resp http.ResponseWriter, last time.Duration) {
lastMsec := uint64(last / time.Millisecond)
resp.Header().Set("X-Consul-LastContact", strconv.FormatUint(lastMsec, 10))
}
// setMeta is used to set the query response meta data
func setMeta(resp http.ResponseWriter, m *structs.QueryMeta) {
setIndex(resp, m.Index)
setLastContact(resp, m.LastContact)
setKnownLeader(resp, m.KnownLeader)
}
// setHeaders is used to set canonical response header fields
func setHeaders(resp http.ResponseWriter, headers map[string]string) {
for field, value := range headers {
resp.Header().Set(http.CanonicalHeaderKey(field), value)
}
}
// parseWait is used to parse the ?wait and ?index query params
// Returns true on error
2014-04-21 19:26:12 +00:00
func parseWait(resp http.ResponseWriter, req *http.Request, b *structs.QueryOptions) bool {
query := req.URL.Query()
if wait := query.Get("wait"); wait != "" {
dur, err := time.ParseDuration(wait)
if err != nil {
2016-02-05 22:06:42 +00:00
resp.WriteHeader(http.StatusBadRequest) // 400
resp.Write([]byte("Invalid wait time"))
return true
}
b.MaxQueryTime = dur
}
if idx := query.Get("index"); idx != "" {
index, err := strconv.ParseUint(idx, 10, 64)
if err != nil {
2016-02-05 22:06:42 +00:00
resp.WriteHeader(http.StatusBadRequest) // 400
resp.Write([]byte("Invalid index"))
return true
}
b.MinQueryIndex = index
}
return false
}
2014-04-21 19:26:12 +00:00
// parseConsistency is used to parse the ?stale and ?consistent query params.
// Returns true on error
func parseConsistency(resp http.ResponseWriter, req *http.Request, b *structs.QueryOptions) bool {
query := req.URL.Query()
if _, ok := query["stale"]; ok {
b.AllowStale = true
}
if _, ok := query["consistent"]; ok {
b.RequireConsistent = true
}
if b.AllowStale && b.RequireConsistent {
2016-02-05 22:06:42 +00:00
resp.WriteHeader(http.StatusBadRequest) // 400
2014-04-21 19:26:12 +00:00
resp.Write([]byte("Cannot specify ?stale with ?consistent, conflicting semantics."))
return true
}
return false
}
// parseDC is used to parse the ?dc query param
func (s *HTTPServer) parseDC(req *http.Request, dc *string) {
if other := req.URL.Query().Get("dc"); other != "" {
*dc = other
} else if *dc == "" {
*dc = s.agent.config.Datacenter
}
}
// parseToken is used to parse the ?token query param or the X-Consul-Token header
2014-08-12 18:35:22 +00:00
func (s *HTTPServer) parseToken(req *http.Request, token *string) {
if other := req.URL.Query().Get("token"); other != "" {
*token = other
2015-02-06 22:10:01 +00:00
return
2014-08-12 18:35:22 +00:00
}
2015-02-06 22:10:01 +00:00
if other := req.Header.Get("X-Consul-Token"); other != "" {
*token = other
return
}
2015-02-06 22:10:01 +00:00
// Set the AtlasACLToken if SCADA
if s.addr == scadaHTTPAddr && s.agent.config.AtlasACLToken != "" {
*token = s.agent.config.AtlasACLToken
return
}
// Set the default ACLToken
*token = s.agent.config.ACLToken
2014-08-12 18:35:22 +00:00
}
// parseSource is used to parse the ?near=<node> query parameter, used for
// sorting by RTT based on a source node. We set the source's DC to the target
// DC in the request, if given, or else the agent's DC.
func (s *HTTPServer) parseSource(req *http.Request, source *structs.QuerySource) {
s.parseDC(req, &source.Datacenter)
if node := req.URL.Query().Get("near"); node != "" {
if node == "_agent" {
source.Node = s.agent.config.NodeName
} else {
source.Node = node
}
}
}
// parse is a convenience method for endpoints that need
// to use both parseWait and parseDC.
2014-04-21 19:26:12 +00:00
func (s *HTTPServer) parse(resp http.ResponseWriter, req *http.Request, dc *string, b *structs.QueryOptions) bool {
s.parseDC(req, dc)
2014-08-12 18:35:22 +00:00
s.parseToken(req, &b.Token)
2014-04-21 19:26:12 +00:00
if parseConsistency(resp, req, b) {
return true
}
return parseWait(resp, req, b)
}