Broadcast Store meta via standard consensus
With this change the cluster metadata (arbitrary key-value data associated with each node) is now broadcast across the cluster using the standard consensus mechanism. Specifically the use case for this metadata is to allow all nodes know the HTTP API address of all other nodes, for the purpose of redirecting requests to the leader. This change removed the need for multiplexing two logical connections over the single Raft TCP connection, which greatly simplifies the networking code generally. Original PR https://github.com/rqlite/rqlite/pull/434master
Philip O'Toole
5 years ago
parent
4a9345cc20
commit
f57ace7da2
@ -1,180 +0,0 @@
|
||||
package cluster
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"net"
|
||||
"os"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
connectionTimeout = 10 * time.Second
|
||||
)
|
||||
|
||||
var respOKMarshalled []byte
|
||||
|
||||
func init() {
|
||||
var err error
|
||||
respOKMarshalled, err = json.Marshal(response{})
|
||||
if err != nil {
|
||||
panic(fmt.Sprintf("unable to JSON marshal OK response: %s", err.Error()))
|
||||
}
|
||||
}
|
||||
|
||||
type response struct {
|
||||
Code int `json:"code,omitempty"`
|
||||
Message string `json:"message,omitempty"`
|
||||
}
|
||||
|
||||
// Transport is the interface the network service must provide.
|
||||
type Transport interface {
|
||||
net.Listener
|
||||
|
||||
// Dial is used to create a new outgoing connection
|
||||
Dial(address string, timeout time.Duration) (net.Conn, error)
|
||||
}
|
||||
|
||||
// Store represents a store of information, managed via consensus.
|
||||
type Store interface {
|
||||
// Leader returns the address of the leader of the consensus system.
|
||||
LeaderAddr() string
|
||||
|
||||
// UpdateAPIPeers updates the API peers on the store.
|
||||
UpdateAPIPeers(peers map[string]string) error
|
||||
}
|
||||
|
||||
// Service allows access to the cluster and associated meta data,
|
||||
// via consensus.
|
||||
type Service struct {
|
||||
tn Transport
|
||||
store Store
|
||||
addr net.Addr
|
||||
|
||||
wg sync.WaitGroup
|
||||
|
||||
logger *log.Logger
|
||||
}
|
||||
|
||||
// NewService returns a new instance of the cluster service
|
||||
func NewService(tn Transport, store Store) *Service {
|
||||
return &Service{
|
||||
tn: tn,
|
||||
store: store,
|
||||
addr: tn.Addr(),
|
||||
logger: log.New(os.Stderr, "[cluster] ", log.LstdFlags),
|
||||
}
|
||||
}
|
||||
|
||||
// Open opens the Service.
|
||||
func (s *Service) Open() error {
|
||||
s.wg.Add(1)
|
||||
go s.serve()
|
||||
s.logger.Println("service listening on", s.tn.Addr())
|
||||
return nil
|
||||
}
|
||||
|
||||
// Close closes the service.
|
||||
func (s *Service) Close() error {
|
||||
s.tn.Close()
|
||||
s.wg.Wait()
|
||||
return nil
|
||||
}
|
||||
|
||||
// Addr returns the address the service is listening on.
|
||||
func (s *Service) Addr() string {
|
||||
return s.addr.String()
|
||||
}
|
||||
|
||||
// SetPeer will set the mapping between raftAddr and apiAddr for the entire cluster.
|
||||
func (s *Service) SetPeer(raftAddr, apiAddr string) error {
|
||||
peer := map[string]string{
|
||||
raftAddr: apiAddr,
|
||||
}
|
||||
|
||||
// Try the local store. It might be the leader.
|
||||
err := s.store.UpdateAPIPeers(peer)
|
||||
if err == nil {
|
||||
// All done! Aren't we lucky?
|
||||
return nil
|
||||
}
|
||||
|
||||
// Try talking to the leader over the network.
|
||||
if leader := s.store.LeaderAddr(); leader == "" {
|
||||
return fmt.Errorf("no leader available")
|
||||
}
|
||||
conn, err := s.tn.Dial(s.store.LeaderAddr(), connectionTimeout)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer conn.Close()
|
||||
|
||||
b, err := json.Marshal(peer)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if _, err := conn.Write(b); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Wait for the response and verify the operation went through.
|
||||
resp := response{}
|
||||
d := json.NewDecoder(conn)
|
||||
err = d.Decode(&resp)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if resp.Code != 0 {
|
||||
return fmt.Errorf(resp.Message)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Service) serve() error {
|
||||
defer s.wg.Done()
|
||||
|
||||
for {
|
||||
conn, err := s.tn.Accept()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
go s.handleConn(conn)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Service) handleConn(conn net.Conn) {
|
||||
s.logger.Printf("received connection from %s", conn.RemoteAddr().String())
|
||||
|
||||
// Only handles peers updates for now.
|
||||
peers := make(map[string]string)
|
||||
d := json.NewDecoder(conn)
|
||||
err := d.Decode(&peers)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Update the peers.
|
||||
if err := s.store.UpdateAPIPeers(peers); err != nil {
|
||||
resp := response{1, err.Error()}
|
||||
b, err := json.Marshal(resp)
|
||||
if err != nil {
|
||||
conn.Close() // Only way left to signal.
|
||||
} else {
|
||||
if _, err := conn.Write(b); err != nil {
|
||||
conn.Close() // Only way left to signal.
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// Let the remote node know everything went OK.
|
||||
if _, err := conn.Write(respOKMarshalled); err != nil {
|
||||
conn.Close() // Only way left to signal.
|
||||
}
|
||||
return
|
||||
}
|
||||
@ -1,194 +0,0 @@
|
||||
package cluster
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func Test_NewServiceOpenClose(t *testing.T) {
|
||||
ml := mustNewMockTransport()
|
||||
ms := &mockStore{}
|
||||
s := NewService(ml, ms)
|
||||
if s == nil {
|
||||
t.Fatalf("failed to create cluster service")
|
||||
}
|
||||
|
||||
if err := s.Open(); err != nil {
|
||||
t.Fatalf("failed to open cluster service")
|
||||
}
|
||||
if err := s.Close(); err != nil {
|
||||
t.Fatalf("failed to close cluster service")
|
||||
}
|
||||
}
|
||||
|
||||
func Test_SetAPIPeer(t *testing.T) {
|
||||
raftAddr, apiAddr := "localhost:4002", "localhost:4001"
|
||||
|
||||
s, _, ms := mustNewOpenService()
|
||||
defer s.Close()
|
||||
if err := s.SetPeer(raftAddr, apiAddr); err != nil {
|
||||
t.Fatalf("failed to set peer: %s", err.Error())
|
||||
}
|
||||
|
||||
if ms.peers[raftAddr] != apiAddr {
|
||||
t.Fatalf("peer not set correctly, exp %s, got %s", apiAddr, ms.peers[raftAddr])
|
||||
}
|
||||
}
|
||||
|
||||
func Test_SetAPIPeerNetwork(t *testing.T) {
|
||||
raftAddr, apiAddr := "localhost:4002", "localhost:4001"
|
||||
|
||||
s, _, ms := mustNewOpenService()
|
||||
defer s.Close()
|
||||
|
||||
raddr, err := net.ResolveTCPAddr("tcp", s.Addr())
|
||||
if err != nil {
|
||||
t.Fatalf("failed to resolve remote uster ervice address: %s", err.Error())
|
||||
}
|
||||
|
||||
conn, err := net.DialTCP("tcp4", nil, raddr)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to connect to remote cluster service: %s", err.Error())
|
||||
}
|
||||
if _, err := conn.Write([]byte(fmt.Sprintf(`{"%s": "%s"}`, raftAddr, apiAddr))); err != nil {
|
||||
t.Fatalf("failed to write to remote cluster service: %s", err.Error())
|
||||
}
|
||||
|
||||
resp := response{}
|
||||
d := json.NewDecoder(conn)
|
||||
err = d.Decode(&resp)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to decode response: %s", err.Error())
|
||||
}
|
||||
|
||||
if resp.Code != 0 {
|
||||
t.Fatalf("response code was non-zero")
|
||||
}
|
||||
|
||||
if ms.peers[raftAddr] != apiAddr {
|
||||
t.Fatalf("peer not set correctly, exp %s, got %s", apiAddr, ms.peers[raftAddr])
|
||||
}
|
||||
}
|
||||
|
||||
func Test_SetAPIPeerFailUpdate(t *testing.T) {
|
||||
raftAddr, apiAddr := "localhost:4002", "localhost:4001"
|
||||
|
||||
s, _, ms := mustNewOpenService()
|
||||
defer s.Close()
|
||||
ms.failUpdateAPIPeers = true
|
||||
|
||||
// Attempt to set peer without a leader
|
||||
if err := s.SetPeer(raftAddr, apiAddr); err == nil {
|
||||
t.Fatalf("no error returned by set peer when no leader")
|
||||
}
|
||||
|
||||
// Start a network server.
|
||||
tn, err := net.Listen("tcp", "localhost:0")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to open test server: %s", err.Error())
|
||||
}
|
||||
ms.leader = tn.Addr().String()
|
||||
|
||||
c := make(chan map[string]string, 1)
|
||||
go func() {
|
||||
conn, err := tn.Accept()
|
||||
if err != nil {
|
||||
t.Fatalf("failed to accept connection from cluster: %s", err.Error())
|
||||
}
|
||||
t.Logf("test server received connection from: %s", conn.RemoteAddr())
|
||||
|
||||
peers := make(map[string]string)
|
||||
d := json.NewDecoder(conn)
|
||||
err = d.Decode(&peers)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to decode message from cluster: %s", err.Error())
|
||||
}
|
||||
|
||||
// Response OK.
|
||||
// Let the remote node know everything went OK.
|
||||
if _, err := conn.Write(respOKMarshalled); err != nil {
|
||||
t.Fatalf("failed to respond to cluster: %s", err.Error())
|
||||
}
|
||||
|
||||
c <- peers
|
||||
}()
|
||||
|
||||
if err := s.SetPeer(raftAddr, apiAddr); err != nil {
|
||||
t.Fatalf("failed to set peer on cluster: %s", err.Error())
|
||||
}
|
||||
|
||||
peers := <-c
|
||||
if peers[raftAddr] != apiAddr {
|
||||
t.Fatalf("peer not set correctly, exp %s, got %s", apiAddr, ms.peers[raftAddr])
|
||||
}
|
||||
}
|
||||
|
||||
func mustNewOpenService() (*Service, *mockTransport, *mockStore) {
|
||||
ml := mustNewMockTransport()
|
||||
ms := newMockStore()
|
||||
s := NewService(ml, ms)
|
||||
if err := s.Open(); err != nil {
|
||||
panic("failed to open new service")
|
||||
}
|
||||
return s, ml, ms
|
||||
}
|
||||
|
||||
type mockTransport struct {
|
||||
tn net.Listener
|
||||
}
|
||||
|
||||
func mustNewMockTransport() *mockTransport {
|
||||
tn, err := net.Listen("tcp", "localhost:0")
|
||||
if err != nil {
|
||||
panic("failed to create mock listener")
|
||||
}
|
||||
return &mockTransport{
|
||||
tn: tn,
|
||||
}
|
||||
}
|
||||
|
||||
func (ml *mockTransport) Accept() (c net.Conn, err error) {
|
||||
return ml.tn.Accept()
|
||||
}
|
||||
|
||||
func (ml *mockTransport) Addr() net.Addr {
|
||||
return ml.tn.Addr()
|
||||
}
|
||||
|
||||
func (ml *mockTransport) Close() (err error) {
|
||||
return ml.tn.Close()
|
||||
}
|
||||
|
||||
func (ml *mockTransport) Dial(addr string, t time.Duration) (net.Conn, error) {
|
||||
return net.DialTimeout("tcp", addr, 5*time.Second)
|
||||
}
|
||||
|
||||
type mockStore struct {
|
||||
leader string
|
||||
peers map[string]string
|
||||
failUpdateAPIPeers bool
|
||||
}
|
||||
|
||||
func newMockStore() *mockStore {
|
||||
return &mockStore{
|
||||
peers: make(map[string]string),
|
||||
}
|
||||
}
|
||||
|
||||
func (ms *mockStore) LeaderAddr() string {
|
||||
return ms.leader
|
||||
}
|
||||
|
||||
func (ms *mockStore) UpdateAPIPeers(peers map[string]string) error {
|
||||
if ms.failUpdateAPIPeers {
|
||||
return fmt.Errorf("forced fail")
|
||||
}
|
||||
|
||||
for k, v := range peers {
|
||||
ms.peers[k] = v
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@ -0,0 +1,12 @@
|
||||
package store
|
||||
|
||||
// DBConfig represents the configuration of the underlying SQLite database.
|
||||
type DBConfig struct {
|
||||
DSN string // Any custom DSN
|
||||
Memory bool // Whether the database is in-memory only.
|
||||
}
|
||||
|
||||
// NewDBConfig returns a new DB config instance.
|
||||
func NewDBConfig(dsn string, memory bool) *DBConfig {
|
||||
return &DBConfig{DSN: dsn, Memory: memory}
|
||||
}
|
||||
@ -0,0 +1,14 @@
|
||||
package store
|
||||
|
||||
// Server represents another node in the cluster.
|
||||
type Server struct {
|
||||
ID string `json:"id,omitempty"`
|
||||
Addr string `json:"addr,omitempty"`
|
||||
}
|
||||
|
||||
// Servers is a set of Servers.
|
||||
type Servers []*Server
|
||||
|
||||
func (s Servers) Less(i, j int) bool { return s[i].ID < s[j].ID }
|
||||
func (s Servers) Len() int { return len(s) }
|
||||
func (s Servers) Swap(i, j int) { s[i], s[j] = s[j], s[i] }
|
||||
@ -0,0 +1,11 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"testing"
|
||||
)
|
||||
|
||||
func Test_NewTransport(t *testing.T) {
|
||||
if NewTransport(nil) == nil {
|
||||
t.Fatal("failed to create new Transport")
|
||||
}
|
||||
}
|
||||
@ -1,4 +1,4 @@
|
||||
/*
|
||||
Package tcp provides various TCP-related utilities. The TCP mux code provided by this package originated with InfluxDB.
|
||||
Package tcp provides the internode communication network layer.
|
||||
*/
|
||||
package tcp
|
||||
|
||||
@ -1,301 +0,0 @@
|
||||
package tcp
|
||||
|
||||
import (
|
||||
"crypto/tls"
|
||||
"crypto/x509"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"log"
|
||||
"net"
|
||||
"os"
|
||||
"strconv"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
// DefaultTimeout is the default length of time to wait for first byte.
|
||||
DefaultTimeout = 30 * time.Second
|
||||
)
|
||||
|
||||
// Layer represents the connection between nodes.
|
||||
type Layer struct {
|
||||
ln net.Listener
|
||||
header byte
|
||||
addr net.Addr
|
||||
|
||||
remoteEncrypted bool
|
||||
skipVerify bool
|
||||
nodeX509CACert string
|
||||
tlsConfig *tls.Config
|
||||
}
|
||||
|
||||
// Addr returns the local address for the layer.
|
||||
func (l *Layer) Addr() net.Addr {
|
||||
return l.addr
|
||||
}
|
||||
|
||||
// Dial creates a new network connection.
|
||||
func (l *Layer) Dial(addr string, timeout time.Duration) (net.Conn, error) {
|
||||
dialer := &net.Dialer{Timeout: timeout}
|
||||
|
||||
var err error
|
||||
var conn net.Conn
|
||||
if l.remoteEncrypted {
|
||||
conn, err = tls.DialWithDialer(dialer, "tcp", addr, l.tlsConfig)
|
||||
} else {
|
||||
conn, err = dialer.Dial("tcp", addr)
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Write a marker byte to indicate message type.
|
||||
_, err = conn.Write([]byte{l.header})
|
||||
if err != nil {
|
||||
conn.Close()
|
||||
return nil, err
|
||||
}
|
||||
return conn, err
|
||||
}
|
||||
|
||||
// Accept waits for the next connection.
|
||||
func (l *Layer) Accept() (net.Conn, error) { return l.ln.Accept() }
|
||||
|
||||
// Close closes the layer.
|
||||
func (l *Layer) Close() error { return l.ln.Close() }
|
||||
|
||||
// Mux multiplexes a network connection.
|
||||
type Mux struct {
|
||||
ln net.Listener
|
||||
addr net.Addr
|
||||
m map[byte]*listener
|
||||
|
||||
wg sync.WaitGroup
|
||||
|
||||
remoteEncrypted bool
|
||||
|
||||
// The amount of time to wait for the first header byte.
|
||||
Timeout time.Duration
|
||||
|
||||
// Out-of-band error logger
|
||||
Logger *log.Logger
|
||||
|
||||
// Path to root X.509 certificate.
|
||||
nodeX509CACert string
|
||||
|
||||
// Path to X509 certificate
|
||||
nodeX509Cert string
|
||||
|
||||
// Path to X509 key.
|
||||
nodeX509Key string
|
||||
|
||||
// Whether to skip verification of other nodes' certificates.
|
||||
InsecureSkipVerify bool
|
||||
|
||||
tlsConfig *tls.Config
|
||||
}
|
||||
|
||||
// NewMux returns a new instance of Mux for ln. If adv is nil,
|
||||
// then the addr of ln is used.
|
||||
func NewMux(ln net.Listener, adv net.Addr) (*Mux, error) {
|
||||
addr := adv
|
||||
if addr == nil {
|
||||
addr = ln.Addr()
|
||||
}
|
||||
|
||||
return &Mux{
|
||||
ln: ln,
|
||||
addr: addr,
|
||||
m: make(map[byte]*listener),
|
||||
Timeout: DefaultTimeout,
|
||||
Logger: log.New(os.Stderr, "[mux] ", log.LstdFlags),
|
||||
}, nil
|
||||
}
|
||||
|
||||
// NewTLSMux returns a new instance of Mux for ln, and encrypts all traffic
|
||||
// using TLS. If adv is nil, then the addr of ln is used.
|
||||
func NewTLSMux(ln net.Listener, adv net.Addr, cert, key, caCert string) (*Mux, error) {
|
||||
mux, err := NewMux(ln, adv)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
mux.tlsConfig, err = createTLSConfig(cert, key, caCert)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
mux.ln = tls.NewListener(ln, mux.tlsConfig)
|
||||
mux.remoteEncrypted = true
|
||||
mux.nodeX509CACert = caCert
|
||||
mux.nodeX509Cert = cert
|
||||
mux.nodeX509Key = key
|
||||
|
||||
return mux, nil
|
||||
}
|
||||
|
||||
// Serve handles connections from ln and multiplexes then across registered listener.
|
||||
func (mux *Mux) Serve() error {
|
||||
mux.Logger.Printf("mux serving on %s, advertising %s", mux.ln.Addr().String(), mux.addr)
|
||||
|
||||
for {
|
||||
// Wait for the next connection.
|
||||
// If it returns a temporary error then simply retry.
|
||||
// If it returns any other error then exit immediately.
|
||||
conn, err := mux.ln.Accept()
|
||||
if err, ok := err.(interface {
|
||||
Temporary() bool
|
||||
}); ok && err.Temporary() {
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
// Wait for all connections to be demuxed
|
||||
mux.wg.Wait()
|
||||
for _, ln := range mux.m {
|
||||
close(ln.c)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// Demux in a goroutine to
|
||||
mux.wg.Add(1)
|
||||
go mux.handleConn(conn)
|
||||
}
|
||||
}
|
||||
|
||||
// Stats returns status of the mux.
|
||||
func (mux *Mux) Stats() (interface{}, error) {
|
||||
s := map[string]string{
|
||||
"addr": mux.addr.String(),
|
||||
"timeout": mux.Timeout.String(),
|
||||
"encrypted": strconv.FormatBool(mux.remoteEncrypted),
|
||||
}
|
||||
|
||||
if mux.remoteEncrypted {
|
||||
s["certificate"] = mux.nodeX509Cert
|
||||
s["key"] = mux.nodeX509Key
|
||||
s["ca_certificate"] = mux.nodeX509CACert
|
||||
s["skip_verify"] = strconv.FormatBool(mux.InsecureSkipVerify)
|
||||
}
|
||||
|
||||
return s, nil
|
||||
}
|
||||
|
||||
func (mux *Mux) handleConn(conn net.Conn) {
|
||||
defer mux.wg.Done()
|
||||
// Set a read deadline so connections with no data don't timeout.
|
||||
if err := conn.SetReadDeadline(time.Now().Add(mux.Timeout)); err != nil {
|
||||
conn.Close()
|
||||
mux.Logger.Printf("tcp.Mux: cannot set read deadline: %s", err)
|
||||
return
|
||||
}
|
||||
|
||||
// Read first byte from connection to determine handler.
|
||||
var typ [1]byte
|
||||
if _, err := io.ReadFull(conn, typ[:]); err != nil {
|
||||
conn.Close()
|
||||
mux.Logger.Printf("tcp.Mux: cannot read header byte: %s", err)
|
||||
return
|
||||
}
|
||||
|
||||
// Reset read deadline and let the listener handle that.
|
||||
if err := conn.SetReadDeadline(time.Time{}); err != nil {
|
||||
conn.Close()
|
||||
mux.Logger.Printf("tcp.Mux: cannot reset set read deadline: %s", err)
|
||||
return
|
||||
}
|
||||
|
||||
// Retrieve handler based on first byte.
|
||||
handler := mux.m[typ[0]]
|
||||
if handler == nil {
|
||||
conn.Close()
|
||||
mux.Logger.Printf("tcp.Mux: handler not registered: %d", typ[0])
|
||||
return
|
||||
}
|
||||
|
||||
// Send connection to handler. The handler is responsible for closing the connection.
|
||||
handler.c <- conn
|
||||
}
|
||||
|
||||
// Listen returns a listener identified by header.
|
||||
// Any connection accepted by mux is multiplexed based on the initial header byte.
|
||||
func (mux *Mux) Listen(header byte) *Layer {
|
||||
// Ensure two listeners are not created for the same header byte.
|
||||
if _, ok := mux.m[header]; ok {
|
||||
panic(fmt.Sprintf("listener already registered under header byte: %d", header))
|
||||
}
|
||||
|
||||
// Create a new listener and assign it.
|
||||
ln := &listener{
|
||||
c: make(chan net.Conn),
|
||||
}
|
||||
mux.m[header] = ln
|
||||
|
||||
layer := &Layer{
|
||||
ln: ln,
|
||||
header: header,
|
||||
addr: mux.addr,
|
||||
remoteEncrypted: mux.remoteEncrypted,
|
||||
skipVerify: mux.InsecureSkipVerify,
|
||||
nodeX509CACert: mux.nodeX509CACert,
|
||||
tlsConfig: mux.tlsConfig,
|
||||
}
|
||||
|
||||
return layer
|
||||
}
|
||||
|
||||
// listener is a receiver for connections received by Mux.
|
||||
type listener struct {
|
||||
c chan net.Conn
|
||||
}
|
||||
|
||||
// Accept waits for and returns the next connection to the listener.
|
||||
func (ln *listener) Accept() (c net.Conn, err error) {
|
||||
conn, ok := <-ln.c
|
||||
if !ok {
|
||||
return nil, errors.New("network connection closed")
|
||||
}
|
||||
return conn, nil
|
||||
}
|
||||
|
||||
// Close is a no-op. The mux's listener should be closed instead.
|
||||
func (ln *listener) Close() error { return nil }
|
||||
|
||||
// Addr always returns nil
|
||||
func (ln *listener) Addr() net.Addr { return nil }
|
||||
|
||||
// newTLSListener returns a net listener which encrypts the traffic using TLS.
|
||||
func newTLSListener(ln net.Listener, certFile, keyFile, caCertFile string) (net.Listener, error) {
|
||||
config, err := createTLSConfig(certFile, keyFile, caCertFile)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return tls.NewListener(ln, config), nil
|
||||
}
|
||||
|
||||
// createTLSConfig returns a TLS config from the given cert and key.
|
||||
func createTLSConfig(certFile, keyFile, caCertFile string) (*tls.Config, error) {
|
||||
var err error
|
||||
config := &tls.Config{}
|
||||
config.Certificates = make([]tls.Certificate, 1)
|
||||
config.Certificates[0], err = tls.LoadX509KeyPair(certFile, keyFile)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if caCertFile != "" {
|
||||
asn1Data, err := ioutil.ReadFile(caCertFile)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
config.RootCAs = x509.NewCertPool()
|
||||
ok := config.RootCAs.AppendCertsFromPEM([]byte(asn1Data))
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("failed to parse root certificate(s) in %s", caCertFile)
|
||||
}
|
||||
}
|
||||
return config, nil
|
||||
}
|
||||
@ -1,235 +0,0 @@
|
||||
package tcp
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"crypto/tls"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"log"
|
||||
"net"
|
||||
"os"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"testing/quick"
|
||||
"time"
|
||||
|
||||
"github.com/rqlite/rqlite/testdata/x509"
|
||||
)
|
||||
|
||||
// Ensure the muxer can split a listener's connections across multiple listeners.
|
||||
func TestMux(t *testing.T) {
|
||||
if err := quick.Check(func(n uint8, msg []byte) bool {
|
||||
if testing.Verbose() {
|
||||
if len(msg) == 0 {
|
||||
log.Printf("n=%d, <no message>", n)
|
||||
} else {
|
||||
log.Printf("n=%d, hdr=%d, len=%d", n, msg[0], len(msg))
|
||||
}
|
||||
}
|
||||
|
||||
var wg sync.WaitGroup
|
||||
|
||||
// Open single listener on random port.
|
||||
tcpListener := mustTCPListener("127.0.0.1:0")
|
||||
defer tcpListener.Close()
|
||||
|
||||
// Setup muxer & listeners.
|
||||
mux, err := NewMux(tcpListener, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to create mux: %s", err.Error())
|
||||
}
|
||||
mux.Timeout = 200 * time.Millisecond
|
||||
if !testing.Verbose() {
|
||||
mux.Logger = log.New(ioutil.Discard, "", 0)
|
||||
}
|
||||
for i := uint8(0); i < n; i++ {
|
||||
ln := mux.Listen(byte(i))
|
||||
|
||||
wg.Add(1)
|
||||
go func(i uint8, ln net.Listener) {
|
||||
defer wg.Done()
|
||||
|
||||
// Wait for a connection for this listener.
|
||||
conn, err := ln.Accept()
|
||||
if conn != nil {
|
||||
defer conn.Close()
|
||||
}
|
||||
|
||||
// If there is no message or the header byte
|
||||
// doesn't match then expect close.
|
||||
if len(msg) == 0 || msg[0] != byte(i) {
|
||||
if err == nil || err.Error() != "network connection closed" {
|
||||
t.Fatalf("unexpected error: %s", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// If the header byte matches this listener
|
||||
// then expect a connection and read the message.
|
||||
var buf bytes.Buffer
|
||||
if _, err := io.CopyN(&buf, conn, int64(len(msg)-1)); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if !bytes.Equal(msg[1:], buf.Bytes()) {
|
||||
t.Fatalf("message mismatch:\n\nexp=%x\n\ngot=%x\n\n", msg[1:], buf.Bytes())
|
||||
}
|
||||
|
||||
// Write response.
|
||||
if _, err := conn.Write([]byte("OK")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}(i, ln)
|
||||
}
|
||||
|
||||
// Begin serving from the listener.
|
||||
go mux.Serve()
|
||||
|
||||
// Write message to TCP listener and read OK response.
|
||||
conn, err := net.Dial("tcp", tcpListener.Addr().String())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
} else if _, err = conn.Write(msg); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Read the response into the buffer.
|
||||
var resp [2]byte
|
||||
_, err = io.ReadFull(conn, resp[:])
|
||||
|
||||
// If the message header is less than n then expect a response.
|
||||
// Otherwise we should get an EOF because the mux closed.
|
||||
if len(msg) > 0 && uint8(msg[0]) < n {
|
||||
if string(resp[:]) != `OK` {
|
||||
t.Fatalf("unexpected response: %s", resp[:])
|
||||
}
|
||||
} else {
|
||||
if err == nil || (err != io.EOF && !(strings.Contains(err.Error(), "connection reset by peer") ||
|
||||
strings.Contains(err.Error(), "closed by the remote host"))) {
|
||||
t.Fatalf("unexpected error: %s", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Close connection.
|
||||
if err := conn.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Close original TCP listener and wait for all goroutines to close.
|
||||
tcpListener.Close()
|
||||
wg.Wait()
|
||||
|
||||
return true
|
||||
}, nil); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMux_Advertise(t *testing.T) {
|
||||
// Setup muxer.
|
||||
tcpListener := mustTCPListener("127.0.0.1:0")
|
||||
defer tcpListener.Close()
|
||||
|
||||
addr := &mockAddr{
|
||||
Nwk: "tcp",
|
||||
Addr: "rqlite.com:8081",
|
||||
}
|
||||
|
||||
mux, err := NewMux(tcpListener, addr)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to create mux: %s", err.Error())
|
||||
}
|
||||
mux.Timeout = 200 * time.Millisecond
|
||||
if !testing.Verbose() {
|
||||
mux.Logger = log.New(ioutil.Discard, "", 0)
|
||||
}
|
||||
|
||||
layer := mux.Listen(1)
|
||||
if layer.Addr().String() != addr.Addr {
|
||||
t.Fatalf("layer advertise address not correct, exp %s, got %s",
|
||||
layer.Addr().String(), addr.Addr)
|
||||
}
|
||||
}
|
||||
|
||||
// Ensure two handlers cannot be registered for the same header byte.
|
||||
func TestMux_Listen_ErrAlreadyRegistered(t *testing.T) {
|
||||
defer func() {
|
||||
if r := recover(); r != `listener already registered under header byte: 5` {
|
||||
t.Fatalf("unexpected recover: %#v", r)
|
||||
}
|
||||
}()
|
||||
|
||||
// Register two listeners with the same header byte.
|
||||
tcpListener := mustTCPListener("127.0.0.1:0")
|
||||
mux, err := NewMux(tcpListener, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to create mux: %s", err.Error())
|
||||
}
|
||||
mux.Listen(5)
|
||||
mux.Listen(5)
|
||||
}
|
||||
|
||||
func TestTLSMux(t *testing.T) {
|
||||
tcpListener := mustTCPListener("127.0.0.1:0")
|
||||
defer tcpListener.Close()
|
||||
|
||||
cert := x509.CertFile()
|
||||
defer os.Remove(cert)
|
||||
key := x509.KeyFile()
|
||||
defer os.Remove(key)
|
||||
|
||||
mux, err := NewTLSMux(tcpListener, nil, cert, key, "")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to create mux: %s", err.Error())
|
||||
}
|
||||
go mux.Serve()
|
||||
|
||||
// Verify that the listener is secured.
|
||||
conn, err := tls.Dial("tcp", tcpListener.Addr().String(), &tls.Config{
|
||||
InsecureSkipVerify: true,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
state := conn.ConnectionState()
|
||||
if !state.HandshakeComplete {
|
||||
t.Fatal("connection handshake failed to complete")
|
||||
}
|
||||
}
|
||||
|
||||
func TestTLSMux_Fail(t *testing.T) {
|
||||
tcpListener := mustTCPListener("127.0.0.1:0")
|
||||
defer tcpListener.Close()
|
||||
|
||||
cert := x509.CertFile()
|
||||
defer os.Remove(cert)
|
||||
key := x509.KeyFile()
|
||||
defer os.Remove(key)
|
||||
|
||||
_, err := NewTLSMux(tcpListener, nil, "xxxx", "yyyy", "")
|
||||
if err == nil {
|
||||
t.Fatalf("created mux unexpectedly with bad resources")
|
||||
}
|
||||
}
|
||||
|
||||
type mockAddr struct {
|
||||
Nwk string
|
||||
Addr string
|
||||
}
|
||||
|
||||
func (m *mockAddr) Network() string {
|
||||
return m.Nwk
|
||||
}
|
||||
|
||||
func (m *mockAddr) String() string {
|
||||
return m.Addr
|
||||
}
|
||||
|
||||
// mustTCPListener returns a listener on bind, or panics.
|
||||
func mustTCPListener(bind string) net.Listener {
|
||||
l, err := net.Listen("tcp", bind)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return l
|
||||
}
|
||||
@ -0,0 +1,101 @@
|
||||
package tcp
|
||||
|
||||
import (
|
||||
"crypto/tls"
|
||||
"fmt"
|
||||
"net"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Transport is the network layer for internode communications.
|
||||
type Transport struct {
|
||||
ln net.Listener
|
||||
|
||||
certFile string // Path to local X.509 cert.
|
||||
certKey string // Path to corresponding X.509 key.
|
||||
remoteEncrypted bool // Remote nodes use encrypted communication.
|
||||
skipVerify bool // Skip verification of remote node certs.
|
||||
}
|
||||
|
||||
// NewTransport returns an initialized unecrypted Transport.
|
||||
func NewTransport() *Transport {
|
||||
return &Transport{}
|
||||
}
|
||||
|
||||
// NewTransport returns an initialized TLS-ecrypted Transport.
|
||||
func NewTLSTransport(certFile, keyPath string, skipVerify bool) *Transport {
|
||||
return &Transport{
|
||||
certFile: certFile,
|
||||
certKey: keyPath,
|
||||
remoteEncrypted: true,
|
||||
skipVerify: skipVerify,
|
||||
}
|
||||
}
|
||||
|
||||
// Open opens the transport, binding to the supplied address.
|
||||
func (t *Transport) Open(addr string) error {
|
||||
ln, err := net.Listen("tcp", addr)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if t.certFile != "" {
|
||||
config, err := createTLSConfig(t.certFile, t.certKey)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
ln = tls.NewListener(ln, config)
|
||||
}
|
||||
|
||||
t.ln = ln
|
||||
return nil
|
||||
}
|
||||
|
||||
// Dial opens a network connection.
|
||||
func (t *Transport) Dial(addr string, timeout time.Duration) (net.Conn, error) {
|
||||
dialer := &net.Dialer{Timeout: timeout}
|
||||
|
||||
var err error
|
||||
var conn net.Conn
|
||||
if t.remoteEncrypted {
|
||||
conf := &tls.Config{
|
||||
InsecureSkipVerify: t.skipVerify,
|
||||
}
|
||||
fmt.Println("doing a TLS dial")
|
||||
conn, err = tls.DialWithDialer(dialer, "tcp", addr, conf)
|
||||
} else {
|
||||
conn, err = dialer.Dial("tcp", addr)
|
||||
}
|
||||
|
||||
return conn, err
|
||||
}
|
||||
|
||||
// Accept waits for the next connection.
|
||||
func (t *Transport) Accept() (net.Conn, error) {
|
||||
c, err := t.ln.Accept()
|
||||
if err != nil {
|
||||
fmt.Println("error accepting: ", err.Error())
|
||||
}
|
||||
return c, err
|
||||
}
|
||||
|
||||
// Close closes the transport
|
||||
func (t *Transport) Close() error {
|
||||
return t.ln.Close()
|
||||
}
|
||||
|
||||
// Addr returns the binding address of the transport.
|
||||
func (t *Transport) Addr() net.Addr {
|
||||
return t.ln.Addr()
|
||||
}
|
||||
|
||||
// createTLSConfig returns a TLS config from the given cert and key.
|
||||
func createTLSConfig(certFile, keyFile string) (*tls.Config, error) {
|
||||
var err error
|
||||
config := &tls.Config{}
|
||||
config.Certificates = make([]tls.Certificate, 1)
|
||||
config.Certificates[0], err = tls.LoadX509KeyPair(certFile, keyFile)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return config, nil
|
||||
}
|
||||
@ -0,0 +1,70 @@
|
||||
package tcp
|
||||
|
||||
import (
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/rqlite/rqlite/testdata/x509"
|
||||
)
|
||||
|
||||
func Test_NewTransport(t *testing.T) {
|
||||
if NewTransport() == nil {
|
||||
t.Fatal("failed to create new Transport")
|
||||
}
|
||||
}
|
||||
|
||||
func Test_TransportOpenClose(t *testing.T) {
|
||||
tn := NewTransport()
|
||||
if err := tn.Open("localhost:0"); err != nil {
|
||||
t.Fatalf("failed to open transport: %s", err.Error())
|
||||
}
|
||||
if tn.Addr().String() == "localhost:0" {
|
||||
t.Fatalf("transport address set incorrectly, got: %s", tn.Addr().String())
|
||||
}
|
||||
if err := tn.Close(); err != nil {
|
||||
t.Fatalf("failed to close transport: %s", err.Error())
|
||||
}
|
||||
}
|
||||
|
||||
func Test_TransportDial(t *testing.T) {
|
||||
tn1 := NewTransport()
|
||||
tn1.Open("localhost:0")
|
||||
go tn1.Accept()
|
||||
tn2 := NewTransport()
|
||||
|
||||
_, err := tn2.Dial(tn1.Addr().String(), time.Second)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to connect to first transport: %s", err.Error())
|
||||
}
|
||||
tn1.Close()
|
||||
}
|
||||
|
||||
func Test_NewTLSTransport(t *testing.T) {
|
||||
c := x509.CertFile()
|
||||
defer os.Remove(c)
|
||||
k := x509.KeyFile()
|
||||
defer os.Remove(k)
|
||||
|
||||
if NewTLSTransport(c, k, true) == nil {
|
||||
t.Fatal("failed to create new TLS Transport")
|
||||
}
|
||||
}
|
||||
|
||||
func Test_TLSTransportOpenClose(t *testing.T) {
|
||||
c := x509.CertFile()
|
||||
defer os.Remove(c)
|
||||
k := x509.KeyFile()
|
||||
defer os.Remove(k)
|
||||
|
||||
tn := NewTLSTransport(c, k, true)
|
||||
if err := tn.Open("localhost:0"); err != nil {
|
||||
t.Fatalf("failed to open TLS transport: %s", err.Error())
|
||||
}
|
||||
if tn.Addr().String() == "localhost:0" {
|
||||
t.Fatalf("TLS transport address set incorrectly, got: %s", tn.Addr().String())
|
||||
}
|
||||
if err := tn.Close(); err != nil {
|
||||
t.Fatalf("failed to close TLS transport: %s", err.Error())
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue