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.
701 lines
20 KiB
Go
701 lines
20 KiB
Go
// Package client is an HTTP client for Controller.
|
|
package client
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"time"
|
|
|
|
"github.com/featurebasedb/featurebase/v3/dax"
|
|
"github.com/featurebasedb/featurebase/v3/dax/computer"
|
|
controllerhttp "github.com/featurebasedb/featurebase/v3/dax/controller/http"
|
|
"github.com/featurebasedb/featurebase/v3/errors"
|
|
"github.com/featurebasedb/featurebase/v3/logger"
|
|
)
|
|
|
|
const (
|
|
defaultScheme = "http"
|
|
)
|
|
|
|
// Ensure type implements interface.
|
|
var _ computer.Registrar = (*Client)(nil)
|
|
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
|
|
httpClient *http.Client
|
|
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,
|
|
httpClient: &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 := c.httpClient.Get(url); err != nil {
|
|
return false
|
|
} else if resp.StatusCode != http.StatusOK {
|
|
defer resp.Body.Close()
|
|
return false
|
|
}
|
|
|
|
return true
|
|
}
|
|
|
|
func (c *Client) CreateDatabase(ctx context.Context, qdb *dax.QualifiedDatabase) error {
|
|
url := fmt.Sprintf("%s/create-database", c.address.WithScheme(defaultScheme))
|
|
|
|
// Encode the request.
|
|
postBody, err := json.Marshal(qdb)
|
|
if err != nil {
|
|
return errors.Wrap(err, "marshalling post request")
|
|
}
|
|
responseBody := bytes.NewBuffer(postBody)
|
|
|
|
// Post the request.
|
|
resp, err := c.httpClient.Post(url, "application/json", responseBody)
|
|
if err != nil {
|
|
return errors.Wrap(err, "posting create database request")
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
return errors.Wrapf(errors.UnmarshalJSON(resp.Body), "status code: %d", resp.StatusCode)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (c *Client) DropDatabase(ctx context.Context, qdbid dax.QualifiedDatabaseID) error {
|
|
url := fmt.Sprintf("%s/drop-database", c.address.WithScheme(defaultScheme))
|
|
|
|
// Encode the request.
|
|
postBody, err := json.Marshal(qdbid)
|
|
if err != nil {
|
|
return errors.Wrap(err, "marshalling post request")
|
|
}
|
|
responseBody := bytes.NewBuffer(postBody)
|
|
|
|
// Post the request.
|
|
resp, err := c.httpClient.Post(url, "application/json", responseBody)
|
|
if err != nil {
|
|
return errors.Wrap(err, "posting drop database request")
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
return errors.Wrapf(errors.UnmarshalJSON(resp.Body), "status code: %d", resp.StatusCode)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (c *Client) DatabaseByID(ctx context.Context, qdbid dax.QualifiedDatabaseID) (*dax.QualifiedDatabase, error) {
|
|
url := fmt.Sprintf("%s/database-by-id", c.address.WithScheme(defaultScheme))
|
|
|
|
// Encode the request.
|
|
postBody, err := json.Marshal(qdbid)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "marshalling post request")
|
|
}
|
|
responseBody := bytes.NewBuffer(postBody)
|
|
|
|
// Post the request.
|
|
c.logger.Debugf("POST database-by-id request: url: %s", url)
|
|
resp, err := c.httpClient.Post(url, "application/json", responseBody)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "posting database-by-id request")
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
return nil, errors.Wrapf(errors.UnmarshalJSON(resp.Body), "status code: %d", resp.StatusCode)
|
|
}
|
|
|
|
var qdb *dax.QualifiedDatabase
|
|
if err := json.NewDecoder(resp.Body).Decode(&qdb); err != nil {
|
|
return nil, errors.Wrap(err, "reading response body")
|
|
}
|
|
|
|
return qdb, nil
|
|
}
|
|
|
|
func (c *Client) DatabaseByName(ctx context.Context, orgID dax.OrganizationID, name dax.DatabaseName) (*dax.QualifiedDatabase, error) {
|
|
url := fmt.Sprintf("%s/database-by-name", c.address.WithScheme(defaultScheme))
|
|
|
|
req := &controllerhttp.DatabaseByNameRequest{
|
|
OrganizationID: orgID,
|
|
Name: name,
|
|
}
|
|
|
|
// Encode the request.
|
|
postBody, err := json.Marshal(req)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "marshalling post request")
|
|
}
|
|
responseBody := bytes.NewBuffer(postBody)
|
|
|
|
// Post the request.
|
|
c.logger.Debugf("POST database-by-name request: url: %s", url)
|
|
resp, err := c.httpClient.Post(url, "application/json", responseBody)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "posting database-by-name request")
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
return nil, errors.Wrapf(errors.UnmarshalJSON(resp.Body), "status code: %d", resp.StatusCode)
|
|
}
|
|
|
|
var qdb *dax.QualifiedDatabase
|
|
if err := json.NewDecoder(resp.Body).Decode(&qdb); err != nil {
|
|
return nil, errors.Wrap(err, "reading response body")
|
|
}
|
|
|
|
return qdb, nil
|
|
}
|
|
|
|
func (c *Client) Databases(ctx context.Context, orgID dax.OrganizationID, ids ...dax.DatabaseID) ([]*dax.QualifiedDatabase, error) {
|
|
url := fmt.Sprintf("%s/databases", c.address.WithScheme(defaultScheme))
|
|
|
|
req := &controllerhttp.DatabasesRequest{
|
|
OrganizationID: orgID,
|
|
DatabaseIDs: ids,
|
|
}
|
|
|
|
// Encode the request.
|
|
postBody, err := json.Marshal(req)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "marshalling post request")
|
|
}
|
|
responseBody := bytes.NewBuffer(postBody)
|
|
|
|
// Post the request.
|
|
c.logger.Debugf("POST databases request: url: %s", url)
|
|
resp, err := c.httpClient.Post(url, "application/json", responseBody)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "posting databases request")
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
return nil, errors.Wrapf(errors.UnmarshalJSON(resp.Body), "status code: %d", resp.StatusCode)
|
|
}
|
|
|
|
var qdbs []*dax.QualifiedDatabase
|
|
if err := json.NewDecoder(resp.Body).Decode(&qdbs); err != nil {
|
|
return nil, errors.Wrap(err, "reading response body")
|
|
}
|
|
|
|
return qdbs, nil
|
|
}
|
|
|
|
func (c *Client) SetDatabaseOption(ctx context.Context, qdbid dax.QualifiedDatabaseID, option string, value string) error {
|
|
url := fmt.Sprintf("%s/database/options", c.address.WithScheme(defaultScheme))
|
|
|
|
req := &controllerhttp.DatabaseOptionRequest{
|
|
QualifiedDatabaseID: qdbid,
|
|
Option: option,
|
|
Value: value,
|
|
}
|
|
|
|
// Encode the request.
|
|
postBody, err := json.Marshal(req)
|
|
if err != nil {
|
|
return errors.Wrap(err, "marshalling post request")
|
|
}
|
|
responseBody := bytes.NewBuffer(postBody)
|
|
|
|
request, err := http.NewRequest(http.MethodPatch, url, responseBody)
|
|
if err != nil {
|
|
return errors.Wrap(err, "creating http request")
|
|
}
|
|
request.Header.Set("Content-Type", "application/json")
|
|
|
|
// Post the request as PATCH.
|
|
c.logger.Debugf("PATCH database/option request: url: %s", url)
|
|
resp, err := c.httpClient.Do(request)
|
|
if err != nil {
|
|
return errors.Wrap(err, "posting database/option request")
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
return errors.Wrapf(errors.UnmarshalJSON(resp.Body), "status code: %d", resp.StatusCode)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// TODO(tlt): collapse Table into this
|
|
func (c *Client) TableByID(ctx context.Context, qtid dax.QualifiedTableID) (*dax.QualifiedTable, error) {
|
|
return c.Table(ctx, qtid)
|
|
}
|
|
|
|
// TODO(tlt): collapse TableID into this
|
|
func (c *Client) TableByName(ctx context.Context, qdbid dax.QualifiedDatabaseID, tname dax.TableName) (*dax.QualifiedTable, error) {
|
|
qtid, err := c.TableID(ctx, qdbid, tname)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "getting table id")
|
|
}
|
|
return c.Table(ctx, qtid)
|
|
}
|
|
|
|
func (c *Client) Table(ctx context.Context, qtid dax.QualifiedTableID) (*dax.QualifiedTable, error) {
|
|
url := fmt.Sprintf("%s/table", c.address.WithScheme(defaultScheme))
|
|
|
|
// Encode the request.
|
|
postBody, err := json.Marshal(qtid)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "marshalling post request")
|
|
}
|
|
responseBody := bytes.NewBuffer(postBody)
|
|
|
|
// Post the request.
|
|
c.logger.Debugf("POST table request: url: %s", url)
|
|
resp, err := c.httpClient.Post(url, "application/json", responseBody)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "posting table request")
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
return nil, errors.Wrapf(errors.UnmarshalJSON(resp.Body), "status code: %d", resp.StatusCode)
|
|
}
|
|
|
|
var qtable *dax.QualifiedTable
|
|
if err := json.NewDecoder(resp.Body).Decode(&qtable); err != nil {
|
|
return nil, errors.Wrap(err, "reading response body")
|
|
}
|
|
|
|
return qtable, nil
|
|
}
|
|
|
|
func (c *Client) TableID(ctx context.Context, qdbid dax.QualifiedDatabaseID, name dax.TableName) (dax.QualifiedTableID, error) {
|
|
url := fmt.Sprintf("%s/table-id", c.address.WithScheme(defaultScheme))
|
|
|
|
dflt := dax.QualifiedTableID{}
|
|
|
|
req := dax.QualifiedTableID{
|
|
QualifiedDatabaseID: qdbid,
|
|
Name: name,
|
|
}
|
|
|
|
// Encode the request.
|
|
postBody, err := json.Marshal(req)
|
|
if err != nil {
|
|
return dflt, errors.Wrap(err, "marshalling post request")
|
|
}
|
|
requestBody := bytes.NewBuffer(postBody)
|
|
|
|
// Post the request.
|
|
resp, err := c.httpClient.Post(url, "application/json", requestBody)
|
|
if err != nil {
|
|
return dflt, errors.Wrap(err, "posting table-id request")
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
return dflt, errors.Wrapf(errors.UnmarshalJSON(resp.Body), "status code: %d", resp.StatusCode)
|
|
}
|
|
|
|
var qtid dax.QualifiedTableID
|
|
if err := json.NewDecoder(resp.Body).Decode(&qtid); err != nil {
|
|
return dflt, errors.Wrap(err, "reading response body")
|
|
}
|
|
|
|
return qtid, nil
|
|
}
|
|
|
|
func (c *Client) Tables(ctx context.Context, qdbid dax.QualifiedDatabaseID, ids ...dax.TableID) ([]*dax.QualifiedTable, error) {
|
|
url := fmt.Sprintf("%s/tables", c.address.WithScheme(defaultScheme))
|
|
|
|
req := controllerhttp.TablesRequest{
|
|
OrganizationID: qdbid.OrganizationID,
|
|
DatabaseID: qdbid.DatabaseID,
|
|
TableIDs: ids,
|
|
}
|
|
|
|
// Encode the request.
|
|
postBody, err := json.Marshal(req)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "marshalling post request")
|
|
}
|
|
responseBody := bytes.NewBuffer(postBody)
|
|
|
|
// Post the request.
|
|
resp, err := c.httpClient.Post(url, "application/json", responseBody)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "posting tables request")
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
return nil, errors.Wrapf(errors.UnmarshalJSON(resp.Body), "Status code: %d", resp.StatusCode)
|
|
}
|
|
|
|
var qtables []*dax.QualifiedTable
|
|
if err := json.NewDecoder(resp.Body).Decode(&qtables); err != nil {
|
|
return nil, errors.Wrap(err, "reading response body")
|
|
}
|
|
|
|
return qtables, nil
|
|
}
|
|
|
|
func (c *Client) CreateTable(ctx context.Context, qtbl *dax.QualifiedTable) error {
|
|
url := fmt.Sprintf("%s/create-table", c.address.WithScheme(defaultScheme))
|
|
|
|
// Encode the request.
|
|
postBody, err := json.Marshal(qtbl)
|
|
if err != nil {
|
|
return errors.Wrap(err, "marshalling post request")
|
|
}
|
|
responseBody := bytes.NewBuffer(postBody)
|
|
|
|
// Post the request.
|
|
resp, err := c.httpClient.Post(url, "application/json", responseBody)
|
|
if err != nil {
|
|
return errors.Wrap(err, "posting create table request")
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
return errors.Wrapf(errors.UnmarshalJSON(resp.Body), "status code: %d", resp.StatusCode)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (c *Client) DropTable(ctx context.Context, qtid dax.QualifiedTableID) error {
|
|
url := fmt.Sprintf("%s/drop-table", c.address.WithScheme(defaultScheme))
|
|
|
|
// Encode the request.
|
|
postBody, err := json.Marshal(qtid)
|
|
if err != nil {
|
|
return errors.Wrap(err, "marshalling post request")
|
|
}
|
|
responseBody := bytes.NewBuffer(postBody)
|
|
|
|
// Post the request.
|
|
resp, err := c.httpClient.Post(url, "application/json", responseBody)
|
|
if err != nil {
|
|
return errors.Wrap(err, "posting drop table request")
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
return errors.Wrapf(errors.UnmarshalJSON(resp.Body), "status code: %d", resp.StatusCode)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (c *Client) CreateField(ctx context.Context, qtid dax.QualifiedTableID, fld *dax.Field) error {
|
|
url := fmt.Sprintf("%s/create-field", c.address.WithScheme(defaultScheme))
|
|
|
|
req := controllerhttp.CreateFieldRequest{
|
|
TableKey: qtid.Key(),
|
|
Field: fld,
|
|
}
|
|
// Encode the request.
|
|
postBody, err := json.Marshal(req)
|
|
if err != nil {
|
|
return errors.Wrap(err, "marshalling post request")
|
|
}
|
|
responseBody := bytes.NewBuffer(postBody)
|
|
|
|
// Post the request.
|
|
resp, err := c.httpClient.Post(url, "application/json", responseBody)
|
|
if err != nil {
|
|
return errors.Wrap(err, "posting create field request")
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
return errors.Wrapf(errors.UnmarshalJSON(resp.Body), "status code: %d", resp.StatusCode)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (c *Client) DropField(ctx context.Context, qtid dax.QualifiedTableID, fldName dax.FieldName) error {
|
|
url := fmt.Sprintf("%s/drop-field", c.address.WithScheme(defaultScheme))
|
|
|
|
// Encode the request.
|
|
req := controllerhttp.DropFieldRequest{
|
|
Table: qtid,
|
|
Field: fldName,
|
|
}
|
|
|
|
postBody, err := json.Marshal(req)
|
|
if err != nil {
|
|
return errors.Wrap(err, "marshalling post request")
|
|
}
|
|
responseBody := bytes.NewBuffer(postBody)
|
|
|
|
// Post the request.
|
|
resp, err := c.httpClient.Post(url, "application/json", responseBody)
|
|
if err != nil {
|
|
return errors.Wrap(err, "posting drop field request")
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
return errors.Wrapf(errors.UnmarshalJSON(resp.Body), "status code: %d", resp.StatusCode)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (c *Client) IngestShard(ctx context.Context, qtid dax.QualifiedTableID, shard dax.ShardNum) (dax.Address, error) {
|
|
url := fmt.Sprintf("%s/ingest-shard", c.address.WithScheme(defaultScheme))
|
|
|
|
var host dax.Address
|
|
|
|
req := &controllerhttp.IngestShardRequest{
|
|
Table: qtid,
|
|
Shard: shard,
|
|
}
|
|
|
|
// Encode the request.
|
|
postBody, err := json.Marshal(req)
|
|
if err != nil {
|
|
return host, errors.Wrap(err, "marshalling post request")
|
|
}
|
|
responseBody := bytes.NewBuffer(postBody)
|
|
|
|
// Post the request.
|
|
resp, err := c.httpClient.Post(url, "application/json", responseBody)
|
|
if err != nil {
|
|
return host, errors.Wrap(err, "posting ingest-shard request")
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
b, _ := io.ReadAll(resp.Body)
|
|
return host, errors.Errorf("status code: %d: %s", resp.StatusCode, b)
|
|
}
|
|
|
|
var isr *controllerhttp.IngestShardResponse
|
|
if err := json.NewDecoder(resp.Body).Decode(&isr); err != nil {
|
|
return host, errors.Wrap(err, "reading response body")
|
|
}
|
|
|
|
return isr.Address, nil
|
|
}
|
|
|
|
func (c *Client) IngestPartition(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum) (dax.Address, error) {
|
|
url := fmt.Sprintf("%s/ingest-partition", c.address.WithScheme(defaultScheme))
|
|
|
|
var host dax.Address
|
|
|
|
req := &controllerhttp.IngestPartitionRequest{
|
|
Table: qtid,
|
|
Partition: partition,
|
|
}
|
|
|
|
// Encode the request.
|
|
postBody, err := json.Marshal(req)
|
|
if err != nil {
|
|
return host, errors.Wrap(err, "marshalling post request")
|
|
}
|
|
responseBody := bytes.NewBuffer(postBody)
|
|
|
|
// Post the request.
|
|
resp, err := c.httpClient.Post(url, "application/json", responseBody)
|
|
if err != nil {
|
|
return host, errors.Wrap(err, "posting ingest-partition request")
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
b, _ := io.ReadAll(resp.Body)
|
|
return host, errors.Errorf("status code: %d: %s", resp.StatusCode, b)
|
|
}
|
|
|
|
var isr *controllerhttp.IngestPartitionResponse
|
|
if err := json.NewDecoder(resp.Body).Decode(&isr); err != nil {
|
|
return host, errors.Wrap(err, "reading response body")
|
|
}
|
|
|
|
return isr.Address, nil
|
|
}
|
|
|
|
func (c *Client) ComputeNodes(ctx context.Context, qtid dax.QualifiedTableID, shards ...dax.ShardNum) ([]dax.ComputeNode, error) {
|
|
url := fmt.Sprintf("%s/compute-nodes", c.address.WithScheme(defaultScheme))
|
|
c.logger.Debugf("ComputeNodes url: %s", url)
|
|
|
|
var nodes []dax.ComputeNode
|
|
|
|
req := &controllerhttp.ComputeNodesRequest{
|
|
Table: qtid,
|
|
Shards: shards,
|
|
}
|
|
|
|
// Encode the request.
|
|
postBody, err := json.Marshal(req)
|
|
if err != nil {
|
|
return nodes, errors.Wrap(err, "marshalling post request")
|
|
}
|
|
responseBody := bytes.NewBuffer(postBody)
|
|
|
|
// Post the request.
|
|
resp, err := c.httpClient.Post(url, "application/json", responseBody)
|
|
if err != nil {
|
|
return nodes, errors.Wrap(err, "posting compute-nodes request")
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
b, _ := io.ReadAll(resp.Body)
|
|
return nodes, errors.Errorf("status code: %d: %s", resp.StatusCode, b)
|
|
}
|
|
|
|
var cnr *controllerhttp.ComputeNodesResponse
|
|
if err := json.NewDecoder(resp.Body).Decode(&cnr); err != nil {
|
|
return nodes, errors.Wrap(err, "reading response body")
|
|
}
|
|
|
|
return cnr.ComputeNodes, nil
|
|
}
|
|
|
|
func (c *Client) TranslateNodes(ctx context.Context, qtid dax.QualifiedTableID, partitions ...dax.PartitionNum) ([]dax.TranslateNode, error) {
|
|
url := fmt.Sprintf("%s/translate-nodes", c.address.WithScheme(defaultScheme))
|
|
c.logger.Debugf("TranslateNodes url: %s", url)
|
|
|
|
var nodes []dax.TranslateNode
|
|
|
|
req := &controllerhttp.TranslateNodesRequest{
|
|
Table: qtid,
|
|
Partitions: partitions,
|
|
}
|
|
|
|
// Encode the request.
|
|
postBody, err := json.Marshal(req)
|
|
if err != nil {
|
|
return nodes, errors.Wrap(err, "marshalling post request")
|
|
}
|
|
responseBody := bytes.NewBuffer(postBody)
|
|
|
|
// Post the request.
|
|
resp, err := c.httpClient.Post(url, "application/json", responseBody)
|
|
if err != nil {
|
|
return nodes, errors.Wrap(err, "posting translate-nodes request")
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
b, _ := io.ReadAll(resp.Body)
|
|
return nodes, errors.Errorf("status code: %d: %s", resp.StatusCode, b)
|
|
}
|
|
|
|
var cnr *controllerhttp.TranslateNodesResponse
|
|
if err := json.NewDecoder(resp.Body).Decode(&cnr); err != nil {
|
|
return nodes, errors.Wrap(err, "reading response body")
|
|
}
|
|
|
|
return cnr.TranslateNodes, nil
|
|
}
|
|
|
|
func (c *Client) RegisterNode(ctx context.Context, node *dax.Node) error {
|
|
url := fmt.Sprintf("%s/register-node", c.address.WithScheme(defaultScheme))
|
|
c.logger.Debugf("RegisterNode: %s, url: %s", node.Address, url)
|
|
|
|
req := &controllerhttp.RegisterNodeRequest{
|
|
Address: node.Address,
|
|
RoleTypes: node.RoleTypes,
|
|
}
|
|
|
|
// Encode the request.
|
|
postBody, err := json.Marshal(req)
|
|
if err != nil {
|
|
return errors.Wrap(err, "marshalling post request")
|
|
}
|
|
responseBody := bytes.NewBuffer(postBody)
|
|
|
|
// Post the request.
|
|
resp, err := c.httpClient.Post(url, "application/json", responseBody)
|
|
if err != nil {
|
|
return errors.Wrap(err, "posting translate-nodes request")
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
return errors.Wrapf(errors.UnmarshalJSON(resp.Body), "registration request to %s status code: %d", url, resp.StatusCode)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (c *Client) CheckInNode(ctx context.Context, node *dax.Node) error {
|
|
url := fmt.Sprintf("%s/check-in-node", c.address.WithScheme(defaultScheme))
|
|
c.logger.Debugf("CheckInNode url: %s", url)
|
|
|
|
req := &controllerhttp.CheckInNodeRequest{
|
|
Address: node.Address,
|
|
RoleTypes: node.RoleTypes,
|
|
}
|
|
|
|
// Encode the request.
|
|
postBody, err := json.Marshal(req)
|
|
if err != nil {
|
|
return errors.Wrap(err, "marshalling post request")
|
|
}
|
|
responseBody := bytes.NewBuffer(postBody)
|
|
|
|
// Post the request.
|
|
resp, err := c.httpClient.Post(url, "application/json", responseBody)
|
|
if err != nil {
|
|
return errors.Wrap(err, "posting translate-nodes request")
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
return errors.Wrapf(errors.UnmarshalJSON(resp.Body), "status code: %d", resp.StatusCode)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (c *Client) SnapshotTable(ctx context.Context, qtid dax.QualifiedTableID) error {
|
|
url := fmt.Sprintf("%s/snapshot", c.address.WithScheme(defaultScheme))
|
|
c.logger.Debugf("Snapshot url: %s", url)
|
|
|
|
// Encode the request.
|
|
postBody, err := json.Marshal(qtid)
|
|
if err != nil {
|
|
return errors.Wrap(err, "marshalling post request")
|
|
}
|
|
responseBody := bytes.NewBuffer(postBody)
|
|
|
|
// Post the request.
|
|
resp, err := c.httpClient.Post(url, "application/json", responseBody)
|
|
if err != nil {
|
|
return errors.Wrap(err, "posting translate-nodes request")
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
b, _ := io.ReadAll(resp.Body)
|
|
return errors.Errorf("status code: %d: %s", resp.StatusCode, b)
|
|
}
|
|
|
|
return nil
|
|
}
|