mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-08-28 10:54:59 +00:00
811 lines
26 KiB
Go
811 lines
26 KiB
Go
// Copyright 2022 Molecula Corp. (DBA FeatureBase).
|
|
// SPDX-License-Identifier: Apache-2.0
|
|
package server
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"io"
|
|
"log"
|
|
"net"
|
|
"net/url"
|
|
"os"
|
|
"path/filepath"
|
|
"runtime"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/featurebasedb/featurebase/v3/authz"
|
|
petcd "github.com/featurebasedb/featurebase/v3/etcd"
|
|
rbfcfg "github.com/featurebasedb/featurebase/v3/rbf/cfg"
|
|
"github.com/featurebasedb/featurebase/v3/storage"
|
|
"github.com/featurebasedb/featurebase/v3/toml"
|
|
"github.com/pkg/errors"
|
|
)
|
|
|
|
const (
|
|
defaultBindPort = "10101"
|
|
defaultBindGRPCPort = "20101"
|
|
defaultDiagnosticsInterval = 1 * time.Hour
|
|
|
|
namespacePilosa = "pilosa"
|
|
namespaceFeaturebase = "featurebase"
|
|
)
|
|
|
|
// TLSConfig contains TLS configuration
|
|
type TLSConfig struct {
|
|
// CertificatePath contains the path to the certificate (.crt or .pem file)
|
|
CertificatePath string `toml:"certificate"`
|
|
// CertificateKeyPath contains the path to the certificate key (.key file)
|
|
CertificateKeyPath string `toml:"key"`
|
|
// CACertPath is the path to a CA certificate (.crt or .pem file)
|
|
CACertPath string `toml:"ca-certificate"`
|
|
// SkipVerify disables verification of server certificates when connecting to another Pilosa node
|
|
SkipVerify bool `toml:"skip-verify"`
|
|
// EnableClientVerification enables verification of client TLS certificates (Mutual TLS)
|
|
EnableClientVerification bool `toml:"enable-client-verification"`
|
|
}
|
|
|
|
// Config represents the configuration for the command.
|
|
type Config struct {
|
|
// Name a unique name for this node in the cluster.
|
|
Name string `toml:"name"`
|
|
|
|
// ControllerAddress is the location at which this node should register
|
|
// itself and retrieve its instructions. For example, after registring, the
|
|
// Controller service might tell this node that it is responsible for
|
|
// specific shards for a particular index.
|
|
ControllerAddress string `toml:"controller-address"`
|
|
|
|
// WriteloggerDir is the location at which this node should
|
|
// read/write change logs. Typically a network mounted filesystem
|
|
// for availability/durability.
|
|
WriteloggerDir string `toml:"writelogger-dir"`
|
|
|
|
// SnapshotterDir is the location at which this node should
|
|
// read/write snapshots. Typically a network mounted filesystem
|
|
// for availability/durability.
|
|
SnapshotterDir string `toml:"snapshotter-dir"`
|
|
|
|
// DataDir is the directory where Pilosa stores both indexed data and
|
|
// running state such as cluster topology information.
|
|
DataDir string `toml:"data-dir"`
|
|
|
|
// Bind is the host:port on which Pilosa will listen.
|
|
Bind string `toml:"bind"`
|
|
|
|
// BindGRPC is the host:port on which Pilosa will bind for gRPC.
|
|
BindGRPC string `toml:"bind-grpc"`
|
|
|
|
// Listener is an already-bound listener to use for http.
|
|
// Listener *net.TCPListener
|
|
Listener net.Listener
|
|
|
|
// GRPCListener is an already-bound listener to use for gRPC.
|
|
// This is for use by test infrastructure, where it's useful to
|
|
// be able to dynamically generate the bindings by actually binding
|
|
// to :0, and avoid "address already in use" errors.
|
|
GRPCListener net.Listener
|
|
|
|
// Advertise is the address advertised by the server to other nodes
|
|
// in the cluster. It should be reachable by all other nodes and should
|
|
// route to an interface that Bind is listening on.
|
|
Advertise string `toml:"advertise"`
|
|
|
|
// AdvertiseGRPC is the address advertised by the server to other nodes
|
|
// in the cluster. It should be reachable by all other nodes and should
|
|
// route to an interface that BindGRPC is listening on.
|
|
AdvertiseGRPC string `toml:"advertise-grpc"`
|
|
|
|
// MaxWritesPerRequest limits the number of mutating commands that can be in
|
|
// a single request to the server. This includes Set, Clear, ClearRow, Store, and SetBit.
|
|
MaxWritesPerRequest int `toml:"max-writes-per-request"`
|
|
|
|
// LogPath configures where Pilosa will write logs.
|
|
LogPath string `toml:"log-path"`
|
|
|
|
// Verbose toggles verbose logging which can be useful for debugging.
|
|
Verbose bool `toml:"verbose"`
|
|
|
|
// HTTP Handler options
|
|
Handler struct {
|
|
// CORS Allowed Origins
|
|
AllowedOrigins []string `toml:"allowed-origins"`
|
|
} `toml:"handler"`
|
|
|
|
// MaxMapCount puts an in-process limit on the number of mmaps. After this
|
|
// is exhausted, Pilosa will fall back to reading the file into memory
|
|
// normally.
|
|
MaxMapCount uint64 `toml:"max-map-count"`
|
|
|
|
// MaxFileCount puts a soft, in-process limit on the number of open fragment
|
|
// files. Once this limit is passed, Pilosa will only keep files open while
|
|
// actively working with them, and will close them afterward. This has a
|
|
// negative effect on performance for workloads which make small appends to
|
|
// lots of fragments.
|
|
MaxFileCount uint64 `toml:"max-file-count"`
|
|
|
|
// TLS
|
|
TLS TLSConfig `toml:"tls"`
|
|
|
|
// WorkerPoolSize controls how many goroutines are created for
|
|
// processing queries. Defaults to runtime.NumCPU(). It is
|
|
// intentionally not defined as a flag... only exposed here so
|
|
// that we can limit the size while running tests in CI so we
|
|
// don't exhaust the goroutine limit.
|
|
WorkerPoolSize int `toml:"-"`
|
|
|
|
// ImportWorkerPoolSize controls how many goroutines are created for
|
|
// processing importRoaring jobs. Defaults to runtime.NumCPU(). It is
|
|
// intentionally not defined as a flag... only exposed here so
|
|
// that we can limit the size while running tests in CI so we
|
|
// don't exhaust the goroutine limit.
|
|
ImportWorkerPoolSize int `toml:"-"`
|
|
|
|
// DirectiveWorkerPoolSize controls how many goroutines are created for
|
|
// processing a Directive (i.e. concurrently loading key/partition/shard
|
|
// data from shapshotter and writelogger) on a compute node. Defaults to
|
|
// runtime.NumCPU().
|
|
DirectiveWorkerPoolSize int `toml:"-"`
|
|
|
|
// Limits the total amount of memory to be used by Extract() & SELECT queries.
|
|
MaxQueryMemory int64 `toml:"max-query-memory"`
|
|
|
|
// On startup, featurebase server contacts a web server to check the latest version.
|
|
// This stores the address for that check
|
|
VerChkAddress string `toml:"verchk-address"`
|
|
|
|
// When checking version, server sends a UUID so that we can keep track of
|
|
// how many unique Featurebase installs are out there. The file is stored in
|
|
// the data directory; this stores the filename to use.
|
|
UUIDFile string `toml:"uuid-file"`
|
|
|
|
Cluster struct {
|
|
ReplicaN int `toml:"replicas"`
|
|
Name string `toml:"name"`
|
|
// This LongQueryTime is deprecated but still exists for backward compatibility
|
|
LongQueryTime toml.Duration `toml:"long-query-time"`
|
|
PartitionToNodeAssignment string `toml:"partition-to-node-assignment"`
|
|
} `toml:"cluster"`
|
|
|
|
// Etcd config is based on embedded etcd.
|
|
Etcd petcd.Options `toml:"etcd"`
|
|
|
|
LongQueryTime toml.Duration `toml:"long-query-time"`
|
|
|
|
Translation struct {
|
|
MapSize int `toml:"map-size"`
|
|
// DEPRECATED: Translation config supports translation store replication.
|
|
PrimaryURL string `toml:"primary-url"`
|
|
} `toml:"translation"`
|
|
|
|
// AntiEntropy config is now deprecated
|
|
AntiEntropy struct {
|
|
Interval toml.Duration `toml:"interval"`
|
|
} `toml:"anti-entropy"`
|
|
|
|
Metric struct {
|
|
// Service can be statsd, prometheus, expvar, or none.
|
|
Service string `toml:"service"`
|
|
// Host tells the statsd client where to write.
|
|
Host string `toml:"host"`
|
|
PollInterval toml.Duration `toml:"poll-interval"`
|
|
// Diagnostics toggles sending some limited diagnostic information to
|
|
// Pilosa's developers.
|
|
Diagnostics bool `toml:"diagnostics"`
|
|
} `toml:"metric"`
|
|
|
|
Tracing struct {
|
|
// SamplerType is the type of sampler to use.
|
|
SamplerType string `toml:"sampler-type"`
|
|
// SamplerParam is the parameter passed to the tracing sampler.
|
|
// Its meaning is dependent on the type of sampler.
|
|
SamplerParam float64 `toml:"sampler-param"`
|
|
// AgentHostPort is the host:port of the local agent.
|
|
AgentHostPort string `toml:"agent-host-port"`
|
|
} `toml:"tracing"`
|
|
|
|
Profile struct {
|
|
// BlockRate is passed directly to runtime.SetBlockProfileRate
|
|
BlockRate int `toml:"block-rate"`
|
|
// MutexFraction is passed directly to runtime.SetMutexProfileFraction
|
|
MutexFraction int `toml:"mutex-fraction"`
|
|
} `toml:"profile"`
|
|
|
|
// CheckInTimeout is the amount of time between compute node check-ins to
|
|
// Controller.
|
|
CheckInInterval time.Duration `toml:"check-in-interval"`
|
|
|
|
// Storage.Backend determines which Tx implementation the holder/Index will
|
|
// use; one of the available transactional-storage engines. Choices are
|
|
// listed in the string constants below. Should be one of "roaring" or
|
|
// "rbf".
|
|
Storage *storage.Config `toml:"storage"`
|
|
|
|
// RBFConfig defines all externally configurable RBF flags.
|
|
RBFConfig *rbfcfg.Config `toml:"rbf"`
|
|
|
|
// QueryHistoryLength sets the maximum number of queries that are maintained
|
|
// for the /query-history endpoint. This parameter is per-node, and the
|
|
// result combines the history from all nodes.
|
|
QueryHistoryLength int `toml:"query-history-length"`
|
|
|
|
// LookupDBDSN is an external database to connect to for `ExternalLookup` queries.
|
|
LookupDBDSN string `toml:"lookup-db-dsn"`
|
|
|
|
// Future flags are used to represent features or functionality which is not
|
|
// yet the default behavior, but will be in a future release.
|
|
Future struct {
|
|
// Rename, if true, will outwardly present the name of the application
|
|
// as FeatureBase instead of Pilosa.
|
|
Rename bool `toml:"rename"`
|
|
} `toml:"future"`
|
|
|
|
DataDog struct {
|
|
Enable bool `toml:"enable"`
|
|
EnableTracing bool `toml:"enable-tracing"`
|
|
Service string `toml:"service"`
|
|
Env string `toml:"env"`
|
|
Version string `toml:"version"`
|
|
Tags string `toml:"tags"`
|
|
CPUProfile bool `toml:"cpu-profile"`
|
|
MutexProfile bool `toml:"mutex-profile"`
|
|
GoroutineProfile bool `toml:"goroutine-profile"`
|
|
BlockProfile bool `toml:"block-profile"`
|
|
HeapProfile bool `toml:"heap-profile"`
|
|
} `toml:"datadog"`
|
|
|
|
Auth Auth
|
|
|
|
Dataframe struct {
|
|
Enable bool `toml:"enable"`
|
|
UseParquet bool `toml:"use-parquet"`
|
|
} `toml:"dataframe"`
|
|
}
|
|
|
|
type Auth struct {
|
|
// Enable AuthZ/AuthN for featurebase server
|
|
Enable bool `toml:"enable"`
|
|
|
|
ClientId string `toml:"client-id"`
|
|
ClientSecret string `toml:"client-secret"`
|
|
AuthorizeURL string `toml:"authorize-url"`
|
|
TokenURL string `toml:"token-url"`
|
|
GroupEndpointURL string `toml:"group-endpoint-url"`
|
|
RedirectBaseURL string `toml:"redirect-base-url"`
|
|
LogoutURL string `toml:"logout-url"`
|
|
Scopes []string `toml:"scopes"`
|
|
SecretKey string `toml:"secret-key"`
|
|
PermissionsFile string `toml:"permissions"`
|
|
QueryLogPath string `toml:"query-log-path"`
|
|
ConfiguredIPs []string `toml:"configured-ips"`
|
|
}
|
|
|
|
// Namespace returns the namespace to use based on the Future flag.
|
|
func (c *Config) Namespace() string {
|
|
if c.Future.Rename {
|
|
return namespaceFeaturebase
|
|
}
|
|
return namespacePilosa
|
|
}
|
|
|
|
// MustValidate checks that all ports in a Config are unique and not zero.
|
|
// We disallow zero because the tests need to be using from the pre-allocated
|
|
// block of ports maintained by the pilosa/test/port port-mapper.
|
|
func (c *Config) MustValidate() {
|
|
err := c.validate()
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
}
|
|
|
|
func (c *Config) validate() error {
|
|
hostPort := []string{
|
|
"Bind", c.Bind, // :10101
|
|
"BindGRPC", c.BindGRPC, // :20101
|
|
"Advertise", c.Advertise, // on hp = 'http://localhost:63002'
|
|
"AdvertiseGRPC", c.AdvertiseGRPC, // on hp = 'http://localhost:63003'
|
|
"Etcd.LClientURL", c.Etcd.LClientURL, // on hp = ':14000'
|
|
"Etcd.AClientURL", c.Etcd.AClientURL, // ""
|
|
"Etcd.LPeerURL", c.Etcd.LPeerURL, // ":"
|
|
"Etcd.APeerURL", c.Etcd.APeerURL, // ""
|
|
"Etcd.ClusterURL", c.Etcd.ClusterURL,
|
|
}
|
|
|
|
ports := make(map[int]bool)
|
|
n := len(hostPort)
|
|
for i := 0; i < n; i += 2 {
|
|
name := hostPort[i]
|
|
hp := hostPort[i+1]
|
|
if hp == "" {
|
|
continue
|
|
}
|
|
if name == "Advertise" && (hp == "" || hp == ":") {
|
|
continue
|
|
}
|
|
if name == "AdvertiseGRPC" && (hp == "" || hp == ":") {
|
|
continue
|
|
}
|
|
|
|
hp = strings.TrimPrefix(hp, "http://")
|
|
hp = strings.TrimPrefix(hp, "https://")
|
|
splt := strings.Split(hp, ":")
|
|
if len(splt) != 2 {
|
|
return fmt.Errorf("'%v' host:port '%v' did not have a colon; all='%#v'", name, hp, hostPort)
|
|
}
|
|
portstring := splt[1]
|
|
port, err := strconv.Atoi(portstring)
|
|
if err != nil {
|
|
return fmt.Errorf("on '%v', could not convert '%v' to int in '%v': '%v'", name, portstring, hp, err)
|
|
}
|
|
if port == 0 {
|
|
return fmt.Errorf("name '%v': zero port found, not allowed. '%v'. all ='%#v'", name, hp, hostPort)
|
|
}
|
|
if ports[port] {
|
|
return fmt.Errorf("name '%v': duplicate port found, not allowed. '%v' with port %v. all ='%#v'", name, hp, port, hostPort)
|
|
}
|
|
ports[port] = true
|
|
}
|
|
return nil
|
|
}
|
|
|
|
const (
|
|
PartitionToNodeJmp string = "jmp-hash"
|
|
PartitionToNodeModulus string = "modulus"
|
|
)
|
|
|
|
// NewConfig returns an instance of Config with default options.
|
|
func NewConfig() *Config {
|
|
c := &Config{
|
|
Name: "pilosa0",
|
|
DataDir: "~/.pilosa",
|
|
Bind: ":" + defaultBindPort,
|
|
BindGRPC: ":" + defaultBindGRPCPort,
|
|
MaxWritesPerRequest: 5000,
|
|
|
|
// We default these Max File/Map counts very high. This is basically a
|
|
// backwards compatibility thing where we don't want to cause different
|
|
// behavior for those who had previously set their system limits high,
|
|
// and weren't experiencing any bad behavior. Ideally you want these set
|
|
// a bit below your system limits.
|
|
MaxMapCount: 1000000,
|
|
MaxFileCount: 1000000,
|
|
|
|
TLS: TLSConfig{},
|
|
|
|
WorkerPoolSize: runtime.NumCPU(),
|
|
ImportWorkerPoolSize: runtime.NumCPU(),
|
|
|
|
DirectiveWorkerPoolSize: runtime.NumCPU(),
|
|
|
|
Storage: storage.NewDefaultConfig(),
|
|
RBFConfig: rbfcfg.NewDefaultConfig(),
|
|
|
|
QueryHistoryLength: 100,
|
|
|
|
LongQueryTime: toml.Duration(-time.Minute),
|
|
|
|
CheckInInterval: 60 * time.Second,
|
|
VerChkAddress: "https://analytics.featurebase.com/v2/featurebase/metrics",
|
|
UUIDFile: ".client_id.txt",
|
|
}
|
|
|
|
// Cluster config.
|
|
c.Cluster.Name = "cluster0"
|
|
c.Cluster.ReplicaN = 1
|
|
c.Cluster.LongQueryTime = toml.Duration(-time.Minute) // TODO remove this once cluster.longQueryTime is fully deprecated
|
|
c.Cluster.PartitionToNodeAssignment = PartitionToNodeJmp
|
|
|
|
// AntiEntropy config.
|
|
c.AntiEntropy.Interval = toml.Duration(0)
|
|
|
|
// Metric config.
|
|
c.Metric.Service = "none"
|
|
c.Metric.PollInterval = toml.Duration(0 * time.Minute)
|
|
c.Metric.Diagnostics = false
|
|
|
|
// Tracing config.
|
|
c.Tracing.SamplerType = "off"
|
|
c.Tracing.SamplerParam = 0.001
|
|
|
|
c.Profile.BlockRate = 10000000 // 1 sample per 10 ms
|
|
c.Profile.MutexFraction = 100 // 1% sampling
|
|
|
|
c.Etcd.AClientURL = ""
|
|
c.Etcd.LClientURL = "http://localhost:10301"
|
|
c.Etcd.APeerURL = ""
|
|
c.Etcd.LPeerURL = "http://localhost:10401"
|
|
c.Etcd.Dir = ""
|
|
c.Etcd.Name = ""
|
|
c.Etcd.ClusterName = ""
|
|
c.Etcd.InitCluster = c.Name + "=" + c.Etcd.LPeerURL
|
|
c.Etcd.HeartbeatTTL = 5
|
|
|
|
c.Etcd.TrustedCAFile = ""
|
|
c.Etcd.ClientCertFile = ""
|
|
c.Etcd.ClientKeyFile = ""
|
|
c.Etcd.PeerCertFile = ""
|
|
c.Etcd.PeerKeyFile = ""
|
|
|
|
// Future flags.
|
|
c.Future.Rename = false
|
|
|
|
return c
|
|
}
|
|
|
|
// validateAddrs controls the address fields in the Config object
|
|
// and fills in any blanks.
|
|
// The addresses fields must be guaranteed by the caller to either be
|
|
// completely empty, or have both a host part and a port part
|
|
// separated by a colon. In the latter case either can be empty to
|
|
// indicate it's left unspecified.
|
|
func (c *Config) validateAddrs(ctx context.Context) error {
|
|
// Validate the advertise address.
|
|
advScheme, advHost, advPort, err := validateAdvertiseAddr(ctx, c.Advertise, c.Bind, defaultBindPort)
|
|
if err != nil {
|
|
return errors.Wrapf(err, "validating advertise address")
|
|
}
|
|
c.Advertise = schemeHostPortString(advScheme, advHost, advPort)
|
|
|
|
// Validate the listen address.
|
|
listenScheme, listenHost, listenPort, err := validateListenAddr(ctx, c.Bind, defaultBindPort)
|
|
if err != nil {
|
|
return errors.Wrap(err, "validating listen address")
|
|
}
|
|
c.Bind = schemeHostPortString(listenScheme, listenHost, listenPort)
|
|
|
|
// Validate the gRPC advertise address.
|
|
_, grpcAdvHost, grpcAdvPort, err := validateAdvertiseAddr(ctx, c.AdvertiseGRPC, c.BindGRPC, defaultBindGRPCPort)
|
|
if err != nil {
|
|
return errors.Wrapf(err, "validating grpc advertise address")
|
|
}
|
|
c.AdvertiseGRPC = schemeHostPortString("grpc", grpcAdvHost, grpcAdvPort)
|
|
|
|
// Validate the gRPC listen address.
|
|
_, grpcListenHost, grpcListenPort, err := validateListenAddr(ctx, c.BindGRPC, defaultBindGRPCPort)
|
|
if err != nil {
|
|
return errors.Wrap(err, "validating grpc listen address")
|
|
}
|
|
c.BindGRPC = schemeHostPortString("grpc", grpcListenHost, grpcListenPort)
|
|
|
|
return nil
|
|
}
|
|
|
|
// validateAdvertiseAddr validates and normalizes an address accessible
|
|
// Ensures that if the "host" part is empty, it gets filled in with
|
|
// the configured listen address if any, otherwise it makes a best
|
|
// guess at the outbound IP address.
|
|
// Returns scheme, host, port as strings.
|
|
func validateAdvertiseAddr(ctx context.Context, advAddr, listenAddr, defaultPort string) (string, string, string, error) {
|
|
listenScheme, listenHost, listenPort, err := splitAddr(listenAddr, defaultPort)
|
|
if err != nil {
|
|
return "", "", "", errors.Wrap(err, "getting listen address")
|
|
}
|
|
|
|
advScheme, advHostPort := splitScheme(advAddr)
|
|
advHost, advPort := "", ""
|
|
if advHostPort != "" {
|
|
var err error
|
|
advHost, advPort, err = net.SplitHostPort(advHostPort)
|
|
if err != nil {
|
|
return "", "", "", errors.Wrapf(err, "splitting host port: %s", advHostPort)
|
|
}
|
|
}
|
|
// If no advertise scheme was specified, use the one from
|
|
// the listen address.
|
|
if advScheme == "" {
|
|
advScheme = listenScheme
|
|
}
|
|
// If there was no port number, reuse the one from the listen
|
|
// address.
|
|
if advPort == "" || advPort == "0" {
|
|
advPort = listenPort
|
|
}
|
|
// Resolve non-numeric to numeric.
|
|
portNumber, err := net.DefaultResolver.LookupPort(ctx, "tcp", advPort)
|
|
if err != nil {
|
|
return "", "", "", errors.Wrapf(err, "looking up non-numeric port: %v", advPort)
|
|
}
|
|
advPort = strconv.Itoa(portNumber)
|
|
|
|
// If the advertise host is empty, then we have two cases.
|
|
if advHost == "" {
|
|
if listenHost == "0.0.0.0" {
|
|
advHost = outboundIP().String()
|
|
} else {
|
|
advHost = listenHost
|
|
}
|
|
}
|
|
return advScheme, advHost, advPort, nil
|
|
}
|
|
|
|
// outboundIP gets the preferred outbound ip of this machine.
|
|
func outboundIP() net.IP {
|
|
// This is not actually making a connection to 8.8.8.8.
|
|
// net.Dial() selects the IP address that would be used
|
|
// if an actual connection to 8.8.8.8 were made, so this
|
|
// choice of address is just meant to ensure that an
|
|
// external address is returned (as opposed to a local
|
|
// address like 127.0.0.1).
|
|
conn, err := net.Dial("udp", "8.8.8.8:80")
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
defer conn.Close()
|
|
|
|
localAddr := conn.LocalAddr().(*net.UDPAddr)
|
|
|
|
return localAddr.IP
|
|
}
|
|
|
|
// validateListenAddr validates and normalizes an address suitable for
|
|
// use with net.Listen(). This accepts an empty "host" part to signify
|
|
// the default (localhost) should be used. Rresolves host names to IP
|
|
// addresses.
|
|
// Returns scheme, host, port as strings.
|
|
func validateListenAddr(ctx context.Context, addr, defaultPort string) (string, string, string, error) {
|
|
scheme, host, port, err := splitAddr(addr, defaultPort)
|
|
if err != nil {
|
|
return "", "", "", errors.Wrap(err, "getting listen address")
|
|
}
|
|
rHost, rPort, err := resolveAddr(ctx, host, port)
|
|
if err != nil {
|
|
return "", "", "", errors.Wrap(err, "resolving address")
|
|
}
|
|
return scheme, rHost, rPort, nil
|
|
}
|
|
|
|
// splitScheme returns two strings: the scheme and the hostPort.
|
|
func splitScheme(addr string) (string, string) {
|
|
parts := strings.SplitN(addr, "://", 2)
|
|
if len(parts) == 1 {
|
|
return "", addr
|
|
}
|
|
return parts[0], parts[1]
|
|
}
|
|
|
|
func schemeHostPortString(scheme, host, port string) string {
|
|
var s string
|
|
if scheme != "" {
|
|
s += fmt.Sprintf("%s://", scheme)
|
|
}
|
|
return s + net.JoinHostPort(host, port)
|
|
}
|
|
|
|
// splitAddr returns scheme, host, port as strings.
|
|
func splitAddr(addr string, defaultPort string) (string, string, string, error) {
|
|
scheme, hostPort := splitScheme(addr)
|
|
host, port := "", ""
|
|
if hostPort != "" {
|
|
var err error
|
|
host, port, err = net.SplitHostPort(hostPort)
|
|
if err != nil {
|
|
return "", "", "", errors.Wrapf(err, "splitting host port: %s", hostPort)
|
|
}
|
|
}
|
|
// It's not ideal to have a default here, but the alterative
|
|
// results in a port of 0, which causes Pilosa to listen on
|
|
// a random port.
|
|
if port == "" {
|
|
port = defaultPort
|
|
}
|
|
return scheme, host, port, nil
|
|
}
|
|
|
|
// resolveAddr resolves non-numeric addresses to numeric (IP, port) addresses.
|
|
func resolveAddr(ctx context.Context, host, port string) (string, string, error) {
|
|
resolver := net.DefaultResolver
|
|
|
|
// Resolve the port number. This may translate service names
|
|
// e.g. "postgresql" to a numeric value.
|
|
portNumber, err := resolver.LookupPort(ctx, "tcp", port)
|
|
if err != nil {
|
|
return "", "", errors.Wrapf(err, "resolving up port: %v", port)
|
|
}
|
|
port = strconv.Itoa(portNumber)
|
|
|
|
// Resolve the address.
|
|
if host == "" || host == "localhost" {
|
|
return host, port, nil
|
|
}
|
|
|
|
addr, err := lookupAddr(ctx, resolver, host)
|
|
if err != nil {
|
|
return "", "", errors.Wrap(err, "looking up address")
|
|
}
|
|
return addr, port, nil
|
|
}
|
|
|
|
// lookupAddr resolves the given address/host to an IP address. If
|
|
// multiple addresses are resolved, it returns the first IPv4 address
|
|
// available if there is one, otherwise the first address.
|
|
func lookupAddr(ctx context.Context, resolver *net.Resolver, host string) (string, error) {
|
|
// Resolve the IP address or hostname to an IP address.
|
|
addrs, err := resolver.LookupIPAddr(ctx, host)
|
|
if err != nil {
|
|
return "", errors.Wrap(err, "looking up IP addresses")
|
|
}
|
|
if len(addrs) == 0 {
|
|
return "", fmt.Errorf("cannot resolve %q to an address", host)
|
|
}
|
|
|
|
// LookupIPAddr() can return a mix of IPv6 and IPv4
|
|
// addresses. Return the first IPv4 address if possible.
|
|
for _, addr := range addrs {
|
|
if ip := addr.IP.To4(); ip != nil {
|
|
return ip.String(), nil
|
|
}
|
|
}
|
|
|
|
// No IPv4 address, return the first resolved address instead.
|
|
return addrs[0].String(), nil
|
|
}
|
|
|
|
func (c *Config) ValidateAuth() (errors []error) {
|
|
if !c.Auth.Enable {
|
|
return
|
|
}
|
|
authConfig := []struct {
|
|
name string
|
|
val string
|
|
}{
|
|
{name: "ClientId", val: c.Auth.ClientId},
|
|
{name: "ClientSecret", val: c.Auth.ClientSecret},
|
|
{name: "AuthorizeURL", val: c.Auth.AuthorizeURL},
|
|
{name: "TokenURL", val: c.Auth.TokenURL},
|
|
{name: "GroupEndpointURL", val: c.Auth.GroupEndpointURL},
|
|
{name: "RedirectBaseURL", val: c.Auth.RedirectBaseURL},
|
|
{name: "LogoutURL", val: c.Auth.LogoutURL},
|
|
{name: "SecretKey", val: c.Auth.SecretKey},
|
|
}
|
|
|
|
for _, configOpt := range authConfig {
|
|
name := configOpt.name
|
|
value := configOpt.val
|
|
if value == "" {
|
|
errors = append(errors, fmt.Errorf("empty string for auth config %s", name))
|
|
continue
|
|
}
|
|
|
|
if name == "SecretKey" {
|
|
if len(value) != 64 {
|
|
errors = append(errors, fmt.Errorf("invalid key length for %s. exp %d, got %d", name, 64, len(value)))
|
|
}
|
|
}
|
|
|
|
if strings.Contains(name, "URL") {
|
|
_, err := url.ParseRequestURI(value)
|
|
if err != nil {
|
|
errors = append(errors, fmt.Errorf("invalid URL for auth config %s: %s", name, err))
|
|
continue
|
|
}
|
|
}
|
|
}
|
|
|
|
if len(c.Auth.Scopes) == 0 {
|
|
errors = append(errors, fmt.Errorf("must provide scope for authentication with IdP - for access and refresh token"))
|
|
}
|
|
|
|
if len(c.Auth.ConfiguredIPs) > 0 {
|
|
for _, IP := range c.Auth.ConfiguredIPs {
|
|
if strings.Contains(IP, ":") {
|
|
errors = append(errors, fmt.Errorf("port is not allowed in IP %v for auth.configured-ips", IP))
|
|
continue
|
|
}
|
|
if IP == "" {
|
|
errors = append(errors, fmt.Errorf("empty string for auth.configured-ips"))
|
|
continue
|
|
}
|
|
if strings.Contains(IP, "localhost") {
|
|
errors = append(errors, fmt.Errorf("%v is not a valid IP for auth.configured-ips, DNS names are not allowed", IP))
|
|
continue
|
|
}
|
|
if strings.EqualFold(IP, "0.0.0.0") {
|
|
errors = append(errors, fmt.Errorf("%v is not a valid IP for auth.configured-ips", IP))
|
|
continue
|
|
}
|
|
// validate CIDR addresses
|
|
if strings.Contains(IP, "/") {
|
|
if _, _, err := net.ParseCIDR(IP); err != nil {
|
|
errors = append(errors, fmt.Errorf("%v is not a valid IP for auth.configured-ips: %v", IP, err))
|
|
continue
|
|
}
|
|
continue
|
|
}
|
|
// validate IP
|
|
if net.ParseIP(IP) == nil {
|
|
errors = append(errors, fmt.Errorf("%v is not a valid IP for auth.configured-ips", IP))
|
|
continue
|
|
}
|
|
}
|
|
}
|
|
return errors
|
|
}
|
|
|
|
func (c *Config) ValidatePermissions(permsFile io.Reader) (errors []error) {
|
|
var p authz.GroupPermissions
|
|
if err := p.ReadPermissionsFile(permsFile); err != nil {
|
|
return append(errors, err)
|
|
}
|
|
|
|
if len(p.Permissions) == 0 {
|
|
return append(errors, fmt.Errorf("no group permissions found in permissions file: %s", c.Auth.PermissionsFile))
|
|
}
|
|
|
|
for groupId, indexPerm := range p.Permissions {
|
|
if groupId == "" {
|
|
errors = append(errors, fmt.Errorf("empty string for group id in permissions file %s", c.Auth.PermissionsFile))
|
|
continue
|
|
}
|
|
|
|
for index, perm := range indexPerm {
|
|
if index == "" {
|
|
errors = append(errors, fmt.Errorf("empty string for index for group id %s in permissions file %s ", groupId, c.Auth.PermissionsFile))
|
|
continue
|
|
}
|
|
|
|
if perm == "" {
|
|
errors = append(errors, fmt.Errorf("empty string for permission for group id %s and index %s in permissions file %s", groupId, index, c.Auth.PermissionsFile))
|
|
continue
|
|
}
|
|
|
|
if !((perm == "write") || (perm == "read")) {
|
|
errors = append(errors, fmt.Errorf("not a valid permission %s for group id %s and index %s in permissions file %s; expected permissions are read or write", perm, groupId, index, c.Auth.PermissionsFile))
|
|
continue
|
|
}
|
|
}
|
|
}
|
|
|
|
if p.Admin == "" {
|
|
errors = append(errors, fmt.Errorf("empty string for admin in permissions file: %s", c.Auth.PermissionsFile))
|
|
}
|
|
|
|
return errors
|
|
}
|
|
|
|
func (c *Config) ValidatePermissionsFile() (err error) {
|
|
if c.Auth.PermissionsFile == "" {
|
|
return fmt.Errorf("empty string for auth config permissions file")
|
|
}
|
|
|
|
fileExt := filepath.Ext(c.Auth.PermissionsFile)
|
|
if (fileExt != ".yaml") && (fileExt != ".yml") {
|
|
return fmt.Errorf("invalid file extension for auth config permissions file: %s", c.Auth.PermissionsFile)
|
|
}
|
|
return
|
|
}
|
|
|
|
func (c *Config) MustValidateAuth() {
|
|
errorsAuth := c.ValidateAuth()
|
|
if len(errorsAuth) > 0 {
|
|
for _, e := range errorsAuth {
|
|
log.Println(e)
|
|
}
|
|
}
|
|
|
|
var errorsPerm []error
|
|
errorsPermFile := c.ValidatePermissionsFile()
|
|
if errorsPermFile == nil {
|
|
permsFile, err := os.Open(c.Auth.PermissionsFile)
|
|
if err != nil {
|
|
log.Println(err)
|
|
}
|
|
|
|
defer permsFile.Close()
|
|
|
|
errorsPerm = c.ValidatePermissions(permsFile)
|
|
if len(errorsPerm) > 0 {
|
|
for _, e := range errorsPerm {
|
|
log.Println(e)
|
|
}
|
|
}
|
|
|
|
} else {
|
|
log.Println(errorsPermFile)
|
|
}
|
|
|
|
if len(errorsAuth) > 0 || len(errorsPerm) > 0 || errorsPermFile != nil {
|
|
log.Fatal(fmt.Errorf("there were errors validating authN/authZ config and/or permissions"))
|
|
}
|
|
}
|