66982a1660
This adds a websocket endpoint for handling `nomad exec`. The endpoint is a websocket interface, as we require a bi-directional streaming (to handle both input and output), which is not very appropriate for plain HTTP 1.0. Using websocket makes implementing the web ui a bit simpler. I considered using golang http hijack capability to treat http request as a plain connection, but the web interface would be too complicated potentially. Furthermore, the API endpoint operates against the raw core nomad exec streaming datastructures, defined in protobuf, with json serializer. Our APIs use json interfaces in general, and protobuf generates json friendly golang structs. Reusing the structs here simplify interface and reduce conversion overhead.
479 lines
14 KiB
Go
479 lines
14 KiB
Go
package agent
|
|
|
|
import (
|
|
"bytes"
|
|
"crypto/tls"
|
|
"encoding/json"
|
|
"fmt"
|
|
"net"
|
|
"net/http"
|
|
"net/http/pprof"
|
|
"os"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/NYTimes/gziphandler"
|
|
assetfs "github.com/elazarl/go-bindata-assetfs"
|
|
"github.com/gorilla/websocket"
|
|
log "github.com/hashicorp/go-hclog"
|
|
"github.com/hashicorp/nomad/helper/tlsutil"
|
|
"github.com/hashicorp/nomad/nomad/structs"
|
|
"github.com/rs/cors"
|
|
"github.com/ugorji/go/codec"
|
|
)
|
|
|
|
const (
|
|
// ErrInvalidMethod is used if the HTTP method is not supported
|
|
ErrInvalidMethod = "Invalid method"
|
|
|
|
// ErrEntOnly is the error returned if accessing an enterprise only
|
|
// endpoint
|
|
ErrEntOnly = "Nomad Enterprise only endpoint"
|
|
)
|
|
|
|
var (
|
|
// Set to false by stub_asset if the ui build tag isn't enabled
|
|
uiEnabled = true
|
|
|
|
// Overridden if the ui build tag isn't enabled
|
|
stubHTML = ""
|
|
|
|
// allowCORS sets permissive CORS headers for a handler
|
|
allowCORS = cors.New(cors.Options{
|
|
AllowedOrigins: []string{"*"},
|
|
AllowedMethods: []string{"HEAD", "GET"},
|
|
AllowedHeaders: []string{"*"},
|
|
})
|
|
)
|
|
|
|
// HTTPServer is used to wrap an Agent and expose it over an HTTP interface
|
|
type HTTPServer struct {
|
|
agent *Agent
|
|
mux *http.ServeMux
|
|
listener net.Listener
|
|
listenerCh chan struct{}
|
|
logger log.Logger
|
|
Addr string
|
|
|
|
wsUpgrader *websocket.Upgrader
|
|
}
|
|
|
|
// NewHTTPServer starts new HTTP server over the agent
|
|
func NewHTTPServer(agent *Agent, config *Config) (*HTTPServer, error) {
|
|
// Start the listener
|
|
lnAddr, err := net.ResolveTCPAddr("tcp", config.normalizedAddrs.HTTP)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
ln, err := config.Listener("tcp", lnAddr.IP.String(), lnAddr.Port)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to start HTTP listener: %v", err)
|
|
}
|
|
|
|
// If TLS is enabled, wrap the listener with a TLS listener
|
|
if config.TLSConfig.EnableHTTP {
|
|
tlsConf, err := tlsutil.NewTLSConfiguration(config.TLSConfig, config.TLSConfig.VerifyHTTPSClient, true)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
tlsConfig, err := tlsConf.IncomingTLSConfig()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
ln = tls.NewListener(tcpKeepAliveListener{ln.(*net.TCPListener)}, tlsConfig)
|
|
}
|
|
|
|
// Create the mux
|
|
mux := http.NewServeMux()
|
|
|
|
wsUpgrader := &websocket.Upgrader{
|
|
ReadBufferSize: 2048,
|
|
WriteBufferSize: 2048,
|
|
}
|
|
|
|
// Create the server
|
|
srv := &HTTPServer{
|
|
agent: agent,
|
|
mux: mux,
|
|
listener: ln,
|
|
listenerCh: make(chan struct{}),
|
|
logger: agent.httpLogger,
|
|
Addr: ln.Addr().String(),
|
|
wsUpgrader: wsUpgrader,
|
|
}
|
|
srv.registerHandlers(config.EnableDebug)
|
|
|
|
// Handle requests with gzip compression
|
|
gzip, err := gziphandler.GzipHandlerWithOpts(gziphandler.MinSize(0))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
go func() {
|
|
defer close(srv.listenerCh)
|
|
http.Serve(ln, gzip(mux))
|
|
}()
|
|
|
|
return srv, nil
|
|
}
|
|
|
|
// 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
|
|
}
|
|
|
|
// Shutdown is used to shutdown the HTTP server
|
|
func (s *HTTPServer) Shutdown() {
|
|
if s != nil {
|
|
s.logger.Debug("shutting down http server")
|
|
s.listener.Close()
|
|
<-s.listenerCh // block until http.Serve has returned.
|
|
}
|
|
}
|
|
|
|
// registerHandlers is used to attach our handlers to the mux
|
|
func (s *HTTPServer) registerHandlers(enableDebug bool) {
|
|
s.mux.HandleFunc("/v1/jobs", s.wrap(s.JobsRequest))
|
|
s.mux.HandleFunc("/v1/jobs/parse", s.wrap(s.JobsParseRequest))
|
|
s.mux.HandleFunc("/v1/job/", s.wrap(s.JobSpecificRequest))
|
|
|
|
s.mux.HandleFunc("/v1/nodes", s.wrap(s.NodesRequest))
|
|
s.mux.HandleFunc("/v1/node/", s.wrap(s.NodeSpecificRequest))
|
|
|
|
s.mux.HandleFunc("/v1/allocations", s.wrap(s.AllocsRequest))
|
|
s.mux.HandleFunc("/v1/allocation/", s.wrap(s.AllocSpecificRequest))
|
|
|
|
s.mux.HandleFunc("/v1/evaluations", s.wrap(s.EvalsRequest))
|
|
s.mux.HandleFunc("/v1/evaluation/", s.wrap(s.EvalSpecificRequest))
|
|
|
|
s.mux.HandleFunc("/v1/deployments", s.wrap(s.DeploymentsRequest))
|
|
s.mux.HandleFunc("/v1/deployment/", s.wrap(s.DeploymentSpecificRequest))
|
|
|
|
s.mux.HandleFunc("/v1/acl/policies", s.wrap(s.ACLPoliciesRequest))
|
|
s.mux.HandleFunc("/v1/acl/policy/", s.wrap(s.ACLPolicySpecificRequest))
|
|
|
|
s.mux.HandleFunc("/v1/acl/bootstrap", s.wrap(s.ACLTokenBootstrap))
|
|
s.mux.HandleFunc("/v1/acl/tokens", s.wrap(s.ACLTokensRequest))
|
|
s.mux.HandleFunc("/v1/acl/token", s.wrap(s.ACLTokenSpecificRequest))
|
|
s.mux.HandleFunc("/v1/acl/token/", s.wrap(s.ACLTokenSpecificRequest))
|
|
|
|
s.mux.Handle("/v1/client/fs/", wrapCORS(s.wrap(s.FsRequest)))
|
|
s.mux.HandleFunc("/v1/client/gc", s.wrap(s.ClientGCRequest))
|
|
s.mux.Handle("/v1/client/stats", wrapCORS(s.wrap(s.ClientStatsRequest)))
|
|
s.mux.Handle("/v1/client/allocation/", wrapCORS(s.wrap(s.ClientAllocRequest)))
|
|
|
|
s.mux.HandleFunc("/v1/agent/self", s.wrap(s.AgentSelfRequest))
|
|
s.mux.HandleFunc("/v1/agent/join", s.wrap(s.AgentJoinRequest))
|
|
s.mux.HandleFunc("/v1/agent/members", s.wrap(s.AgentMembersRequest))
|
|
s.mux.HandleFunc("/v1/agent/force-leave", s.wrap(s.AgentForceLeaveRequest))
|
|
s.mux.HandleFunc("/v1/agent/servers", s.wrap(s.AgentServersRequest))
|
|
s.mux.HandleFunc("/v1/agent/keyring/", s.wrap(s.KeyringOperationRequest))
|
|
s.mux.HandleFunc("/v1/agent/health", s.wrap(s.HealthRequest))
|
|
|
|
s.mux.HandleFunc("/v1/metrics", s.wrap(s.MetricsRequest))
|
|
|
|
s.mux.HandleFunc("/v1/validate/job", s.wrap(s.ValidateJobRequest))
|
|
|
|
s.mux.HandleFunc("/v1/regions", s.wrap(s.RegionListRequest))
|
|
|
|
s.mux.HandleFunc("/v1/status/leader", s.wrap(s.StatusLeaderRequest))
|
|
s.mux.HandleFunc("/v1/status/peers", s.wrap(s.StatusPeersRequest))
|
|
|
|
s.mux.HandleFunc("/v1/search", s.wrap(s.SearchRequest))
|
|
|
|
s.mux.HandleFunc("/v1/operator/raft/", s.wrap(s.OperatorRequest))
|
|
s.mux.HandleFunc("/v1/operator/autopilot/configuration", s.wrap(s.OperatorAutopilotConfiguration))
|
|
s.mux.HandleFunc("/v1/operator/autopilot/health", s.wrap(s.OperatorServerHealth))
|
|
|
|
s.mux.HandleFunc("/v1/system/gc", s.wrap(s.GarbageCollectRequest))
|
|
s.mux.HandleFunc("/v1/system/reconcile/summaries", s.wrap(s.ReconcileJobSummaries))
|
|
|
|
s.mux.HandleFunc("/v1/operator/scheduler/configuration", s.wrap(s.OperatorSchedulerConfiguration))
|
|
|
|
if uiEnabled {
|
|
s.mux.Handle("/ui/", http.StripPrefix("/ui/", handleUI(http.FileServer(&UIAssetWrapper{FileSystem: assetFS()}))))
|
|
} else {
|
|
// Write the stubHTML
|
|
s.mux.HandleFunc("/ui/", func(w http.ResponseWriter, r *http.Request) {
|
|
w.Write([]byte(stubHTML))
|
|
})
|
|
}
|
|
s.mux.Handle("/", handleRootRedirect())
|
|
|
|
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)
|
|
s.mux.HandleFunc("/debug/pprof/trace", pprof.Trace)
|
|
}
|
|
|
|
// Register enterprise endpoints.
|
|
s.registerEnterpriseHandlers()
|
|
}
|
|
|
|
// HTTPCodedError is used to provide the HTTP error code
|
|
type HTTPCodedError interface {
|
|
error
|
|
Code() int
|
|
}
|
|
|
|
type UIAssetWrapper struct {
|
|
FileSystem *assetfs.AssetFS
|
|
}
|
|
|
|
func (fs *UIAssetWrapper) Open(name string) (http.File, error) {
|
|
if file, err := fs.FileSystem.Open(name); err == nil {
|
|
return file, nil
|
|
} else {
|
|
// serve index.html instead of 404ing
|
|
if err == os.ErrNotExist {
|
|
return fs.FileSystem.Open("index.html")
|
|
}
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
func CodedError(c int, s string) HTTPCodedError {
|
|
return &codedError{s, c}
|
|
}
|
|
|
|
type codedError struct {
|
|
s string
|
|
code int
|
|
}
|
|
|
|
func (e *codedError) Error() string {
|
|
return e.s
|
|
}
|
|
|
|
func (e *codedError) Code() int {
|
|
return e.code
|
|
}
|
|
|
|
func handleUI(h http.Handler) http.Handler {
|
|
return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
|
|
header := w.Header()
|
|
header.Add("Content-Security-Policy", "default-src 'none'; connect-src *; img-src 'self' data:; script-src 'self'; style-src 'self' 'unsafe-inline'; form-action 'none'; frame-ancestors 'none'")
|
|
h.ServeHTTP(w, req)
|
|
return
|
|
})
|
|
}
|
|
|
|
func handleRootRedirect() http.Handler {
|
|
return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
|
|
http.Redirect(w, req, "/ui/", 307)
|
|
return
|
|
})
|
|
}
|
|
|
|
// wrap is used to wrap functions to make them more convenient
|
|
func (s *HTTPServer) wrap(handler func(resp http.ResponseWriter, req *http.Request) (interface{}, error)) func(resp http.ResponseWriter, req *http.Request) {
|
|
f := func(resp http.ResponseWriter, req *http.Request) {
|
|
setHeaders(resp, s.agent.config.HTTPAPIResponseHeaders)
|
|
// Invoke the handler
|
|
reqURL := req.URL.String()
|
|
start := time.Now()
|
|
defer func() {
|
|
s.logger.Debug("request complete", "method", req.Method, "path", reqURL, "duration", time.Now().Sub(start))
|
|
}()
|
|
obj, err := handler(resp, req)
|
|
|
|
// Check for an error
|
|
HAS_ERR:
|
|
if err != nil {
|
|
code := 500
|
|
errMsg := err.Error()
|
|
if http, ok := err.(HTTPCodedError); ok {
|
|
code = http.Code()
|
|
} else {
|
|
// RPC errors get wrapped, so manually unwrap by only looking at their suffix
|
|
if strings.HasSuffix(errMsg, structs.ErrPermissionDenied.Error()) {
|
|
errMsg = structs.ErrPermissionDenied.Error()
|
|
code = 403
|
|
} else if strings.HasSuffix(errMsg, structs.ErrTokenNotFound.Error()) {
|
|
errMsg = structs.ErrTokenNotFound.Error()
|
|
code = 403
|
|
}
|
|
}
|
|
|
|
resp.WriteHeader(code)
|
|
resp.Write([]byte(errMsg))
|
|
s.logger.Error("request failed", "method", req.Method, "path", reqURL, "error", err, "code", code)
|
|
return
|
|
}
|
|
|
|
prettyPrint := false
|
|
if v, ok := req.URL.Query()["pretty"]; ok {
|
|
if len(v) > 0 && (len(v[0]) == 0 || v[0] != "0") {
|
|
prettyPrint = true
|
|
}
|
|
}
|
|
|
|
// Write out the JSON object
|
|
if obj != nil {
|
|
var buf bytes.Buffer
|
|
if prettyPrint {
|
|
enc := codec.NewEncoder(&buf, structs.JsonHandlePretty)
|
|
err = enc.Encode(obj)
|
|
if err == nil {
|
|
buf.Write([]byte("\n"))
|
|
}
|
|
} else {
|
|
enc := codec.NewEncoder(&buf, structs.JsonHandle)
|
|
err = enc.Encode(obj)
|
|
}
|
|
if err != nil {
|
|
goto HAS_ERR
|
|
}
|
|
resp.Header().Set("Content-Type", "application/json")
|
|
resp.Write(buf.Bytes())
|
|
}
|
|
}
|
|
return f
|
|
}
|
|
|
|
// decodeBody is used to decode a JSON request body
|
|
func decodeBody(req *http.Request, out interface{}) error {
|
|
dec := json.NewDecoder(req.Body)
|
|
return dec.Decode(&out)
|
|
}
|
|
|
|
// setIndex is used to set the index response header
|
|
func setIndex(resp http.ResponseWriter, index uint64) {
|
|
resp.Header().Set("X-Nomad-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-Nomad-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-Nomad-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
|
|
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 {
|
|
resp.WriteHeader(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 {
|
|
resp.WriteHeader(400)
|
|
resp.Write([]byte("Invalid index"))
|
|
return true
|
|
}
|
|
b.MinQueryIndex = index
|
|
}
|
|
return false
|
|
}
|
|
|
|
// parseConsistency is used to parse the ?stale query params.
|
|
func parseConsistency(req *http.Request, b *structs.QueryOptions) {
|
|
query := req.URL.Query()
|
|
if _, ok := query["stale"]; ok {
|
|
b.AllowStale = true
|
|
}
|
|
}
|
|
|
|
// parsePrefix is used to parse the ?prefix query param
|
|
func parsePrefix(req *http.Request, b *structs.QueryOptions) {
|
|
query := req.URL.Query()
|
|
if prefix := query.Get("prefix"); prefix != "" {
|
|
b.Prefix = prefix
|
|
}
|
|
}
|
|
|
|
// parseRegion is used to parse the ?region query param
|
|
func (s *HTTPServer) parseRegion(req *http.Request, r *string) {
|
|
if other := req.URL.Query().Get("region"); other != "" {
|
|
*r = other
|
|
} else if *r == "" {
|
|
*r = s.agent.config.Region
|
|
}
|
|
}
|
|
|
|
// parseNamespace is used to parse the ?namespace parameter
|
|
func parseNamespace(req *http.Request, n *string) {
|
|
if other := req.URL.Query().Get("namespace"); other != "" {
|
|
*n = other
|
|
} else if *n == "" {
|
|
*n = structs.DefaultNamespace
|
|
}
|
|
}
|
|
|
|
// parseToken is used to parse the X-Nomad-Token param
|
|
func (s *HTTPServer) parseToken(req *http.Request, token *string) {
|
|
if other := req.Header.Get("X-Nomad-Token"); other != "" {
|
|
*token = other
|
|
return
|
|
}
|
|
}
|
|
|
|
// parse is a convenience method for endpoints that need to parse multiple flags
|
|
func (s *HTTPServer) parse(resp http.ResponseWriter, req *http.Request, r *string, b *structs.QueryOptions) bool {
|
|
s.parseRegion(req, r)
|
|
s.parseToken(req, &b.AuthToken)
|
|
parseConsistency(req, b)
|
|
parsePrefix(req, b)
|
|
parseNamespace(req, &b.Namespace)
|
|
return parseWait(resp, req, b)
|
|
}
|
|
|
|
// parseWriteRequest is a convenience method for endpoints that need to parse a
|
|
// write request.
|
|
func (s *HTTPServer) parseWriteRequest(req *http.Request, w *structs.WriteRequest) {
|
|
parseNamespace(req, &w.Namespace)
|
|
s.parseToken(req, &w.AuthToken)
|
|
s.parseRegion(req, &w.Region)
|
|
}
|
|
|
|
// wrapCORS wraps a HandlerFunc in allowCORS and returns a http.Handler
|
|
func wrapCORS(f func(http.ResponseWriter, *http.Request)) http.Handler {
|
|
return allowCORS.Handler(http.HandlerFunc(f))
|
|
}
|