From e201afe241668c147fe4529e8457ec17dc421f45 Mon Sep 17 00:00:00 2001 From: Yuce Tekol Date: Tue, 17 Oct 2017 15:53:42 +0300 Subject: [PATCH] Replaced all http.Clients with InternalClient; updated InternalClient interface. --- client.go | 52 +++++++++++++++++++++++++++++++++++++-------------- ctl/bench.go | 2 +- ctl/import.go | 2 +- fragment.go | 2 +- server.go | 32 +++++++------------------------ 5 files changed, 48 insertions(+), 42 deletions(-) diff --git a/client.go b/client.go index af28b068c..74e051758 100644 --- a/client.go +++ b/client.go @@ -96,7 +96,7 @@ func (c *Client) MaxInverseSliceByIndex(ctx context.Context) (map[string]uint64, // maxSliceByIndex returns the number of slices on a server by index. func (c *Client) maxSliceByIndex(ctx context.Context, inverse bool) (map[string]uint64, error) { // Execute request against the host. - u := uriPathToURL(c.defaultURI, "/slices/max") + u := uriPathToURL(c.clientURI(ctx), "/slices/max") u.RawQuery = (&url.Values{ "inverse": {strconv.FormatBool(inverse)}, }).Encode() @@ -129,10 +129,10 @@ func (c *Client) maxSliceByIndex(ctx context.Context, inverse bool) (map[string] // Schema returns all index and frame schema information. func (c *Client) Schema(ctx context.Context) ([]*IndexInfo, error) { // Execute request against the host. - u := uriPathToURL(c.defaultURI, "/schema") + u := c.defaultURI.Path("/schema") // Build request. - req, err := http.NewRequest("GET", u.String(), nil) + req, err := http.NewRequest("GET", u, nil) if err != nil { return nil, err } @@ -246,11 +246,7 @@ func (c *Client) ExecuteQuery(ctx context.Context, index string, queryRequest *i } // Create HTTP request. - clientURI := c.defaultURI - if contextURI, ok := ctx.Value("uri").(*URI); ok { - clientURI = contextURI - } - u := clientURI.Path(fmt.Sprintf("/index/%s/query", index)) + u := c.clientURI(ctx).Path(fmt.Sprintf("/index/%s/query", index)) req, err := http.NewRequest("POST", u, bytes.NewReader(buf)) if err != nil { return nil, err @@ -294,7 +290,7 @@ func (c *Client) Import(ctx context.Context, index, frame string, slice uint64, return ErrFrameRequired } - buf, err := MarshalImportPayload(index, frame, slice, bits) + buf, err := marshalImportPayload(index, frame, slice, bits) if err != nil { return fmt.Errorf("Error Creating Payload: %s", err) } @@ -331,8 +327,8 @@ func (c *Client) EnsureFrame(ctx context.Context, indexName string, frameName st return err } -// MarshalImportPayload marshalls the import parameters into a protobuf byte slice. -func MarshalImportPayload(index, frame string, slice uint64, bits []Bit) ([]byte, error) { +// marshalImportPayload marshalls the import parameters into a protobuf byte slice. +func marshalImportPayload(index, frame string, slice uint64, bits []Bit) ([]byte, error) { // Separate row and column IDs to reduce allocations. rowIDs := Bits(bits).RowIDs() columnIDs := Bits(bits).ColumnIDs() @@ -399,7 +395,7 @@ func (c *Client) ImportValue(ctx context.Context, index, frame, field string, sl return ErrFrameRequired } - buf, err := MarshalImportValuePayload(index, frame, field, slice, vals) + buf, err := marshalImportValuePayload(index, frame, field, slice, vals) if err != nil { return fmt.Errorf("Error Creating Payload: %s", err) } @@ -420,8 +416,8 @@ func (c *Client) ImportValue(ctx context.Context, index, frame, field string, sl return nil } -// MarshalImportValuePayload marshalls the import parameters into a protobuf byte slice. -func MarshalImportValuePayload(index, frame, field string, slice uint64, vals []FieldValue) ([]byte, error) { +// marshalImportValuePayload marshalls the import parameters into a protobuf byte slice. +func marshalImportValuePayload(index, frame, field string, slice uint64, vals []FieldValue) ([]byte, error) { // Separate row and column IDs to reduce allocations. columnIDs := FieldValues(vals).ColumnIDs() values := FieldValues(vals).Values() @@ -1056,6 +1052,14 @@ func (c *Client) RowAttrDiff(ctx context.Context, index, frame string, blks []At return rsp.Attrs, nil } +func (c *Client) clientURI(ctx context.Context) *URI { + clientURI := c.defaultURI + if contextURI, ok := ctx.Value("uri").(*URI); ok { + clientURI = contextURI + } + return clientURI +} + // Bit represents the location of a single bit. type Bit struct { RowID uint64 @@ -1203,5 +1207,25 @@ func nodePathToURL(node *Node, path string) url.URL { } type InternalClient interface { + MaxSliceByIndex(ctx context.Context) (map[string]uint64, error) + MaxInverseSliceByIndex(ctx context.Context) (map[string]uint64, error) + Schema(ctx context.Context) ([]*IndexInfo, error) + CreateIndex(ctx context.Context, index string, opt IndexOptions) error + FragmentNodes(ctx context.Context, index string, slice uint64) ([]*Node, error) ExecuteQuery(ctx context.Context, index string, queryRequest *internal.QueryRequest) (*internal.QueryResponse, error) + Import(ctx context.Context, index, frame string, slice uint64, bits []Bit) error + EnsureIndex(ctx context.Context, name string, options IndexOptions) error + EnsureFrame(ctx context.Context, indexName string, frameName string, options FrameOptions) error + ImportValue(ctx context.Context, index, frame, field string, slice uint64, vals []FieldValue) error + ExportCSV(ctx context.Context, index, frame, view string, slice uint64, w io.Writer) error + BackupTo(ctx context.Context, w io.Writer, index, frame, view string) error + BackupSlice(ctx context.Context, index, frame, view string, slice uint64) (io.ReadCloser, error) + RestoreFrom(ctx context.Context, r io.Reader, index, frame, view string) error + CreateFrame(ctx context.Context, index, frame string, opt FrameOptions) error + RestoreFrame(ctx context.Context, host, index, frame string) error + FrameViews(ctx context.Context, index, frame string) ([]string, error) + FragmentBlocks(ctx context.Context, index, frame, view string, slice uint64) ([]FragmentBlock, error) + BlockData(ctx context.Context, index, frame, view string, slice uint64, block int) ([]uint64, []uint64, error) + ColumnAttrDiff(ctx context.Context, index string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) + RowAttrDiff(ctx context.Context, index, frame string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) } diff --git a/ctl/bench.go b/ctl/bench.go index 2f1e7a52f..01e07cc14 100644 --- a/ctl/bench.go +++ b/ctl/bench.go @@ -71,7 +71,7 @@ func (cmd *BenchCommand) Run(ctx context.Context) error { } // runSetBit executes a benchmark of random SetBit() operations. -func (cmd *BenchCommand) runSetBit(ctx context.Context, client *pilosa.Client) error { +func (cmd *BenchCommand) runSetBit(ctx context.Context, client pilosa.InternalClient) error { if cmd.N == 0 { return errors.New("operation count required") } else if cmd.Index == "" { diff --git a/ctl/import.go b/ctl/import.go index e3eda260e..d2f888956 100644 --- a/ctl/import.go +++ b/ctl/import.go @@ -58,7 +58,7 @@ type ImportCommand struct { Sort bool `json:"sort"` // Reusable client. - Client *pilosa.Client `json:"-"` + Client pilosa.InternalClient `json:"-"` // Standard input/output *pilosa.CmdIO diff --git a/fragment.go b/fragment.go index cdc8c00ff..148fd6a48 100644 --- a/fragment.go +++ b/fragment.go @@ -1782,7 +1782,7 @@ func (s *FragmentSyncer) syncBlock(id int) error { // Read pairs from each remote block. var pairSets []PairSet - var clients []*Client + var clients []InternalClient for _, node := range s.Cluster.FragmentNodes(f.Index(), f.Slice()) { if s.Host == node.Host { continue diff --git a/server.go b/server.go index e73354cf2..3ce2883c3 100644 --- a/server.go +++ b/server.go @@ -19,7 +19,6 @@ import ( "errors" "fmt" "io" - "io/ioutil" "log" "net" "net/http" @@ -35,6 +34,7 @@ import ( "github.com/CAFxX/gcnotifier" "github.com/gogo/protobuf/proto" "github.com/pilosa/pilosa/internal" + "golang.org/x/net/context" ) // Default server settings. @@ -76,7 +76,7 @@ type Server struct { LogOutput io.Writer - defaultClient *http.Client + defaultClient InternalClient } // NewServer returns a new instance of Server. @@ -483,31 +483,13 @@ func (s *Server) checkMaxSlices(scheme string, hostPort string) (map[string]uint req.Header.Set("Content-Type", "application/x-protobuf") req.Header.Set("User-Agent", "pilosa/"+Version) - resp, err := s.defaultClient.Do(req) + nodeURI, err := NewURIFromAddress(hostPort) if err != nil { return nil, err } - defer resp.Body.Close() - - // Read response into buffer. - body, err := ioutil.ReadAll(resp.Body) - if err != nil { - return nil, err - } - - // Check status code. - if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("invalid status checkMaxSlices: code=%d, err=%s, req=%v", resp.StatusCode, body, req) - } - - // Decode response object. - pb := internal.MaxSlicesResponse{} - - if err = proto.Unmarshal(body, &pb); err != nil { - return nil, err - } - - return pb.MaxSlices, nil + nodeURI.SetScheme(scheme) + ctx := context.WithValue(context.Background(), "uri", nodeURI) + return s.defaultClient.MaxSliceByIndex(ctx) } // monitorRuntime periodically polls the Go runtime metrics. @@ -558,7 +540,7 @@ func (s *Server) createDefaultClient() { if s.TLS != nil { transport.TLSClientConfig = s.TLS } - s.defaultClient = &http.Client{Transport: transport} + s.defaultClient = NewClientFromURI(nil, &ClientOptions{TLS: s.TLS}) } // CountOpenFiles on opperating systems that support lsof