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.
86 lines
2.3 KiB
Go
86 lines
2.3 KiB
Go
// Package client is an HTTP client for the Queryer.
|
|
package client
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"time"
|
|
|
|
featurebase "github.com/featurebasedb/featurebase/v3"
|
|
"github.com/featurebasedb/featurebase/v3/dax"
|
|
"github.com/featurebasedb/featurebase/v3/errors"
|
|
"github.com/featurebasedb/featurebase/v3/logger"
|
|
)
|
|
|
|
const (
|
|
defaultScheme = "http"
|
|
)
|
|
|
|
// 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
|
|
}
|
|
|
|
// New returns a new instance of Client.
|
|
func New(address dax.Address, logger logger.Logger) *Client {
|
|
return &Client{
|
|
address: address,
|
|
logger: logger,
|
|
client: &http.Client{
|
|
Timeout: time.Second * 30,
|
|
},
|
|
}
|
|
}
|
|
|
|
// Health returns true if the client address returns status OK at its /health
|
|
// endpoint.
|
|
func (c *Client) Health() bool {
|
|
url := fmt.Sprintf("%s/health", c.address.WithScheme(defaultScheme))
|
|
|
|
if resp, err := http.Get(url); err != nil {
|
|
return false
|
|
} else if resp.StatusCode != http.StatusOK {
|
|
return false
|
|
}
|
|
|
|
return true
|
|
}
|
|
|
|
func (c *Client) QuerySQL(ctx context.Context, qdbid dax.QualifiedDatabaseID, sql io.Reader) (*featurebase.WireQueryResponse, error) {
|
|
url := fmt.Sprintf("%s/databases/%s/sql", c.address.WithScheme(defaultScheme), qdbid.DatabaseID)
|
|
if qdbid.DatabaseID == "" {
|
|
url = fmt.Sprintf("%s/sql", c.address.WithScheme(defaultScheme))
|
|
}
|
|
|
|
// Post the request.
|
|
c.logger.Debugf("POST query sql request: url: %s", url)
|
|
req, err := http.NewRequest(http.MethodPost, url, sql)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "creating new post request")
|
|
}
|
|
req.Header.Add("Content-Type", "text/plain")
|
|
req.Header.Add("OrganizationID", string(qdbid.OrganizationID))
|
|
|
|
var resp *http.Response
|
|
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)
|
|
}
|
|
|
|
var wireResp *featurebase.WireQueryResponse
|
|
if err := json.NewDecoder(resp.Body).Decode(&wireResp); err != nil {
|
|
return nil, errors.Wrap(err, "reading response body")
|
|
}
|
|
|
|
return wireResp, nil
|
|
}
|