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") }