Feature: Support multiple listeners in configuration
* add ipv6 listening addresses to the default config * ipfsproxy: support multiple listeners. Add default ipv6. * mm * restapi: support multiple listen addresses. enable ipv6 * cluster_config: format default listen addresses * commands: update for multiple listeners. Fix randomports for udp and ipv6. * ipfs-cluster-service: fix randomports test * multiple listeners: fix remaining tests * golint * Disable ipv6 in defaults It is not supported by docker by default. It is not supported in travis-CI build environments. User can enable it now manually. * proxy: disable ipv6 in test * ipfshttp: fix test Co-authored-by: @RubenKelevra <cyrond@gmail.com>
This commit is contained in:
parent
7986d94242
commit
531379b1d9
|
@ -7,6 +7,7 @@ import (
|
|||
"path/filepath"
|
||||
"time"
|
||||
|
||||
ipfsconfig "github.com/ipfs/go-ipfs-config"
|
||||
"github.com/kelseyhightower/envconfig"
|
||||
ma "github.com/multiformats/go-multiaddr"
|
||||
|
||||
|
@ -19,9 +20,13 @@ const (
|
|||
minMaxHeaderBytes = 4096
|
||||
)
|
||||
|
||||
// DefaultListenAddrs contains the default listeners for the proxy.
|
||||
var DefaultListenAddrs = []string{
|
||||
"/ip4/127.0.0.1/tcp/9095",
|
||||
}
|
||||
|
||||
// Default values for Config.
|
||||
const (
|
||||
DefaultListenAddr = "/ip4/127.0.0.1/tcp/9095"
|
||||
DefaultNodeAddr = "/ip4/127.0.0.1/tcp/5001"
|
||||
DefaultNodeHTTPS = false
|
||||
DefaultReadTimeout = 0
|
||||
|
@ -39,7 +44,7 @@ type Config struct {
|
|||
config.Saver
|
||||
|
||||
// Listen parameters for the IPFS Proxy.
|
||||
ListenAddr ma.Multiaddr
|
||||
ListenAddr []ma.Multiaddr
|
||||
|
||||
// Host/Port for the IPFS daemon.
|
||||
NodeAddr ma.Multiaddr
|
||||
|
@ -93,7 +98,7 @@ type Config struct {
|
|||
}
|
||||
|
||||
type jsonConfig struct {
|
||||
ListenMultiaddress string `json:"listen_multiaddress"`
|
||||
ListenMultiaddress ipfsconfig.Strings `json:"listen_multiaddress"`
|
||||
NodeMultiaddress string `json:"node_multiaddress"`
|
||||
NodeHTTPS bool `json:"node_https,omitempty"`
|
||||
|
||||
|
@ -131,10 +136,14 @@ func (cfg *Config) ConfigKey() string {
|
|||
|
||||
// Default sets the fields of this Config to sensible default values.
|
||||
func (cfg *Config) Default() error {
|
||||
proxy, err := ma.NewMultiaddr(DefaultListenAddr)
|
||||
proxy := make([]ma.Multiaddr, 0, len(DefaultListenAddrs))
|
||||
for _, def := range DefaultListenAddrs {
|
||||
a, err := ma.NewMultiaddr(def)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
proxy = append(proxy, a)
|
||||
}
|
||||
node, err := ma.NewMultiaddr(DefaultNodeAddr)
|
||||
if err != nil {
|
||||
return err
|
||||
|
@ -174,7 +183,7 @@ func (cfg *Config) ApplyEnvVars() error {
|
|||
// at least in appearance.
|
||||
func (cfg *Config) Validate() error {
|
||||
var err error
|
||||
if cfg.ListenAddr == nil {
|
||||
if len(cfg.ListenAddr) == 0 {
|
||||
err = errors.New("ipfsproxy.listen_multiaddress not set")
|
||||
}
|
||||
if cfg.NodeAddr == nil {
|
||||
|
@ -230,12 +239,15 @@ func (cfg *Config) LoadJSON(raw []byte) error {
|
|||
}
|
||||
|
||||
func (cfg *Config) applyJSONConfig(jcfg *jsonConfig) error {
|
||||
if jcfg.ListenMultiaddress != "" {
|
||||
proxyAddr, err := ma.NewMultiaddr(jcfg.ListenMultiaddress)
|
||||
if addresses := jcfg.ListenMultiaddress; len(addresses) > 0 {
|
||||
cfg.ListenAddr = make([]ma.Multiaddr, 0, len(addresses))
|
||||
for _, a := range addresses {
|
||||
proxyAddr, err := ma.NewMultiaddr(a)
|
||||
if err != nil {
|
||||
return fmt.Errorf("error parsing proxy listen_multiaddress: %s", err)
|
||||
}
|
||||
cfg.ListenAddr = proxyAddr
|
||||
cfg.ListenAddr = append(cfg.ListenAddr, proxyAddr)
|
||||
}
|
||||
}
|
||||
if jcfg.NodeMultiaddress != "" {
|
||||
nodeAddr, err := ma.NewMultiaddr(jcfg.NodeMultiaddress)
|
||||
|
@ -295,8 +307,13 @@ func (cfg *Config) toJSONConfig() (jcfg *jsonConfig, err error) {
|
|||
|
||||
jcfg = &jsonConfig{}
|
||||
|
||||
addresses := make([]string, 0, len(cfg.ListenAddr))
|
||||
for _, a := range cfg.ListenAddr {
|
||||
addresses = append(addresses, a.String())
|
||||
}
|
||||
|
||||
// Set all configuration fields
|
||||
jcfg.ListenMultiaddress = cfg.ListenAddr.String()
|
||||
jcfg.ListenMultiaddress = addresses
|
||||
jcfg.NodeMultiaddress = cfg.NodeAddr.String()
|
||||
jcfg.ReadTimeout = cfg.ReadTimeout.String()
|
||||
jcfg.ReadHeaderTimeout = cfg.ReadHeaderTimeout.String()
|
||||
|
|
|
@ -40,7 +40,7 @@ func TestLoadJSON(t *testing.T) {
|
|||
|
||||
j := &jsonConfig{}
|
||||
json.Unmarshal(cfgJSON, j)
|
||||
j.ListenMultiaddress = "abc"
|
||||
j.ListenMultiaddress = []string{"abc"}
|
||||
tst, _ := json.Marshal(j)
|
||||
err = cfg.LoadJSON(tst)
|
||||
if err == nil {
|
||||
|
|
|
@ -58,7 +58,7 @@ type Server struct {
|
|||
rpcClient *rpc.Client
|
||||
rpcReady chan struct{}
|
||||
|
||||
listener net.Listener // proxy listener
|
||||
listeners []net.Listener // proxy listener
|
||||
server *http.Server // proxy server
|
||||
ipfsRoundTripper http.RoundTripper // allows to talk to IPFS
|
||||
|
||||
|
@ -126,7 +126,9 @@ func New(cfg *Config) (*Server, error) {
|
|||
return nil, err
|
||||
}
|
||||
|
||||
proxyNet, proxyAddr, err := manet.DialArgs(cfg.ListenAddr)
|
||||
var listeners []net.Listener
|
||||
for _, addr := range cfg.ListenAddr {
|
||||
proxyNet, proxyAddr, err := manet.DialArgs(addr)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -135,6 +137,8 @@ func New(cfg *Config) (*Server, error) {
|
|||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
listeners = append(listeners, l)
|
||||
}
|
||||
|
||||
nodeScheme := "http"
|
||||
if cfg.NodeHTTPS {
|
||||
|
@ -197,7 +201,7 @@ func New(cfg *Config) (*Server, error) {
|
|||
nodeAddr: nodeHTTPAddr,
|
||||
nodeScheme: nodeScheme,
|
||||
rpcReady: make(chan struct{}, 1),
|
||||
listener: l,
|
||||
listeners: listeners,
|
||||
server: s,
|
||||
ipfsRoundTripper: reverseProxy.Transport,
|
||||
}
|
||||
|
@ -284,7 +288,9 @@ func (proxy *Server) Shutdown(ctx context.Context) error {
|
|||
proxy.cancel()
|
||||
close(proxy.rpcReady)
|
||||
proxy.server.SetKeepAlivesEnabled(false)
|
||||
proxy.listener.Close()
|
||||
for _, l := range proxy.listeners {
|
||||
l.Close()
|
||||
}
|
||||
|
||||
proxy.wg.Wait()
|
||||
proxy.shutdown = true
|
||||
|
@ -301,19 +307,27 @@ func (proxy *Server) run() {
|
|||
defer proxy.shutdownLock.Unlock()
|
||||
|
||||
// This launches the proxy
|
||||
proxy.wg.Add(1)
|
||||
go func() {
|
||||
proxy.wg.Add(len(proxy.listeners))
|
||||
for _, l := range proxy.listeners {
|
||||
go func(l net.Listener) {
|
||||
defer proxy.wg.Done()
|
||||
|
||||
maddr, err := manet.FromNetAddr(l.Addr())
|
||||
if err != nil {
|
||||
logger.Error(err)
|
||||
}
|
||||
|
||||
logger.Infof(
|
||||
"IPFS Proxy: %s -> %s",
|
||||
proxy.config.ListenAddr,
|
||||
maddr,
|
||||
proxy.config.NodeAddr,
|
||||
)
|
||||
err := proxy.server.Serve(proxy.listener) // hangs here
|
||||
err = proxy.server.Serve(l) // hangs here
|
||||
if err != nil && !strings.Contains(err.Error(), "closed network connection") {
|
||||
logger.Error(err)
|
||||
}
|
||||
}()
|
||||
}(l)
|
||||
}
|
||||
}
|
||||
|
||||
// ipfsErrorResponder writes an http error response just like IPFS would.
|
||||
|
|
|
@ -34,7 +34,7 @@ func testIPFSProxyWithConfig(t *testing.T, cfg *Config) (*Server, *test.IpfsMock
|
|||
proxyMAddr, _ := ma.NewMultiaddr("/ip4/127.0.0.1/tcp/0")
|
||||
|
||||
cfg.NodeAddr = nodeMAddr
|
||||
cfg.ListenAddr = proxyMAddr
|
||||
cfg.ListenAddr = []ma.Multiaddr{proxyMAddr}
|
||||
cfg.ExtractHeadersExtra = []string{
|
||||
test.IpfsCustomHeaderName,
|
||||
test.IpfsTimeHeaderName,
|
||||
|
@ -716,7 +716,7 @@ func TestProxyError(t *testing.T) {
|
|||
}
|
||||
|
||||
func proxyURL(c *Server) string {
|
||||
addr := c.listener.Addr()
|
||||
addr := c.listeners[0].Addr()
|
||||
return fmt.Sprintf("http://%s/api/v0", addr.String())
|
||||
}
|
||||
|
||||
|
|
|
@ -22,7 +22,7 @@ func testAPI(t *testing.T) *rest.API {
|
|||
|
||||
cfg := &rest.Config{}
|
||||
cfg.Default()
|
||||
cfg.HTTPListenAddr = apiMAddr
|
||||
cfg.HTTPListenAddr = []ma.Multiaddr{apiMAddr}
|
||||
var secret [32]byte
|
||||
prot, err := pnet.NewV1ProtectorFromBytes(&secret)
|
||||
if err != nil {
|
||||
|
@ -54,8 +54,8 @@ func shutdown(a *rest.API) {
|
|||
}
|
||||
|
||||
func apiMAddr(a *rest.API) ma.Multiaddr {
|
||||
listen, _ := a.HTTPAddress()
|
||||
hostPort := strings.Split(listen, ":")
|
||||
listen, _ := a.HTTPAddresses()
|
||||
hostPort := strings.Split(listen[0], ":")
|
||||
|
||||
addr, _ := ma.NewMultiaddr(fmt.Sprintf("/ip4/127.0.0.1/tcp/%s", hostPort[1]))
|
||||
return addr
|
||||
|
|
|
@ -10,14 +10,15 @@ import (
|
|||
"path/filepath"
|
||||
"time"
|
||||
|
||||
"github.com/ipfs/ipfs-cluster/config"
|
||||
|
||||
ipfsconfig "github.com/ipfs/go-ipfs-config"
|
||||
crypto "github.com/libp2p/go-libp2p-core/crypto"
|
||||
peer "github.com/libp2p/go-libp2p-core/peer"
|
||||
ma "github.com/multiformats/go-multiaddr"
|
||||
|
||||
"github.com/kelseyhightower/envconfig"
|
||||
"github.com/rs/cors"
|
||||
|
||||
"github.com/ipfs/ipfs-cluster/config"
|
||||
)
|
||||
|
||||
const configKey = "restapi"
|
||||
|
@ -25,9 +26,13 @@ const envConfigKey = "cluster_restapi"
|
|||
|
||||
const minMaxHeaderBytes = 4096
|
||||
|
||||
// DefaultHTTPListenAddrs contains default listen addresses for the HTTP API.
|
||||
var DefaultHTTPListenAddrs = []string{
|
||||
"/ip4/127.0.0.1/tcp/9094",
|
||||
}
|
||||
|
||||
// These are the default values for Config
|
||||
const (
|
||||
DefaultHTTPListenAddr = "/ip4/127.0.0.1/tcp/9094"
|
||||
DefaultReadTimeout = 0
|
||||
DefaultReadHeaderTimeout = 5 * time.Second
|
||||
DefaultWriteTimeout = 0
|
||||
|
@ -66,7 +71,7 @@ type Config struct {
|
|||
config.Saver
|
||||
|
||||
// Listen address for the HTTP REST API endpoint.
|
||||
HTTPListenAddr ma.Multiaddr
|
||||
HTTPListenAddr []ma.Multiaddr
|
||||
|
||||
// TLS configuration for the HTTP listener
|
||||
TLS *tls.Config
|
||||
|
@ -97,7 +102,7 @@ type Config struct {
|
|||
MaxHeaderBytes int
|
||||
|
||||
// Listen address for the Libp2p REST API endpoint.
|
||||
Libp2pListenAddr ma.Multiaddr
|
||||
Libp2pListenAddr []ma.Multiaddr
|
||||
|
||||
// ID and PrivateKey are used to create a libp2p host if we
|
||||
// want the API component to do it (not by default).
|
||||
|
@ -131,7 +136,7 @@ type Config struct {
|
|||
}
|
||||
|
||||
type jsonConfig struct {
|
||||
HTTPListenMultiaddress string `json:"http_listen_multiaddress"`
|
||||
HTTPListenMultiaddress ipfsconfig.Strings `json:"http_listen_multiaddress"`
|
||||
SSLCertFile string `json:"ssl_cert_file,omitempty"`
|
||||
SSLKeyFile string `json:"ssl_key_file,omitempty"`
|
||||
ReadTimeout string `json:"read_timeout"`
|
||||
|
@ -140,7 +145,7 @@ type jsonConfig struct {
|
|||
IdleTimeout string `json:"idle_timeout"`
|
||||
MaxHeaderBytes int `json:"max_header_bytes"`
|
||||
|
||||
Libp2pListenMultiaddress string `json:"libp2p_listen_multiaddress,omitempty"`
|
||||
Libp2pListenMultiaddress ipfsconfig.Strings `json:"libp2p_listen_multiaddress,omitempty"`
|
||||
ID string `json:"id,omitempty"`
|
||||
PrivateKey string `json:"private_key,omitempty"`
|
||||
|
||||
|
@ -179,8 +184,15 @@ func (cfg *Config) ConfigKey() string {
|
|||
// Default initializes this Config with working values.
|
||||
func (cfg *Config) Default() error {
|
||||
// http
|
||||
httpListen, _ := ma.NewMultiaddr(DefaultHTTPListenAddr)
|
||||
cfg.HTTPListenAddr = httpListen
|
||||
addrs := make([]ma.Multiaddr, 0, len(DefaultHTTPListenAddrs))
|
||||
for _, def := range DefaultHTTPListenAddrs {
|
||||
httpListen, err := ma.NewMultiaddr(def)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
addrs = append(addrs, httpListen)
|
||||
}
|
||||
cfg.HTTPListenAddr = addrs
|
||||
cfg.pathSSLCertFile = ""
|
||||
cfg.pathSSLKeyFile = ""
|
||||
cfg.ReadTimeout = DefaultReadTimeout
|
||||
|
@ -225,7 +237,6 @@ func (cfg *Config) ApplyEnvVars() error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return cfg.applyJSONConfig(jcfg)
|
||||
}
|
||||
|
||||
|
@ -255,9 +266,9 @@ func (cfg *Config) Validate() error {
|
|||
}
|
||||
|
||||
func (cfg *Config) validateLibp2p() error {
|
||||
if cfg.ID != "" || cfg.PrivateKey != nil || cfg.Libp2pListenAddr != nil {
|
||||
if cfg.ID != "" || cfg.PrivateKey != nil || len(cfg.Libp2pListenAddr) > 0 {
|
||||
// if one is set, all should be
|
||||
if cfg.ID == "" || cfg.PrivateKey == nil || cfg.Libp2pListenAddr == nil {
|
||||
if cfg.ID == "" || cfg.PrivateKey == nil || len(cfg.Libp2pListenAddr) == 0 {
|
||||
return errors.New("all ID, private_key and libp2p_listen_multiaddress should be set")
|
||||
}
|
||||
if !cfg.ID.MatchesPrivateKey(cfg.PrivateKey) {
|
||||
|
@ -288,6 +299,7 @@ func (cfg *Config) applyJSONConfig(jcfg *jsonConfig) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
err = cfg.loadLibp2pOptions(jcfg)
|
||||
if err != nil {
|
||||
return err
|
||||
|
@ -302,13 +314,16 @@ func (cfg *Config) applyJSONConfig(jcfg *jsonConfig) error {
|
|||
}
|
||||
|
||||
func (cfg *Config) loadHTTPOptions(jcfg *jsonConfig) error {
|
||||
if httpListen := jcfg.HTTPListenMultiaddress; httpListen != "" {
|
||||
httpAddr, err := ma.NewMultiaddr(httpListen)
|
||||
if addresses := jcfg.HTTPListenMultiaddress; len(addresses) > 0 {
|
||||
cfg.HTTPListenAddr = make([]ma.Multiaddr, 0, len(addresses))
|
||||
for _, addr := range addresses {
|
||||
httpAddr, err := ma.NewMultiaddr(addr)
|
||||
if err != nil {
|
||||
err = fmt.Errorf("error parsing restapi.http_listen_multiaddress: %s", err)
|
||||
return err
|
||||
}
|
||||
cfg.HTTPListenAddr = httpAddr
|
||||
cfg.HTTPListenAddr = append(cfg.HTTPListenAddr, httpAddr)
|
||||
}
|
||||
}
|
||||
|
||||
err := cfg.tlsOptions(jcfg)
|
||||
|
@ -373,13 +388,16 @@ func (cfg *Config) tlsOptions(jcfg *jsonConfig) error {
|
|||
}
|
||||
|
||||
func (cfg *Config) loadLibp2pOptions(jcfg *jsonConfig) error {
|
||||
if libp2pListen := jcfg.Libp2pListenMultiaddress; libp2pListen != "" {
|
||||
libp2pAddr, err := ma.NewMultiaddr(libp2pListen)
|
||||
if addresses := jcfg.Libp2pListenMultiaddress; len(addresses) > 0 {
|
||||
cfg.Libp2pListenAddr = make([]ma.Multiaddr, 0, len(addresses))
|
||||
for _, addr := range addresses {
|
||||
libp2pAddr, err := ma.NewMultiaddr(addr)
|
||||
if err != nil {
|
||||
err = fmt.Errorf("error parsing restapi.libp2p_listen_multiaddress: %s", err)
|
||||
return err
|
||||
}
|
||||
cfg.Libp2pListenAddr = libp2pAddr
|
||||
cfg.Libp2pListenAddr = append(cfg.Libp2pListenAddr, libp2pAddr)
|
||||
}
|
||||
}
|
||||
|
||||
if jcfg.PrivateKey != "" {
|
||||
|
@ -424,8 +442,18 @@ func (cfg *Config) toJSONConfig() (jcfg *jsonConfig, err error) {
|
|||
}
|
||||
}()
|
||||
|
||||
httpAddresses := make([]string, 0, len(cfg.HTTPListenAddr))
|
||||
for _, addr := range cfg.HTTPListenAddr {
|
||||
httpAddresses = append(httpAddresses, addr.String())
|
||||
}
|
||||
|
||||
libp2pAddresses := make([]string, 0, len(cfg.Libp2pListenAddr))
|
||||
for _, addr := range cfg.Libp2pListenAddr {
|
||||
libp2pAddresses = append(libp2pAddresses, addr.String())
|
||||
}
|
||||
|
||||
jcfg = &jsonConfig{
|
||||
HTTPListenMultiaddress: cfg.HTTPListenAddr.String(),
|
||||
HTTPListenMultiaddress: httpAddresses,
|
||||
SSLCertFile: cfg.pathSSLCertFile,
|
||||
SSLKeyFile: cfg.pathSSLKeyFile,
|
||||
ReadTimeout: cfg.ReadTimeout.String(),
|
||||
|
@ -454,8 +482,8 @@ func (cfg *Config) toJSONConfig() (jcfg *jsonConfig, err error) {
|
|||
jcfg.PrivateKey = pKey
|
||||
}
|
||||
}
|
||||
if cfg.Libp2pListenAddr != nil {
|
||||
jcfg.Libp2pListenMultiaddress = cfg.Libp2pListenAddr.String()
|
||||
if len(libp2pAddresses) > 0 {
|
||||
jcfg.Libp2pListenMultiaddress = libp2pAddresses
|
||||
}
|
||||
|
||||
return
|
||||
|
|
|
@ -58,7 +58,7 @@ func TestLoadJSON(t *testing.T) {
|
|||
j := &jsonConfig{}
|
||||
|
||||
json.Unmarshal(cfgJSON, j)
|
||||
j.HTTPListenMultiaddress = "abc"
|
||||
j.HTTPListenMultiaddress = []string{"abc"}
|
||||
tst, _ := json.Marshal(j)
|
||||
err = cfg.LoadJSON(tst)
|
||||
if err == nil {
|
||||
|
@ -103,7 +103,7 @@ func TestLoadJSON(t *testing.T) {
|
|||
|
||||
j = &jsonConfig{}
|
||||
json.Unmarshal(cfgJSON, j)
|
||||
j.Libp2pListenMultiaddress = "abc"
|
||||
j.Libp2pListenMultiaddress = []string{"abc"}
|
||||
tst, _ = json.Marshal(j)
|
||||
err = cfg.LoadJSON(tst)
|
||||
if err == nil {
|
||||
|
@ -178,8 +178,8 @@ func TestLibp2pConfig(t *testing.T) {
|
|||
cfg.ID = pid
|
||||
cfg.PrivateKey = priv
|
||||
addr, _ := ma.NewMultiaddr("/ip4/127.0.0.1/tcp/0")
|
||||
cfg.HTTPListenAddr = addr
|
||||
cfg.Libp2pListenAddr = addr
|
||||
cfg.HTTPListenAddr = []ma.Multiaddr{addr}
|
||||
cfg.Libp2pListenAddr = []ma.Multiaddr{addr}
|
||||
|
||||
err = cfg.Validate()
|
||||
if err != nil {
|
||||
|
|
|
@ -38,7 +38,6 @@ import (
|
|||
libp2pquic "github.com/libp2p/go-libp2p-quic-transport"
|
||||
secio "github.com/libp2p/go-libp2p-secio"
|
||||
libp2ptls "github.com/libp2p/go-libp2p-tls"
|
||||
ma "github.com/multiformats/go-multiaddr"
|
||||
manet "github.com/multiformats/go-multiaddr-net"
|
||||
|
||||
handlers "github.com/gorilla/handlers"
|
||||
|
@ -91,7 +90,7 @@ type API struct {
|
|||
server *http.Server
|
||||
host host.Host
|
||||
|
||||
httpListener net.Listener
|
||||
httpListeners []net.Listener
|
||||
libp2pListener net.Listener
|
||||
|
||||
shutdownLock sync.Mutex
|
||||
|
@ -187,19 +186,19 @@ func NewAPIWithHost(ctx context.Context, cfg *Config, h host.Host) (*API, error)
|
|||
}
|
||||
api.addRoutes(router)
|
||||
|
||||
// Set up api.httpListener if enabled
|
||||
// Set up api.httpListeners if enabled
|
||||
err = api.setupHTTP()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Set up api.libp2pListener if enabled
|
||||
// Set up api.libp2pListeners if enabled
|
||||
err = api.setupLibp2p()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if api.httpListener == nil && api.libp2pListener == nil {
|
||||
if len(api.httpListeners) == 0 && api.libp2pListener == nil {
|
||||
return nil, ErrNoEndpointsEnabled
|
||||
}
|
||||
|
||||
|
@ -208,11 +207,12 @@ func NewAPIWithHost(ctx context.Context, cfg *Config, h host.Host) (*API, error)
|
|||
}
|
||||
|
||||
func (api *API) setupHTTP() error {
|
||||
if api.config.HTTPListenAddr == nil {
|
||||
if len(api.config.HTTPListenAddr) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
n, addr, err := manet.DialArgs(api.config.HTTPListenAddr)
|
||||
for _, listenMAddr := range api.config.HTTPListenAddr {
|
||||
n, addr, err := manet.DialArgs(listenMAddr)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
@ -226,21 +226,22 @@ func (api *API) setupHTTP() error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
api.httpListener = l
|
||||
api.httpListeners = append(api.httpListeners, l)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (api *API) setupLibp2p() error {
|
||||
// Make new host. Override any provided existing one
|
||||
// if we have config for a custom one.
|
||||
if api.config.Libp2pListenAddr != nil {
|
||||
if len(api.config.Libp2pListenAddr) > 0 {
|
||||
// We use a new host context. We will call
|
||||
// Close() on shutdown(). Avoids things like:
|
||||
// https://github.com/ipfs/ipfs-cluster/issues/853
|
||||
h, err := libp2p.New(
|
||||
context.Background(),
|
||||
libp2p.Identity(api.config.PrivateKey),
|
||||
libp2p.ListenAddrs([]ma.Multiaddr{api.config.Libp2pListenAddr}...),
|
||||
libp2p.ListenAddrs(api.config.Libp2pListenAddr...),
|
||||
libp2p.Security(libp2ptls.ID, libp2ptls.New),
|
||||
libp2p.Security(secio.ID, secio.New),
|
||||
libp2p.Transport(libp2pquic.NewTransport),
|
||||
|
@ -264,15 +265,20 @@ func (api *API) setupLibp2p() error {
|
|||
return nil
|
||||
}
|
||||
|
||||
// HTTPAddress returns the HTTP(s) listening address
|
||||
// HTTPAddresses returns the HTTP(s) listening address
|
||||
// in host:port format. Useful when configured to start
|
||||
// on a random port (0). Returns error when the HTTP endpoint
|
||||
// is not enabled.
|
||||
func (api *API) HTTPAddress() (string, error) {
|
||||
if api.httpListener == nil {
|
||||
return "", ErrHTTPEndpointNotEnabled
|
||||
func (api *API) HTTPAddresses() ([]string, error) {
|
||||
if len(api.httpListeners) == 0 {
|
||||
return nil, ErrHTTPEndpointNotEnabled
|
||||
}
|
||||
return api.httpListener.Addr().String(), nil
|
||||
var addrs []string
|
||||
for _, l := range api.httpListeners {
|
||||
addrs = append(addrs, l.Addr().String())
|
||||
}
|
||||
|
||||
return addrs, nil
|
||||
}
|
||||
|
||||
// Host returns the libp2p Host used by the API, if any.
|
||||
|
@ -479,28 +485,38 @@ func (api *API) routes() []route {
|
|||
}
|
||||
|
||||
func (api *API) run(ctx context.Context) {
|
||||
if api.httpListener != nil {
|
||||
api.wg.Add(1)
|
||||
go api.runHTTPServer(ctx)
|
||||
api.wg.Add(len(api.httpListeners))
|
||||
for _, l := range api.httpListeners {
|
||||
go func(l net.Listener) {
|
||||
defer api.wg.Done()
|
||||
api.runHTTPServer(ctx, l)
|
||||
}(l)
|
||||
}
|
||||
|
||||
if api.libp2pListener != nil {
|
||||
api.wg.Add(1)
|
||||
go api.runLibp2pServer(ctx)
|
||||
go func() {
|
||||
defer api.wg.Done()
|
||||
api.runLibp2pServer(ctx)
|
||||
}()
|
||||
}
|
||||
}
|
||||
|
||||
// runs in goroutine from run()
|
||||
func (api *API) runHTTPServer(ctx context.Context) {
|
||||
defer api.wg.Done()
|
||||
func (api *API) runHTTPServer(ctx context.Context, l net.Listener) {
|
||||
select {
|
||||
case <-api.rpcReady:
|
||||
case <-api.ctx.Done():
|
||||
return
|
||||
}
|
||||
|
||||
logger.Infof("REST API (HTTP): %s", api.config.HTTPListenAddr)
|
||||
err := api.server.Serve(api.httpListener)
|
||||
maddr, err := manet.FromNetAddr(l.Addr())
|
||||
if err != nil {
|
||||
logger.Error(err)
|
||||
}
|
||||
|
||||
logger.Infof("REST API (HTTP): %s", maddr)
|
||||
err = api.server.Serve(l)
|
||||
if err != nil && !strings.Contains(err.Error(), "closed network connection") {
|
||||
logger.Error(err)
|
||||
}
|
||||
|
@ -508,8 +524,6 @@ func (api *API) runHTTPServer(ctx context.Context) {
|
|||
|
||||
// runs in goroutine from run()
|
||||
func (api *API) runLibp2pServer(ctx context.Context) {
|
||||
defer api.wg.Done()
|
||||
|
||||
select {
|
||||
case <-api.rpcReady:
|
||||
case <-api.ctx.Done():
|
||||
|
@ -550,9 +564,10 @@ func (api *API) Shutdown(ctx context.Context) error {
|
|||
// Cancel any outstanding ops
|
||||
api.server.SetKeepAlivesEnabled(false)
|
||||
|
||||
if api.httpListener != nil {
|
||||
api.httpListener.Close()
|
||||
for _, l := range api.httpListeners {
|
||||
l.Close()
|
||||
}
|
||||
|
||||
if api.libp2pListener != nil {
|
||||
api.libp2pListener.Close()
|
||||
}
|
||||
|
|
|
@ -50,7 +50,7 @@ func testAPIwithConfig(t *testing.T, cfg *Config, name string) *API {
|
|||
t.Fatal(err)
|
||||
}
|
||||
|
||||
cfg.HTTPListenAddr = apiMAddr
|
||||
cfg.HTTPListenAddr = []ma.Multiaddr{apiMAddr}
|
||||
|
||||
rest, err := NewAPIWithHost(ctx, cfg, h)
|
||||
if err != nil {
|
||||
|
@ -174,8 +174,8 @@ func makeHost(t *testing.T, rest *API) host.Host {
|
|||
type urlF func(a *API) string
|
||||
|
||||
func httpURL(a *API) string {
|
||||
u, _ := a.HTTPAddress()
|
||||
return fmt.Sprintf("http://%s", u)
|
||||
u, _ := a.HTTPAddresses()
|
||||
return fmt.Sprintf("http://%s", u[0])
|
||||
}
|
||||
|
||||
func p2pURL(a *API) string {
|
||||
|
@ -183,8 +183,8 @@ func p2pURL(a *API) string {
|
|||
}
|
||||
|
||||
func httpsURL(a *API) string {
|
||||
u, _ := a.HTTPAddress()
|
||||
return fmt.Sprintf("https://%s", u)
|
||||
u, _ := a.HTTPAddresses()
|
||||
return fmt.Sprintf("https://%s", u[0])
|
||||
}
|
||||
|
||||
func isHTTPS(url string) bool {
|
||||
|
|
|
@ -22,8 +22,11 @@ import (
|
|||
|
||||
const configKey = "cluster"
|
||||
|
||||
// DefaultListenAddrs contains TCP and QUIC listen addresses
|
||||
var DefaultListenAddrs = []string{"/ip4/0.0.0.0/tcp/9096", "/ip4/0.0.0.0/udp/9096/quic"}
|
||||
// DefaultListenAddrs contains TCP and QUIC listen addresses.
|
||||
var DefaultListenAddrs = []string{
|
||||
"/ip4/0.0.0.0/tcp/9096",
|
||||
"/ip4/0.0.0.0/udp/9096/quic",
|
||||
}
|
||||
|
||||
// Configuration defaults
|
||||
const (
|
||||
|
|
|
@ -22,6 +22,7 @@ import (
|
|||
"github.com/ipfs/ipfs-cluster/monitor/pubsubmon"
|
||||
"github.com/ipfs/ipfs-cluster/observations"
|
||||
"github.com/ipfs/ipfs-cluster/pintracker/stateless"
|
||||
"github.com/multiformats/go-multiaddr"
|
||||
"github.com/pkg/errors"
|
||||
cli "github.com/urfave/cli/v2"
|
||||
)
|
||||
|
@ -301,7 +302,7 @@ func runCmd(c *cli.Context) error {
|
|||
if err != nil {
|
||||
return cli.Exit(err, 1)
|
||||
}
|
||||
apiCfg.HTTPListenAddr = listenSocket
|
||||
apiCfg.HTTPListenAddr = []multiaddr.Multiaddr{listenSocket}
|
||||
// Allow customization via env vars
|
||||
err = apiCfg.ApplyEnvVars()
|
||||
if err != nil {
|
||||
|
|
|
@ -328,12 +328,11 @@ the peer IDs in the given multiaddresses.
|
|||
// Generate defaults for all registered components
|
||||
err := cfgHelper.Manager().Default()
|
||||
checkErr("generating default configuration", err)
|
||||
|
||||
if c.Bool("randomports") {
|
||||
cfgs := cfgHelper.Configs()
|
||||
|
||||
for i := range cfgs.Cluster.ListenAddr {
|
||||
cfgs.Cluster.ListenAddr[i], err = cmdutils.RandomizePorts(cfgs.Cluster.ListenAddr[i])
|
||||
}
|
||||
cfgs.Cluster.ListenAddr, err = cmdutils.RandomizePorts(cfgs.Cluster.ListenAddr)
|
||||
checkErr("randomizing ports", err)
|
||||
cfgs.Restapi.HTTPListenAddr, err = cmdutils.RandomizePorts(cfgs.Restapi.HTTPListenAddr)
|
||||
checkErr("randomizing ports", err)
|
||||
|
|
|
@ -9,25 +9,30 @@ import (
|
|||
)
|
||||
|
||||
func TestRandomPorts(t *testing.T) {
|
||||
port := "9096"
|
||||
m1, _ := ma.NewMultiaddr("/ip4/0.0.0.0/tcp/9096")
|
||||
m2, _ := ma.NewMultiaddr("/ip4/0.0.0.0/tcp/9096")
|
||||
m2, _ := ma.NewMultiaddr("/ip6/::/udp/9096")
|
||||
|
||||
m1, err := cmdutils.RandomizePorts(m1)
|
||||
addresses, err := cmdutils.RandomizePorts([]ma.Multiaddr{m1, m2})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
v1, err := m1.ValueForProtocol(ma.P_TCP)
|
||||
v1, err := addresses[0].ValueForProtocol(ma.P_TCP)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
v2, err := m2.ValueForProtocol(ma.P_TCP)
|
||||
v2, err := addresses[1].ValueForProtocol(ma.P_UDP)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if v1 == v2 {
|
||||
t.Error("expected different ports")
|
||||
if v1 == port {
|
||||
t.Error("expected different ipv4 ports")
|
||||
}
|
||||
|
||||
if v2 == port {
|
||||
t.Error("expected different ipv6 ports")
|
||||
}
|
||||
}
|
||||
|
|
|
@ -5,9 +5,11 @@ package cmdutils
|
|||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"os"
|
||||
"os/signal"
|
||||
"strings"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
|
@ -20,10 +22,12 @@ import (
|
|||
)
|
||||
|
||||
// RandomizePorts replaces TCP and UDP ports with random, but valid port
|
||||
// values.
|
||||
func RandomizePorts(m ma.Multiaddr) (ma.Multiaddr, error) {
|
||||
var prev string
|
||||
// values, on the given multiaddresses
|
||||
func RandomizePorts(addrs []ma.Multiaddr) ([]ma.Multiaddr, error) {
|
||||
results := make([]ma.Multiaddr, 0, len(addrs))
|
||||
|
||||
for _, m := range addrs {
|
||||
var prev string
|
||||
var err error
|
||||
components := []ma.Multiaddr{}
|
||||
ma.ForEach(m, func(c ma.Component) bool {
|
||||
|
@ -35,15 +39,26 @@ func RandomizePorts(m ma.Multiaddr) (ma.Multiaddr, error) {
|
|||
return true
|
||||
}
|
||||
|
||||
var ln net.Listener
|
||||
ln, err = net.Listen(c.Protocol().Name, prev+":")
|
||||
var ln io.Closer
|
||||
var port int
|
||||
|
||||
ip := prev
|
||||
if strings.Contains(ip, ":") { // ipv6 needs bracketing
|
||||
ip = "[" + ip + "]"
|
||||
}
|
||||
|
||||
if c.Protocol().Code == ma.P_UDP {
|
||||
ln, port, err = listenUDP(c.Protocol().Name, ip)
|
||||
} else {
|
||||
ln, port, err = listenTCP(c.Protocol().Name, ip)
|
||||
}
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
defer ln.Close()
|
||||
|
||||
var c1 *ma.Component
|
||||
c1, err = ma.NewComponent(c.Protocol().Name, fmt.Sprintf("%d", getPort(ln, code)))
|
||||
c1, err = ma.NewComponent(c.Protocol().Name, fmt.Sprintf("%d", port))
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
|
@ -53,18 +68,33 @@ func RandomizePorts(m ma.Multiaddr) (ma.Multiaddr, error) {
|
|||
|
||||
return true
|
||||
})
|
||||
if err != nil {
|
||||
return results, err
|
||||
}
|
||||
results = append(results, ma.Join(components...))
|
||||
}
|
||||
|
||||
return ma.Join(components...), err
|
||||
return results, nil
|
||||
}
|
||||
|
||||
func getPort(ln net.Listener, code int) int {
|
||||
if code == ma.P_TCP {
|
||||
return ln.Addr().(*net.TCPAddr).Port
|
||||
// returns the listener so it can be closed later and port
|
||||
func listenTCP(name, ip string) (io.Closer, int, error) {
|
||||
ln, err := net.Listen(name, ip+":0")
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
if code == ma.P_UDP {
|
||||
return ln.Addr().(*net.UDPAddr).Port
|
||||
|
||||
return ln, ln.Addr().(*net.TCPAddr).Port, nil
|
||||
}
|
||||
|
||||
// returns the listener so it can be cloesd later and port
|
||||
func listenUDP(name, ip string) (io.Closer, int, error) {
|
||||
ln, err := net.ListenPacket(name, ip+":0")
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
return 0
|
||||
|
||||
return ln, ln.LocalAddr().(*net.UDPAddr).Port, nil
|
||||
}
|
||||
|
||||
// HandleSignals orderly shuts down an IPFS Cluster peer
|
||||
|
|
|
@ -181,9 +181,9 @@ func createComponents(
|
|||
clusterCfg.LeaveOnShutdown = false
|
||||
clusterCfg.SetBaseDir(filepath.Join(testsFolder, host.ID().Pretty()))
|
||||
|
||||
apiCfg.HTTPListenAddr = apiAddr
|
||||
apiCfg.HTTPListenAddr = []ma.Multiaddr{apiAddr}
|
||||
|
||||
ipfsproxyCfg.ListenAddr = proxyAddr
|
||||
ipfsproxyCfg.ListenAddr = []ma.Multiaddr{proxyAddr}
|
||||
ipfsproxyCfg.NodeAddr = nodeAddr
|
||||
|
||||
ipfshttpCfg.NodeAddr = nodeAddr
|
||||
|
|
|
@ -56,8 +56,8 @@ func TestIPFSID(t *testing.T) {
|
|||
if id.ID != test.PeerID1 {
|
||||
t.Error("expected testPeerID")
|
||||
}
|
||||
if len(id.Addresses) != 1 {
|
||||
t.Error("expected 1 address")
|
||||
if len(id.Addresses) != 2 {
|
||||
t.Error("expected 2 address")
|
||||
}
|
||||
if id.Error != "" {
|
||||
t.Error("expected no error")
|
||||
|
|
|
@ -3,7 +3,10 @@
|
|||
"peername": "testname",
|
||||
"secret": "84399cd0be811c2ca372d6ca473ffd73c09034f991c5e306fe9ada6c5fcfb641",
|
||||
"leave_on_shutdown": false,
|
||||
"listen_multiaddress": "/ip4/0.0.0.0/tcp/9096",
|
||||
"listen_multiaddress": [
|
||||
"/ip4/0.0.0.0/tcp/9096",
|
||||
"/ip6/::/tcp/9096"
|
||||
],
|
||||
"state_sync_interval": "1m0s",
|
||||
"replication_factor": -1,
|
||||
"monitor_ping_interval": "15s"
|
||||
|
|
|
@ -3,7 +3,10 @@
|
|||
"peername": "testname",
|
||||
"secret": "84399cd0be811c2ca372d6ca473ffd73c09034f991c5e306fe9ada6c5fcfb641",
|
||||
"leave_on_shutdown": false,
|
||||
"listen_multiaddress": "/ip4/0.0.0.0/tcp/9096",
|
||||
"listen_multiaddress": [
|
||||
"/ip4/0.0.0.0/tcp/9096",
|
||||
"/ip6/::/tcp/9096"
|
||||
],
|
||||
"state_sync_interval": "1m0s",
|
||||
"replication_factor": -1,
|
||||
"monitor_ping_interval": "15s"
|
||||
|
|
|
@ -5,7 +5,10 @@
|
|||
"private_key": "CAASqAkwggSkAgEAAoIBAQC/ZmfWDbwyI0nJdRxgHcTdEaBFQo8sky9E+OOvtwZa5WKoLdHyHOLWxCAdpIHUBbhxz5rkMEWLwPI6ykqLIJToMPO8lJbKVzphOjv4JwpiAPdmeSiYMKLjx5V8MpqU2rwj/Uf3sRL8Gg9/Tei3PZ8cftxN1rkQQeeaOtk0CBxUFZSHEsyut1fbgIeL7TAY+4vCmXW0DBr4wh9fnoES/YivOvSiN9rScgWg6N65LfkI78hzaOJ4Nok2S4vYFCxjTAI9NWFUbhP5eJIFzTU+bZuQZxOn2qsoyw8pNZwuF+JClA/RcgBcCvVZcDH2ueVq/zT++bGCN+EWsAEdvJqJ5bsjAgMBAAECggEAaGDUZ6t94mnUJ4UyQEh7v4OJP7wYkFqEAL0qjfzl/lPyBX1XbQ3Ltwul6AR6uMGV4JszARZCFwDWGLGRDWZrTmTDxyfRQ+9l6vfzFFVWGDQmtz+Dn9uGOWnyX5TJMDxJNec+hBmRHOKpaOd37dYxGz0jr19V9UO7piRJp1J1AHUCypUGv5x1IekioSCu5fEyc7dyWwnmITHBjD08st+bCcjrIUFeXSdJKC8SymYeXdaVE3xH3zVEISKnrfT7bhuKZY1iibZIlXbVLNpyX36LkYJOiCqsMum3u70LH0VvTypkqiDbD4S6qfJ4vvUakpmKpOPutikiP7jkSP+AkaO0AQKBgQDkTuhnDK6+Y0a/HgpHJisji0coO+g2gsIszargHk8nNY2AB8t+EUn7C+Qu8cmrem5V8EXcdxS6z7iAXpJmY1Xepnsz+JP7Q91Lgt3OoqK5EybzUXXKkmNCD65n70Xxn2fEFzm6+GJP3c/HymlDKU2KBCYIyuUeaREjT0Fu3v6tgQKBgQDWnXppJwn4LJHhzFOCeO4zomDJDbLTZCabdKZoFP9r+vtEHAnclDDKx4AYbomSqgERe+DX6HR/tPHRVizP63RYPf7al2mJmPzt1nTkoc1/q5hQoD+oE154dADsW1pUp7AQjwCtys4iq5S0qAwIDpuY8M8bOHwZ+QmBvHYAigJCowKBgQC3HH6TX/2rH463bE2MARXqXSPGJj45sigwrQfW1xhe9zm1LQtN4mn2mvP5nt1D1l82OA6gIzYSGtX8x10eF5/ggqAf78goZ6bOkHh76b8fNzgvQO97eGt5qYAVRjhP8azU/lfEGMEpE1s5/6LrRe41utwSg0C+YkBnlIKDfQDAgQKBgDoBTCF5hK9H1JHzuKpt5uubuo78ndWWnvyrNYKyEirsJddNwLiWcO2NqChyT8qNGkbQdX/Fex89F5KduPTlTYfAEc6g18xxxgK+UM+uj60vArbf6PSTb5gculcnha2VuPdwvx050Cb8uu9s7/uJfzKB+2f/B0O51ID1H+ubYWsDAoGBAKrwGKHyqFTHSPg3XuRA1FgDAoOsfzP9ZJvMEXUWyu/VxjNt+0mRlyGeZ5qb9UZG+K/In4FbC/ux2P/PucCUIbgy/XGPtPXVavMwNbx0MquAcU0FihKXP0CUpi8zwiYc42MF7n/SztQnismxigBMSuJEDurcXXazjfcSRTypduNn",
|
||||
"secret": "84399cd0be811c2ca372d6ca473ffd73c09034f991c5e306fe9ada6c5fcfb641",
|
||||
"leave_on_shutdown": false,
|
||||
"listen_multiaddress": "/ip4/0.0.0.0/tcp/9096",
|
||||
"listen_multiaddress": [
|
||||
"/ip4/0.0.0.0/tcp/9096",
|
||||
"/ip6/::/tcp/9096"
|
||||
],
|
||||
"state_sync_interval": "1m0s",
|
||||
"replication_factor": -1,
|
||||
"monitor_ping_interval": "15s"
|
||||
|
|
|
@ -180,6 +180,7 @@ func (m *IpfsMock) handler(w http.ResponseWriter, r *http.Request) {
|
|||
ID: PeerID1.Pretty(),
|
||||
Addresses: []string{
|
||||
"/ip4/0.0.0.0/tcp/1234",
|
||||
"/ip6/::/tcp/1234",
|
||||
},
|
||||
}
|
||||
j, _ := json.Marshal(resp)
|
||||
|
|
Loading…
Reference in New Issue
Block a user