scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-10-07 11:27:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

reuse clients instead of making new clients

Browse source 
Buckle in, this one's a ride.

This is attached to the same PR as a fix for exiting abruptly
during some tests because I ran into that issue, and comprehended
it, while trying to track down weird and sporadic test failures
that were actually this issue.

The actual, underlying, problem: `make test`, by running all the
tests at once, was hitting a bug that was mostly effectively
triggered by running the `dax/test/dax` tests, and the top-level
`featurebase/v3` tests, at the same time. However, the interaction
was nothing as obvious as temporary files, etcd configuration,
or whatever.

We were running out of port numbers.

The tests were using a bit over 30k simultaneous established TCP
connections, each to different ports, because we were creating
new clients for basically every single operation. For instance,
in a single SQL test that did an import and then a read, we
were creating a new client for each field written to, and then
also creating a new client for each field in results that needed
key translation. And none of these clients were closed or
timed out in any way. In fact, Go doesn't really *do* "closing"
of clients; the closest is that an http.Client can be told to
close idle connections that it has been keeping open.

The worst offenders were both named `fbClient`, and were nigh-identical,
except one of them was implemented as a method on `importer` in the
IDK tree, and one was a standalone function.

It may seem surprising that the method on `importer` is using a shared
client pool for all importers, rather than a new pool for each
importer. This is because we potentially make quite a few importers
during tests.

Before this, running either of the dax tests or the top-level
tests would show well over ten thousand simultaneous ESTABLISHED
connections. After this, the dax tests used nearly twenty.

The problem with port consumption like this, while more noticeable
on MacOS, is also something we could hit on the CI runners, especially
if a single runner ended up with more than one test suite running
at the same time. This probably manifests as sporadic very strange
failures of CI, with messages about "cannot assign requested address".
(Note that an outgoing connection to a successfully-created port
requires *another* port to be assigned for the outbound socket.)

This was complicated dramatically by the fact that, for some
utterly cursed reason, it was *especially* common for the point
at which we hit this, in the top-level featurebase tests, to be
running one of the backup tests in TestVariousQueries, and
specifically, to be hitting it on the dataframe part of the
backup... Which is to say, on the *one* path in the backup function
that called log.Fatal, and thus terminated the featurebase process
abruptly without further commentary.
This commit is contained in:
Seebs 2023-03-01 17:12:24 -06:00 • committed by seebs
parent f3abd11884
commit 244d80753e
5 changed files with 108 additions and 71 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

52
dax/controller/client/client.go
View file

@ -8,6 +8,7 @@ import (
"fmt"
"io"
"net/http"
"time"
"github.com/featurebasedb/featurebase/v3/dax"
"github.com/featurebasedb/featurebase/v3/dax/computer"
@ -27,8 +28,9 @@ var _ dax.Schemar = (*Client)(nil)
// Client is an HTTP client that operates on the Controller endpoints exposed by
// the main Controller service.
type Client struct {
address dax.Address
logger logger.Logger
address dax.Address
httpClient *http.Client
logger logger.Logger
}
// New returns a new instance of Client.
@ -36,6 +38,9 @@ func New(address dax.Address, logger logger.Logger) *Client {
return &Client{
address: address,
logger: logger,
httpClient: &http.Client{
Timeout: time.Second * 30,
},
}
}
@ -44,9 +49,10 @@ func New(address dax.Address, logger logger.Logger) *Client {
func (c *Client) Health() bool {
url := fmt.Sprintf("%s/health", c.address.WithScheme(defaultScheme))
if resp, err := http.Get(url); err != nil {
if resp, err := c.httpClient.Get(url); err != nil {
return false
} else if resp.StatusCode != http.StatusOK {
defer resp.Body.Close()
return false
}
@ -64,7 +70,7 @@ func (c *Client) CreateDatabase(ctx context.Context, qdb *dax.QualifiedDatabase)
responseBody := bytes.NewBuffer(postBody)
// Post the request.
resp, err := http.Post(url, "application/json", responseBody)
resp, err := c.httpClient.Post(url, "application/json", responseBody)
if err != nil {
return errors.Wrap(err, "posting create database request")
}
@ -88,7 +94,7 @@ func (c *Client) DropDatabase(ctx context.Context, qdbid dax.QualifiedDatabaseID
responseBody := bytes.NewBuffer(postBody)
// Post the request.
resp, err := http.Post(url, "application/json", responseBody)
resp, err := c.httpClient.Post(url, "application/json", responseBody)
if err != nil {
return errors.Wrap(err, "posting drop database request")
}
@ -113,7 +119,7 @@ func (c *Client) DatabaseByID(ctx context.Context, qdbid dax.QualifiedDatabaseID
// Post the request.
c.logger.Debugf("POST database-by-id request: url: %s", url)
resp, err := http.Post(url, "application/json", responseBody)
resp, err := c.httpClient.Post(url, "application/json", responseBody)
if err != nil {
return nil, errors.Wrap(err, "posting database-by-id request")
}
@ -148,7 +154,7 @@ func (c *Client) DatabaseByName(ctx context.Context, orgID dax.OrganizationID, n
// Post the request.
c.logger.Debugf("POST database-by-name request: url: %s", url)
resp, err := http.Post(url, "application/json", responseBody)
resp, err := c.httpClient.Post(url, "application/json", responseBody)
if err != nil {
return nil, errors.Wrap(err, "posting database-by-name request")
}
@ -183,7 +189,7 @@ func (c *Client) Databases(ctx context.Context, orgID dax.OrganizationID, ids ..
// Post the request.
c.logger.Debugf("POST databases request: url: %s", url)
resp, err := http.Post(url, "application/json", responseBody)
resp, err := c.httpClient.Post(url, "application/json", responseBody)
if err != nil {
return nil, errors.Wrap(err, "posting databases request")
}
@ -225,7 +231,7 @@ func (c *Client) SetDatabaseOption(ctx context.Context, qdbid dax.QualifiedDatab
// Post the request as PATCH.
c.logger.Debugf("PATCH database/option request: url: %s", url)
resp, err := http.DefaultClient.Do(request)
resp, err := c.httpClient.Do(request)
if err != nil {
return errors.Wrap(err, "posting database/option request")
}
@ -264,7 +270,7 @@ func (c *Client) Table(ctx context.Context, qtid dax.QualifiedTableID) (*dax.Qua
// Post the request.
c.logger.Debugf("POST table request: url: %s", url)
resp, err := http.Post(url, "application/json", responseBody)
resp, err := c.httpClient.Post(url, "application/json", responseBody)
if err != nil {
return nil, errors.Wrap(err, "posting table request")
}
@ -300,7 +306,7 @@ func (c *Client) TableID(ctx context.Context, qdbid dax.QualifiedDatabaseID, nam
requestBody := bytes.NewBuffer(postBody)
// Post the request.
resp, err := http.Post(url, "application/json", requestBody)
resp, err := c.httpClient.Post(url, "application/json", requestBody)
if err != nil {
return dflt, errors.Wrap(err, "posting table-id request")
}
@ -335,7 +341,7 @@ func (c *Client) Tables(ctx context.Context, qdbid dax.QualifiedDatabaseID, ids
responseBody := bytes.NewBuffer(postBody)
// Post the request.
resp, err := http.Post(url, "application/json", responseBody)
resp, err := c.httpClient.Post(url, "application/json", responseBody)
if err != nil {
return nil, errors.Wrap(err, "posting tables request")
}
@ -364,7 +370,7 @@ func (c *Client) CreateTable(ctx context.Context, qtbl *dax.QualifiedTable) erro
responseBody := bytes.NewBuffer(postBody)
// Post the request.
resp, err := http.Post(url, "application/json", responseBody)
resp, err := c.httpClient.Post(url, "application/json", responseBody)
if err != nil {
return errors.Wrap(err, "posting create table request")
}
@ -388,7 +394,7 @@ func (c *Client) DropTable(ctx context.Context, qtid dax.QualifiedTableID) error
responseBody := bytes.NewBuffer(postBody)
// Post the request.
resp, err := http.Post(url, "application/json", responseBody)
resp, err := c.httpClient.Post(url, "application/json", responseBody)
if err != nil {
return errors.Wrap(err, "posting drop table request")
}
@ -416,7 +422,7 @@ func (c *Client) CreateField(ctx context.Context, qtid dax.QualifiedTableID, fld
responseBody := bytes.NewBuffer(postBody)
// Post the request.
resp, err := http.Post(url, "application/json", responseBody)
resp, err := c.httpClient.Post(url, "application/json", responseBody)
if err != nil {
return errors.Wrap(err, "posting create field request")
}
@ -445,7 +451,7 @@ func (c *Client) DropField(ctx context.Context, qtid dax.QualifiedTableID, fldNa
responseBody := bytes.NewBuffer(postBody)
// Post the request.
resp, err := http.Post(url, "application/json", responseBody)
resp, err := c.httpClient.Post(url, "application/json", responseBody)
if err != nil {
return errors.Wrap(err, "posting drop field request")
}
@ -476,7 +482,7 @@ func (c *Client) IngestShard(ctx context.Context, qtid dax.QualifiedTableID, sha
responseBody := bytes.NewBuffer(postBody)
// Post the request.
resp, err := http.Post(url, "application/json", responseBody)
resp, err := c.httpClient.Post(url, "application/json", responseBody)
if err != nil {
return host, errors.Wrap(err, "posting ingest-shard request")
}
@ -513,7 +519,7 @@ func (c *Client) IngestPartition(ctx context.Context, qtid dax.QualifiedTableID,
responseBody := bytes.NewBuffer(postBody)
// Post the request.
resp, err := http.Post(url, "application/json", responseBody)
resp, err := c.httpClient.Post(url, "application/json", responseBody)
if err != nil {
return host, errors.Wrap(err, "posting ingest-partition request")
}
@ -551,7 +557,7 @@ func (c *Client) ComputeNodes(ctx context.Context, qtid dax.QualifiedTableID, sh
responseBody := bytes.NewBuffer(postBody)
// Post the request.
resp, err := http.Post(url, "application/json", responseBody)
resp, err := c.httpClient.Post(url, "application/json", responseBody)
if err != nil {
return nodes, errors.Wrap(err, "posting compute-nodes request")
}
@ -589,7 +595,7 @@ func (c *Client) TranslateNodes(ctx context.Context, qtid dax.QualifiedTableID,
responseBody := bytes.NewBuffer(postBody)
// Post the request.
resp, err := http.Post(url, "application/json", responseBody)
resp, err := c.httpClient.Post(url, "application/json", responseBody)
if err != nil {
return nodes, errors.Wrap(err, "posting translate-nodes request")
}
@ -625,7 +631,7 @@ func (c *Client) RegisterNode(ctx context.Context, node *dax.Node) error {
responseBody := bytes.NewBuffer(postBody)
// Post the request.
resp, err := http.Post(url, "application/json", responseBody)
resp, err := c.httpClient.Post(url, "application/json", responseBody)
if err != nil {
return errors.Wrap(err, "posting translate-nodes request")
}
@ -655,7 +661,7 @@ func (c *Client) CheckInNode(ctx context.Context, node *dax.Node) error {
responseBody := bytes.NewBuffer(postBody)
// Post the request.
resp, err := http.Post(url, "application/json", responseBody)
resp, err := c.httpClient.Post(url, "application/json", responseBody)
if err != nil {
return errors.Wrap(err, "posting translate-nodes request")
}
@ -680,7 +686,7 @@ func (c *Client) SnapshotTable(ctx context.Context, qtid dax.QualifiedTableID) e
responseBody := bytes.NewBuffer(postBody)
// Post the request.
resp, err := http.Post(url, "application/json", responseBody)
resp, err := c.httpClient.Post(url, "application/json", responseBody)
if err != nil {
return errors.Wrap(err, "posting translate-nodes request")
}

12
dax/queryer/client/client.go
View file

@ -22,6 +22,7 @@ const (
// Client is an HTTP client that operates on the Controller endpoints exposed by
// the main Controller service.
type Client struct {
client *http.Client
address dax.Address
logger logger.Logger
}
@ -31,6 +32,9 @@ func New(address dax.Address, logger logger.Logger) *Client {
return &Client{
address: address,
logger: logger,
client: &http.Client{
Timeout: time.Second * 30,
},
}
}
@ -54,10 +58,6 @@ func (c *Client) QuerySQL(ctx context.Context, qdbid dax.QualifiedDatabaseID, sq
url = fmt.Sprintf("%s/sql", c.address.WithScheme(defaultScheme))
}
client := &http.Client{
Timeout: time.Second * 30,
}
// Post the request.
c.logger.Debugf("POST query sql request: url: %s", url)
req, err := http.NewRequest(http.MethodPost, url, sql)
@ -68,10 +68,10 @@ func (c *Client) QuerySQL(ctx context.Context, qdbid dax.QualifiedDatabaseID, sq
req.Header.Add("OrganizationID", string(qdbid.OrganizationID))
var resp *http.Response
if resp, err = client.Do(req); err != nil {
if resp, err = c.client.Do(req); err != nil {
return nil, errors.Wrap(err, "executing post request")
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
b, _ := io.ReadAll(resp.Body)
return nil, errors.Errorf("status code: %d: %s", resp.StatusCode, b)

42
dax/queryer/translator.go
View file

@ -3,9 +3,10 @@ package queryer
import (
"context"
"net/http"
"sync"
pilosa "github.com/featurebasedb/featurebase/v3"
featurebase_client "github.com/featurebasedb/featurebase/v3/client"
fbclient "github.com/featurebasedb/featurebase/v3/client"
"github.com/featurebasedb/featurebase/v3/dax"
"github.com/featurebasedb/featurebase/v3/dax/controller/partitioner"
"github.com/featurebasedb/featurebase/v3/disco"
@ -26,15 +27,28 @@ func NewServerlessTranslator(controller dax.Controller) *serverlessTranslator {
}
}
func fbClient(address dax.Address) (*featurebase_client.Client, error) {
// Set up a FeatureBase client with address.
return featurebase_client.NewClient(address.HostPort(),
featurebase_client.OptClientRetries(2),
featurebase_client.OptClientTotalPoolSize(1000),
featurebase_client.OptClientPoolSizePerRoute(400),
featurebase_client.OptClientPathPrefix(address.Path()),
//featurebase_client.OptClientStatsClient(m.stats),
var fbClientCache = map[dax.Address]*fbclient.Client{}
var fbClientCacheMu sync.Mutex
func fbClient(address dax.Address) (*fbclient.Client, error) {
fbClientCacheMu.Lock()
defer fbClientCacheMu.Unlock()
client := fbClientCache[address]
if client != nil {
return client, nil
}
client, err := fbclient.NewClient(address.HostPort(),
fbclient.OptClientRetries(2),
fbclient.OptClientTotalPoolSize(1000),
fbclient.OptClientPoolSizePerRoute(400),
fbclient.OptClientPathPrefix(address.Path()),
//fbclient.OptClientStatsClient(m.stats),
)
if err != nil {
return nil, err
}
fbClientCache[address] = client
return client, nil
}
func (m *serverlessTranslator) CreateIndexKeys(ctx context.Context, table string, keys []string) (map[string]uint64, error) {
@ -63,7 +77,7 @@ func (m *serverlessTranslator) CreateIndexKeys(ctx context.Context, table string
return nil, errors.Wrap(err, "getting featurebase client")
}
idx := featurebase_client.NewIndex(table)
idx := fbclient.NewIndex(table)
m, err := fbClient.CreateIndexKeys(idx, pMap[pNum]...)
if err != nil {
@ -90,7 +104,7 @@ func (m *serverlessTranslator) CreateFieldKeys(ctx context.Context, table string
return nil, errors.Wrap(err, "getting featurebase client")
}
idx := featurebase_client.NewIndex(table)
idx := fbclient.NewIndex(table)
fld := idx.Field(field)
return fbClient.CreateFieldKeys(fld, keys...)
@ -129,7 +143,7 @@ func (m *serverlessTranslator) FindIndexKeys(ctx context.Context, table string,
return nil, errors.Wrap(err, "getting featurebase client")
}
idx := featurebase_client.NewIndex(table)
idx := fbclient.NewIndex(table)
nodeKeys := []string{}
for _, pNum := range tnode.Partitions {
@ -161,7 +175,7 @@ func (m *serverlessTranslator) FindFieldKeys(ctx context.Context, table, field s
return nil, errors.Wrap(err, "getting featurebase client")
}
idx := featurebase_client.NewIndex(table)
idx := fbclient.NewIndex(table)
fld := idx.Field(field)
return fbClient.FindFieldKeys(fld, keys...)
@ -263,7 +277,7 @@ func (m *serverlessTranslator) TranslateFieldListIDs(ctx context.Context, index,
return makeTranslateIDsRequest(fbClient, index, field, ids)
}
func makeTranslateIDsRequest(fbClient *featurebase_client.Client, table, field string, ids []uint64) ([]string, error) {
func makeTranslateIDsRequest(fbClient *fbclient.Client, table, field string, ids []uint64) ([]string, error) {
method := "POST"
path := "/internal/translate/ids"
headers := map[string]string{

28
dax/test/dax/dax_test.go
View file

@ -8,6 +8,7 @@ import (
"os"
"sort"
"strings"
"sync"
"testing"
"time"
@ -911,13 +912,10 @@ func runTableTests(t *testing.T, queryerAddr dax.Address, cfgs ...tableTestConfi
if cfg.skipQuery {
return
}
for j, sqltest := range cfg.test.SQLTests {
t.Run(sqltest.Name(j), func(t *testing.T) {
for _, sql := range sqltest.SQLs {
t.Run(fmt.Sprintf("sql-%s", sql), func(t *testing.T) {
log.Printf("SQL: %s", sql)
var expRows [][]interface{}
if cfg.querySet == 0 {
expRows = sqltest.ExpRows
@ -928,7 +926,6 @@ func runTableTests(t *testing.T, queryerAddr dax.Address, cfgs ...tableTestConfi
}
expRows = sqltest.ExpRowsPlus1[cfg.querySet-1]
}
resp := runSQL(t, queryerAddr, cfg.qdbid, sql)
headers := resp.Schema.Fields
rows := resp.Data
@ -1116,12 +1113,25 @@ func (c *wireResponseComparer) Equal() bool {
return assert.Equal(c.tb, c.exp, c.got)
}
var sharedClients = map[dax.Address]*queryerclient.Client{}
var clientMu sync.Mutex
func getClient(addr dax.Address) *queryerclient.Client {
clientMu.Lock()
defer clientMu.Unlock()
client := sharedClients[addr]
if client != nil {
return client
}
client = queryerclient.New(addr, logger.StderrLogger)
sharedClients[addr] = client
return client
}
func runSQL(tb testing.TB, queryerAddr dax.Address, qdbid dax.QualifiedDatabaseID, sql string) *featurebase.WireQueryResponse {
tb.Helper()
client := queryerclient.New(queryerAddr, logger.StderrLogger)
resp, err := client.QuerySQL(context.Background(), qdbid, strings.NewReader(sql))
resp, err := getClient(queryerAddr).QuerySQL(context.Background(), qdbid, strings.NewReader(sql))
assert.NoError(tb, err)
return resp
@ -1130,10 +1140,8 @@ func runSQL(tb testing.TB, queryerAddr dax.Address, qdbid dax.QualifiedDatabaseI
func runPQL(tb testing.TB, queryerAddr dax.Address, qdbid dax.QualifiedDatabaseID, table string, pql string) *featurebase.WireQueryResponse {
tb.Helper()
client := queryerclient.New(queryerAddr, logger.StderrLogger)
sqlPQL := fmt.Sprintf("[%s]%s", table, pql)
resp, err := client.QuerySQL(context.Background(), qdbid, strings.NewReader(sqlPQL))
resp, err := getClient(queryerAddr).QuerySQL(context.Background(), qdbid, strings.NewReader(sqlPQL))
assert.NoError(tb, err)
return resp

45
idk/serverless/importer.go
View file

@ -6,7 +6,7 @@ import (
"time"
featurebase "github.com/featurebasedb/featurebase/v3"
featurebaseclient "github.com/featurebasedb/featurebase/v3/client"
fbclient "github.com/featurebasedb/featurebase/v3/client"
"github.com/featurebasedb/featurebase/v3/dax"
"github.com/featurebasedb/featurebase/v3/dax/controller/partitioner"
"github.com/featurebasedb/featurebase/v3/roaring"
@ -33,19 +33,28 @@ func NewImporter(controller dax.Controller, qdbid dax.QualifiedDatabaseID, tbl *
}
}
// fbClient currently returns a new FeatureBase client (for the address) for
// every call to this method. We could cache these connections in a map (keyed
// on address) to avoid creating a new client for an address that we already
// have a client for.
func (m *importer) fbClient(address dax.Address) (*featurebaseclient.Client, error) {
// Set up a FeatureBase client with address.
return featurebaseclient.NewClient(address.HostPort(),
featurebaseclient.OptClientRetries(2),
featurebaseclient.OptClientTotalPoolSize(1000),
featurebaseclient.OptClientPoolSizePerRoute(400),
featurebaseclient.OptClientPathPrefix(address.Path()),
//featurebaseclient.OptClientStatsClient(m.stats),
var fbClientCache = map[dax.Address]*fbclient.Client{}
var fbClientCacheMu sync.Mutex
func (m *importer) fbClient(address dax.Address) (*fbclient.Client, error) {
fbClientCacheMu.Lock()
defer fbClientCacheMu.Unlock()
client := fbClientCache[address]
if client != nil {
return client, nil
}
client, err := fbclient.NewClient(address.HostPort(),
fbclient.OptClientRetries(2),
fbclient.OptClientTotalPoolSize(1000),
fbclient.OptClientPoolSizePerRoute(400),
fbclient.OptClientPathPrefix(address.Path()),
//fbclient.OptClientStatsClient(m.stats),
)
if err != nil {
return nil, err
}
fbClientCache[address] = client
return client, nil
}
func (m *importer) StartTransaction(ctx context.Context, id string, timeout time.Duration, exclusive bool, requestTimeout time.Duration) (*featurebase.Transaction, error) {
@ -85,7 +94,7 @@ func (m *importer) CreateTableKeys(ctx context.Context, tid dax.TableID, keys ..
return nil, errors.Wrap(err, "getting featurebase client")
}
cidx := featurebaseclient.QTableToClientIndex(qtbl)
cidx := fbclient.QTableToClientIndex(qtbl)
stringToIDMap, err := fbClient.CreateIndexKeys(cidx, ks...)
if err != nil {
return nil, errors.Wrapf(err, "creating index keys for partition: %d", partition)
@ -123,7 +132,7 @@ func (m *importer) CreateFieldKeys(ctx context.Context, tid dax.TableID, fname d
return nil, errors.Wrap(err, "getting featurebase client")
}
cfld, err := featurebaseclient.TableFieldToClientField(qtbl, fname)
cfld, err := fbclient.TableFieldToClientField(qtbl, fname)
if err != nil {
return nil, errors.Wrap(err, "converting fieldinfo to client field")
}
@ -148,7 +157,7 @@ func (m *importer) ImportRoaringBitmap(ctx context.Context, tid dax.TableID, fld
return errors.Wrap(err, "getting featurebase client")
}
cfld, err := featurebaseclient.TableFieldToClientField(qtbl, fld.Name)
cfld, err := fbclient.TableFieldToClientField(qtbl, fld.Name)
if err != nil {
return errors.Wrap(err, "converting fieldinfo to client field")
}
@ -193,7 +202,7 @@ func (m *importer) EncodeImportValues(ctx context.Context, tid dax.TableID, fld
return "", nil, errors.Wrap(err, "getting featurebase client")
}
cfld, err := featurebaseclient.TableFieldToClientField(qtbl, fld.Name)
cfld, err := fbclient.TableFieldToClientField(qtbl, fld.Name)
if err != nil {
return "", nil, errors.Wrap(err, "converting fieldinfo to client field")
}
@ -218,7 +227,7 @@ func (m *importer) EncodeImport(ctx context.Context, tid dax.TableID, fld *dax.F
return "", nil, errors.Wrap(err, "getting featurebase client")
}
cfld, err := featurebaseclient.TableFieldToClientField(qtbl, fld.Name)
cfld, err := fbclient.TableFieldToClientField(qtbl, fld.Name)
if err != nil {
return "", nil, errors.Wrap(err, "converting fieldinfo to client field")
}