mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-08-28 10:54:59 +00:00
244d80753e
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.
284 lines
8.9 KiB
Go
284 lines
8.9 KiB
Go
package serverless
|
|
|
|
import (
|
|
"context"
|
|
"sync"
|
|
"time"
|
|
|
|
featurebase "github.com/featurebasedb/featurebase/v3"
|
|
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"
|
|
"github.com/pkg/errors"
|
|
)
|
|
|
|
// Ensure type implements interface.
|
|
var _ featurebase.Importer = &importer{}
|
|
|
|
// importer
|
|
type importer struct {
|
|
controller dax.Controller
|
|
|
|
mu sync.Mutex
|
|
qdbid dax.QualifiedDatabaseID
|
|
tbl *dax.Table
|
|
}
|
|
|
|
func NewImporter(controller dax.Controller, qdbid dax.QualifiedDatabaseID, tbl *dax.Table) *importer {
|
|
return &importer{
|
|
controller: controller,
|
|
qdbid: qdbid,
|
|
tbl: tbl,
|
|
}
|
|
}
|
|
|
|
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) {
|
|
return &featurebase.Transaction{
|
|
ID: "not-used",
|
|
}, nil
|
|
}
|
|
|
|
func (m *importer) FinishTransaction(ctx context.Context, id string) (*featurebase.Transaction, error) {
|
|
return nil, nil
|
|
}
|
|
|
|
func (m *importer) CreateTableKeys(ctx context.Context, tid dax.TableID, keys ...string) (map[string]uint64, error) {
|
|
qtbl, err := m.getQtbl(ctx, tid)
|
|
if err != nil {
|
|
return nil, errors.Wrapf(err, "getting qtbl")
|
|
}
|
|
|
|
out := make(map[string]uint64)
|
|
|
|
partitioner := partitioner.NewPartitioner()
|
|
|
|
// Get the partition for each key (map[int][]string).
|
|
partitions := partitioner.PartitionsForKeys(qtbl.Key(), qtbl.PartitionN, keys...)
|
|
|
|
// TODO: we can be more efficient here by calling IngestPartitions() with
|
|
// all the partitions at once, then getting the distinct list of addresses
|
|
// and looping over that instead.
|
|
for partition, ks := range partitions {
|
|
address, err := m.controller.IngestPartition(context.Background(), qtbl.QualifiedID(), partition)
|
|
if err != nil {
|
|
return nil, errors.Wrapf(err, "calling ingest-partition on table: %s, partition: %d", qtbl, partition)
|
|
}
|
|
|
|
fbClient, err := m.fbClient(address)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "getting featurebase client")
|
|
}
|
|
|
|
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)
|
|
}
|
|
|
|
for str, id := range stringToIDMap {
|
|
out[str] = id
|
|
}
|
|
}
|
|
|
|
return out, nil
|
|
}
|
|
|
|
func (m *importer) CreateFieldKeys(ctx context.Context, tid dax.TableID, fname dax.FieldName, keys ...string) (map[string]uint64, error) {
|
|
qtbl, err := m.getQtbl(ctx, tid)
|
|
if err != nil {
|
|
return nil, errors.Wrapf(err, "getting qtbl")
|
|
}
|
|
|
|
// For now, we are going to direct all field key translation to the same
|
|
// node handling index key translation for the table, partition 0.
|
|
// TODO: we should be able to partition field key translation on fieldName.
|
|
// If we do that, we might also consider whether we want to support a
|
|
// different partitionN for field translation.
|
|
partition := dax.PartitionNum(0)
|
|
|
|
address, err := m.controller.IngestPartition(context.Background(), qtbl.QualifiedID(), partition)
|
|
if err != nil {
|
|
return nil, errors.Wrapf(err, "calling ingest-partition on table: %s, partition: %d", qtbl, partition)
|
|
}
|
|
|
|
// Set up a FeatureBase client with address.
|
|
fbClient, err := m.fbClient(address)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "getting featurebase client")
|
|
}
|
|
|
|
cfld, err := fbclient.TableFieldToClientField(qtbl, fname)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "converting fieldinfo to client field")
|
|
}
|
|
|
|
return fbClient.CreateFieldKeys(cfld, keys...)
|
|
}
|
|
|
|
func (m *importer) ImportRoaringBitmap(ctx context.Context, tid dax.TableID, fld *dax.Field, shard uint64, views map[string]*roaring.Bitmap, clear bool) error {
|
|
qtbl, err := m.getQtbl(ctx, tid)
|
|
if err != nil {
|
|
return errors.Wrapf(err, "getting qtbl")
|
|
}
|
|
|
|
address, err := m.controller.IngestShard(context.Background(), qtbl.QualifiedID(), dax.ShardNum(shard))
|
|
if err != nil {
|
|
return errors.Wrap(err, "calling ingest-shard")
|
|
}
|
|
|
|
// Set up a FeatureBase client with address.
|
|
fbClient, err := m.fbClient(address)
|
|
if err != nil {
|
|
return errors.Wrap(err, "getting featurebase client")
|
|
}
|
|
|
|
cfld, err := fbclient.TableFieldToClientField(qtbl, fld.Name)
|
|
if err != nil {
|
|
return errors.Wrap(err, "converting fieldinfo to client field")
|
|
}
|
|
|
|
return fbClient.ImportRoaringBitmap(cfld, shard, views, clear)
|
|
}
|
|
|
|
func (m *importer) ImportRoaringShard(ctx context.Context, tid dax.TableID, shard uint64, request *featurebase.ImportRoaringShardRequest) error {
|
|
qtbl, err := m.getQtbl(ctx, tid)
|
|
if err != nil {
|
|
return errors.Wrapf(err, "getting qtbl")
|
|
}
|
|
|
|
address, err := m.controller.IngestShard(context.Background(), qtbl.QualifiedID(), dax.ShardNum(shard))
|
|
if err != nil {
|
|
return errors.Wrap(err, "calling ingest-shard")
|
|
}
|
|
|
|
// Set up a FeatureBase client with address.
|
|
fbClient, err := m.fbClient(address)
|
|
if err != nil {
|
|
return errors.Wrap(err, "getting featurebase client")
|
|
}
|
|
|
|
return fbClient.ImportRoaringShard(string(qtbl.Key()), shard, request)
|
|
}
|
|
|
|
func (m *importer) EncodeImportValues(ctx context.Context, tid dax.TableID, fld *dax.Field, shard uint64, vals []int64, ids []uint64, clear bool) (path string, data []byte, err error) {
|
|
qtbl, err := m.getQtbl(ctx, tid)
|
|
if err != nil {
|
|
return "", nil, errors.Wrapf(err, "getting qtbl")
|
|
}
|
|
|
|
address, err := m.controller.IngestShard(context.Background(), qtbl.QualifiedID(), dax.ShardNum(shard))
|
|
if err != nil {
|
|
return "", nil, errors.Wrap(err, "calling ingest-shard")
|
|
}
|
|
|
|
// Set up a FeatureBase client with address.
|
|
fbClient, err := m.fbClient(address)
|
|
if err != nil {
|
|
return "", nil, errors.Wrap(err, "getting featurebase client")
|
|
}
|
|
|
|
cfld, err := fbclient.TableFieldToClientField(qtbl, fld.Name)
|
|
if err != nil {
|
|
return "", nil, errors.Wrap(err, "converting fieldinfo to client field")
|
|
}
|
|
|
|
return fbClient.EncodeImportValues(cfld, shard, vals, ids, clear)
|
|
}
|
|
|
|
func (m *importer) EncodeImport(ctx context.Context, tid dax.TableID, fld *dax.Field, shard uint64, vals, ids []uint64, clear bool) (path string, data []byte, err error) {
|
|
qtbl, err := m.getQtbl(ctx, tid)
|
|
if err != nil {
|
|
return "", nil, errors.Wrapf(err, "getting qtbl")
|
|
}
|
|
|
|
address, err := m.controller.IngestShard(context.Background(), qtbl.QualifiedID(), dax.ShardNum(shard))
|
|
if err != nil {
|
|
return "", nil, errors.Wrap(err, "calling ingest-shard")
|
|
}
|
|
|
|
// Set up a FeatureBase client with address.
|
|
fbClient, err := m.fbClient(address)
|
|
if err != nil {
|
|
return "", nil, errors.Wrap(err, "getting featurebase client")
|
|
}
|
|
|
|
cfld, err := fbclient.TableFieldToClientField(qtbl, fld.Name)
|
|
if err != nil {
|
|
return "", nil, errors.Wrap(err, "converting fieldinfo to client field")
|
|
}
|
|
|
|
return fbClient.EncodeImport(cfld, shard, vals, ids, clear)
|
|
}
|
|
|
|
func (m *importer) DoImport(ctx context.Context, tid dax.TableID, fld *dax.Field, shard uint64, path string, data []byte) error {
|
|
qtbl, err := m.getQtbl(ctx, tid)
|
|
if err != nil {
|
|
return errors.Wrapf(err, "getting qtbl")
|
|
}
|
|
|
|
address, err := m.controller.IngestShard(context.Background(), qtbl.QualifiedID(), dax.ShardNum(shard))
|
|
if err != nil {
|
|
return errors.Wrap(err, "calling ingest-shard")
|
|
}
|
|
|
|
// Set up a FeatureBase client with address.
|
|
fbClient, err := m.fbClient(address)
|
|
if err != nil {
|
|
return errors.Wrap(err, "getting featurebase client")
|
|
}
|
|
|
|
return fbClient.DoImport(string(qtbl.Key()), shard, path, data)
|
|
}
|
|
|
|
// getQtbl takes a table (TableKey) and sets the local m.qtbl value. When we
|
|
// originally set up this type, it was only used by IDK, and the table was known
|
|
// at the beginning of the process, so it could be set on this import. But
|
|
// later, we used this importer in the Queryer, and it gets set up prior to
|
|
// parsing the table out of sql; which means we don't know what the table is
|
|
// yet. So this method allows us to use the table which is passed into each
|
|
// method to determine the table. We look it up from mds schema once and save it
|
|
// in m.qtbl for any further method calls.
|
|
func (m *importer) getQtbl(ctx context.Context, tid dax.TableID) (*dax.QualifiedTable, error) {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
|
|
if m.tbl != nil {
|
|
return dax.NewQualifiedTable(m.qdbid, m.tbl), nil
|
|
}
|
|
|
|
qtid := dax.NewQualifiedTableID(m.qdbid, tid)
|
|
|
|
qtbl, err := m.controller.TableByID(ctx, qtid)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "getting table")
|
|
}
|
|
|
|
m.tbl = &qtbl.Table
|
|
|
|
return qtbl, nil
|
|
}
|