From 244d80753e2973bae5064d569eb1c59daf3f602e Mon Sep 17 00:00:00 2001 From: Seebs Date: Wed, 1 Mar 2023 17:12:24 -0600 Subject: [PATCH] reuse clients instead of making new clients 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. --- dax/controller/client/client.go | 52 ++++++++++++++++++--------------- dax/queryer/client/client.go | 12 ++++---- dax/queryer/translator.go | 42 +++++++++++++++++--------- dax/test/dax/dax_test.go | 28 +++++++++++------- idk/serverless/importer.go | 45 ++++++++++++++++------------ 5 files changed, 108 insertions(+), 71 deletions(-) diff --git a/dax/controller/client/client.go b/dax/controller/client/client.go index b4e2a27af..d5368e7cc 100644 --- a/dax/controller/client/client.go +++ b/dax/controller/client/client.go @@ -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") } diff --git a/dax/queryer/client/client.go b/dax/queryer/client/client.go index e8fa8be91..35fe1d353 100644 --- a/dax/queryer/client/client.go +++ b/dax/queryer/client/client.go @@ -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) diff --git a/dax/queryer/translator.go b/dax/queryer/translator.go index de771942d..7dce60c02 100644 --- a/dax/queryer/translator.go +++ b/dax/queryer/translator.go @@ -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{ diff --git a/dax/test/dax/dax_test.go b/dax/test/dax/dax_test.go index ff0c3d5f4..b9cfaec05 100644 --- a/dax/test/dax/dax_test.go +++ b/dax/test/dax/dax_test.go @@ -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 diff --git a/idk/serverless/importer.go b/idk/serverless/importer.go index 4e8501de7..3fd947d03 100644 --- a/idk/serverless/importer.go +++ b/idk/serverless/importer.go @@ -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") }