simulator.go

package main

import (
	"encoding/json"
	"fmt"
	"io/ioutil"
	"net"
	"net/http"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/fsouza/go-dockerclient"
	"gopkg.in/inconshreveable/log15.v2"
)

// simulationResult represents the results of a simulation run, containing
// various metadata as well as possibly multiple sub-results in case where
// the same simulator tested multiple things in one go.
type simulationResult struct {
	Start   time.Time `json:"start"`           // Time instance when the simulation ended
	End     time.Time `json:"end"`             // Time instance when the simulation ended
	Success bool      `json:"success"`         // Whether the entire simulation succeeded
	Error   error     `json:"error,omitempty"` // Potential hive failure during simulation

	Subresults []simulationSubresult `json:"subresults,omitempty"` // Optional list of subresults to report

}

type simulationResultSummary struct {
	Start   time.Time `json:"start"`           // Time instance when the simulation ended
	End     time.Time `json:"end"`             // Time instance when the simulation ended
	Success bool      `json:"success"`         // Whether the entire simulation succeeded
	Error   error     `json:"error,omitempty"` // Potential hive failure during simulation

	summaryData
}

// simulationSubresult represents a sub-test a simulation may run and report.
type simulationSubresult struct {
	Name string `json:"name"` // Unique name for a sub-test within a simulation

	Success bool            `json:"success"`           // Whether the sub-test succeeded or not
	Error   string          `json:"error,omitempty"`   // Textual details to explain a failure
	Details json.RawMessage `json:"details,omitempty"` // Structured infos a tester mightw wish to surface
}

// simulateClients runs a batch of simulation tests matched by simulatorPattern
// against a set of clients matching clientPattern, where  the simulator decides
// which of those clients to invoke
func simulateClients(daemon *docker.Client, clientList []string, simulatorPattern string, overrides []string, cacher *buildCacher) (map[string]map[string]*simulationResult, error) {
	// Build all the clients matching the validation pattern
	log15.Info("building clients for simulation", "pattern", clientList)
	clients, err := buildClients(daemon, clientList, cacher)
	if err != nil {
		return nil, err
	}
	if len(clients) == 0 {
		return nil, errNoMatchingClients
	}

	// Build all pseudo clients. pseudo-clients need to be available
	// to simulators. pseudo-clients play the role of special types
	// of actor in a network, such as specific types of malicious actor,
	// network relays, for example.
	pseudos, err := buildPseudoClients(daemon, "pseudo", cacher)
	if err != nil {
		return nil, err
	}
	// Build all the simulators known to the test harness
	log15.Info("building simulators for testing", "pattern", simulatorPattern)
	simulators, err := buildSimulators(daemon, simulatorPattern, cacher)
	if err != nil {
		return nil, err
	}

	// The results are a map of clients=>simulators=>results
	results := make(map[string]map[string]*simulationResult)

	//build the per-client simulator result set
	for client := range clients {
		results[client] = make(map[string]*simulationResult)
	}

	//set the end time of the test
	defer func() {
		for _, cv := range results {
			for _, sv := range cv {
				sv.End = time.Now()
			}
		}
	}()

	for simulator, simulatorImage := range simulators {

		logdir, err := makeTestOutputDirectory(strings.Replace(simulator, string(filepath.Separator), "_", -1), "simulator", clients)
		if err != nil {
			return nil, err
		}

		logger := log15.New("simulator", simulator)

		for client := range clients {
			results[client][simulator] = &simulationResult{
				Start: time.Now(),
			}

		}

		err = simulate(daemon, clients, pseudos, simulatorImage, simulator, overrides, logger, logdir, results) //filepath.Join(logdir, strings.Replace(client, string(filepath.Separator), "_", -1)))
		if err != nil {
			return nil, err
		}

	}
	return results, nil
}

// simulate starts a simulator service locally, starts a controlling container
// and executes its commands until torn down. The exit status of the controller
// container will signal whether the simulation passed or failed.
func simulate(daemon *docker.Client, clients map[string]string, pseudos map[string]string, simulator string, simulatorLabel string, overrides []string, logger log15.Logger, logdir string, results map[string]map[string]*simulationResult) error {
	logger.Info("running client simulation")

	// Start the simulator HTTP API
	sim, err := startSimulatorAPI(daemon, clients, pseudos, simulator, simulatorLabel, overrides, logger, logdir, results)
	if err != nil {
		logger.Error("failed to start simulator API", "error", err)
		return err

	}
	defer sim.Close()

	// Start the simulator controller container
	logger.Debug("creating simulator container")
	hostConfig := &docker.HostConfig{Privileged: true, CapAdd: []string{"SYS_PTRACE"}, SecurityOpt: []string{"seccomp=unconfined"}}
	sc, err := daemon.CreateContainer(docker.CreateContainerOptions{
		Config: &docker.Config{
			Image: simulator,
			Env: []string{"HIVE_SIMULATOR=http://" + sim.listener.Addr().String(),
				"HIVE_DEBUG=" + strconv.FormatBool(*hiveDebug),
				"HIVE_PARALLELISM=" + fmt.Sprintf("%d", *simulatorParallelism),
			},
		},
		HostConfig: hostConfig,
	})

	if err != nil {
		logger.Error("failed to create simulator", "error", err)
		return err
	}

	slogger := logger.New("id", sc.ID[:8])
	slogger.Debug("created simulator container")
	defer func() {
		slogger.Debug("deleting simulator container")
		if err := daemon.RemoveContainer(docker.RemoveContainerOptions{ID: sc.ID, Force: true}); err != nil {
			slogger.Error("failed to delete simulator container", "error", err)
		}
	}()

	// Finish configuring the HTTP webserver with the controlled container
	sim.runner = sc

	// Start the tester container and wait until it finishes
	slogger.Debug("running simulator container")
	waiter, err := runContainer(daemon, sc.ID, slogger, filepath.Join(logdir, "simulator.log"), false, *loglevelFlag)
	if err != nil {
		slogger.Error("failed to run simulator", "error", err)
		return err
	}
	waiter.Wait()

	return nil

}

// startSimulatorAPI starts an HTTP webserver listening for simulator commands
// on the docker bridge and executing them until it is torn down.
func startSimulatorAPI(daemon *docker.Client, clients map[string]string, pseudos map[string]string, simulator string, simulatorLabel string, overrides []string, logger log15.Logger, logdir string, results map[string]map[string]*simulationResult) (*simulatorAPIHandler, error) {
	// Find the IP address of the host container
	logger.Debug("looking up docker bridge IP")
	bridge, err := lookupBridgeIP(logger)
	if err != nil {
		logger.Error("failed to lookup bridge IP", "error", err)
		return nil, err
	}
	logger.Debug("docker bridge IP found", "ip", bridge)

	// Start a tiny API webserver for simulators to coordinate with
	logger.Debug("opening TCP socket for simulator")

	addr, _ := net.ResolveTCPAddr("tcp4", fmt.Sprintf("%s:0", bridge))
	listener, err := net.ListenTCP("tcp4", addr)
	if err != nil {
		logger.Error("failed to listen on bridge adapter", "error", err)
		return nil, err
	}
	logger.Debug("listening for simulator commands", "ip", bridge, "port", listener.Addr().(*net.TCPAddr).Port)

	// Serve connections until the listener is terminated
	logger.Debug("starting simulator API server")
	sim := &simulatorAPIHandler{
		listener:         listener,
		daemon:           daemon,
		logger:           logger,
		logdir:           logdir,
		availableClients: clients,
		availablePseudos: pseudos,
		simulator:        simulator,
		simulatorLabel:   simulatorLabel,
		overrides:        overrides,
		nodes:            make(map[string]*containerInfo),
		timedOutNodes:    make(map[string]*containerInfo),
		result:           results, //the simulator now has access to a map of results-by-client. The simulator decides which clients to run/
	}
	go sim.CheckTimeout()
	go http.Serve(listener, sim)

	return sim, nil
}

type containerInfo struct {
	container  *docker.Container
	name       string
	timeout    time.Time
	useTimeout bool
}

// simulatorAPIHandler is the HTTP request handler directing the docker engine
// with the commands from the simulator runner.
type simulatorAPIHandler struct {
	listener *net.TCPListener

	daemon           *docker.Client
	logger           log15.Logger
	logdir           string
	availableClients map[string]string //the client filter specified by the host. Simulations may not execute other clients.
	availablePseudos map[string]string //all pseudo client images
	simulator        string            //the image name
	simulatorLabel   string            //the simulator label
	overrides        []string
	autoID           uint32

	runner        *docker.Container
	nodes         map[string]*containerInfo               // the running containers, keyed by container.ID[:8]
	timedOutNodes map[string]*containerInfo               // timed-out containers
	result        map[string]map[string]*simulationResult //simulation result log per client name
	lock          sync.RWMutex
}

// CheckTimeout is a goroutine that checks if the timeout has passed and stops
// container if it has.
func (h *simulatorAPIHandler) CheckTimeout() {
	for {
		h.lock.Lock()
		for id, cInfo := range h.nodes {
			var err error

			cInfo.container, err = h.daemon.InspectContainer(cInfo.container.ID)
			if err != nil {
				//container already gone
				h.logger.Info("Container already deleted. ", "Container", cInfo.container.ID)

			} else {
				if !cInfo.container.State.Running || (time.Now().Sub(cInfo.timeout) >= 0 && cInfo.useTimeout) {

					h.logger.Info("Timing out. ", "Running", cInfo.container.State.Running)
					h.timeoutContainer(id, nil)

					// remember this container, for when the subresult comes in later
					h.timedOutNodes[id] = cInfo
				}

			}

		}
		h.lock.Unlock()
		time.Sleep(timeoutCheckDuration)
	}
}

// timeoutContainer terminates a container. OBS! It assumes that the caller already holds h.lock
func (h *simulatorAPIHandler) timeoutContainer(id string, w http.ResponseWriter) {
	containerInfo, ok := h.nodes[id]

	if !ok {
		h.logger.Error("unknown client deletion requested", "id", id)
		if w != nil {
			http.Error(w, "not found", http.StatusNotFound)
		}
		return
	}
	delete(h.nodes, id)
	h.logger.Debug("deleting client container on timeout", "id", id)
	if err := h.daemon.RemoveContainer(docker.RemoveContainerOptions{ID: containerInfo.container.ID, Force: true}); err != nil {
		h.logger.Error("failed to delete client ", "id", id, "error", err)
		if w != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
		}
	}
}

// terminateContainer terminates a container. OBS! It assumes that the caller already holds h.lock
func (h *simulatorAPIHandler) terminateContainer(id string, w http.ResponseWriter) {
	containerInfo, ok := h.nodes[id]
	if !ok {
		h.logger.Error("unknown client deletion requested", "id", id)
		if w != nil {
			http.Error(w, "not found", http.StatusNotFound)
		}
		return
	}
	delete(h.nodes, id)
	h.logger.Debug("deleting client container", "id", id)
	if err := h.daemon.RemoveContainer(docker.RemoveContainerOptions{ID: containerInfo.container.ID, Force: true}); err != nil {
		h.logger.Error("failed to delete client ", "id", id, "error", err)
		if w != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
		}
	}
}

func (h *simulatorAPIHandler) newNode(w http.ResponseWriter, r *http.Request, logger log15.Logger, available map[string]string, checkliveness bool, useTimeout bool) {
	// A new node startup was requested, fetch any envvar overrides from simulators
	r.ParseForm()
	envs := make(map[string]string)
	for key, vals := range r.Form {
		envs[key] = vals[0]
	}

	//the simulation controller needs to tell us now what client to run for the test
	clientName, in := envs["CLIENT"]
	if !in {
		logger.Error("Missing client type", "error", nil)
		http.Error(w, "Missing client type", http.StatusBadRequest)
		return
	}

	//default the loglevel to the simulator log level setting (different from the sysem log level setting)
	logLevel := *simloglevelFlag
	logLevelString, in := envs["HIVE_LOGLEVEL"]
	if !in {
		envs["HIVE_LOGLEVEL"] = strconv.Itoa(logLevel)
	} else {
		var err error
		if logLevel, err = strconv.Atoi(logLevelString); err != nil {
			logger.Error("Simulator client HIVE_LOGLEVEL is not an integer", "error", nil)
		}
	}

	//the simulation host may prevent or be unaware of the simulation controller's requested client
	imageName, in := available[clientName]
	if !in {
		logger.Error("Unknown or forbidden client type", "error", nil)
		http.Error(w, "Unknown or forbidden client type", http.StatusBadRequest)
		return
	}

	// Create and start the requested client container
	logger.Debug("starting new client")
	container, err := createClientContainer(h.daemon, imageName, h.simulator, h.runner, h.overrides, envs)
	if err != nil {
		logger.Error("failed to create client", "error", err)
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	containerID := container.ID[:8]
	containerIP := ""
	containerMAC := ""
	logger = logger.New("client started with id", containerID)

	logfile := fmt.Sprintf("client-%s.log", containerID)

	waiter, err := runContainer(h.daemon, container.ID, logger, filepath.Join(h.logdir, strings.Replace(clientName, string(filepath.Separator), "_", -1), logfile), false, logLevel)
	if err != nil {
		logger.Error("failed to start client", "error", err)
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	go func() {
		// Ensure the goroutine started by runContainer exits, so that
		// its resources (e.g. the logfile it creates) can be garbage
		// collected.
		err := waiter.Wait()
		if err == nil {
			logger.Debug("client container finished cleanly")
		} else {
			logger.Error("client container finished with error", "error", err)
		}
	}()

	// Wait for the HTTP/RPC socket to open or the container to fail
	start := time.Now()
	for {
		// Update the container state
		container, err = h.daemon.InspectContainer(container.ID)
		if err != nil {
			logger.Error("failed to inspect client", "error", err)
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}
		if !container.State.Running {
			logger.Error("client container terminated")
			http.Error(w, "terminated unexpectedly", http.StatusInternalServerError)
			return
		}

		containerIP = container.NetworkSettings.IPAddress
		containerMAC = container.NetworkSettings.MacAddress
		if checkliveness {
			logger.Debug("Checking container online....")
			// Container seems to be alive, check whether the RPC is accepting connections
			if conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", containerIP, 8545)); err == nil {
				logger.Debug("client container online", "time", time.Since(start))
				conn.Close()
				break
			}
		} else {
			break
		}

		time.Sleep(100 * time.Millisecond)
	}
	h.lock.Lock()
	h.nodes[containerID] = &containerInfo{
		container:  container,
		name:       clientName,
		timeout:    time.Now().Add(dockerTimeoutDuration),
		useTimeout: useTimeout,
	}
	h.lock.Unlock()
	//  Container online and responsive, return its ID, IP and MAC for later reference
	fmt.Fprintf(w, "%s@%s@%s", containerID, containerIP, containerMAC)
	return
}

// ServeHTTP handles all the simulator API requests and executes them.
func (h *simulatorAPIHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	logger := h.logger.New("req-id", atomic.AddUint32(&h.autoID, 1))
	logger.Debug("new simulator request", "from", r.RemoteAddr, "method", r.Method, "endpoint", r.URL.Path)

	switch r.Method {
	case "GET":
		// Information retrieval, fetch whatever's needed and return it
		switch {
		case r.URL.Path == "/docker":
			// Docker infos requested, gather and send them back
			info, err := h.daemon.Info()
			if err != nil {
				logger.Error("failed to gather docker infos", "error", err)
				http.Error(w, err.Error(), http.StatusInternalServerError)
				return
			}
			out, _ := json.MarshalIndent(info, "", "  ")
			fmt.Fprintf(w, "%s\n", out)

		case strings.HasPrefix(r.URL.Path, "/nodes/"):
			// Node IP retrieval requested
			id := strings.TrimPrefix(r.URL.Path, "/nodes/")
			h.lock.Lock()
			containerInfo, ok := h.nodes[id]
			h.lock.Unlock()
			if !ok {
				logger.Error("unknown client requested", "id", id)
				http.Error(w, "not found", http.StatusNotFound)
				return
			}
			container, err := h.daemon.InspectContainer(containerInfo.container.ID)
			if err != nil {
				logger.Error("failed to inspect client", "error", err)
				http.Error(w, err.Error(), http.StatusInternalServerError)
				return
			}
			fmt.Fprintf(w, "%s", container.NetworkSettings.IPAddress)

		//docker exec container bash -c 'echo "$ENV_VAR"'
		case strings.HasPrefix(r.URL.Path, "/enodes/"):
			// Node IP retrieval requested
			id := strings.TrimPrefix(r.URL.Path, "/enodes/")
			h.lock.Lock()
			containerInfo, ok := h.nodes[id]
			h.lock.Unlock()
			if !ok {
				logger.Error("unknown client for enode", "id", id)
				http.Error(w, "not found", http.StatusNotFound)
				return
			}

			exec, err := h.daemon.CreateExec(docker.CreateExecOptions{
				AttachStdout: true,
				AttachStderr: false,
				Tty:          false,
				Cmd:          []string{"/enode.sh"},
				Container:    containerInfo.container.ID,
			})
			if err != nil {
				logger.Error("failed to create target enode exec", "error", err)
				http.Error(w, err.Error(), http.StatusInternalServerError)
				return
			}

			err = h.daemon.StartExec(exec.ID, docker.StartExecOptions{
				Detach:       false,
				OutputStream: w,
			})
			if err != nil {
				logger.Error("failed to start target enode exec", "error", err)
				http.Error(w, err.Error(), http.StatusInternalServerError)
				return
			}

		case strings.HasPrefix(r.URL.Path, "/clients"):
			w.Header().Set("Content-Type", "application/json")
			clients := make([]string, 0, len(h.availableClients))
			for client := range h.availableClients {
				clients = append(clients, client)
			}
			json.NewEncoder(w).Encode(clients)

		default:
			http.Error(w, "not found", http.StatusNotFound)
		}

	case "POST":
		// Data mutation, execute the request and return the results
		switch r.URL.Path {
		case "/nodes":

			h.newNode(w, r, logger, h.availableClients, true, true)
			return
		case "/pseudos":

			h.newNode(w, r, logger, h.availablePseudos, false, false)
			return
		case "/logs":
			body, _ := ioutil.ReadAll(r.Body)
			h.logger.Info("message from simulator", "log", string(body))

		case "/subresults":
			// Parse the subresult field into a hive struct
			r.ParseMultipartForm(1024 * 1024)

			success, err := strconv.ParseBool(r.Form.Get("success"))
			if err != nil {
				http.Error(w, err.Error(), http.StatusBadRequest)
				return
			}

			//If there has been a failure, update the whole test result
			//which at present means updating the result set for each
			//known client. TODO: the output format should be
			//re-arranged so that it is grouped first by test and then by client instance type
			if !success {
				for _, resultset := range h.result {
					resultset[h.simulatorLabel].Success = false
				}
			}

			nodeid := r.Form.Get("nodeid")
			if err != nil {
				http.Error(w, err.Error(), http.StatusBadRequest)
				return
			}
			var details json.RawMessage
			if blob := r.Form.Get("details"); blob != "" {
				if err := json.Unmarshal([]byte(blob), &details); err != nil {
					http.Error(w, err.Error(), http.StatusBadRequest)
					return
				}
			}
			// If everything parsed correctly, append the subresult
			h.lock.Lock()
			containerInfo, exist := h.nodes[nodeid]
			if !exist {
				// Add an error even so
				if containerInfo, exist := h.timedOutNodes[nodeid]; exist {
					delete(h.timedOutNodes, nodeid)
					res := h.result[containerInfo.name][h.simulatorLabel]
					res.Subresults = append(res.Subresults, simulationSubresult{
						Name:    r.Form.Get("name"),
						Success: success,
						Error:   fmt.Sprintf("%s (killed after timeout by Hive)", r.Form.Get("error")),
						Details: details,
					})
				}
				h.lock.Unlock()
				http.Error(w, fmt.Sprintf("unknown node %v", nodeid), http.StatusBadRequest)
				return
			}
			res := h.result[containerInfo.name][h.simulatorLabel]
			res.Subresults = append(res.Subresults, simulationSubresult{
				Name:    r.Form.Get("name"),
				Success: success,
				Error:   r.Form.Get("error"),
				Details: details,
			})
			// Also terminate the container now
			delete(h.nodes, nodeid)
			logger.Debug("deleting client container", "id", nodeid)
			h.lock.Unlock()
			if err := h.daemon.RemoveContainer(docker.RemoveContainerOptions{ID: containerInfo.container.ID, Force: true}); err != nil {
				logger.Error("failed to delete client ", "id", nodeid, "error", err)
				http.Error(w, err.Error(), http.StatusInternalServerError)
				return
			}

		default:
			http.Error(w, "not found", http.StatusNotFound)
		}

	case "DELETE":
		// Data deletion, execute the request and return the results
		switch {
		case strings.HasPrefix(r.URL.Path, "/nodes/"):
			// Node deletion requested
			id := strings.TrimPrefix(r.URL.Path, "/nodes/")

			h.lock.Lock()
			h.terminateContainer(id, w)
			h.lock.Unlock()

		default:
			http.Error(w, "not found", http.StatusNotFound)
		}

	default:
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
	}
}

// Close terminates all running containers and tears down the API server.
func (h *simulatorAPIHandler) Close() {
	h.logger.Debug("terminating simulator server")
	h.listener.Close()

	for _, containerInfo := range h.nodes {
		id := containerInfo.container.ID
		h.logger.Debug("deleting client container", "id", id[:8])
		if err := h.daemon.RemoveContainer(docker.RemoveContainerOptions{ID: id, Force: true}); err != nil {
			h.logger.Error("failed to delete client container", "id", id[:8], "error", err)
		}
	}
}