featurebase/server/config.go
pokeeffe-molecula 204558b46f added /sql endpoint; implemented SHOW TABLES (#1935)
* squashed 45 commits into one :)

* tlt/sql experiment (#2035)

* Move PlanOperator to sql3/planner/types package

includes:
type PlanOperatorColumn struct
type PlanOperator interface

* Remove planner dependencies from pilosa package

The goal after this is to prevent the planner package (which doesn't exist yet)
from being imported by the pilosa package; we just want it injected into the server
in server/server.go. This is because the planner package uses pilosa types, so we need
to avoid circular dependencies.

Added ExecutionPlannerFn
Make public: pilosa.ExecOptions
Added a pilosa.Executor interface
Added a planner.types.CompilePlanner interface
Isolated the planner calls to:
- Executor.Execute()
- *API.[method]()

* Move executionplanner files into the sql3/planner package.

This required a bit of gymnastics, and there are some things around
FieldOptions which need to be addressed soon.

* Remove the hacky FieldOptions stuff I added earlier

This implementation just uses the pilosa.FieldOption functional options
provided by the API (as opposed to trying to build a FieldOptions
object.

It also changes field types to constants. These are private for now, but
if we need to make them public, we should put them in the planner/types
package.

* Implement the "scale" value from Decimal(scale)

Also, precision and scale were currently reversed in the parser. This
fixes that.

* Modify the parser to handle CACHETYPE <type> SIZE <size>

It's a little odd to me that the cache type values are Tokens, but I
guess it's ok. One thing to keep in mind is that FeatureBase expects
lowercase values, so this commit changes the parser to set the value to
the lowercase version of the type.

* Fix the /sql2 tests

This entailed a combination of commenting out or t.Skip()-ing tests
which covered code in the parser that has been commented out or removed
as not currently supported in sql3.

It also adds some coverage for the sql.Contraint stringers.

* Prevent JSON sql results from containing closing commas

This commit just re-works the existing output code to avoid inserting
closing commas (which results in invalid JSON).

* Enhance the CREATE TABLE test coverage.

In particular, ensure that the fields which get created in FeatureBase
are what we expect based on the fields defined in the CREATE TABLE
statement.

This also ensures that the TIMEQUANTUM and CACHETYPE contraints are not
provided for the same field (since those constraints are not supported
together).

* Adjust the EBNF file to indicate SIZE contraint is optional

A CACHETYPE can be provided without a SIZE. This change indicates that
SIZE is optional.

* Remove `executionplanner_` from file names (#2040)

* implementation of ALTER TABLE (sans column RENAME)

* refactored expression analysis; added more robust type checking; all unary and bin ops function on ints

* added type support for expressions; full bin/unary op support; added cast; more literal support

* cast int to all other types

* all literals (except idset, stringset & timestamp) make it thru; cast to all types with int as source now works

* implemented LIKE/NOT LIKE

* Implemented IS [NOT] NULL

* Move sql2 files into sql3/parser package (#2045)

* Move sql2 files into sql3/parser package

This also removes the sql2 package.

* Fix tests which were typing _id fields as INT intead of ID

* implemented BETWEEN, NOT BETWEEN

* Add featurebase/error package (#2046)

* Add featurebase/error package

I copied the `dax/errors` package which I am starting to use in the DAX
prototype into `featurebase/errors` in order to start using it with the
sql3 package. It's basically a wrapper around `github.com/pkg/errors`,
but it uses a customer coded error.

The sql package can define its own errors based on the
`featurebase/errors` types. Then do things like `Wrap()` and `Is()`.

* Address the linter complaints: shadowed variables, unreachable code

* implemented IN & NOT IN with expression lists

* first cut of CASE

* Fixed some errors from rebase

* updated bnf; removed unused code; tightened up error handling

* first crack at basic CLI for SQL3

Use: `featurebase cli`

Still lots to do here, but for example:

> select count(*) from tremor
+--------------+
|        COUNT |
+--------------+
| 1.158321e+06 |
+--------------+

* Iterate on the CLI (#2057)

Handle the errors.
Add an "exit" command.
Add some general formatting and white space.

Add termination character: ";" (semicolon)

This commit allows a user to provide multiple or partial SQL statements.

Example of multiple statements:
```
show tables; select * from foo;
```

Example of partial (multi-line) statements:
```
select *
from foo;
```

Don't uppercase the header values

* error refactoring; first cut of TOP; remove unused code; use log.Printf instead of fmt.Printf

* fixed a bug with QualifiedRef from refactoring; added bones of INSERT; removal of unused code; tightened up errors more; fixed failing tests

* single value list for INSERT

* Update bnf per discussion with Travis; INSERT now doing the requisite stuff

* Pat's eyes went square - nothing wrong with TOP, Pat needed to learn arrays again.

* improved some errors; fixed tests to suit

* send warnings back in the api; update CLI to display warnings

* start warning on stuff not implemented so we don't get bugged about it

* Tlt/sql experiment (#2063)

* Expresssion -> Expression

* Add SQL planner test

- adds a test to which it is easier to add tables and SQL statments
- un-exports all of the expression types
- removes the planner pointer from the expression types (it can be added
  back later if need be)

* Fix where clause on a string field

Prior to this commit, the binary expression for a where clause on a
string field was building the call by providing a range operator which
is typically used for BSI fields. This changes it to use the call.Args
for string values.

* Update planner tests to handle multiple sql for the same results

* Reorganize SQL tests

Introduce a test/helpers package and move shared MustQueryRows into that
package.

* Add a compatibility map for field types. (#2064)

This is primarily to address the fact that ID fields were previously
incompatible with INT literals.

We should probably consider introducing a custom type for FieldType
which can be used to define compatibilities.

* significantly refactored type checking

* Handle nil (NULL) values in the sql CLI. (#2067)

go-pretty panics if the interface{} field value is nil. This replaces
nil values with a "NULL" string.

* Squash some commits

fixed a still failing test

added line, col to all error messages

refactored source handling to enable table aliases

fixed some copypasta per review

warnings for order by & topn; implemented select as a source

starting to handle in (select...); added stub for optimizer

JSON-encode the sql error and warning strings (#2069)

Error strings with unencoded characters (like double quotes) were
resulting in invalid json.

got insert working; added symbol table; added concrete optimizer; added nascent NestedLoopsOperator; rewrite "where foo in (select..." as inner join

* all about the sets (#2085)

* implemented setcontains()

* implemented set literal; insert set column values; setcontains/all/any both in expr eval and pql filters

* Convert test to use latest framework. (#2086)

* fixed some comments

* removed refactored tests

Co-authored-by: Travis Turner <travis@pilosa.com>

* Add support for Decimal fields to the sql test. (#2090)

* dates (#2094)

* return dates as strings in output; tightened up decimal type checking

* return dates as strings in output; tightened up decimal type checking

* fixed failing tests after decimal changes

* can now insert decimal values

* implemented insert for timestamp data type; implemented current_date, current_timestamp constants

* fixed some failing tests

* handle date literals from strings in insert statements

* changes from feedback

* Fix pointer method error

* sql3 API interface (#2110)

* Introduce API-related interfaces: SchemaAPI, ComputeAPI

The sql3 code was relying on the pointer: *pilosa.API in order to call
API methods directly on the local node. If we want to import and use the
sql3 package in another service (the DAX queryer, for example), we need
to be able to use an implementation of an interface for those API method
calls.

This commit introduces two interfaces, both automatically implemented by
pilosa.API:
- SchemaAPI
- ComputeAPI

* Convert sql3 code to use IndexInfo instead of Index

The sql3 code was relying on a *pilosa.Index and its methods to get
general information like index and field name, type, etc. This commit
converts everything to use a *pilosa.IndexInfo instead.

This allows us to modify the SchemaAPI interface to also return
IndexInfo instead of Index, which will be a lot easier to implement in a
non-pilosa package (like DAX); creating a *pilosa.Index requires
providing things like data directory paths and holders, which are not
necessary for these use cases.

* Unary and Binary Ops R US plus CAST (#2111)

* implemented string literal for timestamp epoch

* fixed failing test

* fixed the failing test again

* refactored tests; implemented unary op tests for all datatypes; implemented binop tests for int/int, int/id, int/decimal & ID/int

* implemented all binary ops for INT & all other types, ID & all other types

* implemented binary ops for DECIMAL types & all other types

* added STRING & BOOL to various tests; implemented all remaining binOp tests

* fix up some stuff after rebasing

* refactored test defs into multiple files; implemented CAST for every datatype

* added tests for like/not like

* addressed review feedback

* addressed type review feedback

* tightened up IS [NOT] NULL behavior plus tests (#2118)

* tightened up IS [NOT] NULL behavior plus tests

* BETWEEN/NOT BETWEEN with all data types

* addressed review feedback

* Handle negative integers in column min/max constraints (#2120)

This commit parses the min/max contraint as an expression, as opposed to
an int literal, so that negative values are treated as Unary
expressions.

There currently isn't support for min/max constraints on `decimal`
fiels, so for now this change only expects +/- integer values.

* Implement the CREATE TABLE keypartitions logic (#2123)

* Execution time, IN/NOT IN & multiple aggregates (#2124)

* added display of execution time

* IN/NOT IN tests for all data types

* fixed date parsing

* removed duplicative tests

* refactoring aggregates

* suport multiple aggregates

* Address review feedback

* final round of feedback

* Add method SchemaAPI.CreateIndexAndFields() (#2127)

In order to support a CREATE TABLE statement as a single command, this
commit alters the SchemaAPI interface to contain a single method which
handles both the index and its fields. It also updates the sql3 code to
use this interface instead of CreateIndex() and CreateField()
indepedently.

* Symbol Handling (Again) (#2129)

* Refactored symbol handling in the planner; re-instated the select as source tests

* removed commented out code

* addressing review feedback

* Move hard-coded _id field out of planner and into interface implementation (#2130)

This commit moves the hard-coded addition of the `_id` field from the
planner to the SchemaAPI.IndexInfo() implementation method.

NOTE: If anything was expecting SchemaAPI.Schema() to also return the
`_id` field as part of its field list in each table, then it would not
be there because the `_id` field is only added in the IndexInfo() method
for now. Currently that's not a problem because nothing is expecting the
`_id` field for `Schema()`.

* Multiple aggregates, all aggregates stand alone and in GROUP BY (#2132)

* handle multiple aggregates in group by queries

* added handling for avg() aggregate both stand alone and in group by

* tightened up sum & avg outside of group by

* added min, max & percentile

* added warnings

* Make MaterializedRowSet implement the PlanOperator interface. (#2133)

This commit refactors the PQLMultiGroupByOperator to have a PlanOperator
as its output. Then, when it initializes, it sets up a
MaterializedRowSet and populates that with the values from the multiple
group by operations.

* added explicit min/max pql operators

* saved a file I forgot to save

* per review

* Un-indent some if/else nesting (#2136)

Co-authored-by: Travis Turner <travis@pilosa.com>

* Add optional `name` argument to test structs.

This commit adds the `name` argument to `tableTest` and `sqlTest` so
that a test can be optionally named. This allows a developer to more
easily run/identify a particular test by name.

* Inbuilt functions (redux) (#2141)

* set functions type parameter type checking

* implemented datepart

* include SQL3 type in SHOW COLUMNS output

* fixed select as source; failing SHOW COLUMNS test

* select in select list

* dump output columns; handle optimization for select list subqueries

* make it an error to return multiple rows for a select list subquery

* added description

* contants and test coverage for datepart function

* SQL3 Refactor-palooza (#2182)

* removed unneeded IsAggregate()

* first cut of working nested loops operator aka INNER JOIN

* remove selectListItemPlanExpression

* added some warnings

* all the tests are passing again!

* addressed some linter complaints

* added basic order by

* bug fixes; added 'or replace'/'replace' to insert

* for insert references should return appropriately

* added back ability to use subquery singleton expressions

* removed dead code; fixed test

* json-able plan, Schema() plus refactoring

* fixed dumb code

* add some tests for time quantum behavior

* Code cleanup during review. Also fixed INSERT to keyed table bug.

This commit contains a lot of minor adjustments made during code review.

It also contains a bug fix that was preventing INSERT into a keyed table
(i.e. _id type STRING) from working.

Co-authored-by: Travis Turner <travis@molecula.com>

* Fix expected min/max on timestamp column test (decimal field)

I don't know why this changed, but presumably something to do with
decimal related work that happened on master.

* Fix compile problem after rebase

* review feedback

Co-authored-by: Matthew Jaffee <jaffee@pilosa.com>
Co-authored-by: Travis Turner <travis@pilosa.com>
Co-authored-by: Travis Turner <travis@molecula.com>
Co-authored-by: Fletcher Haynes <fletcher@capitalprawn.com>
2022-09-30 11:10:24 -07:00

766 lines
24 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"`
// 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"`
// 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.TCPListener
// 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:"-"`
// Limits the total amount of memory to be used by Extract() & SELECT queries.
MaxQueryMemory int64 `toml:"max-query-memory"`
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 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"`
SQL struct {
// EndpointEnabled enables the /sql endpoint.
EndpointEnabled bool `toml:"endpoint-enabled"`
} `toml:"sql"`
// 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"`
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
}
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(),
Storage: storage.NewDefaultConfig(),
RBFConfig: rbfcfg.NewDefaultConfig(),
QueryHistoryLength: 100,
LongQueryTime: toml.Duration(-time.Minute),
}
// 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.SQL.EndpointEnabled = false
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"))
}
}