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

Migrate HTTP handler and client into http subpackage.

Browse source
This commit is contained in:
Cody Soyland 2018-06-12 13:22:40 -05:00
parent 8d4cf1abf3
commit f2c104dfef
24 changed files with 2577 additions and 2387 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
cache.go
View file

@ -409,7 +409,7 @@ func (p Pairs) String() string {
return buf.String()
}
func encodePairs(a Pairs) []*internal.Pair {
func EncodePairs(a Pairs) []*internal.Pair {
other := make([]*internal.Pair, len(a))
for i := range a {
other[i] = encodePair(a[i])

1110
client.go
View file

File diff suppressed because it is too large Load diff

9
cluster.go
View file

@ -266,6 +266,8 @@ type Cluster struct {
//
RemoteClient *http.Client
InternalClient InternalClient
}
// NewCluster returns a new instance of Cluster with defaults.
@ -281,6 +283,8 @@ func NewCluster() *Cluster {
closing: make(chan struct{}),
joining: make(chan struct{}),
InternalClient: NewNopInternalClient(),
Logger: NopLogger,
}
}
@ -1230,9 +1234,6 @@ func (c *Cluster) FollowResizeInstruction(instr *internal.ResizeInstruction) err
return errors.Wrap(err, "applying schema")
}
// Create a client for calling remote nodes.
client := NewInternalHTTPClientFromURI(&c.Node.URI, c.RemoteClient) // TODO: ClientOptions
// Request each source file in ResizeSources.
for _, src := range instr.Sources {
c.Logger.Printf("get slice %d for index %s from host %s", src.Slice, src.Index, src.Node.URI)
@ -1259,7 +1260,7 @@ func (c *Cluster) FollowResizeInstruction(instr *internal.ResizeInstruction) err
// Stream slice from remote node.
c.Logger.Printf("retrieve slice %d for index %s from host %s", src.Slice, src.Index, src.Node.URI)
rd, err := client.RetrieveSliceFromURI(context.Background(), src.Index, src.Field, src.Slice, srcURI)
rd, err := c.InternalClient.RetrieveSliceFromURI(context.Background(), src.Index, src.Field, src.Slice, srcURI)
if err != nil {
// For now it is an acceptable error if the fragment is not found
// on the remote node. This occurs when a slice has been skipped and

6
ctl/common.go
View file

@ -17,7 +17,7 @@ package ctl
import (
"crypto/tls"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/http"
"github.com/pilosa/pilosa/server"
"github.com/pkg/errors"
"github.com/spf13/pflag"
@ -37,7 +37,7 @@ func SetTLSConfig(flags *pflag.FlagSet, certificatePath *string, certificateKeyP
}
// CommandClient returns a pilosa.InternalHTTPClient for the command
func CommandClient(cmd CommandWithTLSSupport) (*pilosa.InternalHTTPClient, error) {
func CommandClient(cmd CommandWithTLSSupport) (*http.InternalHTTPClient, error) {
tlsConfig := cmd.TLSConfiguration()
var TLSConfig *tls.Config
if tlsConfig.CertificatePath != "" && tlsConfig.CertificateKeyPath != "" {
@ -50,7 +50,7 @@ func CommandClient(cmd CommandWithTLSSupport) (*pilosa.InternalHTTPClient, error
InsecureSkipVerify: tlsConfig.SkipVerify,
}
}
client, err := pilosa.NewInternalHTTPClient(cmd.TLSHost(), server.GetHTTPClient(TLSConfig))
client, err := http.NewInternalHTTPClient(cmd.TLSHost(), http.GetHTTPClient(TLSConfig))
if err != nil {
return nil, errors.Wrap(err, "getting internal client")
}

9
ctl/import.go
View file

@ -26,6 +26,7 @@ import (
"time"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/http"
"github.com/pilosa/pilosa/server"
"github.com/pkg/errors"
)
@ -245,12 +246,12 @@ func (cmd *ImportCommand) importBits(ctx context.Context, bits []pilosa.Bit) err
// Group bits by slice.
logger.Printf("grouping %d bits", len(bits))
bitsBySlice := pilosa.Bits(bits).GroupBySlice()
bitsBySlice := http.Bits(bits).GroupBySlice()
// Parse path into bits.
for slice, chunk := range bitsBySlice {
if cmd.Sort {
sort.Sort(pilosa.BitsByPos(chunk))
sort.Sort(http.BitsByPos(chunk))
}
logger.Printf("importing slice: %d, n=%d", slice, len(chunk))
@ -439,12 +440,12 @@ func (cmd *ImportCommand) importValues(ctx context.Context, vals []pilosa.FieldV
// Group vals by slice.
logger.Printf("grouping %d vals", len(vals))
valsBySlice := pilosa.FieldValues(vals).GroupBySlice()
valsBySlice := http.FieldValues(vals).GroupBySlice()
// Parse path into FieldValues.
for slice, vals := range valsBySlice {
if cmd.Sort {
sort.Sort(pilosa.FieldValues(vals))
sort.Sort(http.FieldValues(vals))
}
logger.Printf("importing slice: %d, n=%d", slice, len(vals))

31
executor.go
View file

@ -17,7 +17,6 @@ package pilosa
import (
"context"
"fmt"
"net/http"
"sort"
"time"
@ -47,19 +46,35 @@ type Executor struct {
Cluster *Cluster
// Client used for remote requests.
client InternalClient
client InternalQueryClient
// Maximum number of SetBit() or ClearBit() commands per request.
MaxWritesPerRequest int
}
// NewExecutor returns a new instance of Executor.
func NewExecutor(remoteClient *http.Client) *Executor {
return &Executor{
client: NewInternalHTTPClientFromURI(nil, remoteClient),
type ExecutorOpt func(e *Executor) error
func ExecutorOptInternalQueryClient(c InternalQueryClient) ExecutorOpt {
return func(e *Executor) error {
e.client = c
return nil
}
}
// NewExecutor returns a new instance of Executor.
func NewExecutor(opts ...ExecutorOpt) *Executor {
e := &Executor{
client: NewNopInternalQueryClient(),
}
for _, opt := range opts {
err := opt(e)
if err != nil {
panic(err)
}
}
return e
}
// Execute executes a PQL query.
func (e *Executor) Execute(ctx context.Context, index string, q *pql.Query, slices []uint64, opt *ExecOptions) ([]interface{}, error) {
// Verify that an index is set.
@ -1380,7 +1395,7 @@ func (e *Executor) remoteExec(ctx context.Context, node *Node, index string, q *
case "SetRowAttrs":
case "SetColumnAttrs":
default:
v, err = decodeRow(pb.Results[i].GetRow()), nil
v, err = DecodeRow(pb.Results[i].GetRow()), nil
}
if err != nil {
return nil, err
@ -1619,7 +1634,7 @@ func (vc *ValCount) Add(other ValCount) ValCount {
}
}
func encodeValCount(vc ValCount) *internal.ValCount {
func EncodeValCount(vc ValCount) *internal.ValCount {
return &internal.ValCount{
Val: vc.Val,
Count: vc.Count,

15
fragment.go
View file

@ -1782,8 +1782,7 @@ func (s *FragmentSyncer) SyncFragment() error {
}
// Retrieve remote blocks.
client := NewInternalHTTPClientFromURI(&node.URI, s.RemoteClient)
blocks, err := client.FragmentBlocks(context.Background(), s.Fragment.Index(), s.Fragment.Field(), s.Fragment.Slice())
blocks, err := s.Cluster.InternalClient.FragmentBlocks(context.Background(), nil, s.Fragment.Index(), s.Fragment.Field(), s.Fragment.Slice())
if err != nil && err != ErrFragmentNotFound {
return errors.Wrap(err, "getting blocks")
}
@ -1847,7 +1846,7 @@ func (s *FragmentSyncer) syncBlock(id int) error {
// Read pairs from each remote block.
var pairSets []PairSet
var clients []InternalClient
var uris []*URI
for _, node := range s.Cluster.SliceNodes(f.Index(), f.Slice()) {
if s.Node.ID == node.ID {
continue
@ -1858,11 +1857,11 @@ func (s *FragmentSyncer) syncBlock(id int) error {
return nil
}
client := NewInternalHTTPClientFromURI(&node.URI, s.RemoteClient)
clients = append(clients, client)
uri := &node.URI
uris = append(uris, uri)
// Only sync the standard block.
rowIDs, columnIDs, err := client.BlockData(context.Background(), f.Index(), f.Field(), f.Slice(), id)
rowIDs, columnIDs, err := s.Cluster.InternalClient.BlockData(context.Background(), &node.URI, f.Index(), f.Field(), f.Slice(), id)
if err != nil {
return errors.Wrap(err, "getting block")
}
@ -1885,7 +1884,7 @@ func (s *FragmentSyncer) syncBlock(id int) error {
}
// Write updates to remote blocks.
for i := 0; i < len(clients); i++ {
for i := 0; i < len(uris); i++ {
set, clear := sets[i], clears[i]
count := 0
@ -1924,7 +1923,7 @@ func (s *FragmentSyncer) syncBlock(id int) error {
Query: buffers[k].String(),
Remote: true,
}
_, err := clients[i].Query(context.Background(), f.Index(), queryRequest)
_, err := s.Cluster.InternalClient.QueryNode(context.Background(), uris[i], f.Index(), queryRequest)
if err != nil {
return errors.Wrap(err, "executing")
}

1187
handler.go
View file

File diff suppressed because it is too large Load diff

8
holder.go
View file

@ -662,11 +662,9 @@ func (s *HolderSyncer) syncIndex(index string) error {
// Sync with every other host.
for _, node := range Nodes(s.Cluster.Nodes).FilterID(s.Node.ID) {
client := NewInternalHTTPClientFromURI(&node.URI, s.RemoteClient)
// Retrieve attributes from differing blocks.
// Skip update and recomputation if no attributes have changed.
m, err := client.ColumnAttrDiff(context.Background(), index, blks)
m, err := s.Cluster.InternalClient.ColumnAttrDiff(context.Background(), &node.URI, index, blks)
if err != nil {
return errors.Wrap(err, "getting differing blocks")
} else if len(m) == 0 {
@ -708,11 +706,9 @@ func (s *HolderSyncer) syncField(index, name string) error {
// Sync with every other host.
for _, node := range Nodes(s.Cluster.Nodes).FilterID(s.Node.ID) {
client := NewInternalHTTPClientFromURI(&node.URI, s.RemoteClient)
// Retrieve attributes from differing blocks.
// Skip update and recomputation if no attributes have changed.
m, err := client.RowAttrDiff(context.Background(), index, name, blks)
m, err := s.Cluster.InternalClient.RowAttrDiff(context.Background(), &node.URI, index, name, blks)
if err == ErrFieldNotFound {
continue // field not created remotely yet, skip
} else if err != nil {

27
holder_test.go
View file

@ -24,8 +24,8 @@ import (
"testing"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/http"
"github.com/pilosa/pilosa/pql"
"github.com/pilosa/pilosa/server"
"github.com/pilosa/pilosa/test"
)
@ -360,8 +360,20 @@ func TestHolder_DeleteIndex(t *testing.T) {
// Ensure holder can sync with a remote holder.
func TestHolderSyncer_SyncHolder(t *testing.T) {
s := test.NewServer()
defer s.Close()
uri, err := pilosa.NewURIFromAddress(s.URL)
if err != nil {
t.Fatal(err)
}
cluster := test.NewCluster(2)
client := server.GetHTTPClient(nil)
client := http.GetHTTPClient(nil)
httpClient := http.NewInternalHTTPClientFromURI(uri, client)
cluster.InternalClient = httpClient
cluster.RemoteClient = client
// Create a local holder.
hldr0 := test.MustOpenHolder()
defer hldr0.Close()
@ -369,11 +381,9 @@ func TestHolderSyncer_SyncHolder(t *testing.T) {
// Create a remote holder wrapped by an HTTP
hldr1 := test.MustOpenHolder()
defer hldr1.Close()
s := test.NewServer()
defer s.Close()
s.Handler.API.Holder = hldr1.Holder
s.Handler.Executor.ExecuteFn = func(ctx context.Context, index string, query *pql.Query, slices []uint64, opt *pilosa.ExecOptions) ([]interface{}, error) {
e := pilosa.NewExecutor(client)
e := pilosa.NewExecutor(pilosa.ExecutorOptInternalQueryClient(httpClient))
e.Holder = hldr1.Holder
e.Node = cluster.Nodes[1]
e.Cluster = cluster
@ -383,11 +393,6 @@ func TestHolderSyncer_SyncHolder(t *testing.T) {
// Mock 2-node, fully replicated cluster.
cluster.ReplicaN = 2
uri, err := pilosa.NewURIFromAddress(s.URL)
if err != nil {
t.Fatal(err)
}
cluster.Nodes[0].URI = test.NewURIFromHostPort("localhost", 0)
cluster.Nodes[1].URI = *uri
@ -445,7 +450,7 @@ func TestHolderSyncer_SyncHolder(t *testing.T) {
Holder: hldr0.Holder,
Node: cluster.Nodes[0],
Cluster: cluster,
RemoteClient: server.GetHTTPClient(nil),
RemoteClient: http.GetHTTPClient(nil),
Stats: pilosa.NopStatsClient,
}

1034
http/client.go Normal file
View file

File diff suppressed because it is too large Load diff

27
client_test.go → http/client_test.go
View file

@ -12,20 +12,20 @@
// See the License for the specific language governing permissions and
// limitations under the License.
package pilosa_test
package http_test
import (
"context"
"fmt"
"net/http"
gohttp "net/http"
"reflect"
"testing"
"github.com/davecgh/go-spew/spew"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/http"
"github.com/pilosa/pilosa/internal"
"github.com/pilosa/pilosa/pql"
"github.com/pilosa/pilosa/server"
"github.com/pilosa/pilosa/test"
)
@ -43,10 +43,10 @@ func createCluster(c *pilosa.Cluster) ([]*test.Server, []*test.Holder) {
return server, hldr
}
var defaultClient *http.Client
var defaultClient *gohttp.Client
func init() {
defaultClient = server.GetHTTPClient(nil)
defaultClient = http.GetHTTPClient(nil)
}
@ -61,21 +61,24 @@ func TestClient_MultiNode(t *testing.T) {
}
s[0].Handler.Executor.ExecuteFn = func(ctx context.Context, index string, query *pql.Query, slices []uint64, opt *pilosa.ExecOptions) ([]interface{}, error) {
e := pilosa.NewExecutor(defaultClient)
httpClient := http.NewInternalHTTPClientFromURI(&cluster.Nodes[0].URI, defaultClient)
e := pilosa.NewExecutor(pilosa.ExecutorOptInternalQueryClient(httpClient))
e.Holder = hldr[0].Holder
e.Node = cluster.Nodes[0]
e.Cluster = cluster
return e.Execute(ctx, index, query, slices, opt)
}
s[1].Handler.Executor.ExecuteFn = func(ctx context.Context, index string, query *pql.Query, slices []uint64, opt *pilosa.ExecOptions) ([]interface{}, error) {
e := pilosa.NewExecutor(defaultClient)
httpClient := http.NewInternalHTTPClientFromURI(&cluster.Nodes[0].URI, defaultClient)
e := pilosa.NewExecutor(pilosa.ExecutorOptInternalQueryClient(httpClient))
e.Holder = hldr[1].Holder
e.Node = cluster.Nodes[1]
e.Cluster = cluster
return e.Execute(ctx, index, query, slices, opt)
}
s[2].Handler.Executor.ExecuteFn = func(ctx context.Context, index string, query *pql.Query, slices []uint64, opt *pilosa.ExecOptions) ([]interface{}, error) {
e := pilosa.NewExecutor(defaultClient)
httpClient := http.NewInternalHTTPClientFromURI(&cluster.Nodes[0].URI, defaultClient)
e := pilosa.NewExecutor(pilosa.ExecutorOptInternalQueryClient(httpClient))
e.Holder = hldr[2].Holder
e.Node = cluster.Nodes[2]
e.Cluster = cluster
@ -98,9 +101,9 @@ func TestClient_MultiNode(t *testing.T) {
}
}
baseBit0 := SliceWidth * sliceNums[0]
baseBit1 := SliceWidth * sliceNums[1]
baseBit2 := SliceWidth * sliceNums[2]
baseBit0 := pilosa.SliceWidth * sliceNums[0]
baseBit1 := pilosa.SliceWidth * sliceNums[1]
baseBit2 := pilosa.SliceWidth * sliceNums[2]
maxSlice := uint64(0)
for _, x := range sliceNums {
@ -336,7 +339,7 @@ func TestClient_FragmentBlocks(t *testing.T) {
// Retrieve blocks.
c := test.MustNewClient(s.Host(), defaultClient)
blocks, err := c.FragmentBlocks(context.Background(), "i", "f", 0)
blocks, err := c.FragmentBlocks(context.Background(), nil, "i", "f", 0)
if err != nil {
t.Fatal(err)
} else if len(blocks) != 2 {

1231
http/handler.go Normal file
View file

File diff suppressed because it is too large Load diff

10
handler_internal_test.go → http/handler_internal_test.go
View file

@ -12,12 +12,14 @@
// See the License for the specific language governing permissions and
// limitations under the License.
package pilosa
package http
import (
"encoding/json"
"reflect"
"testing"
"github.com/pilosa/pilosa"
)
// Test custom UnmarshalJSON for postIndexRequest object
@ -27,7 +29,7 @@ func TestPostIndexRequestUnmarshalJSON(t *testing.T) {
expected postIndexRequest
err string
}{
{json: `{"options": {}}`, expected: postIndexRequest{Options: IndexOptions{}}},
{json: `{"options": {}}`, expected: postIndexRequest{Options: pilosa.IndexOptions{}}},
{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"},
@ -62,12 +64,12 @@ func TestPostFieldRequestUnmarshalJSON(t *testing.T) {
expected postFieldRequest
err string
}{
{json: `{"options": {}}`, expected: postFieldRequest{Options: FieldOptions{}}},
{json: `{"options": {}}`, expected: postFieldRequest{Options: pilosa.FieldOptions{}}},
{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"},
{json: `{"options": {"inverseEnabled": true}}`, err: "Unknown key: inverseEnabled:true"},
{json: `{"options": {"cacheType": "type"}}`, expected: postFieldRequest{Options: FieldOptions{CacheType: "type"}}},
{json: `{"options": {"cacheType": "type"}}`, expected: postFieldRequest{Options: pilosa.FieldOptions{CacheType: "type"}}},
{json: `{"options": {"inverse": true, "cacheType": "type"}}`, err: "Unknown key: inverse:true"},
}
for _, test := range tests {

101
handler_test.go → http/handler_test.go
View file

@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
package pilosa_test
package http_test
import (
"bytes"
@ -21,7 +21,7 @@ import (
"fmt"
"io"
"io/ioutil"
"net/http"
gohttp "net/http"
"net/http/httptest"
"reflect"
"strings"
@ -29,6 +29,7 @@ import (
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/http"
"github.com/pilosa/pilosa/internal"
"github.com/pilosa/pilosa/pql"
"github.com/pilosa/pilosa/test"
@ -49,7 +50,7 @@ func TestHandlerPanics(t *testing.T) {
if !bytes.Contains(bufbytes, []byte("PANIC: runtime error: invalid memory address or nil pointer dereference")) {
t.Fatalf("expected panic in log, but got: %s", bufbytes)
}
if w.Code != http.StatusInternalServerError {
if w.Code != gohttp.StatusInternalServerError {
t.Fatalf("expected internal server error, but got: %v", w.Code)
}
bodyBytes := w.Body.Bytes()
@ -69,7 +70,7 @@ func TestHandler_NotFound(t *testing.T) {
w := httptest.NewRecorder()
h.ServeHTTP(w, test.MustNewHTTPRequest("GET", "/no_such_path", nil))
if w.Code != http.StatusNotFound {
if w.Code != gohttp.StatusNotFound {
t.Fatalf("invalid status: %d", w.Code)
}
}
@ -101,7 +102,7 @@ func TestHandler_Schema(t *testing.T) {
h.API.Cluster = test.NewCluster(1)
w := httptest.NewRecorder()
h.ServeHTTP(w, test.MustNewHTTPRequest("GET", "/schema", nil))
if w.Code != http.StatusOK {
if w.Code != gohttp.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{"indexes":[{"name":"i0","fields":[{"name":"f0"},{"name":"f1","views":[{"name":"standard"}]}]},{"name":"i1","fields":[{"name":"f0","views":[{"name":"standard"}]}]}]}`+"\n" {
} else if body := w.Body.String(); body != `{"indexes":[{"name":"i0","fields":[{"name":"f0","options":{"cacheType":"ranked","cacheSize":50000}},{"name":"f1","options":{"cacheType":"ranked","cacheSize":50000},"views":[{"name":"standard"}]}]},{"name":"i1","fields":[{"name":"f0","options":{"cacheType":"ranked","cacheSize":50000},"views":[{"name":"standard"}]}]}]}`+"\n" {
@ -142,7 +143,7 @@ func TestHandler_Status(t *testing.T) {
w := httptest.NewRecorder()
h.ServeHTTP(w, test.MustNewHTTPRequest("GET", "/status", nil))
if w.Code != http.StatusOK {
if w.Code != gohttp.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{"state":"NORMAL","nodes":[{"id":"node0","uri":{"scheme":"http","host":"host0"},"isCoordinator":false}],"localID":"node0"}`+"\n" {
t.Fatalf("unexpected body: %s", body)
@ -156,9 +157,9 @@ func TestHandler_Info(t *testing.T) {
w := httptest.NewRecorder()
h.ServeHTTP(w, test.MustNewHTTPRequest("GET", "/info", nil))
if w.Code != http.StatusOK {
if w.Code != gohttp.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != fmt.Sprintf("{\"sliceWidth\":%d}\n", SliceWidth) {
} else if body := w.Body.String(); body != fmt.Sprintf("{\"sliceWidth\":%d}\n", pilosa.SliceWidth) {
t.Fatalf("unexpected body: %s", body)
}
}
@ -173,7 +174,7 @@ func TestHandler_ClusterResizeAbort(t *testing.T) {
w := httptest.NewRecorder()
h.ServeHTTP(w, test.MustNewHTTPRequest("POST", "/cluster/resize/abort", nil))
if w.Code != http.StatusOK {
if w.Code != gohttp.StatusOK {
bod, err := ioutil.ReadAll(w.Body)
t.Fatalf("unexpected status code: %d, bod: %s, readerr: %v", w.Code, bod, err)
} else if body := w.Body.String(); body != `{"info":"complete current job: no resize job currently running"}`+"\n" {
@ -188,20 +189,20 @@ func TestHandler_MaxSlices(t *testing.T) {
hldr := test.MustOpenHolder()
defer hldr.Close()
hldr.MustCreateFragmentIfNotExists("i0", "f0", pilosa.ViewStandard, 1).MustSetBits(30, (1*SliceWidth)+1)
hldr.MustCreateFragmentIfNotExists("i0", "f0", pilosa.ViewStandard, 1).MustSetBits(30, (1*SliceWidth)+2)
hldr.MustCreateFragmentIfNotExists("i0", "f0", pilosa.ViewStandard, 3).MustSetBits(30, (3*SliceWidth)+4)
hldr.MustCreateFragmentIfNotExists("i0", "f0", pilosa.ViewStandard, 1).MustSetBits(30, (1*pilosa.SliceWidth)+1)
hldr.MustCreateFragmentIfNotExists("i0", "f0", pilosa.ViewStandard, 1).MustSetBits(30, (1*pilosa.SliceWidth)+2)
hldr.MustCreateFragmentIfNotExists("i0", "f0", pilosa.ViewStandard, 3).MustSetBits(30, (3*pilosa.SliceWidth)+4)
hldr.MustCreateFragmentIfNotExists("i1", "f1", pilosa.ViewStandard, 0).MustSetBits(40, (0*SliceWidth)+1)
hldr.MustCreateFragmentIfNotExists("i1", "f1", pilosa.ViewStandard, 0).MustSetBits(40, (0*SliceWidth)+2)
hldr.MustCreateFragmentIfNotExists("i1", "f1", pilosa.ViewStandard, 0).MustSetBits(40, (0*SliceWidth)+8)
hldr.MustCreateFragmentIfNotExists("i1", "f1", pilosa.ViewStandard, 0).MustSetBits(40, (0*pilosa.SliceWidth)+1)
hldr.MustCreateFragmentIfNotExists("i1", "f1", pilosa.ViewStandard, 0).MustSetBits(40, (0*pilosa.SliceWidth)+2)
hldr.MustCreateFragmentIfNotExists("i1", "f1", pilosa.ViewStandard, 0).MustSetBits(40, (0*pilosa.SliceWidth)+8)
h := test.MustNewHandler()
h.API.Holder = hldr.Holder
h.API.Cluster = test.NewCluster(1)
w := httptest.NewRecorder()
h.ServeHTTP(w, test.MustNewHTTPRequest("GET", "/slices/max", nil))
if w.Code != http.StatusOK {
if w.Code != gohttp.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{"standard":{"i0":3,"i1":0}}`+"\n" {
t.Fatalf("unexpected body: %s", body)
@ -229,7 +230,7 @@ func TestHandler_Query_Args_URL(t *testing.T) {
w := httptest.NewRecorder()
h.ServeHTTP(w, test.MustNewHTTPRequest("POST", "/index/idx0/query?slices=0,1", strings.NewReader("Count( Bitmap( id=100))")))
if w.Code != http.StatusOK {
if w.Code != gohttp.StatusOK {
t.Fatalf("unexpected status code: %d %s", w.Code, w.Body.String())
} else if body := w.Body.String(); body != `{"results":[100]}`+"\n" {
t.Fatalf("unexpected body: %q", body)
@ -270,7 +271,7 @@ func TestHandler_Query_Args_Protobuf(t *testing.T) {
w := httptest.NewRecorder()
h.ServeHTTP(w, req)
if w.Code != http.StatusOK {
if w.Code != gohttp.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
}
}
@ -286,7 +287,7 @@ func TestHandler_Query_Args_Err(t *testing.T) {
h.API.Holder = hldr.Holder
h.ServeHTTP(w, test.MustNewHTTPRequest("POST", "/index/idx0/query?slices=a,b", strings.NewReader("Bitmap(id=100)")))
if w.Code != http.StatusBadRequest {
if w.Code != gohttp.StatusBadRequest {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{"error":"invalid slice argument"}`+"\n" {
t.Fatalf("unexpected body: %q", body)
@ -295,7 +296,7 @@ func TestHandler_Query_Args_Err(t *testing.T) {
func TestHandler_Query_Params_Err(t *testing.T) {
w := httptest.NewRecorder()
test.MustNewHandler().ServeHTTP(w, test.MustNewHTTPRequest("POST", "/index/idx0/query?slices=0,1&db=sample", strings.NewReader("Bitmap(id=100)")))
if w.Code != http.StatusBadRequest {
if w.Code != gohttp.StatusBadRequest {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{"error":"db is not a valid argument"}`+"\n" {
t.Fatalf("unexpected body: %q", body)
@ -317,7 +318,7 @@ func TestHandler_Query_Uint64_JSON(t *testing.T) {
w := httptest.NewRecorder()
h.ServeHTTP(w, test.MustNewHTTPRequest("POST", "/index/idx0/query?slices=0,1", strings.NewReader("Count( Bitmap( id=100))")))
if w.Code != http.StatusOK {
if w.Code != gohttp.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{"results":[100]}`+"\n" {
t.Fatalf("unexpected body: %q", body)
@ -340,14 +341,14 @@ func TestHandler_Query_Uint64_Protobuf(t *testing.T) {
r := test.MustNewHTTPRequest("POST", "/index/i/query", strings.NewReader("Count(Bitmap(id=100))"))
r.Header.Set("Accept", "application/x-protobuf")
h.ServeHTTP(w, r)
if w.Code != http.StatusOK {
if w.Code != gohttp.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
}
var resp internal.QueryResponse
if err := proto.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
} else if rt := resp.Results[0].Type; rt != pilosa.QueryResultTypeUint64 {
} else if rt := resp.Results[0].Type; rt != http.QueryResultTypeUint64 {
t.Fatalf("unexpected response type: %d", resp.Results[0].Type)
} else if n := resp.Results[0].N; n != 100 {
t.Fatalf("unexpected n: %d", n)
@ -370,7 +371,7 @@ func TestHandler_Query_Bitmap_JSON(t *testing.T) {
w := httptest.NewRecorder()
h.ServeHTTP(w, test.MustNewHTTPRequest("POST", "/index/i/query", strings.NewReader("Bitmap(id=100)")))
if w.Code != http.StatusOK {
if w.Code != gohttp.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{"results":[{"attrs":{"a":"b","c":1,"d":true},"columns":[1,3,66,1048577]}]}`+"\n" {
t.Fatalf("unexpected body: %s", body)
@ -403,7 +404,7 @@ func TestHandler_Query_Row_ColumnAttrs_JSON(t *testing.T) {
w := httptest.NewRecorder()
h.ServeHTTP(w, test.MustNewHTTPRequest("POST", "/index/i/query?columnAttrs=true", strings.NewReader("Bitmap(id=100)")))
if w.Code != http.StatusOK {
if w.Code != gohttp.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{"results":[{"attrs":{"a":"b","c":1,"d":true},"columns":[1,3,66,1048577]}],"columnAttrs":[{"id":3,"attrs":{"x":"y"}},{"id":66,"attrs":{"y":123,"z":false}}]}`+"\n" {
t.Fatalf("unexpected body: %s", body)
@ -428,16 +429,16 @@ func TestHandler_Query_Row_Protobuf(t *testing.T) {
r := test.MustNewHTTPRequest("POST", "/index/i/query", strings.NewReader("Bitmap(id=100)"))
r.Header.Set("Accept", "application/x-protobuf")
h.ServeHTTP(w, r)
if w.Code != http.StatusOK {
if w.Code != gohttp.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
}
var resp internal.QueryResponse
if err := proto.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
} else if rt := resp.Results[0].Type; rt != pilosa.QueryResultTypeRow {
} else if rt := resp.Results[0].Type; rt != http.QueryResultTypeRow {
t.Fatalf("unexpected response type: %d", resp.Results[0].Type)
} else if columns := resp.Results[0].Row.Columns; !reflect.DeepEqual(columns, []uint64{1, SliceWidth + 1}) {
} else if columns := resp.Results[0].Row.Columns; !reflect.DeepEqual(columns, []uint64{1, pilosa.SliceWidth + 1}) {
t.Fatalf("unexpected columns: %+v", columns)
} else if attrs := resp.Results[0].Row.Attrs; len(attrs) != 3 {
t.Fatalf("unexpected attr length: %d", len(attrs))
@ -486,7 +487,7 @@ func TestHandler_Query_Row_ColumnAttrs_Protobuf(t *testing.T) {
r.Header.Set("Content-Type", "application/x-protobuf")
r.Header.Set("Accept", "application/x-protobuf")
h.ServeHTTP(w, r)
if w.Code != http.StatusOK {
if w.Code != gohttp.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
}
@ -494,9 +495,9 @@ func TestHandler_Query_Row_ColumnAttrs_Protobuf(t *testing.T) {
if err := proto.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
}
if columns := resp.Results[0].Row.Columns; !reflect.DeepEqual(columns, []uint64{1, SliceWidth + 1}) {
if columns := resp.Results[0].Row.Columns; !reflect.DeepEqual(columns, []uint64{1, pilosa.SliceWidth + 1}) {
t.Fatalf("unexpected columns: %+v", columns)
} else if rt := resp.Results[0].Type; rt != pilosa.QueryResultTypeRow {
} else if rt := resp.Results[0].Type; rt != http.QueryResultTypeRow {
t.Fatalf("unexpected response type: %d", resp.Results[0].Type)
} else if attrs := resp.Results[0].Row.Attrs; len(attrs) != 3 {
t.Fatalf("unexpected attr length: %d", len(attrs))
@ -536,7 +537,7 @@ func TestHandler_Query_Pairs_JSON(t *testing.T) {
w := httptest.NewRecorder()
h.ServeHTTP(w, test.MustNewHTTPRequest("POST", "/index/i/query", strings.NewReader(`TopN(field=x, n=2)`)))
if w.Code != http.StatusOK {
if w.Code != gohttp.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{"results":[[{"id":1,"count":2},{"id":3,"count":4}]]}`+"\n" {
t.Fatalf("unexpected body: %q", body)
@ -562,14 +563,14 @@ func TestHandler_Query_Pairs_Protobuf(t *testing.T) {
r := test.MustNewHTTPRequest("POST", "/index/i/query", strings.NewReader(`TopN(field=x, n=2)`))
r.Header.Set("Accept", "application/x-protobuf")
h.ServeHTTP(w, r)
if w.Code != http.StatusOK {
if w.Code != gohttp.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
}
var resp internal.QueryResponse
if err := proto.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
} else if rt := resp.Results[0].Type; rt != pilosa.QueryResultTypePairs {
} else if rt := resp.Results[0].Type; rt != http.QueryResultTypePairs {
t.Fatalf("unexpected response type: %d", resp.Results[0].Type)
} else if a := resp.Results[0].GetPairs(); len(a) != 2 {
t.Fatalf("unexpected pair length: %d", len(a))
@ -590,7 +591,7 @@ func TestHandler_Query_Err_JSON(t *testing.T) {
w := httptest.NewRecorder()
h.ServeHTTP(w, test.MustNewHTTPRequest("POST", "/index/i/query", strings.NewReader(`Bitmap(id=100)`)))
if w.Code != http.StatusBadRequest {
if w.Code != gohttp.StatusBadRequest {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{"error":"executing: marker"}`+"\n" {
t.Fatalf("unexpected body: %q", body)
@ -613,7 +614,7 @@ func TestHandler_Query_Err_Protobuf(t *testing.T) {
r := test.MustNewHTTPRequest("POST", "/index/i/query", strings.NewReader(`TopN(field=x, n=2)`))
r.Header.Set("Accept", "application/x-protobuf")
h.ServeHTTP(w, r)
if w.Code != http.StatusBadRequest {
if w.Code != gohttp.StatusBadRequest {
t.Fatalf("unexpected status code: %d", w.Code)
}
@ -635,7 +636,7 @@ func TestHandler_Query_MethodNotAllowed(t *testing.T) {
h.API.Holder = hldr.Holder
w := httptest.NewRecorder()
h.ServeHTTP(w, test.MustNewHTTPRequest("GET", "/index/i/query", nil))
if w.Code != http.StatusMethodNotAllowed {
if w.Code != gohttp.StatusMethodNotAllowed {
t.Fatalf("invalid status: %d", w.Code)
}
}
@ -650,7 +651,7 @@ func TestHandler_Query_ErrParse(t *testing.T) {
h.API.Holder = hldr.Holder
w := httptest.NewRecorder()
h.ServeHTTP(w, test.MustNewHTTPRequest("POST", "/index/idx0/query?slices=0,1", strings.NewReader("bad_fn(")))
if w.Code != http.StatusBadRequest {
if w.Code != gohttp.StatusBadRequest {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{"error":"parsing: expected comma, right paren, or identifier, found \"\" occurred at line 1, char 8"}`+"\n" {
t.Fatalf("unexpected body: %s", body)
@ -672,14 +673,14 @@ func TestHandler_Index_Delete(t *testing.T) {
}
// Send request to delete index.
resp, err := http.DefaultClient.Do(test.MustNewHTTPRequest("DELETE", s.URL+"/index/i", strings.NewReader("")))
resp, err := gohttp.DefaultClient.Do(test.MustNewHTTPRequest("DELETE", s.URL+"/index/i", strings.NewReader("")))
if err != nil {
t.Fatal(err)
}
defer resp.Body.Close()
// Verify body response.
if resp.StatusCode != http.StatusOK {
if resp.StatusCode != gohttp.StatusOK {
t.Fatalf("unexpected status: %d", resp.StatusCode)
} else if buf, err := ioutil.ReadAll(resp.Body); err != nil {
t.Fatal(err)
@ -707,7 +708,7 @@ func TestHandler_DeleteField(t *testing.T) {
h.API.Cluster = test.NewCluster(1)
w := httptest.NewRecorder()
h.ServeHTTP(w, test.MustNewHTTPRequest("DELETE", "/index/i0/field/f1", strings.NewReader("")))
if w.Code != http.StatusOK {
if w.Code != gohttp.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{}`+"\n" {
t.Fatalf("unexpected body: %s", body)
@ -749,7 +750,7 @@ func TestHandler_Index_AttrStore_Diff(t *testing.T) {
blks[1].Checksum = []byte("MISMATCHED_CHECKSUM")
// Send block checksums to determine diff.
resp, err := http.Post(
resp, err := gohttp.Post(
s.URL+"/index/i/attr/diff",
"application/json",
strings.NewReader(`{"blocks":`+string(test.MustMarshalJSON(blks))+`}`),
@ -799,7 +800,7 @@ func TestHandler_Field_AttrStore_Diff(t *testing.T) {
blks[1].Checksum = []byte("MISMATCHED_CHECKSUM")
// Send block checksums to determine diff.
resp, err := http.Post(
resp, err := gohttp.Post(
s.URL+"/index/i/field/meta/attr/diff",
"application/json",
strings.NewReader(`{"blocks":`+string(test.MustMarshalJSON(blks))+`}`),
@ -831,7 +832,7 @@ func TestHandler_Version(t *testing.T) {
if strings.HasPrefix(version, "v") {
version = version[1:]
}
if w.Code != http.StatusOK {
if w.Code != gohttp.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if w.Body.String() != `{"version":"`+version+`"}`+"\n" {
t.Fatalf("unexpected body: %q", w.Body.String())
@ -851,7 +852,7 @@ func TestHandler_Fragment_Nodes(t *testing.T) {
w := httptest.NewRecorder()
r := test.MustNewHTTPRequest("GET", "/fragment/nodes?index=X&slice=0", nil)
h.ServeHTTP(w, r)
if w.Code != http.StatusOK {
if w.Code != gohttp.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `[{"id":"node2","uri":{"scheme":"http","host":"host2"},"isCoordinator":false},{"id":"node0","uri":{"scheme":"http","host":"host0"},"isCoordinator":false}]`+"\n" {
t.Fatalf("unexpected body: %q", body)
@ -861,7 +862,7 @@ func TestHandler_Fragment_Nodes(t *testing.T) {
w = httptest.NewRecorder()
r = test.MustNewHTTPRequest("GET", "/fragment/nodes?db=X&slice=0", nil)
h.ServeHTTP(w, r)
if w.Code != http.StatusBadRequest {
if w.Code != gohttp.StatusBadRequest {
t.Fatalf("unexpected status code: %d", w.Code)
}
@ -869,7 +870,7 @@ func TestHandler_Fragment_Nodes(t *testing.T) {
w = httptest.NewRecorder()
r = test.MustNewHTTPRequest("GET", "/fragment/nodes?slice=0", nil)
h.ServeHTTP(w, r)
if w.Code != http.StatusBadRequest {
if w.Code != gohttp.StatusBadRequest {
t.Fatalf("unexpected status code: %d", w.Code)
}
}
@ -885,7 +886,7 @@ func TestHandler_Expvars(t *testing.T) {
w := httptest.NewRecorder()
r := test.MustNewHTTPRequest("GET", "/debug/vars", nil)
h.ServeHTTP(w, r)
if w.Code != http.StatusOK {
if w.Code != gohttp.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
}
}
@ -908,7 +909,7 @@ func TestHandler_RecalculateCaches(t *testing.T) {
w := httptest.NewRecorder()
h.ServeHTTP(w, test.MustNewHTTPRequest("POST", "/recalculate-caches", nil))
if w.Code != http.StatusNoContent {
if w.Code != gohttp.StatusNoContent {
t.Fatalf("unexpected status code: %d", w.Code)
}
@ -939,7 +940,7 @@ func TestHandler_CORS(t *testing.T) {
}
// CORS config should allow preflight response
handler = test.MustNewHandler(pilosa.OptHandlerAllowedOrigins([]string{"http://test/"}))
handler = test.MustNewHandler(http.OptHandlerAllowedOrigins([]string{"http://test/"}))
w = httptest.NewRecorder()
handler.ServeHTTP(w, req)
result = w.Result()

10
pilosa.go
View file

@ -88,17 +88,17 @@ type ColumnAttrSet struct {
Attrs map[string]interface{} `json:"attrs,omitempty"`
}
// encodeColumnAttrSets converts a into its internal representation.
func encodeColumnAttrSets(a []*ColumnAttrSet) []*internal.ColumnAttrSet {
// EncodeColumnAttrSets converts a into its internal representation.
func EncodeColumnAttrSets(a []*ColumnAttrSet) []*internal.ColumnAttrSet {
other := make([]*internal.ColumnAttrSet, len(a))
for i := range a {
other[i] = encodeColumnAttrSet(a[i])
other[i] = EncodeColumnAttrSet(a[i])
}
return other
}
// encodeColumnAttrSet converts set into its internal representation.
func encodeColumnAttrSet(set *ColumnAttrSet) *internal.ColumnAttrSet {
// EncodeColumnAttrSet converts set into its internal representation.
func EncodeColumnAttrSet(set *ColumnAttrSet) *internal.ColumnAttrSet {
return &internal.ColumnAttrSet{
ID: set.ID,
Attrs: encodeAttrs(set.Attrs),

8
row.go
View file

@ -261,8 +261,8 @@ func (r *Row) Columns() []uint64 {
return a
}
// encodeRow converts r into its internal representation.
func encodeRow(r *Row) *internal.Row {
// EncodeRow converts r into its internal representation.
func EncodeRow(r *Row) *internal.Row {
if r == nil {
return nil
}
@ -273,8 +273,8 @@ func encodeRow(r *Row) *internal.Row {
}
}
// decodeRow converts r from its internal representation.
func decodeRow(pr *internal.Row) *Row {
// DecodeRow converts r from its internal representation.
func DecodeRow(pr *internal.Row) *Row {
if pr == nil {
return nil
}

45
server.go
View file

@ -59,7 +59,7 @@ type Server struct {
executor *Executor
// External
handler *Handler
handler Handlerer
Broadcaster Broadcaster
BroadcastReceiver BroadcastReceiver
Gossiper Gossiper
@ -127,7 +127,7 @@ func OptServerLongQueryTime(dur time.Duration) ServerOption {
}
}
func OptServerHandler(h *Handler) ServerOption {
func OptServerHandler(h Handlerer) ServerOption {
return func(s *Server) error {
s.handler = h
return nil
@ -162,16 +162,24 @@ func OptServerGCNotifier(gcn GCNotifier) ServerOption {
}
}
// TODO: Remove RemoteClient
func OptServerRemoteClient(c *http.Client) ServerOption {
return func(s *Server) error {
s.executor = NewExecutor(c)
s.remoteClient = c
s.defaultClient = NewInternalHTTPClientFromURI(nil, c)
s.Cluster.RemoteClient = c
return nil
}
}
func OptServerInternalClient(c InternalClient) ServerOption {
return func(s *Server) error {
s.executor = NewExecutor(ExecutorOptInternalQueryClient(c))
s.defaultClient = c
s.Cluster.InternalClient = c
return nil
}
}
func OptServerStatsClient(sc StatsClient) ServerOption {
return func(s *Server) error {
s.Holder.Stats = sc
@ -203,15 +211,15 @@ func OptServerURI(uri *URI) ServerOption {
// NewServer returns a new instance of Server.
func NewServer(opts ...ServerOption) (*Server, error) {
handler, err := NewHandler()
if err != nil {
return nil, errors.Wrap(err, "initializing handler")
}
//handler, err := NewNopHandler()
//if err != nil {
// return nil, errors.Wrap(err, "initializing handler")
//}
s := &Server{
closing: make(chan struct{}),
Cluster: NewCluster(),
Holder: NewHolder(),
handler: handler,
closing: make(chan struct{}),
Cluster: NewCluster(),
Holder: NewHolder(),
//handler: handler,
Broadcaster: NopBroadcaster,
BroadcastReceiver: NopBroadcastReceiver,
diagnostics: NewDiagnosticsCollector(DefaultDiagnosticServer),
@ -268,7 +276,7 @@ func NewServer(opts ...ServerOption) (*Server, error) {
s.executor.Node = node
s.executor.Cluster = s.Cluster
s.executor.MaxWritesPerRequest = s.maxWritesPerRequest
s.handler.API.Executor = s.executor
s.handler.GetAPI().Executor = s.executor
return s, nil
}
@ -300,11 +308,12 @@ func (s *Server) Open() error {
s.Cluster.MaxWritesPerRequest = s.maxWritesPerRequest
// Initialize HTTP handler.
s.handler.API.Holder = s.Holder
s.handler.API.Broadcaster = s.Broadcaster
s.handler.API.BroadcastHandler = s
s.handler.API.StatusHandler = s
s.handler.API.Cluster = s.Cluster
api := s.handler.GetAPI()
api.Holder = s.Holder
api.Broadcaster = s.Broadcaster
api.BroadcastHandler = s
api.StatusHandler = s
api.Cluster = s.Cluster
// Initialize Holder.
s.Holder.Broadcaster = s.Broadcaster

39
server/server.go
View file

@ -25,7 +25,6 @@ import (
"log"
"math/rand"
"net"
"net/http"
"os"
"os/signal"
"strconv"
@ -39,6 +38,7 @@ import (
"github.com/pilosa/pilosa/gcnotify"
"github.com/pilosa/pilosa/gopsutil"
"github.com/pilosa/pilosa/gossip"
"github.com/pilosa/pilosa/http"
"github.com/pilosa/pilosa/statsd"
"github.com/pkg/errors"
)
@ -164,13 +164,17 @@ func (m *Command) SetupServer() error {
}
m.logger.Printf("%s %s, build time %s\n", productName, pilosa.Version, pilosa.BuildTime)
handler, err := pilosa.NewHandler(pilosa.OptHandlerAllowedOrigins(m.Config.Handler.AllowedOrigins))
api := pilosa.NewAPI()
api.Logger = m.logger
handler, err := http.NewHandler(
http.OptHandlerAllowedOrigins(m.Config.Handler.AllowedOrigins),
http.OptHandlerAPI(api),
http.OptHandlerLogger(m.logger),
)
if err != nil {
return errors.Wrap(err, "wrapping handler")
}
handler.Logger = m.logger
handler.API = pilosa.NewAPI()
handler.API.Logger = m.logger
uri, err := pilosa.AddressWithDefaults(m.Config.Bind)
if err != nil {
@ -211,8 +215,8 @@ func (m *Command) SetupServer() error {
return errors.Wrap(err, "getting listener")
}
c := GetHTTPClient(TLSConfig)
handler.API.RemoteClient = c
c := http.GetHTTPClient(TLSConfig)
api.RemoteClient = c
m.Server, err = pilosa.NewServer(
pilosa.OptServerAntiEntropyInterval(time.Duration(m.Config.AntiEntropy.Interval)),
@ -232,31 +236,12 @@ func (m *Command) SetupServer() error {
pilosa.OptServerListener(ln),
pilosa.OptServerURI(uri),
pilosa.OptServerRemoteClient(c),
pilosa.OptServerInternalClient(http.NewInternalHTTPClientFromURI(uri, c)),
)
return errors.Wrap(err, "new server")
}
func GetHTTPClient(t *tls.Config) *http.Client {
transport := &http.Transport{
Proxy: http.ProxyFromEnvironment,
DialContext: (&net.Dialer{
Timeout: 30 * time.Second,
KeepAlive: 30 * time.Second,
DualStack: true,
}).DialContext,
MaxIdleConns: 1000,
MaxIdleConnsPerHost: 200,
IdleConnTimeout: 90 * time.Second,
TLSHandshakeTimeout: 10 * time.Second,
ExpectContinueTimeout: 1 * time.Second,
}
if t != nil {
transport.TLSClientConfig = t
}
return &http.Client{Transport: transport}
}
// SetupNetworking sets up internode communication based on the configuration.
func (m *Command) SetupNetworking() error {

3
server/server_test.go
View file

@ -29,6 +29,7 @@ import (
"github.com/pelletier/go-toml"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/http"
"github.com/pilosa/pilosa/server"
"github.com/pilosa/pilosa/test"
)
@ -44,7 +45,7 @@ func TestMain_Set_Quick(t *testing.T) {
defer m.Close()
// Create client.
client, err := pilosa.NewInternalHTTPClient(m.Server.URI.HostPort(), server.GetHTTPClient(nil))
client, err := http.NewInternalHTTPClient(m.Server.URI.HostPort(), http.GetHTTPClient(nil))
if err != nil {
t.Fatal(err)
}

10
test/client.go
View file

@ -15,19 +15,19 @@
package test
import (
"net/http"
gohttp "net/http"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/http"
)
// Client represents a test wrapper for pilosa.Client.
type Client struct {
*pilosa.InternalHTTPClient
*http.InternalHTTPClient
}
// MustNewClient returns a new instance of Client. Panic on error.
func MustNewClient(host string, h *http.Client) *Client {
c, err := pilosa.NewInternalHTTPClient(host, h)
func MustNewClient(host string, h *gohttp.Client) *Client {
c, err := http.NewInternalHTTPClient(host, h)
if err != nil {
panic(err)
}

11
test/executor.go
View file

@ -15,12 +15,12 @@
package test
import (
"net/http"
gohttp "net/http"
"strings"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/http"
"github.com/pilosa/pilosa/pql"
"github.com/pilosa/pilosa/server"
)
// Executor represents a test wrapper for pilosa.Executor.
@ -28,16 +28,17 @@ type Executor struct {
*pilosa.Executor
}
var remoteClient *http.Client
var remoteClient *gohttp.Client
func init() {
remoteClient = server.GetHTTPClient(nil)
remoteClient = http.GetHTTPClient(nil)
}
// NewExecutor returns a new instance of Executor.
// The executor always matches the uri of the first cluster node.
func NewExecutor(holder *pilosa.Holder, cluster *pilosa.Cluster) *Executor {
executor := pilosa.NewExecutor(remoteClient)
client := http.NewInternalHTTPClientFromURI(nil, remoteClient)
executor := pilosa.NewExecutor(pilosa.ExecutorOptInternalQueryClient(client))
e := &Executor{Executor: executor}
e.Holder = holder
e.Cluster = cluster

15
test/handler.go
View file

@ -19,25 +19,26 @@ import (
"encoding/json"
"io"
"io/ioutil"
"net/http"
gohttp "net/http"
"net/http/httptest"
"net/url"
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/http"
"github.com/pilosa/pilosa/internal"
"github.com/pilosa/pilosa/pql"
)
// Handler represents a test wrapper for pilosa.Handler.
type Handler struct {
*pilosa.Handler
*http.Handler
Executor HandlerExecutor
}
// NewHandler returns a new instance of Handler.
func NewHandler(opts ...pilosa.HandlerOption) (*Handler, error) {
handler, err := pilosa.NewHandler(opts...)
func NewHandler(opts ...http.HandlerOption) (*Handler, error) {
handler, err := http.NewHandler(opts...)
if err != nil {
return nil, err
}
@ -55,7 +56,7 @@ func NewHandler(opts ...pilosa.HandlerOption) (*Handler, error) {
}
// MustNewHandler returns a new instance of Handler.
func MustNewHandler(opts ...pilosa.HandlerOption) *Handler {
func MustNewHandler(opts ...http.HandlerOption) *Handler {
h, err := NewHandler(opts...)
if err != nil {
panic(err)
@ -145,8 +146,8 @@ func MustParseURLHost(rawurl string) string {
}
// MustNewHTTPRequest creates a new HTTP request. Panic on error.
func MustNewHTTPRequest(method, urlStr string, body io.Reader) *http.Request {
req, err := http.NewRequest(method, urlStr, body)
func MustNewHTTPRequest(method, urlStr string, body io.Reader) *gohttp.Request {
req, err := gohttp.NewRequest(method, urlStr, body)
if err != nil {
panic(err)
}

16
test/pilosa.go
View file

@ -19,15 +19,15 @@ import (
"fmt"
"io"
"io/ioutil"
"net/http"
gohttp "net/http"
"os"
"strings"
"testing"
"time"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/boltdb"
"github.com/pilosa/pilosa/gossip"
"github.com/pilosa/pilosa/http"
"github.com/pilosa/pilosa/server"
"github.com/pilosa/pilosa/toml"
"github.com/pkg/errors"
@ -238,8 +238,8 @@ func (m *Main) RunWithTransport(host string, bindPort int, joinSeeds []string) (
func (m *Main) URL() string { return "http://" + m.Server.Addr().String() }
// Client returns a client to connect to the program.
func (m *Main) Client() *pilosa.InternalHTTPClient {
client, err := pilosa.NewInternalHTTPClient(m.Server.URI.HostPort(), server.GetHTTPClient(nil))
func (m *Main) Client() *http.InternalHTTPClient {
client, err := http.NewInternalHTTPClient(m.Server.URI.HostPort(), http.GetHTTPClient(nil))
if err != nil {
panic(err)
}
@ -249,7 +249,7 @@ func (m *Main) Client() *pilosa.InternalHTTPClient {
// Query executes a query against the program through the HTTP API.
func (m *Main) Query(index, rawQuery, query string) (string, error) {
resp := MustDo("POST", m.URL()+fmt.Sprintf("/index/%s/query?", index)+rawQuery, query)
if resp.StatusCode != http.StatusOK {
if resp.StatusCode != gohttp.StatusOK {
return "", fmt.Errorf("invalid status: %d, body=%s", resp.StatusCode, resp.Body)
}
return resp.Body, nil
@ -267,11 +267,11 @@ func (m *Main) RecalculateCaches() error {
// MustDo executes http.Do() with an http.NewRequest(). Panic on error.
func MustDo(method, urlStr string, body string) *httpResponse {
req, err := http.NewRequest(method, urlStr, strings.NewReader(body))
req, err := gohttp.NewRequest(method, urlStr, strings.NewReader(body))
if err != nil {
panic(err)
}
resp, err := http.DefaultClient.Do(req)
resp, err := gohttp.DefaultClient.Do(req)
if err != nil {
panic(err)
}
@ -287,6 +287,6 @@ func MustDo(method, urlStr string, body string) *httpResponse {
// httpResponse is a wrapper for http.Response that holds the Body as a string.
type httpResponse struct {
*http.Response
*gohttp.Response
Body string
}