scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-10 15:01:03 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

https with signed certificates work

Browse source
This commit is contained in:
Yuce Tekol 2017-10-08 21:11:19 +03:00
parent 85e19a1155
commit 3501732b19
No known key found for this signature in database
GPG key ID: CB59E46D2FB90573
17 changed files with 182 additions and 76 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

25
client.go
View file

@ -31,13 +31,20 @@ import (
"strconv"
"time"
"crypto/tls"
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa/internal"
)
// ClientOptions represents the configuration for a Client
type ClientOptions struct {
TLS *tls.Config
}
// Client represents a client to the Pilosa cluster.
type Client struct {
host *URI
host *URI
options *ClientOptions
// The client to use for HTTP communication.
// Defaults to the http.DefaultClient.
@ -45,7 +52,7 @@ type Client struct {
}
// NewClient returns a new instance of Client to connect to host.
func NewClient(host string) (*Client, error) {
func NewClient(host string, options *ClientOptions) (*Client, error) {
if host == "" {
return nil, ErrHostRequired
}
@ -55,16 +62,24 @@ func NewClient(host string) (*Client, error) {
return nil, err
}
return NewClientFromURI(uri)
return NewClientFromURI(uri, options)
}
func NewClientFromURI(uri *URI) (*Client, error) {
func NewClientFromURI(uri *URI, options *ClientOptions) (*Client, error) {
if uri == nil {
return nil, ErrHostRequired
}
if options == nil {
options = &ClientOptions{}
}
transport := &http.Transport{}
if options.TLS != nil {
transport.TLSClientConfig = options.TLS
}
client := &http.Client{Transport: transport}
return &Client{
host: uri,
HTTPClient: http.DefaultClient,
HTTPClient: client,
}, nil
}

19
config.go
View file

@ -14,7 +14,9 @@
package pilosa
import "time"
import (
"time"
)
// Cluster types.
const (
@ -120,7 +122,7 @@ func (c *Config) Validate() error {
if err != nil {
return err
}
if !foundItem(c.Cluster.Hosts, bindWithDefaults.HostPort()) {
if !c.foundHost(bindWithDefaults) {
return ErrConfigHostsMissing
}
}
@ -129,6 +131,19 @@ func (c *Config) Validate() error {
return nil
}
func (c *Config) foundHost(host *URI) bool {
for _, clusterHost := range c.Cluster.Hosts {
uri, err := NewURIFromAddress(clusterHost)
if err != nil {
continue
}
if host.Equals(uri) {
return true
}
}
return false
}
// Duration is a TOML wrapper type for time.Duration.
type Duration time.Duration

4
ctl/backup.go
View file

@ -38,6 +38,8 @@ type BackupCommand struct {
// Standard input/output
*pilosa.CmdIO
TLS pilosa.TLSConfig
}
// NewBackupCommand returns a new instance of BackupCommand.
@ -55,7 +57,7 @@ func (cmd *BackupCommand) Run(ctx context.Context) error {
}
// Create a client to the server.
client, err := pilosa.NewClient(cmd.Host)
client, err := pilosa.NewClient(cmd.Host, nil)
if err != nil {
return err
}

2
ctl/bench.go
View file

@ -52,7 +52,7 @@ func NewBenchCommand(stdin io.Reader, stdout, stderr io.Writer) *BenchCommand {
// Run executes the bench command.
func (cmd *BenchCommand) Run(ctx context.Context) error {
// Create a client to the server.
client, err := pilosa.NewClient(cmd.Host)
client, err := pilosa.NewClient(cmd.Host, nil)
if err != nil {
return err
}

2
ctl/export.go
View file

@ -73,7 +73,7 @@ func (cmd *ExportCommand) Run(ctx context.Context) error {
}
// Create a client to the server.
client, err := pilosa.NewClient(cmd.Host)
client, err := pilosa.NewClient(cmd.Host, nil)
if err != nil {
return err
}

2
ctl/import.go
View file

@ -86,7 +86,7 @@ func (cmd *ImportCommand) Run(ctx context.Context) error {
return errors.New("path required")
}
// Create a client to the server.
client, err := pilosa.NewClient(cmd.Host)
client, err := pilosa.NewClient(cmd.Host, nil)
if err != nil {
return err
}

2
ctl/restore.go
View file

@ -55,7 +55,7 @@ func (cmd *RestoreCommand) Run(ctx context.Context) error {
}
// Create a client to the server.
client, err := pilosa.NewClient(cmd.Host)
client, err := pilosa.NewClient(cmd.Host, nil)
if err != nil {
return err
}

1
executor_test.go
View file

@ -75,7 +75,6 @@ func TestExecutor_Execute_Bitmap(t *testing.T) {
} else if bits := res[0].(*pilosa.Bitmap).Bits(); !reflect.DeepEqual(bits, []uint64{3, SliceWidth + 1}) {
t.Fatalf("unexpected bits: %+v", bits)
} else if attrs := res[0].(*pilosa.Bitmap).Attrs; !reflect.DeepEqual(attrs, map[string]interface{}{}) {
fmt.Println("ATTRS", attrs)
t.Fatalf("unexpected attrs: %s", spew.Sdump(attrs))
}
})

9
fragment.go
View file

@ -1654,8 +1654,9 @@ func (h *blockHasher) WriteValue(v uint64) {
type FragmentSyncer struct {
Fragment *Fragment
Host string
Cluster *Cluster
Host string
Cluster *Cluster
ClientOptions *ClientOptions
Closing <-chan struct{}
}
@ -1690,7 +1691,7 @@ func (s *FragmentSyncer) SyncFragment() error {
}
// Retrieve remote blocks.
client, err := NewClient(node.Host)
client, err := NewClient(node.Host, s.ClientOptions)
if err != nil {
return err
}
@ -1769,7 +1770,7 @@ func (s *FragmentSyncer) syncBlock(id int) error {
return nil
}
client, err := NewClient(node.Host)
client, err := NewClient(node.Host, s.ClientOptions)
if err != nil {
return err
}

4
gossip/gossip.go
View file

@ -48,7 +48,7 @@ type GossipNodeSet struct {
func (g *GossipNodeSet) Nodes() []*pilosa.Node {
a := make([]*pilosa.Node, 0, g.memberlist.NumMembers())
for _, n := range g.memberlist.Members() {
a = append(a, &pilosa.Node{Host: n.Name})
a = append(a, &pilosa.Node{Scheme: "gossip", Host: n.Name})
}
return a
}
@ -77,7 +77,7 @@ func (g *GossipNodeSet) Open() error {
}
// attach to gossip seed node
nodes := []*pilosa.Node{&pilosa.Node{Host: g.config.gossipSeed}} //TODO: support a list of seeds
nodes := []*pilosa.Node{&pilosa.Node{Scheme: "gossip", Host: g.config.gossipSeed}} //TODO: support a list of seeds
err = g.joinWithRetry(pilosa.Nodes(nodes).Hosts())
if err != nil {
return err

3
handler.go
View file

@ -58,6 +58,7 @@ type Handler struct {
// Local hostname & cluster configuration.
Host *URI
Cluster *Cluster
ClientOptions *ClientOptions
Router *mux.Router
@ -1504,7 +1505,7 @@ func (h *Handler) handlePostFrameRestore(w http.ResponseWriter, r *http.Request)
}
// Create a client for the remote cluster.
client, err := NewClientFromURI(host)
client, err := NewClientFromURI(host, h.ClientOptions)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return

18
holder.go
View file

@ -417,8 +417,9 @@ func (h *Holder) logger() *log.Logger { return log.New(h.LogOutput, "", log.Lstd
type HolderSyncer struct {
Holder *Holder
Host *URI
Cluster *Cluster
Host *URI
Cluster *Cluster
ClientOptions *ClientOptions
// Signals that the sync should stop.
Closing <-chan struct{}
@ -504,7 +505,7 @@ func (s *HolderSyncer) syncIndex(index string) error {
// Sync with every other host.
for _, node := range Nodes(s.Cluster.Nodes).FilterHost(s.Host.HostPort()) {
client, err := NewClient(node.Host)
client, err := NewClient(node.Host, s.ClientOptions)
if err != nil {
return err
}
@ -549,7 +550,7 @@ func (s *HolderSyncer) syncFrame(index, name string) error {
// Sync with every other host.
for _, node := range Nodes(s.Cluster.Nodes).FilterHost(s.Host.HostPort()) {
client, err := NewClient(node.Host)
client, err := NewClient(node.Host, s.ClientOptions)
if err != nil {
return err
}
@ -602,10 +603,11 @@ func (s *HolderSyncer) syncFragment(index, frame, view string, slice uint64) err
// Sync fragments together.
fs := FragmentSyncer{
Fragment: frag,
Host: s.Host.HostPort(),
Cluster: s.Cluster,
Closing: s.Closing,
Fragment: frag,
Host: s.Host.HostPort(),
Cluster: s.Cluster,
Closing: s.Closing,
ClientOptions: s.ClientOptions,
}
if err := fs.SyncFragment(); err != nil {
return err

18
server.go
View file

@ -69,7 +69,7 @@ type Server struct {
MetricInterval time.Duration
// TLS configuration
TLS TLSConfig
TLS *tls.Config
// Misc options.
MaxWritesPerRequest int
@ -107,19 +107,8 @@ func (s *Server) Open() error {
var err error
// If bind URI has the https scheme, enable TLS
if s.Host.Scheme() == "https" {
if s.TLS.CertificatePath == "" {
return errors.New("certificate path is required for TLS sockets")
}
if s.TLS.CertificateKeyPath == "" {
return errors.New("certificate key path is required for TLS sockets")
}
cert, err := tls.LoadX509KeyPair(s.TLS.CertificatePath, s.TLS.CertificateKeyPath)
if err != nil {
return err
}
config := tls.Config{Certificates: []tls.Certificate{cert}}
ln, err = tls.Listen("tcp", s.Host.HostPort(), &config)
if s.Host.Scheme() == "https" && s.TLS != nil {
ln, err = tls.Listen("tcp", s.Host.HostPort(), s.TLS)
if err != nil {
return err
}
@ -255,6 +244,7 @@ func (s *Server) monitorAntiEntropy() {
syncer.Host = s.Host
syncer.Cluster = s.Cluster
syncer.Closing = s.closing
syncer.ClientOptions = &ClientOptions{TLS: s.TLS}
// Sync holders.
if err := syncer.SyncHolder(); err != nil {

41
server/server.go
View file

@ -29,6 +29,7 @@ import (
"strings"
"time"
"crypto/tls"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/gossip"
"github.com/pilosa/pilosa/statsd"
@ -110,11 +111,24 @@ func (m *Command) SetupServer() error {
return err
}
uri, err := pilosa.AddressWithDefaults(m.Config.Bind)
if err != nil {
return err
}
m.Server.Host = uri
cluster := pilosa.NewCluster()
cluster.ReplicaN = m.Config.Cluster.ReplicaN
for _, hostport := range m.Config.Cluster.Hosts {
cluster.Nodes = append(cluster.Nodes, &pilosa.Node{Host: hostport})
for _, address := range m.Config.Cluster.Hosts {
uri, err := pilosa.NewURIFromAddress(address)
if err != nil {
return err
}
cluster.Nodes = append(cluster.Nodes, &pilosa.Node{
Scheme: uri.Scheme(),
Host: uri.HostPort(),
})
}
m.Server.Cluster = cluster
@ -138,11 +152,21 @@ func (m *Command) SetupServer() error {
// Copy configuration flags.
m.Server.MaxWritesPerRequest = m.Config.MaxWritesPerRequest
bindWithDefaults, err := pilosa.AddressWithDefaults(m.Config.Bind)
if err != nil {
return err
// Setup TLS
if uri.Scheme() == "https" {
if m.Config.TLS.CertificatePath == "" {
return errors.New("certificate path is required for TLS sockets")
}
if m.Config.TLS.CertificateKeyPath == "" {
return errors.New("certificate key path is required for TLS sockets")
}
cert, err := tls.LoadX509KeyPair(m.Config.TLS.CertificatePath, m.Config.TLS.CertificateKeyPath)
if err != nil {
return err
}
m.Server.TLS = &tls.Config{Certificates: []tls.Certificate{cert}}
m.Server.Handler.ClientOptions = &pilosa.ClientOptions{TLS: m.Server.TLS}
}
m.Server.Host = bindWithDefaults
// Set internal port (string).
gossipPortStr := pilosa.DefaultGossipPort
@ -162,8 +186,8 @@ func (m *Command) SetupServer() error {
}
// get the host portion of addr to use for binding
gossipHost := bindWithDefaults.Host()
gossipNodeSet := gossip.NewGossipNodeSet(bindWithDefaults.HostPort(), gossipHost, gossipPort, gossipSeed, m.Server)
gossipHost := uri.Host()
gossipNodeSet := gossip.NewGossipNodeSet(uri.HostPort(), gossipHost, gossipPort, gossipSeed, m.Server)
m.Server.Cluster.NodeSet = gossipNodeSet
m.Server.Broadcaster = gossipNodeSet
m.Server.BroadcastReceiver = gossipNodeSet
@ -182,7 +206,6 @@ func (m *Command) SetupServer() error {
// Set configuration options.
m.Server.AntiEntropyInterval = time.Duration(m.Config.AntiEntropy.Interval)
m.Server.Cluster.LongQueryTime = time.Duration(m.Config.Cluster.LongQueryTime)
m.Server.TLS = m.Config.TLS
return nil
}

6
server/server_test.go
View file

@ -53,7 +53,7 @@ func TestMain_Set_Quick(t *testing.T) {
defer m.Close()
// Create client.
client, err := pilosa.NewClient(m.Server.Host.HostPort())
client, err := pilosa.NewClient(m.Server.Host.HostPort(), nil)
if err != nil {
t.Fatal(err)
}
@ -326,7 +326,7 @@ func TestMain_FrameRestore(t *testing.T) {
defer m2.Close()
// Import from first cluster.
client, err := pilosa.NewClient(m2.Server.Host.HostPort())
client, err := pilosa.NewClient(m2.Server.Host.HostPort(), nil)
if err != nil {
t.Fatal(err)
} else if err := m2.Client().CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists {
@ -697,7 +697,7 @@ func (m *Main) URL() string { return "http://" + m.Server.Addr().String() }
// Client returns a client to connect to the program.
func (m *Main) Client() *pilosa.Client {
client, err := pilosa.NewClient(m.Server.Host.HostPort())
client, err := pilosa.NewClient(m.Server.Host.HostPort(), nil)
if err != nil {
panic(err)
}

2
test/client.go
View file

@ -11,7 +11,7 @@ type Client struct {
// MustNewClient returns a new instance of Client. Panic on error.
func MustNewClient(host string) *Client {
c, err := pilosa.NewClient(host)
c, err := pilosa.NewClient(host, nil)
if err != nil {
panic(err)
}

100
uri.go
View file

@ -22,12 +22,14 @@ import (
"strings"
)
var addressRegexp = regexp.MustCompile("^(([+a-z]+):\\/\\/)?([0-9a-z.-]+)?(:([0-9]+))?$")
var schemeRegexp = regexp.MustCompile("^[+a-z]+$")
var hostRegexp = regexp.MustCompile("^[0-9a-z.-]+$|^\\[[:0-9a-fA-F]+\\]$")
var addressRegexp = regexp.MustCompile("^(([+a-z]+):\\/\\/)?([0-9a-z.-]+|\\[[:0-9a-fA-F]+\\])?(:([0-9]+))?$")
// URI represents a Pilosa URI.
// A Pilosa URI consists of three parts:
// 1) Scheme: Protocol of the URI. Default: http.
// 2) Host: Hostname or IP URI. Default: localhost.
// 2) Host: Hostname or IP URI. Default: localhost. IPv6 addresses should be written in brackets, e.g., `[fd42:4201:f86b:7e09:216:3eff:fefa:ed80]`.
// 3) Port: Port of the URI. Default: 10101.
//
// All parts of the URI are optional. The following are equivalent:
@ -41,6 +43,7 @@ type URI struct {
scheme string
host string
port uint16
error error
}
// DefaultURI creates and returns the default URI.
@ -54,17 +57,28 @@ func DefaultURI() *URI {
// NewURIFromHostPort returns a URI with specified host and port.
func NewURIFromHostPort(host string, port uint16) (*URI, error) {
// TODO: validate host
return &URI{
scheme: "http",
host: host,
port: port,
}, nil
uri := DefaultURI()
err := uri.SetHost(host)
if err != nil {
return nil, err
}
uri.SetPort(port)
return uri, nil
}
// NewURIFromAddress parses the passed address and returns a URI.
func NewURIFromAddress(address string) (*URI, error) {
return parseAddress(address)
uri, err := parseAddress(address)
if err != nil {
return &URI{error: err}, err
}
return uri, err
}
// URIFromAddress creates a URI from the given address.
func URIFromAddress(host string) *URI {
uri, _ := NewURIFromAddress(host)
return uri
}
// Scheme returns the scheme of this URI.
@ -72,8 +86,14 @@ func (u *URI) Scheme() string {
return u.scheme
}
func (u *URI) SetScheme(scheme string) {
// SetScheme sets the scheme of this URI.
func (u *URI) SetScheme(scheme string) error {
m := schemeRegexp.FindStringSubmatch(scheme)
if m == nil {
return errors.New("invalid scheme")
}
u.scheme = scheme
return nil
}
// Host returns the host of this URI.
@ -81,16 +101,36 @@ func (u *URI) Host() string {
return u.host
}
// SetHost sets the host of this URI.
func (u *URI) SetHost(host string) error {
m := hostRegexp.FindStringSubmatch(host)
if m == nil {
return errors.New("invalid host")
}
u.host = host
return nil
}
// Port returns the port of this URI.
func (u *URI) Port() uint16 {
return u.port
}
// SetPort updates the port
// SetPort sets the port of this URI.
func (u *URI) SetPort(port uint16) {
u.port = port
}
// HostPort returns `Host:Port`
func (u *URI) HostPort() string {
// XXX: The following is just to make TestHandler_Status; remove it
if u == nil {
return ""
}
s := fmt.Sprintf("%s:%d", u.host, u.port)
return s
}
// Normalize returns the address in a form usable by a HTTP client.
func (u *URI) Normalize() string {
scheme := u.scheme
@ -101,15 +141,6 @@ func (u *URI) Normalize() string {
return fmt.Sprintf("%s://%s:%d", scheme, u.host, u.port)
}
// HostPort returns the address suitable for passing to `net.Listener.Listen`
func (u *URI) HostPort() string {
if u == nil {
return ""
}
s := fmt.Sprintf("%s:%d", u.host, u.port)
return s
}
// Equals returns true if the checked URI is equivalent to this URI.
func (u URI) Equals(other *URI) bool {
if other == nil {
@ -120,6 +151,33 @@ func (u URI) Equals(other *URI) bool {
u.port == other.port
}
// Error returns the error if this URI has one.
func (u *URI) Error() error {
return u.error
}
// Valid returns true if this is a valid URI.
func (u *URI) Valid() bool {
return u != nil && u.error == nil
}
// The following methods are required to implement pflag Value interface.
// Set sets the time quantum value.
func (u *URI) Set(value string) error {
uri, err := NewURIFromAddress(value)
if err != nil {
return err
}
*u = *uri
return nil
}
// Type returns the type of a time quantum value.
func (u URI) Type() string {
return "URI"
}
func parseAddress(address string) (uri *URI, err error) {
m := addressRegexp.FindStringSubmatch(address)
if m == nil {
@ -137,7 +195,7 @@ func parseAddress(address string) (uri *URI, err error) {
if m[5] != "" {
port, err = strconv.Atoi(m[5])
if err != nil {
return nil, errors.New("error converting port string to int")
return nil, errors.New("converting port string to int")
}
}
uri = &URI{