node: refactor package node (#21105)

This PR significantly changes the APIs for instantiating Ethereum nodes in a Go program. The new APIs are not backwards-compatible, but we feel that this is made up for by the much simpler way of registering services on node.Node. You can find more information and rationale in the design document: https://gist.github.com/renaynay/5bec2de19fde66f4d04c535fd24f0775. There is also a new feature in Node's Go API: it is now possible to register arbitrary handlers on the user-facing HTTP server. In geth, this facility is used to enable GraphQL. There is a single minor change relevant for geth users in this PR: The GraphQL API is no longer available separately from the JSON-RPC HTTP server. If you want GraphQL, you need to enable it using the ./geth --http --graphql flag combination. The --graphql.port and --graphql.addr flags are no longer available.
2020-08-03 19:40:46 +02:00
parent b2b14e6ce3
commit c0c01612e9
63 changed files with 2606 additions and 2887 deletions
--- a/node/rpcstack.go
+++ b/node/rpcstack.go
@ -18,17 +18,304 @@ package node

 import (
 	"compress/gzip"
+	"context"
+	"fmt"
 	"io"
 	"io/ioutil"
 	"net"
 	"net/http"
+	"sort"
 	"strings"
 	"sync"
+	"sync/atomic"

 	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/rs/cors"
 )

+// httpConfig is the JSON-RPC/HTTP configuration.
+type httpConfig struct {
+	Modules            []string
+	CorsAllowedOrigins []string
+	Vhosts             []string
+}
+
+// wsConfig is the JSON-RPC/Websocket configuration
+type wsConfig struct {
+	Origins []string
+	Modules []string
+}
+
+type rpcHandler struct {
+	http.Handler
+	server *rpc.Server
+}
+
+type httpServer struct {
+	log      log.Logger
+	timeouts rpc.HTTPTimeouts
+	mux      http.ServeMux // registered handlers go here
+
+	mu       sync.Mutex
+	server   *http.Server
+	listener net.Listener // non-nil when server is running
+
+	// HTTP RPC handler things.
+	httpConfig  httpConfig
+	httpHandler atomic.Value // *rpcHandler
+
+	// WebSocket handler things.
+	wsConfig  wsConfig
+	wsHandler atomic.Value // *rpcHandler
+
+	// These are set by setListenAddr.
+	endpoint string
+	host     string
+	port     int
+
+	handlerNames map[string]string
+}
+
+func newHTTPServer(log log.Logger, timeouts rpc.HTTPTimeouts) *httpServer {
+	h := &httpServer{log: log, timeouts: timeouts, handlerNames: make(map[string]string)}
+	h.httpHandler.Store((*rpcHandler)(nil))
+	h.wsHandler.Store((*rpcHandler)(nil))
+	return h
+}
+
+// setListenAddr configures the listening address of the server.
+// The address can only be set while the server isn't running.
+func (h *httpServer) setListenAddr(host string, port int) error {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	if h.listener != nil && (host != h.host || port != h.port) {
+		return fmt.Errorf("HTTP server already running on %s", h.endpoint)
+	}
+
+	h.host, h.port = host, port
+	h.endpoint = fmt.Sprintf("%s:%d", host, port)
+	return nil
+}
+
+// listenAddr returns the listening address of the server.
+func (h *httpServer) listenAddr() string {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	if h.listener != nil {
+		return h.listener.Addr().String()
+	}
+	return h.endpoint
+}
+
+// start starts the HTTP server if it is enabled and not already running.
+func (h *httpServer) start() error {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	if h.endpoint == "" || h.listener != nil {
+		return nil // already running or not configured
+	}
+
+	// Initialize the server.
+	h.server = &http.Server{Handler: h}
+	if h.timeouts != (rpc.HTTPTimeouts{}) {
+		CheckTimeouts(&h.timeouts)
+		h.server.ReadTimeout = h.timeouts.ReadTimeout
+		h.server.WriteTimeout = h.timeouts.WriteTimeout
+		h.server.IdleTimeout = h.timeouts.IdleTimeout
+	}
+
+	// Start the server.
+	listener, err := net.Listen("tcp", h.endpoint)
+	if err != nil {
+		// If the server fails to start, we need to clear out the RPC and WS
+		// configuration so they can be configured another time.
+		h.disableRPC()
+		h.disableWS()
+		return err
+	}
+	h.listener = listener
+	go h.server.Serve(listener)
+
+	// if server is websocket only, return after logging
+	if h.wsAllowed() && !h.rpcAllowed() {
+		h.log.Info("WebSocket enabled", "url", fmt.Sprintf("ws://%v", listener.Addr()))
+		return nil
+	}
+	// Log http endpoint.
+	h.log.Info("HTTP server started",
+		"endpoint", listener.Addr(),
+		"cors", strings.Join(h.httpConfig.CorsAllowedOrigins, ","),
+		"vhosts", strings.Join(h.httpConfig.Vhosts, ","),
+	)
+
+	// Log all handlers mounted on server.
+	var paths []string
+	for path := range h.handlerNames {
+		paths = append(paths, path)
+	}
+	sort.Strings(paths)
+	logged := make(map[string]bool, len(paths))
+	for _, path := range paths {
+		name := h.handlerNames[path]
+		if !logged[name] {
+			log.Info(name+" enabled", "url", "http://"+listener.Addr().String()+path)
+			logged[name] = true
+		}
+	}
+	return nil
+}
+
+func (h *httpServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	rpc := h.httpHandler.Load().(*rpcHandler)
+	if r.RequestURI == "/" {
+		// Serve JSON-RPC on the root path.
+		ws := h.wsHandler.Load().(*rpcHandler)
+		if ws != nil && isWebsocket(r) {
+			ws.ServeHTTP(w, r)
+			return
+		}
+		if rpc != nil {
+			rpc.ServeHTTP(w, r)
+			return
+		}
+	} else if rpc != nil {
+		// Requests to a path below root are handled by the mux,
+		// which has all the handlers registered via Node.RegisterHandler.
+		// These are made available when RPC is enabled.
+		h.mux.ServeHTTP(w, r)
+		return
+	}
+	w.WriteHeader(404)
+}
+
+// stop shuts down the HTTP server.
+func (h *httpServer) stop() {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.doStop()
+}
+
+func (h *httpServer) doStop() {
+	if h.listener == nil {
+		return // not running
+	}
+
+	// Shut down the server.
+	httpHandler := h.httpHandler.Load().(*rpcHandler)
+	wsHandler := h.httpHandler.Load().(*rpcHandler)
+	if httpHandler != nil {
+		h.httpHandler.Store((*rpcHandler)(nil))
+		httpHandler.server.Stop()
+	}
+	if wsHandler != nil {
+		h.wsHandler.Store((*rpcHandler)(nil))
+		wsHandler.server.Stop()
+	}
+	h.server.Shutdown(context.Background())
+	h.listener.Close()
+	h.log.Info("HTTP server stopped", "endpoint", h.listener.Addr())
+
+	// Clear out everything to allow re-configuring it later.
+	h.host, h.port, h.endpoint = "", 0, ""
+	h.server, h.listener = nil, nil
+}
+
+// enableRPC turns on JSON-RPC over HTTP on the server.
+func (h *httpServer) enableRPC(apis []rpc.API, config httpConfig) error {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	if h.rpcAllowed() {
+		return fmt.Errorf("JSON-RPC over HTTP is already enabled")
+	}
+
+	// Create RPC server and handler.
+	srv := rpc.NewServer()
+	if err := RegisterApisFromWhitelist(apis, config.Modules, srv, false); err != nil {
+		return err
+	}
+	h.httpConfig = config
+	h.httpHandler.Store(&rpcHandler{
+		Handler: NewHTTPHandlerStack(srv, config.CorsAllowedOrigins, config.Vhosts),
+		server:  srv,
+	})
+	return nil
+}
+
+// disableRPC stops the HTTP RPC handler. This is internal, the caller must hold h.mu.
+func (h *httpServer) disableRPC() bool {
+	handler := h.httpHandler.Load().(*rpcHandler)
+	if handler != nil {
+		h.httpHandler.Store((*rpcHandler)(nil))
+		handler.server.Stop()
+	}
+	return handler != nil
+}
+
+// enableWS turns on JSON-RPC over WebSocket on the server.
+func (h *httpServer) enableWS(apis []rpc.API, config wsConfig) error {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	if h.wsAllowed() {
+		return fmt.Errorf("JSON-RPC over WebSocket is already enabled")
+	}
+
+	// Create RPC server and handler.
+	srv := rpc.NewServer()
+	if err := RegisterApisFromWhitelist(apis, config.Modules, srv, false); err != nil {
+		return err
+	}
+	h.wsConfig = config
+	h.wsHandler.Store(&rpcHandler{
+		Handler: srv.WebsocketHandler(config.Origins),
+		server:  srv,
+	})
+	return nil
+}
+
+// stopWS disables JSON-RPC over WebSocket and also stops the server if it only serves WebSocket.
+func (h *httpServer) stopWS() {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	if h.disableWS() {
+		if !h.rpcAllowed() {
+			h.doStop()
+		}
+	}
+}
+
+// disableWS disables the WebSocket handler. This is internal, the caller must hold h.mu.
+func (h *httpServer) disableWS() bool {
+	ws := h.wsHandler.Load().(*rpcHandler)
+	if ws != nil {
+		h.wsHandler.Store((*rpcHandler)(nil))
+		ws.server.Stop()
+	}
+	return ws != nil
+}
+
+// rpcAllowed returns true when JSON-RPC over HTTP is enabled.
+func (h *httpServer) rpcAllowed() bool {
+	return h.httpHandler.Load().(*rpcHandler) != nil
+}
+
+// wsAllowed returns true when JSON-RPC over WebSocket is enabled.
+func (h *httpServer) wsAllowed() bool {
+	return h.wsHandler.Load().(*rpcHandler) != nil
+}
+
+// isWebsocket checks the header of an http request for a websocket upgrade request.
+func isWebsocket(r *http.Request) bool {
+	return strings.ToLower(r.Header.Get("Upgrade")) == "websocket" &&
+		strings.ToLower(r.Header.Get("Connection")) == "upgrade"
+}
+
 // NewHTTPHandlerStack returns wrapped http-related handlers
 func NewHTTPHandlerStack(srv http.Handler, cors []string, vhosts []string) http.Handler {
 	// Wrap the CORS-handler within a host-handler
@ -45,8 +332,8 @@ func newCorsHandler(srv http.Handler, allowedOrigins []string) http.Handler {
 	c := cors.New(cors.Options{
 		AllowedOrigins: allowedOrigins,
 		AllowedMethods: []string{http.MethodPost, http.MethodGet},
-		MaxAge:         600,
 		AllowedHeaders: []string{"*"},
+		MaxAge:         600,
 	})
 	return c.Handler(srv)
 }
@ -138,22 +425,68 @@ func newGzipHandler(next http.Handler) http.Handler {
 	})
 }

-// NewWebsocketUpgradeHandler returns a websocket handler that serves an incoming request only if it contains an upgrade
-// request to the websocket protocol. If not, serves the the request with the http handler.
-func NewWebsocketUpgradeHandler(h http.Handler, ws http.Handler) http.Handler {
-	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if isWebsocket(r) {
-			ws.ServeHTTP(w, r)
-			log.Debug("serving websocket request")
-			return
+type ipcServer struct {
+	log      log.Logger
+	endpoint string
+
+	mu       sync.Mutex
+	listener net.Listener
+	srv      *rpc.Server
+}
+
+func newIPCServer(log log.Logger, endpoint string) *ipcServer {
+	return &ipcServer{log: log, endpoint: endpoint}
+}
+
+// Start starts the httpServer's http.Server
+func (is *ipcServer) start(apis []rpc.API) error {
+	is.mu.Lock()
+	defer is.mu.Unlock()
+
+	if is.listener != nil {
+		return nil // already running
+	}
+	listener, srv, err := rpc.StartIPCEndpoint(is.endpoint, apis)
+	if err != nil {
+		return err
+	}
+	is.log.Info("IPC endpoint opened", "url", is.endpoint)
+	is.listener, is.srv = listener, srv
+	return nil
+}
+
+func (is *ipcServer) stop() error {
+	is.mu.Lock()
+	defer is.mu.Unlock()
+
+	if is.listener == nil {
+		return nil // not running
+	}
+	err := is.listener.Close()
+	is.srv.Stop()
+	is.listener, is.srv = nil, nil
+	is.log.Info("IPC endpoint closed", "url", is.endpoint)
+	return err
+}
+
+// RegisterApisFromWhitelist checks the given modules' availability, generates a whitelist based on the allowed modules,
+// and then registers all of the APIs exposed by the services.
+func RegisterApisFromWhitelist(apis []rpc.API, modules []string, srv *rpc.Server, exposeAll bool) error {
+	if bad, available := checkModuleAvailability(modules, apis); len(bad) > 0 {
+		log.Error("Unavailable modules in HTTP API list", "unavailable", bad, "available", available)
+	}
+	// Generate the whitelist based on the allowed modules
+	whitelist := make(map[string]bool)
+	for _, module := range modules {
+		whitelist[module] = true
+	}
+	// Register all the APIs exposed by the services
+	for _, api := range apis {
+		if exposeAll || whitelist[api.Namespace] || (len(whitelist) == 0 && api.Public) {
+			if err := srv.RegisterName(api.Namespace, api.Service); err != nil {
+				return err
+			}
 		}
-
-		h.ServeHTTP(w, r)
-	})
-}
-
-// isWebsocket checks the header of an http request for a websocket upgrade request.
-func isWebsocket(r *http.Request) bool {
-	return strings.ToLower(r.Header.Get("Upgrade")) == "websocket" &&
-		strings.ToLower(r.Header.Get("Connection")) == "upgrade"
+	}
+	return nil
 }