scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-07 09:05:55 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

remove http subpackage and bring implementations into core

Browse source 
remove interfaces as necessary
This commit is contained in:
Matthew Jaffee 2022-02-03 16:23:09 -06:00
parent 0f0e418763
commit 254bacc40c
34 changed files with 554 additions and 756 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
api_test.go
View file

@ -22,7 +22,6 @@ import (
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/authn"
"github.com/molecula/featurebase/v3/boltdb"
"github.com/molecula/featurebase/v3/http"
"github.com/molecula/featurebase/v3/server"
"github.com/molecula/featurebase/v3/shardwidth"
"github.com/molecula/featurebase/v3/test"
@ -36,21 +35,21 @@ func TestAPI_Import(t *testing.T) {
pilosa.OptServerNodeID("node0"),
pilosa.OptServerClusterHasher(&offsetModHasher{}),
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderFunc(nil)),
)},
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerNodeID("node1"),
pilosa.OptServerClusterHasher(&offsetModHasher{}),
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderFunc(nil)),
)},
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerNodeID("node2"),
pilosa.OptServerClusterHasher(&offsetModHasher{}),
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderFunc(nil)),
)},
)
defer c.Close()
@ -222,19 +221,19 @@ func TestAPI_ImportValue(t *testing.T) {
server.OptCommandServerOptions(
pilosa.OptServerNodeID("node0"),
pilosa.OptServerClusterHasher(&offsetModHasher{}),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderFunc(nil)),
)},
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerNodeID("node1"),
pilosa.OptServerClusterHasher(&offsetModHasher{}),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderFunc(nil)),
)},
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerNodeID("node2"),
pilosa.OptServerClusterHasher(&offsetModHasher{}),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderFunc(nil)),
)},
)
defer c.Close()
@ -529,7 +528,7 @@ func TestAPI_Ingest(t *testing.T) {
server.OptCommandServerOptions(
pilosa.OptServerNodeID("node0"),
pilosa.OptServerClusterHasher(&offsetModHasher{}),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderFunc(nil)),
)},
)
defer c.Close()
@ -648,7 +647,7 @@ func BenchmarkIngest(b *testing.B) {
server.OptCommandServerOptions(
pilosa.OptServerNodeID("node0"),
pilosa.OptServerClusterHasher(&offsetModHasher{}),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderFunc(nil)),
)},
)
defer c.Close()
@ -709,7 +708,7 @@ func TestAPI_ClearFlagForImportAndImportValues(t *testing.T) {
server.OptCommandServerOptions(
pilosa.OptServerNodeID("node0"),
pilosa.OptServerClusterHasher(&offsetModHasher{}),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderFunc(nil)),
)},
)
defer c.Close()
@ -1430,7 +1429,7 @@ func TestAPI_RBFDebugInfo(t *testing.T) {
server.OptCommandServerOptions(
pilosa.OptServerNodeID("node0"),
pilosa.OptServerClusterHasher(&offsetModHasher{}),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderFunc(nil)),
)},
)
defer c.Close()

211
client.go
View file

@ -3,12 +3,8 @@ package pilosa
import (
"context"
"io"
"time"
"github.com/molecula/featurebase/v3/ingest"
pnet "github.com/molecula/featurebase/v3/net"
"github.com/molecula/featurebase/v3/topology"
)
// Bit represents the intersection of a row and a column. It can be specified by
@ -29,74 +25,6 @@ type FieldValue struct {
Value int64
}
// InternalClient should be implemented by any struct that enables any transport between nodes
// TODO: Refactor
// Note from Travis: Typically an interface containing more than two or three methods is an indication that
// something hasn't been architected correctly.
// While I understand that putting the entire Client behind an interface might require this many methods,
// I don't want to let it go unquestioned.
// Another note from Travis: I think we eventually want to unify `InternalClient` with
// the `github.com/molecula/featurebase/v3/client` client.
// Doing that may obviate the need to refactor this.
type InternalClient interface {
InternalQueryClient
AvailableShards(ctx context.Context, indexName string) ([]uint64, error)
MaxShardByIndex(ctx context.Context) (map[string]uint64, error)
Schema(ctx context.Context) ([]*IndexInfo, error)
PostSchema(ctx context.Context, uri *pnet.URI, s *Schema, remote bool) error
CreateIndex(ctx context.Context, index string, opt IndexOptions) error
FragmentNodes(ctx context.Context, index string, shard uint64) ([]*topology.Node, error)
PartitionNodes(ctx context.Context, partitionID int) ([]*topology.Node, error)
Nodes(ctx context.Context) ([]*topology.Node, error)
Query(ctx context.Context, index string, queryRequest *QueryRequest) (*QueryResponse, error)
Import(ctx context.Context, qcx *Qcx, req *ImportRequest, options *ImportOptions) error
EnsureIndex(ctx context.Context, name string, options IndexOptions) error
EnsureField(ctx context.Context, indexName string, fieldName string) error
EnsureFieldWithOptions(ctx context.Context, index, field string, opt FieldOptions) error
ImportValue(ctx context.Context, qcx *Qcx, req *ImportValueRequest, options *ImportOptions) error
ExportCSV(ctx context.Context, index, field string, shard uint64, w io.Writer) error
CreateField(ctx context.Context, index, field string) error
CreateFieldWithOptions(ctx context.Context, index, field string, opt FieldOptions) error
FragmentBlocks(ctx context.Context, uri *pnet.URI, index, field, view string, shard uint64) ([]FragmentBlock, error)
BlockData(ctx context.Context, uri *pnet.URI, index, field, view string, shard uint64, block int) ([]uint64, []uint64, error)
SendMessage(ctx context.Context, uri *pnet.URI, msg []byte) error
RetrieveShardFromURI(ctx context.Context, index, field, view string, shard uint64, uri pnet.URI) (io.ReadCloser, error)
RetrieveTranslatePartitionFromURI(ctx context.Context, index string, partition int, uri pnet.URI) (io.ReadCloser, error)
ImportRoaring(ctx context.Context, uri *pnet.URI, index, field string, shard uint64, remote bool, req *ImportRoaringRequest) error
ShardReader(ctx context.Context, index string, shard uint64) (io.ReadCloser, error)
MutexCheck(ctx context.Context, uri *pnet.URI, index string, field string, details bool, limit int) (map[uint64]map[uint64][]uint64, error)
IngestNodeOperations(ctx context.Context, uri *pnet.URI, indexName string, ireq *ingest.ShardedRequest) error
IDAllocDataReader(ctx context.Context) (io.ReadCloser, error)
IDAllocDataWriter(ctx context.Context, f io.Reader, primary *topology.Node) error
IndexTranslateDataReader(ctx context.Context, index string, partitionID int) (io.ReadCloser, error)
FieldTranslateDataReader(ctx context.Context, index, field string) (io.ReadCloser, error)
StartTransaction(ctx context.Context, id string, timeout time.Duration, exclusive bool) (*Transaction, error)
FinishTransaction(ctx context.Context, id string) (*Transaction, error)
Transactions(ctx context.Context) (map[string]*Transaction, error)
GetTransaction(ctx context.Context, id string) (*Transaction, error)
GetNodeUsage(ctx context.Context, uri *pnet.URI) (map[string]NodeUsage, error)
GetPastQueries(ctx context.Context, uri *pnet.URI) ([]PastQueryStatus, error)
// ImportFieldKeys and ImportIndexKeys are mainly used when
// restoring a backup. They take a readerFunc which returns a
// reader rather than taking an io.Reader directly to allow for
// efficient retries (rather than reading the entire request body
// into a buffer and reusing it). Reader returned from the func
// must be properly closed by the implementation.
ImportFieldKeys(ctx context.Context, uri *pnet.URI, index, field string, remote bool, readerFunc func() (io.Reader, error)) error
ImportIndexKeys(ctx context.Context, uri *pnet.URI, index string, partitionID int, remote bool, readerFunc func() (io.Reader, error)) error
// SetInternalAPI tells the client the API it should use for internal/loopback ops
// where applicable.
SetInternalAPI(api *API)
}
//===============
// InternalQueryClient is the internal interface for querying a node.
type InternalQueryClient interface {
SchemaNode(ctx context.Context, uri *pnet.URI, views bool) ([]*IndexInfo, error)
@ -159,142 +87,3 @@ func newNopInternalQueryClient() nopInternalQueryClient {
}
var _ InternalQueryClient = newNopInternalQueryClient()
//===============
type nopInternalClient struct{ nopInternalQueryClient }
func newNopInternalClient() nopInternalClient {
return nopInternalClient{}
}
var _ InternalClient = newNopInternalClient()
func (n nopInternalClient) AvailableShards(ctx context.Context, indexName string) ([]uint64, error) {
return nil, nil
}
func (n nopInternalClient) MaxShardByIndex(context.Context) (map[string]uint64, error) {
return nil, nil
}
func (n nopInternalClient) Schema(ctx context.Context) ([]*IndexInfo, error) { return nil, nil }
func (n nopInternalClient) PostSchema(ctx context.Context, uri *pnet.URI, s *Schema, remote bool) error {
return nil
}
func (n nopInternalClient) CreateIndex(ctx context.Context, index string, opt IndexOptions) error {
return nil
}
func (n nopInternalClient) FragmentNodes(ctx context.Context, index string, shard uint64) ([]*topology.Node, error) {
return nil, nil
}
func (n nopInternalClient) PartitionNodes(ctx context.Context, partitionID int) ([]*topology.Node, error) {
return nil, nil
}
func (n nopInternalClient) Nodes(ctx context.Context) ([]*topology.Node, error) {
return nil, nil
}
func (n nopInternalClient) Query(ctx context.Context, index string, queryRequest *QueryRequest) (*QueryResponse, error) {
return nil, nil
}
func (n nopInternalClient) Import(ctx context.Context, qcx *Qcx, req *ImportRequest, options *ImportOptions) error {
return nil
}
func (n nopInternalClient) ImportValue(ctx context.Context, qcx *Qcx, req *ImportValueRequest, options *ImportOptions) error {
return nil
}
func (n nopInternalClient) ImportRoaring(ctx context.Context, uri *pnet.URI, index, field string, shard uint64, remote bool, req *ImportRoaringRequest) error {
return nil
}
func (n nopInternalClient) MutexCheck(ctx context.Context, uri *pnet.URI, index, field string, details bool, limit int) (map[uint64]map[uint64][]uint64, error) {
return nil, nil
}
func (n nopInternalClient) IngestNodeOperations(ctx context.Context, uri *pnet.URI, indexName string, ireq *ingest.ShardedRequest) error {
return nil
}
func (n nopInternalClient) ShardReader(ctx context.Context, index string, shard uint64) (io.ReadCloser, error) {
return nil, nil
}
func (n nopInternalClient) IDAllocDataReader(ctx context.Context) (io.ReadCloser, error) {
return nil, nil
}
func (n nopInternalClient) IDAllocDataWriter(cctx context.Context, f io.Reader, primary *topology.Node) error {
return nil
}
func (n nopInternalClient) IndexTranslateDataReader(ctx context.Context, index string, partitionID int) (io.ReadCloser, error) {
return nil, nil
}
func (n nopInternalClient) FieldTranslateDataReader(ctx context.Context, index, field string) (io.ReadCloser, error) {
return nil, nil
}
func (n nopInternalClient) EnsureIndex(ctx context.Context, name string, options IndexOptions) error {
return nil
}
func (n nopInternalClient) EnsureField(ctx context.Context, indexName string, fieldName string) error {
return nil
}
func (n nopInternalClient) EnsureFieldWithOptions(ctx context.Context, index, field string, opt FieldOptions) error {
return nil
}
func (n nopInternalClient) ExportCSV(ctx context.Context, index, field string, shard uint64, w io.Writer) error {
return nil
}
func (n nopInternalClient) CreateField(ctx context.Context, index, field string) error { return nil }
func (n nopInternalClient) CreateFieldWithOptions(ctx context.Context, index, field string, opt FieldOptions) error {
return nil
}
func (n nopInternalClient) FragmentBlocks(ctx context.Context, uri *pnet.URI, index, field, view string, shard uint64) ([]FragmentBlock, error) {
return nil, nil
}
func (n nopInternalClient) BlockData(ctx context.Context, uri *pnet.URI, index, field, view string, shard uint64, block int) ([]uint64, []uint64, error) {
return nil, nil, nil
}
func (n nopInternalClient) SendMessage(ctx context.Context, uri *pnet.URI, msg []byte) error {
return nil
}
func (n nopInternalClient) RetrieveShardFromURI(ctx context.Context, index, field, view string, shard uint64, uri pnet.URI) (io.ReadCloser, error) {
return nil, nil
}
func (n nopInternalClient) RetrieveTranslatePartitionFromURI(ctx context.Context, index string, partition int, uri pnet.URI) (io.ReadCloser, error) {
return nil, nil
}
func (n nopInternalClient) StartTransaction(ctx context.Context, id string, timeout time.Duration, exclusive bool) (*Transaction, error) {
return nil, nil
}
func (n nopInternalClient) FinishTransaction(ctx context.Context, id string) (*Transaction, error) {
return nil, nil
}
func (n nopInternalClient) Transactions(ctx context.Context) (map[string]*Transaction, error) {
return nil, nil
}
func (n nopInternalClient) GetTransaction(ctx context.Context, id string) (*Transaction, error) {
return nil, nil
}
func (n nopInternalClient) GetNodeUsage(ctx context.Context, uri *pnet.URI) (map[string]NodeUsage, error) {
return nil, nil
}
func (n nopInternalClient) GetPastQueries(ctx context.Context, uri *pnet.URI) ([]PastQueryStatus, error) {
return nil, nil
}
func (c nopInternalClient) ImportFieldKeys(ctx context.Context, uri *pnet.URI, index, field string, remote bool, readerFunc func() (io.Reader, error)) error {
return nil
}
func (c nopInternalClient) ImportIndexKeys(ctx context.Context, uri *pnet.URI, index string, partitionID int, remote bool, readerFunc func() (io.Reader, error)) error {
return nil
}
func (c nopInternalClient) SetInternalAPI(api *API) {
}

4
cluster.go
View file

@ -102,7 +102,7 @@ type cluster struct { // nolint: maligned
logger logger.Logger
InternalClient InternalClient
InternalClient *InternalClient
confirmDownRetries int
confirmDownSleep time.Duration
@ -120,7 +120,7 @@ func newCluster() *cluster {
translationSyncer: NopTranslationSyncer,
InternalClient: newNopInternalClient(),
InternalClient: &InternalClient{}, // TODO might have to fill this out a bit
logger: logger.NopLogger,

6
cmd/badloader/badloader.go
View file

@ -13,7 +13,7 @@ import (
gohttp "net/http"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/http"
"github.com/molecula/featurebase/v3/encoding/proto"
pnet "github.com/molecula/featurebase/v3/net"
"github.com/molecula/featurebase/v3/vprint"
@ -22,7 +22,7 @@ import (
"strings"
)
func UploadTar(srcFile string, client *http.InternalClient) error {
func UploadTar(srcFile string, client *pilosa.InternalClient) error {
t0 := time.Now()
f, err := os.Open(srcFile)
if err != nil {
@ -114,7 +114,7 @@ func main() {
host := "127.0.0.1:10101"
h := &gohttp.Client{}
c, err := http.NewInternalClient(host, h)
c, err := pilosa.NewInternalClient(host, h, pilosa.WithSerializer(proto.Serializer{}))
vprint.PanicOn(err)
tarSrcPath := "q2.tar.gz"

10
cmd/pilosa-bench/main.go
View file

@ -16,8 +16,8 @@ import (
"strings"
"time"
"github.com/molecula/featurebase/v3"
phttp "github.com/molecula/featurebase/v3/http"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/encoding/proto"
"golang.org/x/sync/errgroup"
)
@ -78,7 +78,7 @@ func run(ctx context.Context, args []string) (err error) {
rand.Seed(0)
// Setup connection to pilosa.
client, err := phttp.NewInternalClient(*hostport, http.DefaultClient)
client, err := pilosa.NewInternalClient(*hostport, http.DefaultClient, pilosa.WithSerializer(proto.Serializer{}))
if err != nil {
return err
}
@ -270,7 +270,7 @@ func generateTopKQuery(index, field string, from, to time.Time) string {
}
// loadFields returns a mapping of index/field names to field info & identifiers.
func loadFields(ctx context.Context, client *phttp.InternalClient) (map[fieldKey]*fieldInfo, error) {
func loadFields(ctx context.Context, client *pilosa.InternalClient) (map[fieldKey]*fieldInfo, error) {
indexes, err := client.Schema(ctx)
if err != nil {
return nil, err
@ -299,7 +299,7 @@ func loadFields(ctx context.Context, client *phttp.InternalClient) (map[fieldKey
}
// fetchFieldIDs returns a list of field IDs or keys.
func fetchFieldIDs(ctx context.Context, client *phttp.InternalClient, indexName, fieldName string) (*pilosa.RowIdentifiers, error) {
func fetchFieldIDs(ctx context.Context, client *pilosa.InternalClient, indexName, fieldName string) (*pilosa.RowIdentifiers, error) {
resp, err := client.Query(ctx, indexName, &pilosa.QueryRequest{Index: indexName, Query: `Rows(` + fieldName + `)`})
if err != nil {
return nil, err

5
cmd/random-query/main.go
View file

@ -16,9 +16,10 @@ import (
"time"
"github.com/gogo/protobuf/proto"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/client"
"github.com/molecula/featurebase/v3/http"
fb_proto "github.com/molecula/featurebase/v3/encoding/proto"
"github.com/molecula/featurebase/v3/pb"
"github.com/molecula/featurebase/v3/pql"
"github.com/molecula/featurebase/v3/vprint"
@ -162,7 +163,7 @@ func main() {
func (cfg *RandomQueryConfig) Run() (err error) {
remoteClient := nethttp.DefaultClient
cli, err := http.NewInternalClient(cfg.HostPort, remoteClient)
cli, err := pilosa.NewInternalClient(cfg.HostPort, remoteClient, pilosa.WithSerializer(fb_proto.Serializer{}))
if err != nil {
return err
}

3
cmd/random-query/main_test.go
View file

@ -9,7 +9,6 @@ import (
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/boltdb"
"github.com/molecula/featurebase/v3/http"
"github.com/molecula/featurebase/v3/server"
"github.com/molecula/featurebase/v3/test"
. "github.com/molecula/featurebase/v3/vprint" // nolint:staticcheck
@ -35,7 +34,7 @@ func Test_RandomQuery(t *testing.T) {
server.OptCommandServerOptions(
pilosa.OptServerNodeID(nodeid[0]),
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerReplicaN(nReplicas),
)},
)

8
cmd/slurp/slurp.go
View file

@ -18,7 +18,7 @@ import (
"time"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/http"
"github.com/molecula/featurebase/v3/encoding/proto"
pnet "github.com/molecula/featurebase/v3/net"
"github.com/molecula/featurebase/v3/vprint"
)
@ -32,7 +32,7 @@ type stateMachine struct {
lastField string
lastShard uint64
state string
client *http.InternalClient
client *pilosa.InternalClient
start time.Time
profile string
@ -136,7 +136,7 @@ func (r *stateMachine) Upload() error {
return nil
}
func UploadTar(srcFile string, client *http.InternalClient, profile, host string) error {
func UploadTar(srcFile string, client *pilosa.InternalClient, profile, host string) error {
f, err := os.Open(srcFile)
if err != nil {
@ -192,7 +192,7 @@ func main() {
if profile != "" {
startProfile(host)
}
c, err := http.NewInternalClient(host, h)
c, err := pilosa.NewInternalClient(host, h, pilosa.WithSerializer(proto.Serializer{}))
vprint.PanicOn(err)
t0 := time.Now()

13
ctl/backup.go
View file

@ -13,7 +13,7 @@ import (
"time"
pilosa "github.com/molecula/featurebase/v3"
fb_http "github.com/molecula/featurebase/v3/http"
"github.com/molecula/featurebase/v3/encoding/proto"
"github.com/molecula/featurebase/v3/server"
"github.com/molecula/featurebase/v3/topology"
"github.com/pkg/errors"
@ -49,7 +49,7 @@ type BackupCommand struct { // nolint: maligned
Pprof string `json:"pprof"`
// Reusable client.
client pilosa.InternalClient
client *pilosa.InternalClient
// Standard input/output
*pilosa.CmdIO
@ -93,7 +93,7 @@ func (cmd *BackupCommand) Run(ctx context.Context) (err error) {
}
// Create a client to the server.
client, err := commandClient(cmd, fb_http.WithClientRetryPeriod(cmd.RetryPeriod), fb_http.ClientResponseHeaderTimeoutOption(cmd.HeaderTimeout))
client, err := commandClient(cmd, pilosa.WithClientRetryPeriod(cmd.RetryPeriod), pilosa.ClientResponseHeaderTimeoutOption(cmd.HeaderTimeout))
if err != nil {
return fmt.Errorf("creating client: %w", err)
}
@ -289,9 +289,10 @@ func (cmd *BackupCommand) backupShardNode(ctx context.Context, indexName string,
logger := cmd.Logger()
logger.Printf("backing up shard: index=%q id=%d", indexName, shard)
client := fb_http.NewInternalClientFromURI(&node.URI,
fb_http.GetHTTPClient(cmd.tlsConfig, fb_http.ClientResponseHeaderTimeoutOption(cmd.HeaderTimeout)),
fb_http.WithClientRetryPeriod(cmd.RetryPeriod))
client := pilosa.NewInternalClientFromURI(&node.URI,
pilosa.GetHTTPClient(cmd.tlsConfig, pilosa.ClientResponseHeaderTimeoutOption(cmd.HeaderTimeout)),
pilosa.WithClientRetryPeriod(cmd.RetryPeriod),
pilosa.WithSerializer(proto.Serializer{}))
rc, err := client.ShardReader(ctx, indexName, shard)
if err != nil {
return fmt.Errorf("fetching shard reader: %w", err)

2
ctl/chksum.go
View file

@ -20,7 +20,7 @@ type ChkSumCommand struct { // nolint: maligned
Host string `json:"host"`
// Reusable client.
client pilosa.InternalClient
client *pilosa.InternalClient
// Standard input/output
*pilosa.CmdIO

24
ctl/common.go
View file

@ -4,7 +4,8 @@ package ctl
import (
"time"
"github.com/molecula/featurebase/v3/http"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/encoding/proto"
"github.com/molecula/featurebase/v3/logger"
"github.com/molecula/featurebase/v3/server"
"github.com/pkg/errors"
@ -27,14 +28,14 @@ func SetTLSConfig(flags *pflag.FlagSet, prefix string, certificatePath *string,
flags.BoolVarP(enableClientVerification, prefix+"tls.enable-client-verification", "", false, "Enable TLS certificate client verification for incoming connections")
}
// AnyClientOption can be either http.InternalClientOption or
// http.ClientOption. The internal options are specific to the
// AnyClientOption can be either pilosa.InternalClientOption or
// pilosa.ClientOption. The internal options are specific to the
// featurebase client, whereas the client options are applied to the
// Go HTTP client that gets used under the hood.
type AnyClientOption interface{}
// commandClient returns a pilosa.InternalHTTPClient for the command
func commandClient(cmd CommandWithTLSSupport, opts ...AnyClientOption) (*http.InternalClient, error) {
func commandClient(cmd CommandWithTLSSupport, opts ...AnyClientOption) (*pilosa.InternalClient, error) {
internalopts, clientopts, err := separateOptions(opts...)
if err != nil {
return nil, errors.Wrap(err, "separating client options")
@ -42,13 +43,14 @@ func commandClient(cmd CommandWithTLSSupport, opts ...AnyClientOption) (*http.In
// we default dial timeout to 3s in commandClient, but prepend it
// to the option list so other options can override it.
clientopts = append([]http.ClientOption{http.ClientDialTimeoutOption(time.Second * 3)}, clientopts...)
clientopts = append([]pilosa.ClientOption{pilosa.ClientDialTimeoutOption(time.Second * 3)}, clientopts...)
internalopts = append([]pilosa.InternalClientOption{pilosa.WithSerializer(proto.Serializer{})}, internalopts...)
tls := cmd.TLSConfiguration()
tlsConfig, err := server.GetTLSConfig(&tls, cmd.Logger())
if err != nil {
return nil, errors.Wrap(err, "getting tls config")
}
client, err := http.NewInternalClient(cmd.TLSHost(), http.GetHTTPClient(tlsConfig, clientopts...), internalopts...)
client, err := pilosa.NewInternalClient(cmd.TLSHost(), pilosa.GetHTTPClient(tlsConfig, clientopts...), internalopts...)
if err != nil {
return nil, errors.Wrap(err, "getting internal client")
}
@ -57,15 +59,15 @@ func commandClient(cmd CommandWithTLSSupport, opts ...AnyClientOption) (*http.In
// separateOptions splits the list of AnyClientOption into the two
// possible types.
func separateOptions(opts ...AnyClientOption) ([]http.InternalClientOption, []http.ClientOption, error) {
internalopts := []http.InternalClientOption{}
clientopts := []http.ClientOption{}
func separateOptions(opts ...AnyClientOption) ([]pilosa.InternalClientOption, []pilosa.ClientOption, error) {
internalopts := []pilosa.InternalClientOption{}
clientopts := []pilosa.ClientOption{}
for _, opt := range opts {
if iopt, ok := opt.(http.InternalClientOption); ok {
if iopt, ok := opt.(pilosa.InternalClientOption); ok {
internalopts = append(internalopts, iopt)
continue
}
if copt, ok := opt.(http.ClientOption); ok {
if copt, ok := opt.(pilosa.ClientOption); ok {
clientopts = append(clientopts, copt)
continue
}

4
ctl/import.go
View file

@ -11,7 +11,7 @@ import (
"strconv"
"time"
"github.com/molecula/featurebase/v3"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/pql"
"github.com/molecula/featurebase/v3/server"
"github.com/pkg/errors"
@ -48,7 +48,7 @@ type ImportCommand struct { // nolint: maligned
Sort bool `json:"sort"`
// Reusable client.
client pilosa.InternalClient
client *pilosa.InternalClient
// Standard input/output
*pilosa.CmdIO

5
ctl/restore.go
View file

@ -18,7 +18,6 @@ import (
"github.com/hashicorp/go-retryablehttp"
pilosa "github.com/molecula/featurebase/v3"
fb_http "github.com/molecula/featurebase/v3/http"
"github.com/molecula/featurebase/v3/logger"
"github.com/molecula/featurebase/v3/server"
"github.com/molecula/featurebase/v3/topology"
@ -44,7 +43,7 @@ type RestoreCommand struct {
Pprof string `json:"pprof"`
// Reusable client.
client pilosa.InternalClient
client *pilosa.InternalClient
// Standard input/output
*pilosa.CmdIO
@ -86,7 +85,7 @@ func (cmd *RestoreCommand) Run(ctx context.Context) (err error) {
return fmt.Errorf("parsing tls config: %w", err)
}
// Create a client to the server.
client, err := commandClient(cmd, fb_http.WithClientRetryPeriod(cmd.RetryPeriod))
client, err := commandClient(cmd, pilosa.WithClientRetryPeriod(cmd.RetryPeriod))
if err != nil {
return fmt.Errorf("creating client: %w", err)
}

5
dbshard_test.go
View file

@ -7,9 +7,8 @@ import (
"reflect"
"testing"
"github.com/molecula/featurebase/v3"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/boltdb"
"github.com/molecula/featurebase/v3/http"
"github.com/molecula/featurebase/v3/server"
"github.com/molecula/featurebase/v3/test"
. "github.com/molecula/featurebase/v3/vprint" // nolint:staticcheck
@ -23,7 +22,7 @@ func TestAPI_SimplerOneNode_ImportColumnKey(t *testing.T) {
pilosa.OptServerNodeID("node0"),
pilosa.OptServerClusterHasher(&offsetModHasher{}),
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderFunc(nil)),
)},
)
defer c.Close()

5
executor_test.go
View file

@ -29,7 +29,6 @@ import (
"github.com/molecula/featurebase/v3/boltdb"
"github.com/molecula/featurebase/v3/ctl"
"github.com/molecula/featurebase/v3/disco"
"github.com/molecula/featurebase/v3/http"
"github.com/molecula/featurebase/v3/pql"
"github.com/molecula/featurebase/v3/proto"
"github.com/molecula/featurebase/v3/server"
@ -3826,7 +3825,7 @@ func TestExecutor_Execute_Existence(t *testing.T) {
c := test.MustRunCluster(t, 1, []server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderFunc(nil)),
),
})
defer c.Close()
@ -4217,7 +4216,7 @@ func TestExecutor_Execute_All(t *testing.T) {
c := test.MustRunCluster(t, 1, []server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderFunc(nil)),
),
})
defer c.Close()

6
handler.go
View file

@ -76,9 +76,9 @@ func (resp *QueryResponse) MarshalJSON() ([]byte, error) {
})
}
// Handler is the interface for the data handler, a wrapper around
// HandlerI is the interface for the data handler, a wrapper around
// Pilosa's data store.
type Handler interface {
type HandlerI interface {
Serve() error
Close() error
}
@ -94,7 +94,7 @@ func (n nopHandler) Close() error {
}
// NopHandler is a no-op implementation of the Handler interface.
var NopHandler Handler = nopHandler{}
var NopHandler HandlerI = nopHandler{}
// ImportValueRequest describes the import request structure
// for a value (BSI) import.

13
http/error.go
View file

@ -1,13 +0,0 @@
// Copyright 2021 Molecula Corp. All rights reserved.
package http
// Error defines a standard application error.
type Error struct {
// Human-readable message.
Message string `json:"message"`
}
// Error returns the string representation of the error message.
func (e *Error) Error() string {
return e.Message
}

428
http/handler.go → http_handler.go
View file

File diff suppressed because it is too large Load diff

27
http/handler_internal_test.go → http_handler_internal_test.go
View file

@ -1,5 +1,5 @@
// Copyright 2021 Molecula Corp. All rights reserved.
package http
package pilosa
import (
"bytes"
@ -17,7 +17,6 @@ import (
"time"
"github.com/golang-jwt/jwt"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/authn"
"golang.org/x/oauth2"
@ -33,9 +32,9 @@ func TestPostIndexRequestUnmarshalJSON(t *testing.T) {
expected postIndexRequest
err string
}{
{json: `{"options": {}}`, expected: postIndexRequest{Options: pilosa.IndexOptions{TrackExistence: true}}},
{json: `{"options": {"trackExistence": false}}`, expected: postIndexRequest{Options: pilosa.IndexOptions{TrackExistence: false}}},
{json: `{"options": {"keys": true}}`, expected: postIndexRequest{Options: pilosa.IndexOptions{Keys: true, TrackExistence: true}}},
{json: `{"options": {}}`, expected: postIndexRequest{Options: IndexOptions{TrackExistence: true}}},
{json: `{"options": {"trackExistence": false}}`, expected: postIndexRequest{Options: IndexOptions{TrackExistence: false}}},
{json: `{"options": {"keys": true}}`, expected: postIndexRequest{Options: IndexOptions{Keys: true, TrackExistence: true}}},
{json: `{"options": 4}`, err: "options is not map[string]interface{}"},
{json: `{"option": {}}`, err: "unknown key: option:map[]"},
{json: `{"options": {"badKey": "test"}}`, err: "unknown key: badKey:test"},
@ -107,8 +106,8 @@ func decimalPtr(d pql.Decimal) *pql.Decimal {
// Test fieldOption validation.
func TestFieldOptionValidation(t *testing.T) {
timeQuantum := pilosa.TimeQuantum("YMD")
defaultCacheSize := uint32(pilosa.DefaultCacheSize)
timeQuantum := TimeQuantum("YMD")
defaultCacheSize := uint32(DefaultCacheSize)
tests := []struct {
json string
expected postFieldRequest
@ -116,17 +115,17 @@ func TestFieldOptionValidation(t *testing.T) {
}{
// FieldType: Set
{json: `{"options": {}}`, expected: postFieldRequest{Options: fieldOptions{
Type: pilosa.FieldTypeSet,
CacheType: stringPtr(pilosa.DefaultCacheType),
Type: FieldTypeSet,
CacheType: stringPtr(DefaultCacheType),
CacheSize: &defaultCacheSize,
}}},
{json: `{"options": {"type": "set"}}`, expected: postFieldRequest{Options: fieldOptions{
Type: pilosa.FieldTypeSet,
CacheType: stringPtr(pilosa.DefaultCacheType),
Type: FieldTypeSet,
CacheType: stringPtr(DefaultCacheType),
CacheSize: &defaultCacheSize,
}}},
{json: `{"options": {"type": "set", "cacheType": "lru"}}`, expected: postFieldRequest{Options: fieldOptions{
Type: pilosa.FieldTypeSet,
Type: FieldTypeSet,
CacheType: stringPtr("lru"),
CacheSize: &defaultCacheSize,
}}},
@ -138,7 +137,7 @@ func TestFieldOptionValidation(t *testing.T) {
{json: `{"options": {"type": "int"}}`, err: "min is required for field type int"},
{json: `{"options": {"type": "int", "min": 0}}`, err: "max is required for field type int"},
{json: `{"options": {"type": "int", "min": 0, "max": 1001}}`, expected: postFieldRequest{Options: fieldOptions{
Type: pilosa.FieldTypeInt,
Type: FieldTypeInt,
Min: decimalPtr(pql.NewDecimal(0, 0)),
Max: decimalPtr(pql.NewDecimal(1001, 0)),
}}},
@ -149,7 +148,7 @@ func TestFieldOptionValidation(t *testing.T) {
// FieldType: Time
{json: `{"options": {"type": "time"}}`, err: "timeQuantum is required for field type time"},
{json: `{"options": {"type": "time", "timeQuantum": "YMD"}}`, expected: postFieldRequest{Options: fieldOptions{
Type: pilosa.FieldTypeTime,
Type: FieldTypeTime,
TimeQuantum: &timeQuantum,
}}},
{json: `{"options": {"type": "time", "timeQuantum": "YMD", "min": 0}}`, err: "min does not apply to field type time"},

24
http/handler_test.go → http_handler_test.go
View file

@ -1,5 +1,5 @@
// Copyright 2021 Molecula Corp. All rights reserved.
package http_test
package pilosa_test
import (
"encoding/json"
@ -10,17 +10,17 @@ import (
"testing"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/http"
"github.com/molecula/featurebase/v3/encoding/proto"
"github.com/molecula/featurebase/v3/server"
"github.com/molecula/featurebase/v3/test"
)
func TestHandlerOptions(t *testing.T) {
_, err := http.NewHandler()
_, err := pilosa.NewHandler()
if err == nil {
t.Fatalf("expected error making handler without options, got nil")
}
_, err = http.NewHandler(http.OptHandlerAPI(&pilosa.API{}))
_, err = pilosa.NewHandler(pilosa.OptHandlerAPI(&pilosa.API{}))
if err == nil {
t.Fatalf("expected error making handler without options, got nil")
}
@ -30,24 +30,30 @@ func TestHandlerOptions(t *testing.T) {
t.Fatalf("creating listener: %v", err)
}
_, err = http.NewHandler(http.OptHandlerListener(ln, ln.Addr().String()))
_, err = pilosa.NewHandler(pilosa.OptHandlerListener(ln, ln.Addr().String()))
if err == nil {
t.Fatalf("expected error making handler without options, got nil")
}
_, err = pilosa.NewHandler(pilosa.OptHandlerListener(ln, ln.Addr().String()), pilosa.OptHandlerSerializer(proto.Serializer{}), pilosa.OptHandlerSerializer(proto.RoaringSerializer))
if err == nil {
t.Fatalf("expected error making handler without enough options, got nil")
}
}
func TestMarshalUnmarshalTransactionResponse(t *testing.T) {
tests := []struct {
name string
tr *http.TransactionResponse
tr *pilosa.TransactionResponse
}{
{
name: "nil transaction",
tr: &http.TransactionResponse{},
tr: &pilosa.TransactionResponse{},
},
{
name: "empty transaction",
tr: &http.TransactionResponse{Transaction: &pilosa.Transaction{}},
tr: &pilosa.TransactionResponse{Transaction: &pilosa.Transaction{}},
},
}
@ -58,7 +64,7 @@ func TestMarshalUnmarshalTransactionResponse(t *testing.T) {
t.Fatalf("marshaling: %v", err)
}
mytr := &http.TransactionResponse{}
mytr := &pilosa.TransactionResponse{}
err = json.Unmarshal(data, mytr)
if err != nil {
t.Fatalf("unmarshalling: %v", err)

29
http/translator.go → http_translator.go
View file

@ -1,5 +1,5 @@
// Copyright 2021 Molecula Corp. All rights reserved.
package http
package pilosa
import (
"bytes"
@ -12,25 +12,24 @@ import (
"reflect"
"sync"
"github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/logger"
)
func GetOpenTranslateReaderFunc(client *http.Client) pilosa.OpenTranslateReaderFunc {
func GetOpenTranslateReaderFunc(client *http.Client) OpenTranslateReaderFunc {
return GetOpenTranslateReaderWithLockerFunc(client, nopLocker{})
}
func GetOpenTranslateReaderWithLockerFunc(client *http.Client, locker sync.Locker) pilosa.OpenTranslateReaderFunc {
func GetOpenTranslateReaderWithLockerFunc(client *http.Client, locker sync.Locker) OpenTranslateReaderFunc {
lockType := reflect.TypeOf(locker)
if lockType.Kind() == reflect.Ptr {
lockType = lockType.Elem()
}
return func(ctx context.Context, nodeURL string, offsets pilosa.TranslateOffsetMap) (pilosa.TranslateEntryReader, error) {
return func(ctx context.Context, nodeURL string, offsets TranslateOffsetMap) (TranslateEntryReader, error) {
return openTranslateReader(ctx, nodeURL, offsets, client, reflect.New(lockType).Interface().(sync.Locker))
}
}
func openTranslateReader(ctx context.Context, nodeURL string, offsets pilosa.TranslateOffsetMap, client *http.Client, locker sync.Locker) (pilosa.TranslateEntryReader, error) {
func openTranslateReader(ctx context.Context, nodeURL string, offsets TranslateOffsetMap, client *http.Client, locker sync.Locker) (TranslateEntryReader, error) {
r := NewTranslateEntryReader(ctx, client)
r.locker = locker
@ -47,9 +46,9 @@ type nopLocker struct{}
func (nopLocker) Lock() {}
func (nopLocker) Unlock() {}
// TranslateEntryReader represents an implementation of pilosa.TranslateEntryReader.
// TranslateEntryReader represents an implementation of TranslateEntryReader.
// It consolidates all index & field translate entries into a single reader.
type TranslateEntryReader struct {
type HTTPTranslateEntryReader struct {
locker sync.Locker
ctx context.Context
@ -60,7 +59,7 @@ type TranslateEntryReader struct {
// Lookup of offsets for each index & field.
// Must be set before calling Open().
Offsets pilosa.TranslateOffsetMap
Offsets TranslateOffsetMap
// URL to stream entries from.
// Must be set before calling Open().
@ -72,17 +71,17 @@ type TranslateEntryReader struct {
}
// NewTranslateEntryReader returns a new instance of TranslateEntryReader.
func NewTranslateEntryReader(ctx context.Context, client *http.Client) *TranslateEntryReader {
func NewTranslateEntryReader(ctx context.Context, client *http.Client) *HTTPTranslateEntryReader {
if client == nil {
client = http.DefaultClient
}
r := &TranslateEntryReader{locker: nopLocker{}, HTTPClient: client, Logger: logger.NopLogger}
r := &HTTPTranslateEntryReader{locker: nopLocker{}, HTTPClient: client, Logger: logger.NopLogger}
r.ctx, r.cancel = context.WithCancel(ctx)
return r
}
// Open initiates the reader.
func (r *TranslateEntryReader) Open() error {
func (r *HTTPTranslateEntryReader) Open() error {
// Serialize map of offsets to request body.
requestBody, err := json.Marshal(r.Offsets)
if err != nil {
@ -107,7 +106,7 @@ func (r *TranslateEntryReader) Open() error {
// Handle error codes.
if resp.StatusCode == http.StatusNotImplemented {
r.body.Close()
return pilosa.ErrNotImplemented
return ErrNotImplemented
} else if resp.StatusCode != http.StatusOK {
body, _ := ioutil.ReadAll(resp.Body)
r.body.Close()
@ -117,7 +116,7 @@ func (r *TranslateEntryReader) Open() error {
}
// Close stops the reader.
func (r *TranslateEntryReader) Close() error {
func (r *HTTPTranslateEntryReader) Close() error {
if r.cancel != nil {
r.cancel()
}
@ -132,7 +131,7 @@ func (r *TranslateEntryReader) Close() error {
// ReadEntry reads the next entry from the stream into entry.
// Returns io.EOF at the end of the stream.
func (r *TranslateEntryReader) ReadEntry(entry *pilosa.TranslateEntry) error {
func (r *HTTPTranslateEntryReader) ReadEntry(entry *TranslateEntry) error {
r.locker.Lock()
defer r.locker.Unlock()

11
http/translator_test.go → http_translator_test.go
View file

@ -1,5 +1,5 @@
// Copyright 2021 Molecula Corp. All rights reserved.
package http_test
package pilosa_test
import (
"context"
@ -8,8 +8,7 @@ import (
"testing"
"time"
"github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/http"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/test"
)
@ -41,7 +40,7 @@ func TestTranslateStore_EntryReader(t *testing.T) {
}
// Connect to server and stream all available data.
r := http.NewTranslateEntryReader(context.Background(), nil)
r := pilosa.NewTranslateEntryReader(context.Background(), nil)
r.URL = primary.URL()
// Wait to ensure writes make it to translate store
@ -123,7 +122,7 @@ func BenchmarkReadEntryNoMutex(b *testing.B) {
defer teardown()
for n := 0; n < b.N; n++ {
r, err := http.GetOpenTranslateReaderFunc(nil)(ctx, url, offset)
r, err := pilosa.GetOpenTranslateReaderFunc(nil)(ctx, url, offset)
if err != nil {
b.Fatalf("opening translate reader: %+v", err)
}
@ -138,7 +137,7 @@ func BenchmarkReadEntryWithMutex(b *testing.B) {
defer teardown()
for n := 0; n < b.N; n++ {
r, err := http.GetOpenTranslateReaderWithLockerFunc(nil, &sync.Mutex{})(ctx, url, offset)
r, err := pilosa.GetOpenTranslateReaderWithLockerFunc(nil, &sync.Mutex{})(ctx, url, offset)
if err != nil {
b.Fatalf("opening translate reader: %+v", err)
}

12
internal/clustertests/cluster_test.go
View file

@ -17,7 +17,7 @@ import (
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/authn"
"github.com/molecula/featurebase/v3/disco"
picli "github.com/molecula/featurebase/v3/http"
"github.com/molecula/featurebase/v3/encoding/proto"
"github.com/molecula/featurebase/v3/logger"
"github.com/pkg/errors"
)
@ -87,15 +87,15 @@ func TestClusterStuff(t *testing.T) {
auth = true
}
cli1, err := picli.NewInternalClient("pilosa1:10101", picli.GetHTTPClient(nil))
cli1, err := pilosa.NewInternalClient("pilosa1:10101", pilosa.GetHTTPClient(nil), pilosa.WithSerializer(proto.Serializer{}))
if err != nil {
t.Fatalf("getting client: %v", err)
}
cli2, err := picli.NewInternalClient("pilosa2:10101", picli.GetHTTPClient(nil))
cli2, err := pilosa.NewInternalClient("pilosa2:10101", pilosa.GetHTTPClient(nil), pilosa.WithSerializer(proto.Serializer{}))
if err != nil {
t.Fatalf("getting client: %v", err)
}
cli3, err := picli.NewInternalClient("pilosa3:10101", picli.GetHTTPClient(nil))
cli3, err := pilosa.NewInternalClient("pilosa3:10101", pilosa.GetHTTPClient(nil), pilosa.WithSerializer(proto.Serializer{}))
if err != nil {
t.Fatalf("getting client: %v", err)
}
@ -134,7 +134,7 @@ func TestClusterStuff(t *testing.T) {
}
// Check query results from each node.
for i, cli := range []*picli.InternalClient{cli1, cli2, cli3} {
for i, cli := range []*pilosa.InternalClient{cli1, cli2, cli3} {
r, err := cli.Query(ctx, "testidx", &pilosa.QueryRequest{Index: "testidx", Query: "Count(Row(testf=0))"})
if err != nil {
t.Fatalf("count querying pilosa%d: %v", i, err)
@ -157,7 +157,7 @@ func TestClusterStuff(t *testing.T) {
t.Log("done waiting for stability")
// Check query results from each node.
for i, cli := range []*picli.InternalClient{cli1, cli2, cli3} {
for i, cli := range []*pilosa.InternalClient{cli1, cli2, cli3} {
r, err := cli.Query(ctx, "testidx", &pilosa.QueryRequest{Index: "testidx", Query: "Count(Row(testf=0))"})
if err != nil {
t.Fatalf("count querying pilosa%d: %v", i, err)

14
internal/clustertests/pause_node_test.go
View file

@ -17,7 +17,7 @@ import (
pilosa "github.com/molecula/featurebase/v3"
boltdb "github.com/molecula/featurebase/v3/boltdb"
"github.com/molecula/featurebase/v3/disco"
"github.com/molecula/featurebase/v3/http"
"github.com/molecula/featurebase/v3/encoding/proto"
"github.com/molecula/featurebase/v3/net"
"github.com/molecula/featurebase/v3/topology"
"github.com/pkg/errors"
@ -54,7 +54,7 @@ func pauseNode(t *testing.T, node string) error {
}
type keyInserter struct {
client *http.InternalClient
client *pilosa.InternalClient
uri *net.URI
index string
keys []string
@ -69,10 +69,10 @@ func getAddress(node string) string {
return node + ":10101"
}
func getClients(addrs []string) ([]*http.InternalClient, error) {
clients := make([]*http.InternalClient, 0, len(addrs))
func getClients(addrs []string) ([]*pilosa.InternalClient, error) {
clients := make([]*pilosa.InternalClient, 0, len(addrs))
for _, addr := range addrs {
c, err := http.NewInternalClient(addr, http.GetHTTPClient(nil))
c, err := pilosa.NewInternalClient(addr, pilosa.GetHTTPClient(nil), pilosa.WithSerializer(proto.Serializer{}))
if err != nil {
return nil, err
}
@ -93,7 +93,7 @@ func getURIsFromAddresses(addrs []string) ([]*net.URI, error) {
return uris, nil
}
func readIndexTranslateData(ctx context.Context, client *http.InternalClient, dirPath, index string, partition int) error {
func readIndexTranslateData(ctx context.Context, client *pilosa.InternalClient, dirPath, index string, partition int) error {
// read translateStore contents from endpoint
r, err := client.IndexTranslateDataReader(ctx, index, partition)
if err != nil {
@ -177,7 +177,7 @@ var errOpRetriable = errors.New("If operation failed on this error, it can be re
func verifyNodeHasGivenKeys(ctx context.Context, node, index, dirPath string, keys []string) error {
// get client that's connected to node
address := getAddress(node)
client, err := http.NewInternalClient(address, http.GetHTTPClient(nil))
client, err := pilosa.NewInternalClient(address, pilosa.GetHTTPClient(nil), pilosa.WithSerializer(proto.Serializer{}))
if err != nil {
return err
}

272
http/client.go → internal_client.go
View file

@ -1,5 +1,5 @@
// Copyright 2021 Molecula Corp. All rights reserved.
package http
// Copyright 2022 Molecula Corp. All rights reserved.
package pilosa
import (
"bytes"
@ -20,9 +20,7 @@ import (
"time"
"github.com/hashicorp/go-retryablehttp"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/authn"
"github.com/molecula/featurebase/v3/encoding/proto"
"github.com/molecula/featurebase/v3/ingest"
"github.com/molecula/featurebase/v3/logger"
pnet "github.com/molecula/featurebase/v3/net"
@ -34,7 +32,7 @@ import (
// InternalClient represents a client to the Pilosa cluster.
type InternalClient struct {
defaultURI *pnet.URI
serializer pilosa.Serializer
serializer Serializer
log logger.Logger
@ -42,7 +40,7 @@ type InternalClient struct {
httpClient *http.Client
retryableClient *retryablehttp.Client
// the local node's API, used for operations that we can short-circuit that way
api *pilosa.API
api *API
// secret Key for auth across nodes
secretKey string
@ -53,7 +51,7 @@ type InternalClient struct {
// of going through http.
func NewInternalClient(host string, remoteClient *http.Client, opts ...InternalClientOption) (*InternalClient, error) {
if host == "" {
return nil, pilosa.ErrHostRequired
return nil, ErrHostRequired
}
uri, err := pnet.NewURIFromAddress(host)
@ -67,6 +65,12 @@ func NewInternalClient(host string, remoteClient *http.Client, opts ...InternalC
type InternalClientOption func(c *InternalClient)
func WithSerializer(s Serializer) InternalClientOption {
return func(c *InternalClient) {
c.serializer = s
}
}
// WithSecretKey adds the secretKey used for inter-node communication when auth
// is enabled
func WithSecretKey(secretKey string) InternalClientOption {
@ -122,7 +126,6 @@ func retryWith400Policy(ctx context.Context, resp *http.Response, err error) (bo
func NewInternalClientFromURI(defaultURI *pnet.URI, remoteClient *http.Client, opts ...InternalClientOption) *InternalClient {
ic := &InternalClient{
defaultURI: defaultURI,
serializer: proto.Serializer{},
httpClient: remoteClient,
log: logger.NewStandardLogger(os.Stderr),
}
@ -167,7 +170,7 @@ func (c *InternalClient) maxShardByIndex(ctx context.Context) (map[string]uint64
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req.Header.Set("Accept", "application/json")
req = AddAuthToken(ctx, req)
@ -200,7 +203,7 @@ func (c *InternalClient) AvailableShards(ctx context.Context, indexName string)
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req.Header.Set("Accept", "application/json")
req = AddAuthToken(ctx, req)
@ -220,7 +223,7 @@ func (c *InternalClient) AvailableShards(ctx context.Context, indexName string)
// SchemaNode returns all index and field schema information from the specified
// node.
func (c *InternalClient) SchemaNode(ctx context.Context, uri *pnet.URI, views bool) ([]*pilosa.IndexInfo, error) {
func (c *InternalClient) SchemaNode(ctx context.Context, uri *pnet.URI, views bool) ([]*IndexInfo, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Schema")
defer span.Finish()
@ -234,7 +237,7 @@ func (c *InternalClient) SchemaNode(ctx context.Context, uri *pnet.URI, views bo
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req.Header.Set("Accept", "application/json")
req = AddAuthToken(ctx, req)
@ -253,7 +256,7 @@ func (c *InternalClient) SchemaNode(ctx context.Context, uri *pnet.URI, views bo
}
// Schema returns all index and field schema information.
func (c *InternalClient) Schema(ctx context.Context) ([]*pilosa.IndexInfo, error) {
func (c *InternalClient) Schema(ctx context.Context) ([]*IndexInfo, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Schema")
defer span.Finish()
@ -266,7 +269,7 @@ func (c *InternalClient) Schema(ctx context.Context) ([]*pilosa.IndexInfo, error
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req.Header.Set("Accept", "application/json")
req = AddAuthToken(ctx, req)
@ -304,7 +307,7 @@ func (c *InternalClient) IngestSchema(ctx context.Context, uri *pnet.URI, buf []
req.Header.Set("Content-Length", strconv.Itoa(len(buf)))
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "application/json")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req = AddAuthToken(ctx, req)
resp, err := c.executeRequest(req.WithContext(ctx), giveRawResponse(true))
@ -320,7 +323,7 @@ func (c *InternalClient) IngestSchema(ctx context.Context, uri *pnet.URI, buf []
var msg string
// try to decode a JSON response
var sr successResponse
qr := &pilosa.QueryResponse{}
qr := &QueryResponse{}
if err = json.Unmarshal(buf, &sr); err == nil {
msg = sr.Error.Error()
} else if err := c.serializer.Unmarshal(buf, qr); err == nil {
@ -355,7 +358,7 @@ func (c *InternalClient) IngestOperations(ctx context.Context, uri *pnet.URI, in
req.Header.Set("Content-Length", strconv.Itoa(len(buf)))
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "application/json")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req = AddAuthToken(ctx, req)
resp, err := c.executeRequest(req.WithContext(ctx))
@ -388,7 +391,7 @@ func (c *InternalClient) IngestNodeOperations(ctx context.Context, uri *pnet.URI
req.Header.Set("Content-Length", strconv.Itoa(len(buf)))
req.Header.Set("Content-Type", "application/x-protobuf")
req.Header.Set("Accept", "application/x-protobuf")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req = AddAuthToken(ctx, req)
resp, err := c.executeRequest(req.WithContext(ctx))
@ -417,7 +420,7 @@ func (c *InternalClient) MutexCheck(ctx context.Context, uri *pnet.URI, indexNam
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("Accept", "application/json")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req = AddAuthToken(ctx, req)
resp, err := c.executeRequest(req.WithContext(ctx))
@ -434,7 +437,7 @@ func (c *InternalClient) MutexCheck(ctx context.Context, uri *pnet.URI, indexNam
return out, err
}
func (c *InternalClient) PostSchema(ctx context.Context, uri *pnet.URI, s *pilosa.Schema, remote bool) error {
func (c *InternalClient) PostSchema(ctx context.Context, uri *pnet.URI, s *Schema, remote bool) error {
u := uri.Path(fmt.Sprintf("/schema?remote=%v", remote))
buf, err := json.Marshal(s)
if err != nil {
@ -448,7 +451,7 @@ func (c *InternalClient) PostSchema(ctx context.Context, uri *pnet.URI, s *pilos
req.Header.Set("Content-Length", strconv.Itoa(len(buf)))
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "application/json")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req = AddAuthToken(ctx, req)
resp, err := c.executeRequest(req.WithContext(ctx))
@ -463,7 +466,7 @@ func (c *InternalClient) PostSchema(ctx context.Context, uri *pnet.URI, s *pilos
}
// CreateIndex creates a new index on the server.
func (c *InternalClient) CreateIndex(ctx context.Context, index string, opt pilosa.IndexOptions) error {
func (c *InternalClient) CreateIndex(ctx context.Context, index string, opt IndexOptions) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.CreateIndex")
defer span.Finish()
@ -495,14 +498,14 @@ func (c *InternalClient) CreateIndex(ctx context.Context, index string, opt pilo
req.Header.Set("Content-Length", strconv.Itoa(len(buf)))
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "application/json")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req = AddAuthToken(ctx, req)
// Execute request against the host.
resp, err := c.executeRequest(req.WithContext(ctx))
if err != nil {
if resp != nil && resp.StatusCode == http.StatusConflict {
return pilosa.ErrIndexExists
return ErrIndexExists
}
return err
}
@ -524,7 +527,7 @@ func (c *InternalClient) FragmentNodes(ctx context.Context, index string, shard
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req.Header.Set("Accept", "application/json")
req = AddAuthToken(ctx, req)
@ -556,7 +559,7 @@ func (c *InternalClient) Nodes(ctx context.Context) ([]*topology.Node, error) {
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req.Header.Set("Accept", "application/json")
req = AddAuthToken(ctx, req)
@ -575,21 +578,21 @@ func (c *InternalClient) Nodes(ctx context.Context) ([]*topology.Node, error) {
}
// Query executes query against the index.
func (c *InternalClient) Query(ctx context.Context, index string, queryRequest *pilosa.QueryRequest) (*pilosa.QueryResponse, error) {
func (c *InternalClient) Query(ctx context.Context, index string, queryRequest *QueryRequest) (*QueryResponse, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Query")
defer span.Finish()
return c.QueryNode(ctx, c.defaultURI, index, queryRequest)
}
// QueryNode executes query against the index, sending the request to the node specified.
func (c *InternalClient) QueryNode(ctx context.Context, uri *pnet.URI, index string, queryRequest *pilosa.QueryRequest) (*pilosa.QueryResponse, error) {
func (c *InternalClient) QueryNode(ctx context.Context, uri *pnet.URI, index string, queryRequest *QueryRequest) (*QueryResponse, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "QueryNode")
defer span.Finish()
if index == "" {
return nil, pilosa.ErrIndexRequired
return nil, ErrIndexRequired
} else if queryRequest.Query == "" {
return nil, pilosa.ErrQueryRequired
return nil, ErrQueryRequired
}
buf, err := c.serializer.Marshal(queryRequest)
if err != nil {
@ -615,7 +618,7 @@ func (c *InternalClient) QueryNode(ctx context.Context, uri *pnet.URI, index str
req.Header.Set("Content-Type", "application/x-protobuf")
req.Header.Set("Accept", "application/x-protobuf")
req.Header.Set("X-Pilosa-Row", "roaring")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
// Execute request against the host.
resp, err := c.executeRequest(req.WithContext(ctx))
@ -630,7 +633,7 @@ func (c *InternalClient) QueryNode(ctx context.Context, uri *pnet.URI, index str
return nil, errors.Wrap(err, "reading")
}
qresp := &pilosa.QueryResponse{}
qresp := &QueryResponse{}
if err := c.serializer.Unmarshal(body, qresp); err != nil {
return nil, fmt.Errorf("unmarshal response: %s", err)
} else if qresp.Err != nil {
@ -649,12 +652,12 @@ func getPrimaryNode(nodes []*topology.Node) *topology.Node {
return nil
}
func (c *InternalClient) EnsureIndex(ctx context.Context, name string, options pilosa.IndexOptions) error {
func (c *InternalClient) EnsureIndex(ctx context.Context, name string, options IndexOptions) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.EnsureIndex")
defer span.Finish()
err := c.CreateIndex(ctx, name, options)
if err == nil || errors.Cause(err) == pilosa.ErrIndexExists {
if err == nil || errors.Cause(err) == ErrIndexExists {
return nil
}
return err
@ -663,21 +666,21 @@ func (c *InternalClient) EnsureIndex(ctx context.Context, name string, options p
func (c *InternalClient) EnsureField(ctx context.Context, indexName string, fieldName string) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.EnsureField")
defer span.Finish()
return c.EnsureFieldWithOptions(ctx, indexName, fieldName, pilosa.FieldOptions{})
return c.EnsureFieldWithOptions(ctx, indexName, fieldName, FieldOptions{})
}
func (c *InternalClient) EnsureFieldWithOptions(ctx context.Context, indexName string, fieldName string, opt pilosa.FieldOptions) error {
func (c *InternalClient) EnsureFieldWithOptions(ctx context.Context, indexName string, fieldName string, opt FieldOptions) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.EnsureFieldWithOptions")
defer span.Finish()
err := c.CreateFieldWithOptions(ctx, indexName, fieldName, opt)
if err == nil || errors.Cause(err) == pilosa.ErrFieldExists {
if err == nil || errors.Cause(err) == ErrFieldExists {
return nil
}
return err
}
// importNode sends a pre-marshaled import request to a node.
func (c *InternalClient) importNode(ctx context.Context, node *topology.Node, index, field string, buf []byte, opts *pilosa.ImportOptions) error {
func (c *InternalClient) importNode(ctx context.Context, node *topology.Node, index, field string, buf []byte, opts *ImportOptions) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.importNode")
defer span.Finish()
@ -702,7 +705,7 @@ func (c *InternalClient) importNode(ctx context.Context, node *topology.Node, in
req.Header.Set("Content-Type", "application/x-protobuf")
req.Header.Set("Accept", "application/x-protobuf")
req.Header.Set("X-Pilosa-Row", "roaring")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req = AddAuthToken(ctx, req)
// Execute request against the host.
@ -718,7 +721,7 @@ func (c *InternalClient) importNode(ctx context.Context, node *topology.Node, in
return errors.Wrap(err, "reading")
}
var isresp pilosa.ImportResponse
var isresp ImportResponse
if err := c.serializer.Unmarshal(body, &isresp); err != nil {
return fmt.Errorf("unmarshal import response: %s", err)
} else if s := isresp.Err; s != "" {
@ -736,7 +739,7 @@ func (c *InternalClient) importNode(ctx context.Context, node *topology.Node, in
// that in here with a type switch seems messy. Similarly, index/field/shard
// exist because we can't access those members of the two slightly different
// structs.
func (c *InternalClient) importHelper(ctx context.Context, req pilosa.Message, process func() error, index string, field string, shard uint64, options *pilosa.ImportOptions) error {
func (c *InternalClient) importHelper(ctx context.Context, req Message, process func() error, index string, field string, shard uint64, options *ImportOptions) error {
// If we don't actually know what shards we're sending to, and we have
// a local API and a qcx, we'll have a process function that uses the local
// API. Otherwise, even if we have an API
@ -846,7 +849,7 @@ func (c *InternalClient) importHelper(ctx context.Context, req pilosa.Message, p
//
// If we get a non-nil qcx, and have an associated API, we'll use that API
// directly for the local shard.
func (c *InternalClient) Import(ctx context.Context, qcx *pilosa.Qcx, req *pilosa.ImportRequest, options *pilosa.ImportOptions) error {
func (c *InternalClient) Import(ctx context.Context, qcx *Qcx, req *ImportRequest, options *ImportOptions) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Import")
defer span.Finish()
@ -874,7 +877,7 @@ func (c *InternalClient) Import(ctx context.Context, qcx *pilosa.Qcx, req *pilos
//
// If we get a non-nil qcx, and have an associated API, we'll use that API
// directly for the local shard.
func (c *InternalClient) ImportValue(ctx context.Context, qcx *pilosa.Qcx, req *pilosa.ImportValueRequest, options *pilosa.ImportOptions) error {
func (c *InternalClient) ImportValue(ctx context.Context, qcx *Qcx, req *ImportValueRequest, options *ImportOptions) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Import")
defer span.Finish()
@ -892,14 +895,14 @@ func (c *InternalClient) ImportValue(ctx context.Context, qcx *pilosa.Qcx, req *
// ImportRoaring does fast import of raw bits in roaring format (pilosa or
// official format, see API.ImportRoaring).
func (c *InternalClient) ImportRoaring(ctx context.Context, uri *pnet.URI, index, field string, shard uint64, remote bool, req *pilosa.ImportRoaringRequest) error {
func (c *InternalClient) ImportRoaring(ctx context.Context, uri *pnet.URI, index, field string, shard uint64, remote bool, req *ImportRoaringRequest) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.ImportRoaring")
defer span.Finish()
if index == "" {
return pilosa.ErrIndexRequired
return ErrIndexRequired
} else if field == "" {
return pilosa.ErrFieldRequired
return ErrFieldRequired
}
if uri == nil {
uri = c.defaultURI
@ -923,7 +926,7 @@ func (c *InternalClient) ImportRoaring(ctx context.Context, uri *pnet.URI, index
httpReq.Header.Set("Content-Type", "application/x-protobuf")
httpReq.Header.Set("Accept", "application/x-protobuf")
httpReq.Header.Set("X-Pilosa-Row", "roaring")
httpReq.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
httpReq.Header.Set("User-Agent", "pilosa/"+Version)
httpReq = AddAuthToken(ctx, httpReq)
// Execute request against the host.
@ -934,7 +937,7 @@ func (c *InternalClient) ImportRoaring(ctx context.Context, uri *pnet.URI, index
defer resp.Body.Close()
dec := json.NewDecoder(resp.Body)
rbody := &pilosa.ImportResponse{}
rbody := &ImportResponse{}
err = dec.Decode(rbody)
// Decode can return EOF when no error occurred. helpful!
if err != nil && err != io.EOF {
@ -952,9 +955,9 @@ func (c *InternalClient) ExportCSV(ctx context.Context, index, field string, sha
defer span.Finish()
if index == "" {
return pilosa.ErrIndexRequired
return ErrIndexRequired
} else if field == "" {
return pilosa.ErrFieldRequired
return ErrFieldRequired
}
// Retrieve a list of nodes that own the shard.
@ -998,7 +1001,7 @@ func (c *InternalClient) exportNodeCSV(ctx context.Context, node *topology.Node,
return errors.Wrap(err, "creating request")
}
req.Header.Set("Accept", "text/csv")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req = AddAuthToken(ctx, req)
// Execute request against the host.
@ -1041,14 +1044,14 @@ func (c *InternalClient) RetrieveShardFromURI(ctx context.Context, index, field,
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req = AddAuthToken(ctx, req)
// Execute request.
resp, err := c.executeRequest(req.WithContext(ctx))
if err != nil {
if resp != nil && resp.StatusCode == http.StatusNotFound {
return nil, pilosa.ErrFragmentNotFound
return nil, ErrFragmentNotFound
}
return nil, err
}
@ -1059,19 +1062,19 @@ func (c *InternalClient) RetrieveShardFromURI(ctx context.Context, index, field,
func (c *InternalClient) CreateField(ctx context.Context, index, field string) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.CreateField")
defer span.Finish()
return c.CreateFieldWithOptions(ctx, index, field, pilosa.FieldOptions{})
return c.CreateFieldWithOptions(ctx, index, field, FieldOptions{})
}
// CreateFieldWithOptions creates a new field on the server.
func (c *InternalClient) CreateFieldWithOptions(ctx context.Context, index, field string, opt pilosa.FieldOptions) error {
func (c *InternalClient) CreateFieldWithOptions(ctx context.Context, index, field string, opt FieldOptions) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.CreateFieldWithOptions")
defer span.Finish()
if index == "" {
return pilosa.ErrIndexRequired
return ErrIndexRequired
}
// convert pilosa.FieldOptions to fieldOptions
// convert FieldOptions to fieldOptions
//
// TODO this kind of sucks because it's one more place that needs
// changes when we change anything with field options (and there
@ -1082,23 +1085,23 @@ func (c *InternalClient) CreateFieldWithOptions(ctx context.Context, index, fiel
Type: opt.Type,
}
switch fieldOpt.Type {
case pilosa.FieldTypeSet, pilosa.FieldTypeMutex:
case FieldTypeSet, FieldTypeMutex:
fieldOpt.CacheType = &opt.CacheType
fieldOpt.CacheSize = &opt.CacheSize
fieldOpt.Keys = &opt.Keys
case pilosa.FieldTypeInt:
case FieldTypeInt:
fieldOpt.Min = &opt.Min
fieldOpt.Max = &opt.Max
case pilosa.FieldTypeTime:
case FieldTypeTime:
fieldOpt.TimeQuantum = &opt.TimeQuantum
case pilosa.FieldTypeBool:
case FieldTypeBool:
// pass
case pilosa.FieldTypeDecimal:
case FieldTypeDecimal:
fieldOpt.Min = &opt.Min
fieldOpt.Max = &opt.Max
fieldOpt.Scale = &opt.Scale
default:
fieldOpt.Type = pilosa.DefaultFieldType
fieldOpt.Type = DefaultFieldType
fieldOpt.Keys = &opt.Keys
}
@ -1131,14 +1134,14 @@ func (c *InternalClient) CreateFieldWithOptions(ctx context.Context, index, fiel
req.Header.Set("Content-Length", strconv.Itoa(len(buf)))
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "application/json")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req = AddAuthToken(ctx, req)
// Execute request against the host.
resp, err := c.executeRequest(req.WithContext(ctx))
if err != nil {
if resp != nil && resp.StatusCode == http.StatusConflict {
return pilosa.ErrFieldExists
return ErrFieldExists
}
return err
}
@ -1148,7 +1151,7 @@ func (c *InternalClient) CreateFieldWithOptions(ctx context.Context, index, fiel
// FragmentBlocks returns a list of block checksums for a fragment on a host.
// Only returns blocks which contain data.
func (c *InternalClient) FragmentBlocks(ctx context.Context, uri *pnet.URI, index, field, view string, shard uint64) ([]pilosa.FragmentBlock, error) {
func (c *InternalClient) FragmentBlocks(ctx context.Context, uri *pnet.URI, index, field, view string, shard uint64) ([]FragmentBlock, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.FragmentBlocks")
defer span.Finish()
@ -1169,7 +1172,7 @@ func (c *InternalClient) FragmentBlocks(ctx context.Context, uri *pnet.URI, inde
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req.Header.Set("Accept", "application/json")
req = AddAuthToken(ctx, req)
@ -1178,7 +1181,7 @@ func (c *InternalClient) FragmentBlocks(ctx context.Context, uri *pnet.URI, inde
if err != nil {
// Return the appropriate error.
if resp != nil && resp.StatusCode == http.StatusNotFound {
return nil, pilosa.ErrFragmentNotFound
return nil, ErrFragmentNotFound
}
return nil, err
}
@ -1200,7 +1203,7 @@ func (c *InternalClient) BlockData(ctx context.Context, uri *pnet.URI, index, fi
if uri == nil {
panic("need to pass a URI to BlockData")
}
buf, err := c.serializer.Marshal(&pilosa.BlockDataRequest{
buf, err := c.serializer.Marshal(&BlockDataRequest{
Index: index,
Field: field,
View: view,
@ -1220,7 +1223,7 @@ func (c *InternalClient) BlockData(ctx context.Context, uri *pnet.URI, index, fi
req.Header.Set("Content-Length", strconv.Itoa(len(buf)))
req.Header.Set("Accept", "application/protobuf")
req.Header.Set("X-Pilosa-Row", "roaring")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req = AddAuthToken(ctx, req)
resp, err := c.executeRequest(req.WithContext(ctx))
@ -1233,7 +1236,7 @@ func (c *InternalClient) BlockData(ctx context.Context, uri *pnet.URI, index, fi
defer resp.Body.Close()
// Decode response object.
var rsp pilosa.BlockDataResponse
var rsp BlockDataResponse
if body, err := ioutil.ReadAll(resp.Body); err != nil {
return nil, nil, errors.Wrap(err, "reading")
} else if err := c.serializer.Unmarshal(body, &rsp); err != nil {
@ -1254,7 +1257,7 @@ func (c *InternalClient) SendMessage(ctx context.Context, uri *pnet.URI, msg []b
}
req.Header.Set("Content-Type", "application/x-protobuf")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req.Header.Set("Accept", "application/json")
req.Header.Set("Connection", "keep-alive")
if c.secretKey != "" {
@ -1272,16 +1275,16 @@ func (c *InternalClient) SendMessage(ctx context.Context, uri *pnet.URI, msg []b
}
// TranslateKeysNode function is mainly called to translate keys from primary node.
// If primary node returns 404 error the function wraps it with pilosa.ErrTranslatingKeyNotFound.
// If primary node returns 404 error the function wraps it with ErrTranslatingKeyNotFound.
func (c *InternalClient) TranslateKeysNode(ctx context.Context, uri *pnet.URI, index, field string, keys []string, writable bool) ([]uint64, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "TranslateKeysNode")
defer span.Finish()
if index == "" {
return nil, pilosa.ErrIndexRequired
return nil, ErrIndexRequired
}
buf, err := c.serializer.Marshal(&pilosa.TranslateKeysRequest{
buf, err := c.serializer.Marshal(&TranslateKeysRequest{
Index: index,
Field: field,
Keys: keys,
@ -1302,14 +1305,14 @@ func (c *InternalClient) TranslateKeysNode(ctx context.Context, uri *pnet.URI, i
req.Header.Set("Content-Type", "application/x-protobuf")
req.Header.Set("Accept", "application/x-protobuf")
req.Header.Set("X-Pilosa-Row", "roaring")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req = AddAuthToken(ctx, req)
// Execute request against the host.
resp, err := c.executeRequest(req.WithContext(ctx))
if err != nil {
if resp != nil && resp.StatusCode == http.StatusNotFound {
return nil, errors.Wrap(pilosa.ErrTranslatingKeyNotFound, err.Error())
return nil, errors.Wrap(ErrTranslatingKeyNotFound, err.Error())
}
return nil, err
}
@ -1321,7 +1324,7 @@ func (c *InternalClient) TranslateKeysNode(ctx context.Context, uri *pnet.URI, i
return nil, errors.Wrap(err, "reading")
}
tkresp := &pilosa.TranslateKeysResponse{}
tkresp := &TranslateKeysResponse{}
if err := c.serializer.Unmarshal(body, tkresp); err != nil {
return nil, fmt.Errorf("unmarshal response: %s", err)
}
@ -1334,10 +1337,10 @@ func (c *InternalClient) TranslateIDsNode(ctx context.Context, uri *pnet.URI, in
defer span.Finish()
if index == "" {
return nil, pilosa.ErrIndexRequired
return nil, ErrIndexRequired
}
buf, err := c.serializer.Marshal(&pilosa.TranslateIDsRequest{
buf, err := c.serializer.Marshal(&TranslateIDsRequest{
Index: index,
Field: field,
IDs: ids,
@ -1357,7 +1360,7 @@ func (c *InternalClient) TranslateIDsNode(ctx context.Context, uri *pnet.URI, in
req.Header.Set("Content-Type", "application/x-protobuf")
req.Header.Set("Accept", "application/x-protobuf")
req.Header.Set("X-Pilosa-Row", "roaring")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req = AddAuthToken(ctx, req)
// Execute request against the host.
@ -1373,7 +1376,7 @@ func (c *InternalClient) TranslateIDsNode(ctx context.Context, uri *pnet.URI, in
return nil, errors.Wrap(err, "reading")
}
tkresp := &pilosa.TranslateIDsResponse{}
tkresp := &TranslateIDsResponse{}
if err := c.serializer.Unmarshal(body, tkresp); err != nil {
return nil, fmt.Errorf("unmarshal response: %s", err)
}
@ -1381,7 +1384,7 @@ func (c *InternalClient) TranslateIDsNode(ctx context.Context, uri *pnet.URI, in
}
// GetNodeUsage retrieves the size-on-disk information for the specified node.
func (c *InternalClient) GetNodeUsage(ctx context.Context, uri *pnet.URI) (map[string]pilosa.NodeUsage, error) {
func (c *InternalClient) GetNodeUsage(ctx context.Context, uri *pnet.URI) (map[string]NodeUsage, error) {
u := uri.Path("/ui/usage?remote=true")
req, err := http.NewRequest("GET", u, nil)
if err != nil {
@ -1389,7 +1392,7 @@ func (c *InternalClient) GetNodeUsage(ctx context.Context, uri *pnet.URI) (map[s
}
req.Header.Set("Accept", "application/json")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req = AddAuthToken(ctx, req)
// Execute request against the host.
@ -1405,7 +1408,7 @@ func (c *InternalClient) GetNodeUsage(ctx context.Context, uri *pnet.URI) (map[s
return nil, errors.Wrap(err, "reading")
}
nodeUsages := make(map[string]pilosa.NodeUsage) // map of size 1
nodeUsages := make(map[string]NodeUsage) // map of size 1
if err := json.Unmarshal(body, &nodeUsages); err != nil {
return nil, fmt.Errorf("unmarshal response: %s", err)
}
@ -1413,7 +1416,7 @@ func (c *InternalClient) GetNodeUsage(ctx context.Context, uri *pnet.URI) (map[s
}
// GetPastQueries retrieves the query history log for the specified node.
func (c *InternalClient) GetPastQueries(ctx context.Context, uri *pnet.URI) ([]pilosa.PastQueryStatus, error) {
func (c *InternalClient) GetPastQueries(ctx context.Context, uri *pnet.URI) ([]PastQueryStatus, error) {
u := uri.Path("/query-history?remote=true")
req, err := http.NewRequest("GET", u, nil)
if err != nil {
@ -1421,7 +1424,7 @@ func (c *InternalClient) GetPastQueries(ctx context.Context, uri *pnet.URI) ([]p
}
req.Header.Set("Accept", "application/json")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req = AddAuthToken(ctx, req)
// Execute request against the host.
@ -1437,7 +1440,7 @@ func (c *InternalClient) GetPastQueries(ctx context.Context, uri *pnet.URI) ([]p
return nil, errors.Wrap(err, "reading")
}
queries := make([]pilosa.PastQueryStatus, 100)
queries := make([]PastQueryStatus, 100)
if err := json.Unmarshal(body, &queries); err != nil {
return nil, fmt.Errorf("unmarshal response: %s", err)
}
@ -1463,7 +1466,7 @@ func (c *InternalClient) FindIndexKeysNode(ctx context.Context, uri *pnet.URI, i
req.Header.Set("Content-Length", strconv.Itoa(len(reqData)))
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "application/json")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req = AddAuthToken(ctx, req)
// Send the request.
@ -1512,7 +1515,7 @@ func (c *InternalClient) FindFieldKeysNode(ctx context.Context, uri *pnet.URI, i
req.Header.Set("Content-Length", strconv.Itoa(len(reqData)))
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "application/json")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req = AddAuthToken(ctx, req)
// Send the request.
@ -1562,7 +1565,7 @@ func (c *InternalClient) CreateIndexKeysNode(ctx context.Context, uri *pnet.URI,
req.Header.Set("Content-Length", strconv.Itoa(len(reqData)))
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "application/json")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req = AddAuthToken(ctx, req)
// Send the request.
@ -1615,7 +1618,7 @@ func (c *InternalClient) CreateFieldKeysNode(ctx context.Context, uri *pnet.URI,
req.Header.Set("Content-Length", strconv.Itoa(len(reqData)))
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "application/json")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req = AddAuthToken(ctx, req)
// Send the request.
@ -1660,7 +1663,7 @@ func (c *InternalClient) MatchFieldKeysNode(ctx context.Context, uri *pnet.URI,
// Apply headers.
req.Header.Set("Content-Length", strconv.Itoa(len(like)))
req.Header.Set("Accept", "application/json")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req = AddAuthToken(ctx, req)
// Send the request.
@ -1690,7 +1693,7 @@ func (c *InternalClient) MatchFieldKeysNode(ctx context.Context, uri *pnet.URI,
return matches, nil
}
func (c *InternalClient) Transactions(ctx context.Context) (map[string]*pilosa.Transaction, error) {
func (c *InternalClient) Transactions(ctx context.Context) (map[string]*Transaction, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Transactions")
defer span.Finish()
@ -1700,7 +1703,7 @@ func (c *InternalClient) Transactions(ctx context.Context) (map[string]*pilosa.T
return nil, errors.Wrap(err, "creating transactions request")
}
req.Header.Set("Accept", "application/json")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req = AddAuthToken(ctx, req)
resp, err := c.executeRequest(req.WithContext(ctx))
@ -1711,15 +1714,15 @@ func (c *InternalClient) Transactions(ctx context.Context) (map[string]*pilosa.T
_, _ = io.Copy(ioutil.Discard, resp.Body)
_ = resp.Body.Close()
}()
trnsMap := make(map[string]*pilosa.Transaction)
trnsMap := make(map[string]*Transaction)
err = json.NewDecoder(resp.Body).Decode(&trnsMap)
return trnsMap, errors.Wrap(err, "json decoding")
}
func (c *InternalClient) StartTransaction(ctx context.Context, id string, timeout time.Duration, exclusive bool) (*pilosa.Transaction, error) {
func (c *InternalClient) StartTransaction(ctx context.Context, id string, timeout time.Duration, exclusive bool) (*Transaction, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.StartTransaction")
defer span.Finish()
buf, err := json.Marshal(&pilosa.Transaction{
buf, err := json.Marshal(&Transaction{
ID: id,
Timeout: timeout,
Exclusive: exclusive,
@ -1739,7 +1742,7 @@ func (c *InternalClient) StartTransaction(ctx context.Context, id string, timeou
req.Header.Set("Content-Length", strconv.Itoa(len(buf)))
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "application/json")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req = AddAuthToken(ctx, req)
resp, err := c.executeRequest(req.WithContext(ctx), giveRawResponse(true))
@ -1756,14 +1759,14 @@ func (c *InternalClient) StartTransaction(ctx context.Context, id string, timeou
return nil, errors.Wrap(err, "decoding response")
}
if resp.StatusCode == 409 {
err = pilosa.ErrTransactionExclusive
err = ErrTransactionExclusive
} else if tr.Error != "" {
err = errors.New(tr.Error)
}
return tr.Transaction, err
}
func (c *InternalClient) FinishTransaction(ctx context.Context, id string) (*pilosa.Transaction, error) {
func (c *InternalClient) FinishTransaction(ctx context.Context, id string) (*Transaction, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.FinishTransaction")
defer span.Finish()
@ -1774,7 +1777,7 @@ func (c *InternalClient) FinishTransaction(ctx context.Context, id string) (*pil
}
req.Header.Set("Accept", "application/json")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req = AddAuthToken(ctx, req)
resp, err := c.executeRequest(req.WithContext(ctx), giveRawResponse(true))
@ -1797,7 +1800,7 @@ func (c *InternalClient) FinishTransaction(ctx context.Context, id string) (*pil
return tr.Transaction, err
}
func (c *InternalClient) GetTransaction(ctx context.Context, id string) (*pilosa.Transaction, error) {
func (c *InternalClient) GetTransaction(ctx context.Context, id string) (*Transaction, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.GetTransaction")
defer span.Finish()
@ -1811,7 +1814,7 @@ func (c *InternalClient) GetTransaction(ctx context.Context, id string) (*pilosa
return nil, errors.Wrap(err, "creating get transaction request")
}
req.Header.Set("Accept", "application/json")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req = AddAuthToken(ctx, req)
resp, err := c.executeRequest(req.WithContext(ctx), giveRawResponse(true))
@ -1922,7 +1925,7 @@ func (c *InternalClient) handleResponse(req *http.Request, eo *executeOpts, resp
var msg string
// try to decode a JSON response
var sr successResponse
qr := &pilosa.QueryResponse{}
qr := &QueryResponse{}
if err = json.Unmarshal(buf, &sr); err == nil {
msg = sr.Error.Error()
} else if err := c.serializer.Unmarshal(buf, qr); err == nil {
@ -1936,7 +1939,7 @@ func (c *InternalClient) handleResponse(req *http.Request, eo *executeOpts, resp
}
// Bits is a slice of Bit.
type Bits []pilosa.Bit
type Bits []Bit
func (p Bits) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
func (p Bits) Len() int { return len(p) }
@ -2029,10 +2032,10 @@ func (p Bits) Timestamps() []int64 {
}
// GroupByShard returns a map of bits by shard.
func (p Bits) GroupByShard() map[uint64][]pilosa.Bit {
m := make(map[uint64][]pilosa.Bit)
func (p Bits) GroupByShard() map[uint64][]Bit {
m := make(map[uint64][]Bit)
for _, bit := range p {
shard := bit.ColumnID / pilosa.ShardWidth
shard := bit.ColumnID / ShardWidth
m[shard] = append(m[shard], bit)
}
@ -2045,7 +2048,7 @@ func (p Bits) GroupByShard() map[uint64][]pilosa.Bit {
}
// FieldValues represents a slice of field values.
type FieldValues []pilosa.FieldValue
type FieldValues []FieldValue
func (p FieldValues) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
func (p FieldValues) Len() int { return len(p) }
@ -2098,10 +2101,10 @@ func (p FieldValues) Values() []int64 {
}
// GroupByShard returns a map of field values by shard.
func (p FieldValues) GroupByShard() map[uint64][]pilosa.FieldValue {
m := make(map[uint64][]pilosa.FieldValue)
func (p FieldValues) GroupByShard() map[uint64][]FieldValue {
m := make(map[uint64][]FieldValue)
for _, val := range p {
shard := val.ColumnID / pilosa.ShardWidth
shard := val.ColumnID / ShardWidth
m[shard] = append(m[shard], val)
}
@ -2114,7 +2117,7 @@ func (p FieldValues) GroupByShard() map[uint64][]pilosa.FieldValue {
}
// BitsByPos is a slice of bits sorted row then column.
type BitsByPos []pilosa.Bit
type BitsByPos []Bit
func (p BitsByPos) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
func (p BitsByPos) Len() int { return len(p) }
@ -2126,11 +2129,6 @@ func (p BitsByPos) Less(i, j int) bool {
return p0 < p1
}
// pos returns the row position of a row/column pair.
func pos(rowID, columnID uint64) uint64 {
return (rowID * pilosa.ShardWidth) + (columnID % pilosa.ShardWidth)
}
func uriPathToURL(uri *pnet.URI, path string) url.URL {
return url.URL{
Scheme: uri.Scheme,
@ -2170,14 +2168,14 @@ func (c *InternalClient) RetrieveTranslatePartitionFromURI(ctx context.Context,
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req = AddAuthToken(ctx, req)
// Execute request.
resp, err := c.executeRequest(req.WithContext(ctx))
if err != nil {
if resp != nil && resp.StatusCode == http.StatusNotFound {
return nil, pilosa.ErrFragmentNotFound
return nil, ErrFragmentNotFound
}
return nil, err
}
@ -2189,7 +2187,7 @@ func (c *InternalClient) ImportIndexKeys(ctx context.Context, uri *pnet.URI, ind
defer span.Finish()
if index == "" {
return pilosa.ErrIndexRequired
return ErrIndexRequired
}
if uri == nil {
@ -2205,7 +2203,7 @@ func (c *InternalClient) ImportIndexKeys(ctx context.Context, uri *pnet.URI, ind
if err != nil {
return errors.Wrap(err, "creating request")
}
httpReq.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
httpReq.Header.Set("User-Agent", "pilosa/"+Version)
token, ok := ctx.Value("token").(string)
if ok && token != "" {
httpReq.Header.Set("Authorization", token)
@ -2225,7 +2223,7 @@ func (c *InternalClient) ImportFieldKeys(ctx context.Context, uri *pnet.URI, ind
defer span.Finish()
if index == "" {
return pilosa.ErrIndexRequired
return ErrIndexRequired
}
if uri == nil {
@ -2241,7 +2239,7 @@ func (c *InternalClient) ImportFieldKeys(ctx context.Context, uri *pnet.URI, ind
if err != nil {
return errors.Wrap(err, "creating request")
}
httpReq.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
httpReq.Header.Set("User-Agent", "pilosa/"+Version)
token, ok := ctx.Value("token").(string)
if ok && token != "" {
@ -2271,7 +2269,7 @@ func (c *InternalClient) ShardReader(ctx context.Context, index string, shard ui
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req.Header.Set("Accept", "application/octet-stream")
req = AddAuthToken(ctx, req)
@ -2294,7 +2292,7 @@ func (c *InternalClient) IDAllocDataReader(ctx context.Context) (io.ReadCloser,
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req.Header.Set("Accept", "application/octet-stream")
req = AddAuthToken(ctx, req)
@ -2318,7 +2316,7 @@ func (c *InternalClient) IDAllocDataWriter(ctx context.Context, f io.Reader, pri
return errors.Wrap(err, "creating request")
}
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req.Header.Set("Accept", "application/octet-stream")
req = AddAuthToken(ctx, req)
@ -2345,7 +2343,7 @@ func (c *InternalClient) IndexTranslateDataReader(ctx context.Context, index str
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req.Header.Set("Accept", "application/octet-stream")
req = AddAuthToken(ctx, req)
@ -2353,7 +2351,7 @@ func (c *InternalClient) IndexTranslateDataReader(ctx context.Context, index str
resp, err := c.executeRequest(req.WithContext(ctx), forwardAuthHeader(true))
if resp != nil && resp.StatusCode == http.StatusNotFound {
resp.Body.Close()
return nil, pilosa.ErrTranslateStoreNotFound
return nil, ErrTranslateStoreNotFound
} else if err != nil {
return nil, err
}
@ -2375,7 +2373,7 @@ func (c *InternalClient) FieldTranslateDataReader(ctx context.Context, index, fi
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req.Header.Set("Accept", "application/octet-stream")
req = AddAuthToken(ctx, req)
@ -2383,16 +2381,14 @@ func (c *InternalClient) FieldTranslateDataReader(ctx context.Context, index, fi
resp, err := c.executeRequest(req.WithContext(ctx))
if resp != nil && resp.StatusCode == http.StatusNotFound {
resp.Body.Close()
return nil, pilosa.ErrTranslateStoreNotFound
return nil, ErrTranslateStoreNotFound
} else if err != nil {
return nil, err
}
return resp.Body, nil
}
// Status function is just a public function for this particular implementation of InternalClient.
// It's not require by pilosa.InternalClient interface.
// The function returns pilosa cluster state as a string ("NORMAL", "DEGRADED", "DOWN", "RESIZING", ...)
// Status returns pilosa cluster state as a string ("NORMAL", "DEGRADED", "DOWN", "RESIZING", ...)
func (c *InternalClient) Status(ctx context.Context) (string, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Status")
defer span.Finish()
@ -2406,7 +2402,7 @@ func (c *InternalClient) Status(ctx context.Context) (string, error) {
return "", errors.Wrap(err, "creating request")
}
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req.Header.Set("Accept", "application/json")
req = AddAuthToken(ctx, req)
@ -2438,7 +2434,7 @@ func (c *InternalClient) PartitionNodes(ctx context.Context, partitionID int) ([
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("User-Agent", "pilosa/"+Version)
req.Header.Set("Accept", "application/json")
req = AddAuthToken(ctx, req)
@ -2456,6 +2452,6 @@ func (c *InternalClient) PartitionNodes(ctx context.Context, partitionID int) ([
return a, nil
}
func (c *InternalClient) SetInternalAPI(api *pilosa.API) {
func (c *InternalClient) SetInternalAPI(api *API) {
c.api = api
}

48
http/client_test.go → internal_client_test.go
View file

@ -1,5 +1,5 @@
// Copyright 2021 Molecula Corp. All rights reserved.
package http_test
package pilosa_test
import (
"bufio"
@ -15,7 +15,7 @@ import (
"github.com/davecgh/go-spew/spew"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/http"
"github.com/molecula/featurebase/v3/encoding/proto"
"github.com/molecula/featurebase/v3/pql"
"github.com/molecula/featurebase/v3/server"
"github.com/molecula/featurebase/v3/test"
@ -122,9 +122,9 @@ func TestClient_MultiNode(t *testing.T) {
// Connect to each node to compare results.
client := make([]*Client, 3)
client[0] = MustNewClient(c.GetNode(0).URL(), http.GetHTTPClient(nil))
client[1] = MustNewClient(c.GetNode(1).URL(), http.GetHTTPClient(nil))
client[2] = MustNewClient(c.GetNode(2).URL(), http.GetHTTPClient(nil))
client[0] = MustNewClient(c.GetNode(0).URL(), pilosa.GetHTTPClient(nil))
client[1] = MustNewClient(c.GetNode(1).URL(), pilosa.GetHTTPClient(nil))
client[2] = MustNewClient(c.GetNode(2).URL(), pilosa.GetHTTPClient(nil))
topN := 4
queryRequest := &pilosa.QueryRequest{
@ -188,7 +188,7 @@ func TestClient_Export(t *testing.T) {
cmd.MustCreateField(t, "unkeyed", "keyedf", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 1000), pilosa.OptFieldKeys())
cmd.MustCreateField(t, "unkeyed", "unkeyedf", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 1000))
c := MustNewClient(host, http.GetHTTPClient(nil))
c := MustNewClient(host, pilosa.GetHTTPClient(nil))
data := []pilosa.Bit{
{RowID: 1, ColumnID: 100, RowKey: "row1", ColumnKey: "col100"},
{RowID: 1, ColumnID: 101, RowKey: "row1", ColumnKey: "col101"},
@ -376,7 +376,7 @@ func TestClient_Import(t *testing.T) {
recIDs := []uint64{0, 3, 7}
valueIDs := []uint64{0, 3, 7}
c := MustNewClient(host, http.GetHTTPClient(nil))
c := MustNewClient(host, pilosa.GetHTTPClient(nil))
// set API to point at the local node
c.SetInternalAPI(cmd.API)
@ -532,7 +532,7 @@ func TestClient_ImportRoaring(t *testing.T) {
// Send import request.
host := cluster.GetNode(0).URL()
c := MustNewClient(host, http.GetHTTPClient(nil))
c := MustNewClient(host, pilosa.GetHTTPClient(nil))
// [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 65537]
roaringReq := makeImportRoaringRequest(false, "3B3001000100000900010000000100010009000100")
if err := c.ImportRoaring(context.Background(), &cluster.GetNode(0).API.Node().URI, "i", "f", 0, false, roaringReq); err != nil {
@ -656,7 +656,7 @@ func TestClient_ImportRoaring_MultiView(t *testing.T) {
// Send import request.
host := cluster.GetNode(0).URL()
c := MustNewClient(host, http.GetHTTPClient(nil))
c := MustNewClient(host, pilosa.GetHTTPClient(nil))
req := &pilosa.ImportRoaringRequest{Views: map[string][]byte{}}
req.Views["a"], _ = hex.DecodeString("3B3001000100000900010000000100010009000100")
req.Views["b"], _ = hex.DecodeString("3B3001000100000900010000000100010009000100")
@ -681,7 +681,7 @@ func TestClient_ImportKeys(t *testing.T) {
cmd.MustCreateField(t, "unkeyed", "keyedf", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 1000), pilosa.OptFieldKeys())
// Send import request.
c := MustNewClient(host, http.GetHTTPClient(nil))
c := MustNewClient(host, pilosa.GetHTTPClient(nil))
baseReq := &pilosa.ImportRequest{
Index: "keyed",
Field: "keyedf",
@ -774,8 +774,8 @@ func TestClient_ImportKeys(t *testing.T) {
cmd0.MustCreateField(t, "keyed", "keyedf1", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 1000), pilosa.OptFieldKeys())
// Send import request.
c0 := MustNewClient(host0, http.GetHTTPClient(nil))
c1 := MustNewClient(host1, http.GetHTTPClient(nil))
c0 := MustNewClient(host0, pilosa.GetHTTPClient(nil))
c1 := MustNewClient(host1, pilosa.GetHTTPClient(nil))
// Import to node0.
t.Run("Import node0", func(t *testing.T) {
@ -852,7 +852,7 @@ func TestClient_ImportKeys(t *testing.T) {
}
// Send import request.
c := MustNewClient(host, http.GetHTTPClient(nil))
c := MustNewClient(host, pilosa.GetHTTPClient(nil))
req := &pilosa.ImportValueRequest{
Index: "i",
Field: "f",
@ -931,7 +931,7 @@ func TestClient_ImportIDs(t *testing.T) {
}
// Send import request.
c := MustNewClient(host, http.GetHTTPClient(nil))
c := MustNewClient(host, pilosa.GetHTTPClient(nil))
req := &pilosa.ImportValueRequest{
Index: idxName,
Field: fldName,
@ -999,7 +999,7 @@ func TestClient_ImportValue(t *testing.T) {
}
// Send import request.
c := MustNewClient(host, http.GetHTTPClient(nil))
c := MustNewClient(host, pilosa.GetHTTPClient(nil))
req := &pilosa.ImportValueRequest{
Index: "i",
Field: "f",
@ -1078,7 +1078,7 @@ func TestClient_ImportExistence(t *testing.T) {
}
// Send import request.
c := MustNewClient(host, http.GetHTTPClient(nil))
c := MustNewClient(host, pilosa.GetHTTPClient(nil))
req := &pilosa.ImportRequest{
Index: "iset",
Field: "fset",
@ -1114,7 +1114,7 @@ func TestClient_ImportExistence(t *testing.T) {
}
// Send import request.
c := MustNewClient(host, http.GetHTTPClient(nil))
c := MustNewClient(host, pilosa.GetHTTPClient(nil))
req := &pilosa.ImportValueRequest{
Index: "iint",
Field: "fint",
@ -1155,7 +1155,7 @@ func TestClient_FragmentBlocks(t *testing.T) {
// Set a bit on a different shard.
hldr.SetBit("i", "f", 0, 1)
c := MustNewClient(cmd.URL(), http.GetHTTPClient(nil))
c := MustNewClient(cmd.URL(), pilosa.GetHTTPClient(nil))
blocks, err := c.FragmentBlocks(context.Background(), nil, "i", "f", "standard", 0)
if err != nil {
t.Fatal(err)
@ -1180,7 +1180,7 @@ func TestClient_CreateDecimalField(t *testing.T) {
defer cluster.Close()
cmd := cluster.GetNode(0)
c := MustNewClient(cmd.URL(), http.GetHTTPClient(nil))
c := MustNewClient(cmd.URL(), pilosa.GetHTTPClient(nil))
index := "cdf"
err := c.CreateIndex(context.Background(), index, pilosa.IndexOptions{})
@ -1290,8 +1290,8 @@ func TestClientTransactions(t *testing.T) {
coord := c.GetPrimary()
other := c.GetNonPrimary()
client0 := MustNewClient(coord.URL(), http.GetHTTPClient(nil))
client1 := MustNewClient(other.URL(), http.GetHTTPClient(nil))
client0 := MustNewClient(coord.URL(), pilosa.GetHTTPClient(nil))
client1 := MustNewClient(other.URL(), pilosa.GetHTTPClient(nil))
// can create, list, get, and finish a transaction
var expDeadline time.Time
@ -1444,12 +1444,12 @@ func TestClientTransactions(t *testing.T) {
// Client represents a test wrapper for pilosa.Client.
type Client struct {
*http.InternalClient
*pilosa.InternalClient
}
// MustNewClient returns a new instance of Client. Panic on error.
func MustNewClient(host string, h *gohttp.Client) *Client {
c, err := http.NewInternalClient(host, h)
c, err := pilosa.NewInternalClient(host, h, pilosa.WithSerializer(proto.Serializer{}))
if err != nil {
panic(err)
}
@ -1497,7 +1497,7 @@ func TestClient_ImportRoaringExists(t *testing.T) {
}
// Send import request.
host := node.URL()
c := MustNewClient(host, http.GetHTTPClient(nil))
c := MustNewClient(host, pilosa.GetHTTPClient(nil))
// [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 65537]
roaringReq := makeImportRoaringRequest(false, "3B3001000100000900010000000100010009000100")

8
server.go
View file

@ -86,7 +86,7 @@ type Server struct { // nolint: maligned
// HolderConfig stashes server options that are really Holder options.
holderConfig *HolderConfig
defaultClient InternalClient
defaultClient *InternalClient
dataDir string
// Threshold for logging long-running queries
@ -193,7 +193,7 @@ func OptServerGCNotifier(gcn GCNotifier) ServerOption {
// OptServerInternalClient is a functional option on Server
// used to set the implementation of InternalClient.
func OptServerInternalClient(c InternalClient) ServerOption {
func OptServerInternalClient(c *InternalClient) ServerOption {
return func(s *Server) error {
s.defaultClient = c
s.cluster.InternalClient = c
@ -405,7 +405,7 @@ func NewServer(opts ...ServerOption) (*Server, error) {
cluster: cluster,
diagnostics: newDiagnosticsCollector(defaultDiagnosticServer),
systemInfo: newNopSystemInfo(),
defaultClient: nopInternalClient{},
defaultClient: &InternalClient{}, // TODO may need to make this a valid thing
gcNotifier: NopGCNotifier,
@ -511,7 +511,7 @@ func NewServer(opts ...ServerOption) (*Server, error) {
return s, nil
}
func (s *Server) InternalClient() InternalClient {
func (s *Server) InternalClient() *InternalClient {
return s.defaultClient
}

15
server/handler_test.go
View file

@ -21,7 +21,6 @@ import (
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/boltdb"
"github.com/molecula/featurebase/v3/encoding/proto"
"github.com/molecula/featurebase/v3/http"
"github.com/molecula/featurebase/v3/pql"
"github.com/molecula/featurebase/v3/server"
"github.com/molecula/featurebase/v3/test"
@ -31,7 +30,7 @@ func TestHandler_PostSchemaCluster(t *testing.T) {
cluster := test.MustRunCluster(t, 3)
defer cluster.Close()
cmd := cluster.GetNode(0)
h := cmd.Handler.(*http.Handler).Handler
h := cmd.Handler.(*pilosa.Handler).Handler
t.Run("PostSchema", func(t *testing.T) {
w := httptest.NewRecorder()
@ -70,7 +69,7 @@ func TestHandler_Endpoints(t *testing.T) {
cluster := test.MustRunCluster(t, 1)
defer cluster.Close()
cmd := cluster.GetNode(0)
h := cmd.Handler.(*http.Handler).Handler
h := cmd.Handler.(*pilosa.Handler).Handler
holder := cmd.Server.Holder()
hldr := test.Holder{Holder: holder}
@ -1120,7 +1119,7 @@ func TestHandler_Endpoints(t *testing.T) {
clus := test.MustRunCluster(t, 1, []server.CommandOption{test.OptAllowedOrigins([]string{"http://test/"})})
defer clus.Close()
w = httptest.NewRecorder()
h1 := clus.GetNode(0).Handler.(*http.Handler).Handler
h1 := clus.GetNode(0).Handler.(*pilosa.Handler).Handler
h1.ServeHTTP(w, req)
result = w.Result()
@ -1383,7 +1382,7 @@ func TestHandler_Endpoints(t *testing.T) {
clus := test.MustRunCluster(t, 1, []server.CommandOption{test.OptAllowedOrigins([]string{"http://test/"})})
defer clus.Close()
w = httptest.NewRecorder()
h := clus.GetNode(0).Handler.(*http.Handler).Handler
h := clus.GetNode(0).Handler.(*pilosa.Handler).Handler
h.ServeHTTP(w, req)
result = w.Result()
@ -1402,7 +1401,7 @@ func TestCluster_TranslateStore(t *testing.T) {
cluster.Nodes[0] = test.NewCommandNode(t,
server.OptCommandServerOptions(
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderWithLockerFunc(nil, &sync.Mutex{})),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderWithLockerFunc(nil, &sync.Mutex{})),
),
)
@ -1423,7 +1422,7 @@ func TestClusterTranslator(t *testing.T) {
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderWithLockerFunc(nil, &sync.Mutex{})),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderWithLockerFunc(nil, &sync.Mutex{})),
)},
[]server.CommandOption{
server.OptCommandServerOptions(
@ -1487,7 +1486,7 @@ func TestClusterTranslator(t *testing.T) {
// defer cluster.Close()
// cmd := cluster.GetNode(0)
// h := cmd.Handler.(*http.Handler).Handler
// h := cmd.Handler.(*pilosa.Handler).Handler
// w := httptest.NewRecorder()

35
server/server.go
View file

@ -36,7 +36,6 @@ import (
petcd "github.com/molecula/featurebase/v3/etcd"
"github.com/molecula/featurebase/v3/gcnotify"
"github.com/molecula/featurebase/v3/gopsutil"
"github.com/molecula/featurebase/v3/http"
"github.com/molecula/featurebase/v3/logger"
pnet "github.com/molecula/featurebase/v3/net"
"github.com/molecula/featurebase/v3/prometheus"
@ -74,7 +73,7 @@ type Command struct {
logger loggerLogger
queryLogger loggerLogger
Handler pilosa.Handler
Handler pilosa.HandlerI
grpcServer *grpcServer
grpcLn net.Listener
API *pilosa.API
@ -405,7 +404,7 @@ func (m *Command) SetupServer() error {
// Save listenURI for later reference.
m.listenURI = uri
c := http.GetHTTPClient(m.tlsConfig)
c := pilosa.GetHTTPClient(m.tlsConfig)
// Get advertise address as uri.
advertiseURI, err := pilosa.AddressWithDefaults(m.Config.Advertise)
@ -473,7 +472,7 @@ func (m *Command) SetupServer() error {
pilosa.OptServerDiagnosticsInterval(diagnosticsInterval),
pilosa.OptServerExecutorPoolSize(m.Config.WorkerPoolSize),
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderWithLockerFunc(c, &sync.Mutex{})),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderWithLockerFunc(c, &sync.Mutex{})),
pilosa.OptServerOpenIDAllocator(pilosa.OpenIDAllocator),
pilosa.OptServerLogger(m.logger),
pilosa.OptServerQueryLogger(m.queryLogger),
@ -498,9 +497,9 @@ func (m *Command) SetupServer() error {
serverOptions = append(serverOptions, m.serverOptions...)
if m.Config.Auth.Enable {
serverOptions = append(serverOptions, pilosa.OptServerInternalClient(http.NewInternalClientFromURI(uri, c, http.WithSecretKey(m.Config.Auth.SecretKey))))
serverOptions = append(serverOptions, pilosa.OptServerInternalClient(pilosa.NewInternalClientFromURI(uri, c, pilosa.WithSecretKey(m.Config.Auth.SecretKey), pilosa.WithSerializer(proto.Serializer{}))))
} else {
serverOptions = append(serverOptions, pilosa.OptServerInternalClient(http.NewInternalClientFromURI(uri, c)))
serverOptions = append(serverOptions, pilosa.OptServerInternalClient(pilosa.NewInternalClientFromURI(uri, c, pilosa.WithSerializer(proto.Serializer{}))))
}
m.Server, err = pilosa.NewServer(serverOptions...)
@ -573,17 +572,19 @@ func (m *Command) SetupServer() error {
OptGRPCServerQueryLogger(m.queryLogger),
)
m.Handler, err = http.NewHandler(
http.OptHandlerAllowedOrigins(m.Config.Handler.AllowedOrigins),
http.OptHandlerAPI(m.API),
http.OptHandlerLogger(m.logger),
http.OptHandlerQueryLogger(m.queryLogger),
http.OptHandlerFileSystem(&statik.FileSystem{}),
http.OptHandlerListener(m.ln, m.Config.Advertise),
http.OptHandlerCloseTimeout(m.closeTimeout),
http.OptHandlerMiddleware(m.grpcServer.middleware(m.Config.Handler.AllowedOrigins)),
http.OptHandlerAuthN(m.auth),
http.OptHandlerAuthZ(&p),
m.Handler, err = pilosa.NewHandler(
pilosa.OptHandlerAllowedOrigins(m.Config.Handler.AllowedOrigins),
pilosa.OptHandlerAPI(m.API),
pilosa.OptHandlerLogger(m.logger),
pilosa.OptHandlerQueryLogger(m.queryLogger),
pilosa.OptHandlerFileSystem(&statik.FileSystem{}),
pilosa.OptHandlerListener(m.ln, m.Config.Advertise),
pilosa.OptHandlerCloseTimeout(m.closeTimeout),
pilosa.OptHandlerMiddleware(m.grpcServer.middleware(m.Config.Handler.AllowedOrigins)),
pilosa.OptHandlerAuthN(m.auth),
pilosa.OptHandlerAuthZ(&p),
pilosa.OptHandlerSerializer(proto.Serializer{}),
pilosa.OptHandlerRoaringSerializer(proto.RoaringSerializer),
)
return errors.Wrap(err, "new handler")
}

6
server/server_test.go
View file

@ -19,7 +19,7 @@ import (
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/disco"
"github.com/molecula/featurebase/v3/http"
"github.com/molecula/featurebase/v3/encoding/proto"
"github.com/molecula/featurebase/v3/pql"
"github.com/molecula/featurebase/v3/roaring"
"github.com/molecula/featurebase/v3/server"
@ -54,7 +54,7 @@ func TestMain_Set_Quick(t *testing.T) {
defer m.Close()
// Create client.
client, err := http.NewInternalClient(m.API.Node().URI.HostPort(), http.GetHTTPClient(nil))
client, err := pilosa.NewInternalClient(m.API.Node().URI.HostPort(), pilosa.GetHTTPClient(nil), pilosa.WithSerializer(proto.Serializer{}))
client.SetInternalAPI(m.API)
if err != nil {
t.Fatal(err)
@ -904,7 +904,7 @@ func TestQueryingWithQuotesAndStuff(t *testing.T) {
m := test.RunCommand(t)
defer m.Close()
client, err := http.NewInternalClient(m.API.Node().URI.HostPort(), http.GetHTTPClient(nil))
client, err := pilosa.NewInternalClient(m.API.Node().URI.HostPort(), pilosa.GetHTTPClient(nil), pilosa.WithSerializer(proto.Serializer{}))
client.SetInternalAPI(m.API)
if err != nil {
t.Fatal(err)

5
stats/stats_test.go
View file

@ -9,8 +9,7 @@ import (
"testing"
"time"
"github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/http"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/logger"
"github.com/molecula/featurebase/v3/stats"
"github.com/molecula/featurebase/v3/test"
@ -143,7 +142,7 @@ func TestStatsCount_APICalls(t *testing.T) {
cluster := test.MustRunCluster(t, 1)
defer cluster.Close()
cmd := cluster.GetNode(0)
h := cmd.Handler.(*http.Handler).Handler
h := cmd.Handler.(*pilosa.Handler).Handler
holder := cmd.Server.Holder()
hldr := test.Holder{Holder: holder}

5
test/pilosa.go
View file

@ -16,7 +16,6 @@ import (
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/disco"
"github.com/molecula/featurebase/v3/encoding/proto"
"github.com/molecula/featurebase/v3/http"
"github.com/molecula/featurebase/v3/server"
"github.com/molecula/featurebase/v3/testhook"
)
@ -165,8 +164,8 @@ func (m *Command) IsPrimary() bool {
}
// Client returns a client to connect to the program.
func (m *Command) Client() *http.InternalClient {
return m.Server.InternalClient().(*http.InternalClient)
func (m *Command) Client() *pilosa.InternalClient {
return m.Server.InternalClient()
}
// Query executes a query against the program through the HTTP API.

23
translator_test.go
View file

@ -13,7 +13,6 @@ import (
"github.com/google/go-cmp/cmp"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/boltdb"
"github.com/molecula/featurebase/v3/http"
"github.com/molecula/featurebase/v3/mock"
"github.com/molecula/featurebase/v3/server"
"github.com/molecula/featurebase/v3/test"
@ -156,25 +155,25 @@ func TestTranslation_KeyNotFound(t *testing.T) {
server.OptCommandServerOptions(
pilosa.OptServerNodeID("node0"),
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderFunc(nil)),
)},
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerNodeID("node1"),
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderFunc(nil)),
)},
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerNodeID("node2"),
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderFunc(nil)),
)},
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerNodeID("node3"),
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderFunc(nil)),
)},
)
defer c.Close()
@ -312,19 +311,19 @@ func TestTranslation_Primary(t *testing.T) {
server.OptCommandServerOptions(
pilosa.OptServerNodeID("node0"),
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderFunc(nil)),
)},
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerNodeID("node1"),
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderFunc(nil)),
)},
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerNodeID("node2"),
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderFunc(nil)),
)},
)
defer c.Close()
@ -388,25 +387,25 @@ func TestTranslation_TranslateIDsOnCluster(t *testing.T) {
server.OptCommandServerOptions(
pilosa.OptServerNodeID("node0"),
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderFunc(nil)),
)},
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerNodeID("node1"),
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderFunc(nil)),
)},
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerNodeID("node2"),
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderFunc(nil)),
)},
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerNodeID("node3"),
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderFunc(nil)),
)},
)
defer c.Close()

3
tx_test.go
View file

@ -7,7 +7,6 @@ import (
"testing"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/http"
"github.com/molecula/featurebase/v3/server"
"github.com/molecula/featurebase/v3/test"
. "github.com/molecula/featurebase/v3/vprint" // nolint:staticcheck
@ -51,7 +50,7 @@ func TestAPI_ImportAtomicRecord(t *testing.T) {
server.OptCommandServerOptions(
pilosa.OptServerNodeID("node0"),
pilosa.OptServerClusterHasher(&offsetModHasher{}),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderFunc(nil)),
)},
)
defer c.Close()