ae8bb7365a
Currently when you submit a manual request to the alloc lifecycle API with a version of Curl that will submit empty bodies, the alloc restart api will fail with an EOF error. This behaviour is undesired, as it is reasonable to not submit a body at all when restarting an entire allocation rather than an individual task. This fixes it by ignoring EOF (not unexpected EOF) errors and treating them as entire task restarts.
543 lines
15 KiB
Go
543 lines
15 KiB
Go
package agent
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"net"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
|
|
"github.com/golang/snappy"
|
|
"github.com/gorilla/websocket"
|
|
cstructs "github.com/hashicorp/nomad/client/structs"
|
|
"github.com/hashicorp/nomad/nomad/structs"
|
|
"github.com/hashicorp/nomad/plugins/drivers"
|
|
"github.com/ugorji/go/codec"
|
|
)
|
|
|
|
const (
|
|
allocNotFoundErr = "allocation not found"
|
|
resourceNotFoundErr = "resource not found"
|
|
)
|
|
|
|
func (s *HTTPServer) AllocsRequest(resp http.ResponseWriter, req *http.Request) (interface{}, error) {
|
|
if req.Method != "GET" {
|
|
return nil, CodedError(405, ErrInvalidMethod)
|
|
}
|
|
|
|
args := structs.AllocListRequest{}
|
|
if s.parse(resp, req, &args.Region, &args.QueryOptions) {
|
|
return nil, nil
|
|
}
|
|
|
|
var out structs.AllocListResponse
|
|
if err := s.agent.RPC("Alloc.List", &args, &out); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
setMeta(resp, &out.QueryMeta)
|
|
if out.Allocations == nil {
|
|
out.Allocations = make([]*structs.AllocListStub, 0)
|
|
}
|
|
for _, alloc := range out.Allocations {
|
|
alloc.SetEventDisplayMessages()
|
|
}
|
|
return out.Allocations, nil
|
|
}
|
|
|
|
func (s *HTTPServer) AllocSpecificRequest(resp http.ResponseWriter, req *http.Request) (interface{}, error) {
|
|
reqSuffix := strings.TrimPrefix(req.URL.Path, "/v1/allocation/")
|
|
|
|
// tokenize the suffix of the path to get the alloc id and find the action
|
|
// invoked on the alloc id
|
|
tokens := strings.Split(reqSuffix, "/")
|
|
if len(tokens) > 2 || len(tokens) < 1 {
|
|
return nil, CodedError(404, resourceNotFoundErr)
|
|
}
|
|
allocID := tokens[0]
|
|
|
|
if len(tokens) == 1 {
|
|
return s.allocGet(allocID, resp, req)
|
|
}
|
|
|
|
switch tokens[1] {
|
|
case "stop":
|
|
return s.allocStop(allocID, resp, req)
|
|
}
|
|
|
|
return nil, CodedError(404, resourceNotFoundErr)
|
|
}
|
|
|
|
func (s *HTTPServer) allocGet(allocID string, resp http.ResponseWriter, req *http.Request) (interface{}, error) {
|
|
if req.Method != "GET" {
|
|
return nil, CodedError(405, ErrInvalidMethod)
|
|
}
|
|
|
|
args := structs.AllocSpecificRequest{
|
|
AllocID: allocID,
|
|
}
|
|
if s.parse(resp, req, &args.Region, &args.QueryOptions) {
|
|
return nil, nil
|
|
}
|
|
|
|
var out structs.SingleAllocResponse
|
|
if err := s.agent.RPC("Alloc.GetAlloc", &args, &out); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
setMeta(resp, &out.QueryMeta)
|
|
if out.Alloc == nil {
|
|
return nil, CodedError(404, "alloc not found")
|
|
}
|
|
|
|
// Decode the payload if there is any
|
|
alloc := out.Alloc
|
|
if alloc.Job != nil && len(alloc.Job.Payload) != 0 {
|
|
decoded, err := snappy.Decode(nil, alloc.Job.Payload)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
alloc = alloc.Copy()
|
|
alloc.Job.Payload = decoded
|
|
}
|
|
alloc.SetEventDisplayMessages()
|
|
|
|
return alloc, nil
|
|
}
|
|
|
|
func (s *HTTPServer) allocStop(allocID string, resp http.ResponseWriter, req *http.Request) (interface{}, error) {
|
|
if !(req.Method == "POST" || req.Method == "PUT") {
|
|
return nil, CodedError(405, ErrInvalidMethod)
|
|
}
|
|
|
|
sr := &structs.AllocStopRequest{
|
|
AllocID: allocID,
|
|
}
|
|
s.parseWriteRequest(req, &sr.WriteRequest)
|
|
|
|
var out structs.AllocStopResponse
|
|
err := s.agent.RPC("Alloc.Stop", &sr, &out)
|
|
return &out, err
|
|
}
|
|
|
|
func (s *HTTPServer) ClientAllocRequest(resp http.ResponseWriter, req *http.Request) (interface{}, error) {
|
|
reqSuffix := strings.TrimPrefix(req.URL.Path, "/v1/client/allocation/")
|
|
|
|
// tokenize the suffix of the path to get the alloc id and find the action
|
|
// invoked on the alloc id
|
|
tokens := strings.Split(reqSuffix, "/")
|
|
if len(tokens) != 2 {
|
|
return nil, CodedError(404, resourceNotFoundErr)
|
|
}
|
|
allocID := tokens[0]
|
|
switch tokens[1] {
|
|
case "stats":
|
|
return s.allocStats(allocID, resp, req)
|
|
case "exec":
|
|
return s.allocExec(allocID, resp, req)
|
|
case "snapshot":
|
|
if s.agent.client == nil {
|
|
return nil, clientNotRunning
|
|
}
|
|
return s.allocSnapshot(allocID, resp, req)
|
|
case "restart":
|
|
return s.allocRestart(allocID, resp, req)
|
|
case "gc":
|
|
return s.allocGC(allocID, resp, req)
|
|
case "signal":
|
|
return s.allocSignal(allocID, resp, req)
|
|
}
|
|
|
|
return nil, CodedError(404, resourceNotFoundErr)
|
|
}
|
|
|
|
func (s *HTTPServer) ClientGCRequest(resp http.ResponseWriter, req *http.Request) (interface{}, error) {
|
|
// Get the requested Node ID
|
|
requestedNode := req.URL.Query().Get("node_id")
|
|
|
|
// Build the request and parse the ACL token
|
|
args := structs.NodeSpecificRequest{
|
|
NodeID: requestedNode,
|
|
}
|
|
s.parse(resp, req, &args.QueryOptions.Region, &args.QueryOptions)
|
|
|
|
// Determine the handler to use
|
|
useLocalClient, useClientRPC, useServerRPC := s.rpcHandlerForNode(requestedNode)
|
|
|
|
// Make the RPC
|
|
var reply structs.GenericResponse
|
|
var rpcErr error
|
|
if useLocalClient {
|
|
rpcErr = s.agent.Client().ClientRPC("Allocations.GarbageCollectAll", &args, &reply)
|
|
} else if useClientRPC {
|
|
rpcErr = s.agent.Client().RPC("ClientAllocations.GarbageCollectAll", &args, &reply)
|
|
} else if useServerRPC {
|
|
rpcErr = s.agent.Server().RPC("ClientAllocations.GarbageCollectAll", &args, &reply)
|
|
} else {
|
|
rpcErr = CodedError(400, "No local Node and node_id not provided")
|
|
}
|
|
|
|
if rpcErr != nil {
|
|
if structs.IsErrNoNodeConn(rpcErr) {
|
|
rpcErr = CodedError(404, rpcErr.Error())
|
|
}
|
|
}
|
|
|
|
return nil, rpcErr
|
|
}
|
|
|
|
func (s *HTTPServer) allocRestart(allocID string, resp http.ResponseWriter, req *http.Request) (interface{}, error) {
|
|
// Build the request and parse the ACL token
|
|
args := structs.AllocRestartRequest{
|
|
AllocID: allocID,
|
|
TaskName: "",
|
|
}
|
|
s.parse(resp, req, &args.QueryOptions.Region, &args.QueryOptions)
|
|
|
|
// Explicitly parse the body separately to disallow overriding AllocID in req Body.
|
|
var reqBody struct {
|
|
TaskName string
|
|
}
|
|
err := json.NewDecoder(req.Body).Decode(&reqBody)
|
|
if err != nil && err != io.EOF {
|
|
return nil, err
|
|
}
|
|
if reqBody.TaskName != "" {
|
|
args.TaskName = reqBody.TaskName
|
|
}
|
|
|
|
// Determine the handler to use
|
|
useLocalClient, useClientRPC, useServerRPC := s.rpcHandlerForAlloc(allocID)
|
|
|
|
// Make the RPC
|
|
var reply structs.GenericResponse
|
|
var rpcErr error
|
|
if useLocalClient {
|
|
rpcErr = s.agent.Client().ClientRPC("Allocations.Restart", &args, &reply)
|
|
} else if useClientRPC {
|
|
rpcErr = s.agent.Client().RPC("ClientAllocations.Restart", &args, &reply)
|
|
} else if useServerRPC {
|
|
rpcErr = s.agent.Server().RPC("ClientAllocations.Restart", &args, &reply)
|
|
} else {
|
|
rpcErr = CodedError(400, "No local Node and node_id not provided")
|
|
}
|
|
|
|
if rpcErr != nil {
|
|
if structs.IsErrNoNodeConn(rpcErr) || structs.IsErrUnknownAllocation(rpcErr) {
|
|
rpcErr = CodedError(404, rpcErr.Error())
|
|
}
|
|
}
|
|
|
|
return reply, rpcErr
|
|
}
|
|
|
|
func (s *HTTPServer) allocGC(allocID string, resp http.ResponseWriter, req *http.Request) (interface{}, error) {
|
|
// Build the request and parse the ACL token
|
|
args := structs.AllocSpecificRequest{
|
|
AllocID: allocID,
|
|
}
|
|
s.parse(resp, req, &args.QueryOptions.Region, &args.QueryOptions)
|
|
|
|
// Determine the handler to use
|
|
useLocalClient, useClientRPC, useServerRPC := s.rpcHandlerForAlloc(allocID)
|
|
|
|
// Make the RPC
|
|
var reply structs.GenericResponse
|
|
var rpcErr error
|
|
if useLocalClient {
|
|
rpcErr = s.agent.Client().ClientRPC("Allocations.GarbageCollect", &args, &reply)
|
|
} else if useClientRPC {
|
|
rpcErr = s.agent.Client().RPC("ClientAllocations.GarbageCollect", &args, &reply)
|
|
} else if useServerRPC {
|
|
rpcErr = s.agent.Server().RPC("ClientAllocations.GarbageCollect", &args, &reply)
|
|
} else {
|
|
rpcErr = CodedError(400, "No local Node and node_id not provided")
|
|
}
|
|
|
|
if rpcErr != nil {
|
|
if structs.IsErrNoNodeConn(rpcErr) || structs.IsErrUnknownAllocation(rpcErr) {
|
|
rpcErr = CodedError(404, rpcErr.Error())
|
|
}
|
|
}
|
|
|
|
return nil, rpcErr
|
|
}
|
|
|
|
func (s *HTTPServer) allocSignal(allocID string, resp http.ResponseWriter, req *http.Request) (interface{}, error) {
|
|
if !(req.Method == "POST" || req.Method == "PUT") {
|
|
return nil, CodedError(405, ErrInvalidMethod)
|
|
}
|
|
|
|
// Build the request and parse the ACL token
|
|
args := structs.AllocSignalRequest{}
|
|
err := decodeBody(req, &args)
|
|
if err != nil {
|
|
return nil, CodedError(400, fmt.Sprintf("Failed to decode body: %v", err))
|
|
}
|
|
s.parse(resp, req, &args.QueryOptions.Region, &args.QueryOptions)
|
|
args.AllocID = allocID
|
|
|
|
// Determine the handler to use
|
|
useLocalClient, useClientRPC, useServerRPC := s.rpcHandlerForAlloc(allocID)
|
|
|
|
// Make the RPC
|
|
var reply structs.GenericResponse
|
|
var rpcErr error
|
|
if useLocalClient {
|
|
rpcErr = s.agent.Client().ClientRPC("Allocations.Signal", &args, &reply)
|
|
} else if useClientRPC {
|
|
rpcErr = s.agent.Client().RPC("ClientAllocations.Signal", &args, &reply)
|
|
} else if useServerRPC {
|
|
rpcErr = s.agent.Server().RPC("ClientAllocations.Signal", &args, &reply)
|
|
} else {
|
|
rpcErr = CodedError(400, "No local Node and node_id not provided")
|
|
}
|
|
|
|
if rpcErr != nil {
|
|
if structs.IsErrNoNodeConn(rpcErr) || structs.IsErrUnknownAllocation(rpcErr) {
|
|
rpcErr = CodedError(404, rpcErr.Error())
|
|
}
|
|
}
|
|
|
|
return reply, rpcErr
|
|
}
|
|
|
|
func (s *HTTPServer) allocSnapshot(allocID string, resp http.ResponseWriter, req *http.Request) (interface{}, error) {
|
|
var secret string
|
|
s.parseToken(req, &secret)
|
|
if !s.agent.Client().ValidateMigrateToken(allocID, secret) {
|
|
return nil, structs.ErrPermissionDenied
|
|
}
|
|
|
|
allocFS, err := s.agent.Client().GetAllocFS(allocID)
|
|
if err != nil {
|
|
return nil, fmt.Errorf(allocNotFoundErr)
|
|
}
|
|
if err := allocFS.Snapshot(resp); err != nil {
|
|
return nil, fmt.Errorf("error making snapshot: %v", err)
|
|
}
|
|
return nil, nil
|
|
}
|
|
|
|
func (s *HTTPServer) allocStats(allocID string, resp http.ResponseWriter, req *http.Request) (interface{}, error) {
|
|
|
|
// Build the request and parse the ACL token
|
|
task := req.URL.Query().Get("task")
|
|
args := cstructs.AllocStatsRequest{
|
|
AllocID: allocID,
|
|
Task: task,
|
|
}
|
|
s.parse(resp, req, &args.QueryOptions.Region, &args.QueryOptions)
|
|
|
|
// Determine the handler to use
|
|
useLocalClient, useClientRPC, useServerRPC := s.rpcHandlerForAlloc(allocID)
|
|
|
|
// Make the RPC
|
|
var reply cstructs.AllocStatsResponse
|
|
var rpcErr error
|
|
if useLocalClient {
|
|
rpcErr = s.agent.Client().ClientRPC("Allocations.Stats", &args, &reply)
|
|
} else if useClientRPC {
|
|
rpcErr = s.agent.Client().RPC("ClientAllocations.Stats", &args, &reply)
|
|
} else if useServerRPC {
|
|
rpcErr = s.agent.Server().RPC("ClientAllocations.Stats", &args, &reply)
|
|
} else {
|
|
rpcErr = CodedError(400, "No local Node and node_id not provided")
|
|
}
|
|
|
|
if rpcErr != nil {
|
|
if structs.IsErrNoNodeConn(rpcErr) || structs.IsErrUnknownAllocation(rpcErr) {
|
|
rpcErr = CodedError(404, rpcErr.Error())
|
|
}
|
|
}
|
|
|
|
return reply.Stats, rpcErr
|
|
}
|
|
|
|
func (s *HTTPServer) allocExec(allocID string, resp http.ResponseWriter, req *http.Request) (interface{}, error) {
|
|
// Build the request and parse the ACL token
|
|
task := req.URL.Query().Get("task")
|
|
cmdJsonStr := req.URL.Query().Get("command")
|
|
var command []string
|
|
err := json.Unmarshal([]byte(cmdJsonStr), &command)
|
|
if err != nil {
|
|
// this shouldn't happen, []string is always be serializable to json
|
|
return nil, fmt.Errorf("failed to marshal command into json: %v", err)
|
|
}
|
|
|
|
ttyB := false
|
|
if tty := req.URL.Query().Get("tty"); tty != "" {
|
|
ttyB, err = strconv.ParseBool(tty)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("tty value is not a boolean: %v", err)
|
|
}
|
|
}
|
|
|
|
args := cstructs.AllocExecRequest{
|
|
AllocID: allocID,
|
|
Task: task,
|
|
Cmd: command,
|
|
Tty: ttyB,
|
|
}
|
|
s.parse(resp, req, &args.QueryOptions.Region, &args.QueryOptions)
|
|
|
|
conn, err := s.wsUpgrader.Upgrade(resp, req, nil)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to upgrade connection: %v", err)
|
|
}
|
|
|
|
return s.execStreamImpl(conn, &args)
|
|
}
|
|
|
|
func (s *HTTPServer) execStreamImpl(ws *websocket.Conn, args *cstructs.AllocExecRequest) (interface{}, error) {
|
|
allocID := args.AllocID
|
|
method := "Allocations.Exec"
|
|
|
|
// Get the correct handler
|
|
localClient, remoteClient, localServer := s.rpcHandlerForAlloc(allocID)
|
|
var handler structs.StreamingRpcHandler
|
|
var handlerErr error
|
|
if localClient {
|
|
handler, handlerErr = s.agent.Client().StreamingRpcHandler(method)
|
|
} else if remoteClient {
|
|
handler, handlerErr = s.agent.Client().RemoteStreamingRpcHandler(method)
|
|
} else if localServer {
|
|
handler, handlerErr = s.agent.Server().StreamingRpcHandler(method)
|
|
}
|
|
|
|
if handlerErr != nil {
|
|
return nil, CodedError(500, handlerErr.Error())
|
|
}
|
|
|
|
// Create a pipe connecting the (possibly remote) handler to the http response
|
|
httpPipe, handlerPipe := net.Pipe()
|
|
decoder := codec.NewDecoder(httpPipe, structs.MsgpackHandle)
|
|
encoder := codec.NewEncoder(httpPipe, structs.MsgpackHandle)
|
|
|
|
// Create a goroutine that closes the pipe if the connection closes.
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
go func() {
|
|
<-ctx.Done()
|
|
httpPipe.Close()
|
|
|
|
// don't close ws - wait to drain messages
|
|
}()
|
|
|
|
// Create a channel that decodes the results
|
|
errCh := make(chan HTTPCodedError, 2)
|
|
|
|
// stream response
|
|
go func() {
|
|
defer cancel()
|
|
|
|
// Send the request
|
|
if err := encoder.Encode(args); err != nil {
|
|
errCh <- CodedError(500, err.Error())
|
|
return
|
|
}
|
|
|
|
go forwardExecInput(encoder, ws, errCh)
|
|
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
errCh <- nil
|
|
return
|
|
default:
|
|
}
|
|
|
|
var res cstructs.StreamErrWrapper
|
|
err := decoder.Decode(&res)
|
|
if isClosedError(err) {
|
|
ws.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""))
|
|
errCh <- nil
|
|
return
|
|
}
|
|
|
|
if err != nil {
|
|
errCh <- CodedError(500, err.Error())
|
|
return
|
|
}
|
|
decoder.Reset(httpPipe)
|
|
|
|
if err := res.Error; err != nil {
|
|
code := 500
|
|
if err.Code != nil {
|
|
code = int(*err.Code)
|
|
}
|
|
errCh <- CodedError(code, err.Error())
|
|
return
|
|
}
|
|
|
|
if err := ws.WriteMessage(websocket.TextMessage, res.Payload); err != nil {
|
|
errCh <- CodedError(500, err.Error())
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
|
|
// start streaming request to streaming RPC - returns when streaming completes or errors
|
|
handler(handlerPipe)
|
|
// stop streaming background goroutines for streaming - but not websocket activity
|
|
cancel()
|
|
// retreieve any error and/or wait until goroutine stop and close errCh connection before
|
|
// closing websocket connection
|
|
codedErr := <-errCh
|
|
|
|
if isClosedError(codedErr) {
|
|
codedErr = nil
|
|
} else if codedErr != nil {
|
|
ws.WriteMessage(websocket.CloseMessage,
|
|
websocket.FormatCloseMessage(toWsCode(codedErr.Code()), codedErr.Error()))
|
|
}
|
|
ws.Close()
|
|
|
|
return nil, codedErr
|
|
}
|
|
|
|
func toWsCode(httpCode int) int {
|
|
switch httpCode {
|
|
case 500:
|
|
return websocket.CloseInternalServerErr
|
|
default:
|
|
// placeholder error code
|
|
return websocket.ClosePolicyViolation
|
|
}
|
|
}
|
|
|
|
func isClosedError(err error) bool {
|
|
if err == nil {
|
|
return false
|
|
}
|
|
|
|
return err == io.EOF ||
|
|
err == io.ErrClosedPipe ||
|
|
strings.Contains(err.Error(), "closed") ||
|
|
strings.Contains(err.Error(), "EOF")
|
|
}
|
|
|
|
// forwardExecInput forwards exec input (e.g. stdin) from websocket connection
|
|
// to the streaming RPC connection to client
|
|
func forwardExecInput(encoder *codec.Encoder, ws *websocket.Conn, errCh chan<- HTTPCodedError) {
|
|
for {
|
|
sf := &drivers.ExecTaskStreamingRequestMsg{}
|
|
err := ws.ReadJSON(sf)
|
|
if err == io.EOF {
|
|
return
|
|
}
|
|
|
|
if err != nil {
|
|
errCh <- CodedError(500, err.Error())
|
|
return
|
|
}
|
|
|
|
err = encoder.Encode(sf)
|
|
if err != nil {
|
|
errCh <- CodedError(500, err.Error())
|
|
}
|
|
}
|
|
}
|