diff --git a/Gopkg.lock b/Gopkg.lock index dc3ddc49a..4a6068840 100644 --- a/Gopkg.lock +++ b/Gopkg.lock @@ -154,10 +154,10 @@ revision = "ef8a98b0bbce4a65b5aa4c368430a80ddc533168" [[projects]] - branch = "master" name = "github.com/hashicorp/memberlist" packages = ["."] - revision = "2288bf30e9c8d7b5f6549bf62e07120d72fd4b6c" + revision = "ce8abaa0c60c2d6bee7219f5ddf500e0a1457b28" + version = "v0.1.0" [[projects]] name = "github.com/inconshreveable/mousetrap" @@ -324,6 +324,6 @@ [solve-meta] analyzer-name = "dep" analyzer-version = 1 - inputs-digest = "6ae38dc521f55a89507ec21c575e0b65505de9514d1db6255a1b9b99fcf19ab3" + inputs-digest = "8290156ce8b4066c46ab83d743f4c81df0a17e148415bb1ee8409a51ac4c3ba4" solver-name = "gps-cdcl" solver-version = 1 diff --git a/Gopkg.toml b/Gopkg.toml index f5f81ad68..459775e54 100644 --- a/Gopkg.toml +++ b/Gopkg.toml @@ -17,11 +17,6 @@ # Only one of "branch", "version" or "revision" can be specified. branch = "master" -[[constraint]] - name = "github.com/hashicorp/memberlist" - branch = "master" - - [[constraint]] name = "github.com/gorilla/handlers" version = "=1.3.0" diff --git a/api.go b/api.go index bc71cc5a2..da337252d 100644 --- a/api.go +++ b/api.go @@ -136,9 +136,9 @@ func (api *API) Query(ctx context.Context, req *QueryRequest) (QueryResponse, er } // Translate column attributes, if necessary. - if api.server.translateFile != nil { + if api.holder.translateFile != nil { for _, col := range resp.ColumnAttrSets { - v, err := api.server.translateFile.TranslateColumnToString(req.Index, col.ID) + v, err := api.holder.translateFile.TranslateColumnToString(req.Index, col.ID) if err != nil { return resp, err } @@ -391,7 +391,7 @@ func (api *API) FragmentBlockData(_ context.Context, body io.Reader) ([]byte, er } // Retrieve fragment from holder. - f := api.holder.fragment(req.Index, req.Field, viewStandard, req.Shard) + f := api.holder.fragment(req.Index, req.Field, req.View, req.Shard) if f == nil { return nil, ErrFragmentNotFound } @@ -409,13 +409,13 @@ func (api *API) FragmentBlockData(_ context.Context, body io.Reader) ([]byte, er } // FragmentBlocks returns the checksums and block ids for all blocks in the specified fragment. -func (api *API) FragmentBlocks(_ context.Context, indexName string, fieldName string, shard uint64) ([]FragmentBlock, error) { +func (api *API) FragmentBlocks(_ context.Context, indexName, fieldName, viewName string, shard uint64) ([]FragmentBlock, error) { if err := api.validate(apiFragmentBlocks); err != nil { return nil, errors.Wrap(err, "validating api method") } // Retrieve fragment from holder. - f := api.holder.fragment(indexName, fieldName, viewStandard, shard) + f := api.holder.fragment(indexName, fieldName, viewName, shard) if f == nil { return nil, ErrFragmentNotFound } @@ -610,11 +610,36 @@ func (api *API) Import(_ context.Context, req *ImportRequest) error { return errors.Wrap(err, "validating api method") } + index := api.holder.Index(req.Index) + if index == nil { + return newNotFoundError(ErrIndexNotFound) + } + field, err := api.indexField(req.Index, req.Field, req.Shard) if err != nil { return errors.Wrap(err, "getting field") } + // Translate row keys. + if field.keys() { + if len(req.RowIDs) != 0 { + return errors.New("row ids cannot be used because field uses string keys") + } + if req.RowIDs, err = api.holder.translateFile.TranslateRowsToUint64(index.Name(), field.Name(), req.RowKeys); err != nil { + return errors.Wrap(err, "translating rows") + } + } + + // Translate column keys. + if index.Keys() { + if len(req.ColumnIDs) != 0 { + return errors.New("column ids cannot be used because index uses string keys") + } + if req.ColumnIDs, err = api.holder.translateFile.TranslateColumnsToUint64(index.Name(), req.ColumnKeys); err != nil { + return errors.Wrap(err, "translating columns") + } + } + // Convert timestamps to time.Time. timestamps := make([]*time.Time, len(req.Timestamps)) for i, ts := range req.Timestamps { @@ -639,10 +664,26 @@ func (api *API) ImportValue(_ context.Context, req *ImportValueRequest) error { return errors.Wrap(err, "validating api method") } + index := api.holder.Index(req.Index) + if index == nil { + return newNotFoundError(ErrIndexNotFound) + } + field, err := api.indexField(req.Index, req.Field, req.Shard) if err != nil { return errors.Wrap(err, "getting field") } + + // Translate column keys. + if index.Keys() { + if len(req.ColumnIDs) != 0 { + return errors.New("column ids cannot be used because index uses string keys") + } + if req.ColumnIDs, err = api.holder.translateFile.TranslateColumnsToUint64(index.Name(), req.ColumnKeys); err != nil { + return errors.Wrap(err, "translating columns") + } + } + // Import into fragment. err = field.importValue(req.ColumnIDs, req.Values) if err != nil { @@ -764,7 +805,7 @@ func (api *API) ResizeAbort() error { // GetTranslateData provides a reader for key translation logs starting at offset. func (api *API) GetTranslateData(ctx context.Context, offset int64) (io.ReadCloser, error) { - rc, err := api.server.translateFile.Reader(ctx, offset) + rc, err := api.holder.translateFile.Reader(ctx, offset) if err != nil { return nil, errors.Wrap(err, "read from translate store") } diff --git a/client.go b/client.go index 32f6ad226..a0b89f604 100644 --- a/client.go +++ b/client.go @@ -18,8 +18,9 @@ type Bit struct { // FieldValue represents the value for a column within a // range-encoded field. type FieldValue struct { - ColumnID uint64 - Value int64 + ColumnID uint64 + ColumnKey string + Value int64 } // InternalClient should be implemented by any struct that enables any transport between nodes @@ -33,6 +34,7 @@ type InternalClient interface { Schema(ctx context.Context) ([]*IndexInfo, error) CreateIndex(ctx context.Context, index string, opt IndexOptions) error FragmentNodes(ctx context.Context, index string, shard uint64) ([]*Node, error) + Nodes(ctx context.Context) ([]*Node, error) Query(ctx context.Context, index string, queryRequest *QueryRequest) (*QueryResponse, error) QueryNode(ctx context.Context, uri *URI, index string, queryRequest *QueryRequest) (*QueryResponse, error) Import(ctx context.Context, index, field string, shard uint64, bits []Bit) error @@ -40,10 +42,11 @@ type InternalClient interface { EnsureIndex(ctx context.Context, name string, options IndexOptions) error EnsureField(ctx context.Context, indexName string, fieldName string) error ImportValue(ctx context.Context, index, field string, shard uint64, vals []FieldValue) error + ImportValueK(ctx context.Context, index, field string, vals []FieldValue) error ExportCSV(ctx context.Context, index, field string, shard uint64, w io.Writer) error CreateField(ctx context.Context, index, field string) error - FragmentBlocks(ctx context.Context, uri *URI, index, field string, shard uint64) ([]FragmentBlock, error) - BlockData(ctx context.Context, uri *URI, index, field string, shard uint64, block int) ([]uint64, []uint64, error) + FragmentBlocks(ctx context.Context, uri *URI, index, field, view string, shard uint64) ([]FragmentBlock, error) + BlockData(ctx context.Context, uri *URI, index, field, view string, shard uint64, block int) ([]uint64, []uint64, error) ColumnAttrDiff(ctx context.Context, uri *URI, index string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) RowAttrDiff(ctx context.Context, uri *URI, index, field string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) SendMessage(ctx context.Context, uri *URI, msg []byte) error @@ -88,6 +91,9 @@ func (n nopInternalClient) CreateIndex(ctx context.Context, index string, opt In func (n nopInternalClient) FragmentNodes(ctx context.Context, index string, shard uint64) ([]*Node, error) { return nil, nil } +func (n nopInternalClient) Nodes(ctx context.Context) ([]*Node, error) { + return nil, nil +} func (n nopInternalClient) Query(ctx context.Context, index string, queryRequest *QueryRequest) (*QueryResponse, error) { return nil, nil } @@ -109,14 +115,17 @@ func (n nopInternalClient) EnsureField(ctx context.Context, indexName string, fi func (n nopInternalClient) ImportValue(ctx context.Context, index, field string, shard uint64, vals []FieldValue) error { return nil } +func (n nopInternalClient) ImportValueK(ctx context.Context, index, field string, vals []FieldValue) 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) FragmentBlocks(ctx context.Context, uri *URI, index, field string, shard uint64) ([]FragmentBlock, error) { +func (n nopInternalClient) FragmentBlocks(ctx context.Context, uri *URI, index, field, view string, shard uint64) ([]FragmentBlock, error) { return nil, nil } -func (n nopInternalClient) BlockData(ctx context.Context, uri *URI, index, field string, shard uint64, block int) ([]uint64, []uint64, error) { +func (n nopInternalClient) BlockData(ctx context.Context, uri *URI, index, field, view string, shard uint64, block int) ([]uint64, []uint64, error) { return nil, nil, nil } func (n nopInternalClient) ColumnAttrDiff(ctx context.Context, uri *URI, index string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) { diff --git a/cluster.go b/cluster.go index 84f648d91..3c2f8070d 100644 --- a/cluster.go +++ b/cluster.go @@ -169,7 +169,7 @@ type nodeAction struct { type cluster struct { // nolint: maligned id string Node *Node - nodes []*Node // TODO phase this out? + nodes []*Node // Hashing algorithm used to assign partitions to nodes. Hasher Hasher @@ -489,12 +489,7 @@ func (c *cluster) receiveNodeState(nodeID string, state string) error { c.Topology.mu.Unlock() c.logger.Printf("received state %s (%s)", state, nodeID) - // Set cluster state to NORMAL. - if c.haveTopologyAgreement() && c.allNodesReady() { - return c.unprotectedSetStateAndBroadcast(ClusterStateNormal) - } - - return nil + return c.unprotectedSetStateAndBroadcast(c.determineClusterState()) } // determineClusterState is unprotected. @@ -983,7 +978,6 @@ func (c *cluster) allNodesReady() (ret bool) { } func (c *cluster) handleNodeAction(nodeAction nodeAction) error { - c.mu.Lock() j, err := c.unprotectedGenerateResizeJob(nodeAction) c.mu.Unlock() @@ -1007,7 +1001,7 @@ func (c *cluster) handleNodeAction(nodeAction nodeAction) error { c.logger.Printf("wait for jobResult") jobResult := <-j.result - // Make sure j.Run() didn't return an error. + // Make sure j.run() didn't return an error. if eg.Wait() != nil { return errors.Wrap(err, "running job") } @@ -1616,11 +1610,6 @@ func (c *cluster) considerTopology() error { return fmt.Errorf("coordinator %s is not in topology: %v", c.Node.ID, c.Topology.nodeIDs) } - // If local node is the only thing in .topology, continue. - //if len(c.Topology.NodeIDs) == 1 { - // return nil - //} - // Keep the cluster in state "STARTING" until hearing from all nodes. // Topology contains 2+ hosts. return nil @@ -1832,6 +1821,10 @@ func (c *cluster) mergeClusterStatus(cs *ClusterStatus) error { } } + // If the cluster membership has changed, reset the primary for + // translate store replication. + c.holder.setPrimaryTranslateStore(c.unprotectedPreviousNode()) + c.unprotectedSetState(cs.State) c.markAsJoined() @@ -1839,6 +1832,24 @@ func (c *cluster) mergeClusterStatus(cs *ClusterStatus) error { return nil } +// unprotectedPreviousNode returns the node listed before the current node in c.Nodes. +// If there is only one node in the cluster, returns nil. +// If the current node is the first node in the list, returns the last node. +func (c *cluster) unprotectedPreviousNode() *Node { + if len(c.nodes) <= 1 { + return nil + } + + pos := c.nodePositionByID(c.Node.ID) + if pos == -1 { + return nil + } else if pos == 0 { + return c.nodes[len(c.nodes)-1] + } else { + return c.nodes[pos-1] + } +} + // setStatic is unprotected, but only called before the cluster has been started // (and therefore not concurrently). func (c *cluster) setStatic(hosts []string) error { diff --git a/cluster_internal_test.go b/cluster_internal_test.go index 8bfd364bc..757e6c9b4 100644 --- a/cluster_internal_test.go +++ b/cluster_internal_test.go @@ -450,6 +450,60 @@ func TestCluster_Nodes(t *testing.T) { }) } +func TestCluster_PreviousNode(t *testing.T) { + node0 := &Node{ID: "node0"} + node1 := &Node{ID: "node1"} + node2 := &Node{ID: "node2"} + + t.Run("OneNode", func(t *testing.T) { + c := newCluster() + c.addNodeBasicSorted(node0) + + c.Node = node0 + if prev := c.unprotectedPreviousNode(); prev != nil { + t.Errorf("expected: nil, but got: %v", prev) + } + }) + + t.Run("TwoNode", func(t *testing.T) { + c := newCluster() + c.addNodeBasicSorted(node0) + c.addNodeBasicSorted(node1) + + c.Node = node0 + if prev := c.unprotectedPreviousNode(); prev != node1 { + t.Errorf("expected: node1, but got: %v", prev) + } + + c.Node = node1 + if prev := c.unprotectedPreviousNode(); prev != node0 { + t.Errorf("expected: node0, but got: %v", prev) + } + }) + + t.Run("ThreeNode", func(t *testing.T) { + c := newCluster() + c.addNodeBasicSorted(node0) + c.addNodeBasicSorted(node1) + c.addNodeBasicSorted(node2) + + c.Node = node0 + if prev := c.unprotectedPreviousNode(); prev != node2 { + t.Errorf("expected: node2, but got: %v", prev) + } + + c.Node = node1 + if prev := c.unprotectedPreviousNode(); prev != node0 { + t.Errorf("expected: node0, but got: %v", prev) + } + + c.Node = node2 + if prev := c.unprotectedPreviousNode(); prev != node1 { + t.Errorf("expected: node1, but got: %v", prev) + } + }) +} + // NEXT: move this test to internal and unexport IsCoordinator func TestCluster_Coordinator(t *testing.T) { uri1 := NewTestURIFromHostPort("node1", 0) diff --git a/cmd/import.go b/cmd/import.go index 22e9b0117..7e82bfd46 100644 --- a/cmd/import.go +++ b/cmd/import.go @@ -51,7 +51,7 @@ omitted. If it is present then its format should be YYYY-MM-DDTHH:MM. flags.StringVarP(&Importer.Host, "host", "", "localhost:10101", "host:port of Pilosa.") flags.StringVarP(&Importer.Index, "index", "i", "", "Pilosa index to import into.") flags.StringVarP(&Importer.Field, "field", "f", "", "Field to import into.") - flags.BoolVar(&Importer.StringKeys, "string-keys", false, "Treat payload as string keys.") + flags.BoolVar(&Importer.StringKeys, "string-keys", false, "REMOVED (key type is now determined by index/field configuration): Treat payload as string keys.") flags.IntVarP(&Importer.BufferSize, "buffer-size", "s", 10000000, "Number of bits to buffer/sort before importing.") flags.BoolVarP(&Importer.Sort, "sort", "", false, "Enables sorting before import.") flags.BoolVarP(&Importer.CreateSchema, "create", "e", false, "Create the schema if it does not exist before import.") diff --git a/ctl/import.go b/ctl/import.go index d3472acf1..2969b581b 100644 --- a/ctl/import.go +++ b/ctl/import.go @@ -46,7 +46,8 @@ type ImportCommand struct { // nolint: maligned // CreateSchema ensures the schema exists before import CreateSchema bool - // Indicates that the payload should be treated as string keys. + // REMOVED: Indicates that the payload should be treated as string keys. + // TODO: remove this in a future release StringKeys bool `json:"StringKeys"` // Filenames to import from. @@ -79,6 +80,11 @@ func NewImportCommand(stdin io.Reader, stdout, stderr io.Writer) *ImportCommand func (cmd *ImportCommand) Run(ctx context.Context) error { logger := log.New(cmd.Stderr, "", log.LstdFlags) + // REMOVED: warning that --string-keys flag has been deprecated. + if cmd.StringKeys { + logger.Printf("REMOVED: The string-keys flag is no longer used.") + } + // Validate arguments. // Index and field are validated early before the files are parsed. if cmd.Index == "" { @@ -108,20 +114,26 @@ func (cmd *ImportCommand) Run(ctx context.Context) error { if err != nil { return errors.Wrap(err, "getting schema") } + + var useColumnKeys, useRowKeys bool for _, index := range schema { if index.Name == cmd.Index { + useColumnKeys = index.Options.Keys for _, field := range index.Fields { if field.Name == cmd.Field { + useRowKeys = field.Options.Keys fieldType = field.Options.Type + break } } + break } } // Import each path and import by shard. for _, path := range cmd.Paths { logger.Printf("parsing: %s", path) - if err := cmd.importPath(ctx, fieldType, path); err != nil { + if err := cmd.importPath(ctx, fieldType, useColumnKeys, useRowKeys, path); err != nil { return err } } @@ -142,21 +154,16 @@ func (cmd *ImportCommand) ensureSchema(ctx context.Context) error { } // importPath parses a path into bits and imports it to the server. -func (cmd *ImportCommand) importPath(ctx context.Context, fieldType, path string) error { +func (cmd *ImportCommand) importPath(ctx context.Context, fieldType string, useColumnKeys, useRowKeys bool, path string) error { // If fieldType is `int`, treat the import data as values to be range-encoded. if fieldType == pilosa.FieldTypeInt { - return cmd.bufferValues(ctx, path) - } else { - if cmd.StringKeys { - return cmd.bufferBitsK(ctx, path) - } else { - return cmd.bufferBits(ctx, path) - } + return cmd.bufferValues(ctx, useColumnKeys, path) } + return cmd.bufferBits(ctx, useColumnKeys, useRowKeys, path) } // bufferBits buffers slices of bits to be imported as a batch. -func (cmd *ImportCommand) bufferBits(ctx context.Context, path string) error { +func (cmd *ImportCommand) bufferBits(ctx context.Context, useColumnKeys, useRowKeys bool, path string) error { a := make([]pilosa.Bit, 0, cmd.BufferSize) var r *csv.Reader @@ -198,18 +205,22 @@ func (cmd *ImportCommand) bufferBits(ctx context.Context, path string) error { var bit pilosa.Bit // Parse row id. - rowID, err := strconv.ParseUint(record[0], 10, 64) - if err != nil { - return fmt.Errorf("invalid row id on row %d: %q", rnum, record[0]) + if useRowKeys { + bit.RowKey = record[0] + } else { + if bit.RowID, err = strconv.ParseUint(record[0], 10, 64); err != nil { + return fmt.Errorf("invalid row id on row %d: %q", rnum, record[0]) + } } - bit.RowID = rowID // Parse column id. - columnID, err := strconv.ParseUint(record[1], 10, 64) - if err != nil { - return fmt.Errorf("invalid column id on row %d: %q", rnum, record[1]) + if useColumnKeys { + bit.ColumnKey = record[1] + } else { + if bit.ColumnID, err = strconv.ParseUint(record[1], 10, 64); err != nil { + return fmt.Errorf("invalid column id on row %d: %q", rnum, record[1]) + } } - bit.ColumnID = columnID // Parse time, if exists. if len(record) > 2 && record[2] != "" { @@ -224,7 +235,7 @@ func (cmd *ImportCommand) bufferBits(ctx context.Context, path string) error { // If we've reached the buffer size then import bits. if len(a) == cmd.BufferSize { - if err := cmd.importBits(ctx, a); err != nil { + if err := cmd.importBits(ctx, useColumnKeys, useRowKeys, a); err != nil { return err } a = a[:0] @@ -232,13 +243,22 @@ func (cmd *ImportCommand) bufferBits(ctx context.Context, path string) error { } // If there are still bits in the buffer then flush them. - return cmd.importBits(ctx, a) + return cmd.importBits(ctx, useColumnKeys, useRowKeys, a) } // importBits sends batches of bits to the server. -func (cmd *ImportCommand) importBits(ctx context.Context, bits []pilosa.Bit) error { +func (cmd *ImportCommand) importBits(ctx context.Context, useColumnKeys, useRowKeys bool, bits []pilosa.Bit) error { logger := log.New(cmd.Stderr, "", log.LstdFlags) + // If keys are used, all bits are sent to the primary translate store (i.e. coordinator). + if useColumnKeys || useRowKeys { + logger.Printf("importing keys: n=%d", len(bits)) + if err := cmd.client.ImportK(ctx, cmd.Index, cmd.Field, bits); err != nil { + return errors.Wrap(err, "importing keys") + } + return nil + } + // Group bits by shard. logger.Printf("grouping %d bits", len(bits)) bitsByShard := http.Bits(bits).GroupByShard() @@ -258,100 +278,8 @@ func (cmd *ImportCommand) importBits(ctx context.Context, bits []pilosa.Bit) err return nil } -// bufferBitsK buffers slices of keys to be imported as a batch. -func (cmd *ImportCommand) bufferBitsK(ctx context.Context, path string) error { - a := make([]pilosa.Bit, 0, cmd.BufferSize) - - var r *csv.Reader - - if path != "-" { - // Open file for reading. - f, err := os.Open(path) - if err != nil { - return errors.Wrap(err, "opening file") - } - defer f.Close() - - // Read rows as bits. - r = csv.NewReader(f) - } else { - r = csv.NewReader(cmd.Stdin) - } - - r.FieldsPerRecord = -1 - rnum := 0 - for { - rnum++ - - // Read CSV row. - record, err := r.Read() - if err == io.EOF { - break - } else if err != nil { - return errors.Wrap(err, "reading") - } - - // Ignore blank rows. - if record[0] == "" { - continue - } else if len(record) < 2 { - return fmt.Errorf("bad column count on row %d: col=%d", rnum, len(record)) - } - - var bit pilosa.Bit - - // Parse row key. - if record[0] == "" { - return fmt.Errorf("invalid row key on row %d: %q", rnum, record[0]) - } - bit.RowKey = record[0] - - // Parse column key. - if record[1] == "" { - return fmt.Errorf("invalid column id on row %d: %q", rnum, record[1]) - } - bit.ColumnKey = record[1] - - // Parse time, if exists. - if len(record) > 2 && record[2] != "" { - t, err := time.Parse(pilosa.TimeFormat, record[2]) - if err != nil { - return fmt.Errorf("invalid timestamp on row %d: %q", rnum, record[2]) - } - bit.Timestamp = t.UnixNano() - } - - a = append(a, bit) - - // If we've reached the buffer size then import bits. - if len(a) == cmd.BufferSize { - if err := cmd.importBitsK(ctx, a); err != nil { - return err - } - a = a[:0] - } - } - - // If there are still bitKs in the buffer then flush them. - return cmd.importBitsK(ctx, a) -} - -// importBitsK sends batches of bitKs to the server. -func (cmd *ImportCommand) importBitsK(ctx context.Context, bits []pilosa.Bit) error { - logger := log.New(cmd.Stderr, "", log.LstdFlags) - - // TODO: does it help to sort the rowKeys? - - logger.Printf("importing keys: n=%d", len(bits)) - if err := cmd.client.ImportK(ctx, cmd.Index, cmd.Field, bits); err != nil { - return errors.Wrap(err, "importing keys") - } - - return nil -} - // bufferValues buffers slices of FieldValues to be imported as a batch. -func (cmd *ImportCommand) bufferValues(ctx context.Context, path string) error { +func (cmd *ImportCommand) bufferValues(ctx context.Context, useColumnKeys bool, path string) error { a := make([]pilosa.FieldValue, 0, cmd.BufferSize) var r *csv.Reader @@ -393,11 +321,13 @@ func (cmd *ImportCommand) bufferValues(ctx context.Context, path string) error { var val pilosa.FieldValue // Parse column id. - columnID, err := strconv.ParseUint(record[0], 10, 64) - if err != nil { - return fmt.Errorf("invalid column id on row %d: %q", rnum, record[0]) + if useColumnKeys { + val.ColumnKey = record[0] + } else { + if val.ColumnID, err = strconv.ParseUint(record[0], 10, 64); err != nil { + return fmt.Errorf("invalid column id on row %d: %q", rnum, record[0]) + } } - val.ColumnID = columnID // Parse FieldValue. value, err := strconv.ParseInt(record[1], 10, 64) @@ -410,7 +340,7 @@ func (cmd *ImportCommand) bufferValues(ctx context.Context, path string) error { // If we've reached the buffer size then import FieldValues. if len(a) == cmd.BufferSize { - if err := cmd.importValues(ctx, a); err != nil { + if err := cmd.importValues(ctx, useColumnKeys, a); err != nil { return err } a = a[:0] @@ -418,13 +348,22 @@ func (cmd *ImportCommand) bufferValues(ctx context.Context, path string) error { } // If there are still values in the buffer then flush them. - return cmd.importValues(ctx, a) + return cmd.importValues(ctx, useColumnKeys, a) } // importValues sends batches of FieldValues to the server. -func (cmd *ImportCommand) importValues(ctx context.Context, vals []pilosa.FieldValue) error { +func (cmd *ImportCommand) importValues(ctx context.Context, useColumnKeys bool, vals []pilosa.FieldValue) error { logger := log.New(cmd.Stderr, "", log.LstdFlags) + // If keys are used, all values are sent to the primary translate store (i.e. coordinator). + if useColumnKeys { + logger.Printf("importing keyed values: n=%d", len(vals)) + if err := cmd.client.ImportValueK(ctx, cmd.Index, cmd.Field, vals); err != nil { + return errors.Wrap(err, "importing keys") + } + return nil + } + // Group vals by shard. logger.Printf("grouping %d vals", len(vals)) valsByShard := http.FieldValues(vals).GroupByShard() diff --git a/ctl/import_test.go b/ctl/import_test.go index a41f6d525..cd75a5d0a 100644 --- a/ctl/import_test.go +++ b/ctl/import_test.go @@ -101,6 +101,60 @@ func TestImportCommand_RunValue(t *testing.T) { } } +// Ensure that import with keys runs. +func TestImportCommand_RunKeys(t *testing.T) { + buf := bytes.Buffer{} + stdin, stdout, stderr := GetIO(buf) + cm := NewImportCommand(stdin, stdout, stderr) + file, err := ioutil.TempFile("", "import-key.csv") + file.Write([]byte("foo1,bar2\nfoo3,bar4\nfoo5,bar6")) + ctx := context.Background() + if err != nil { + t.Fatal(err) + } + + cmd := test.MustRunCluster(t, 1)[0] + cm.Host = cmd.API.Node().URI.HostPort() + + http.DefaultClient.Do(MustNewHTTPRequest("POST", "http://"+cm.Host+"/index/i", strings.NewReader(`{"options":{"keys": true}}`))) + http.DefaultClient.Do(MustNewHTTPRequest("POST", "http://"+cm.Host+"/index/i/field/f", strings.NewReader(`{"options":{"keys": true}}`))) + + cm.Index = "i" + cm.Field = "f" + cm.Paths = []string{file.Name()} + err = cm.Run(ctx) + if err != nil { + t.Fatalf("Import Run with keys doesn't work: %s", err) + } +} + +// Ensure that integer import with keys runs. +func TestImportCommand_RunValueKeys(t *testing.T) { + buf := bytes.Buffer{} + stdin, stdout, stderr := GetIO(buf) + cm := NewImportCommand(stdin, stdout, stderr) + file, err := ioutil.TempFile("", "import-key.csv") + file.Write([]byte("foo1,2\nfoo3,4\nfoo5,6")) + ctx := context.Background() + if err != nil { + t.Fatal(err) + } + + cmd := test.MustRunCluster(t, 1)[0] + cm.Host = cmd.API.Node().URI.HostPort() + + http.DefaultClient.Do(MustNewHTTPRequest("POST", "http://"+cm.Host+"/index/i", strings.NewReader(`{"options":{"keys": true}}`))) + http.DefaultClient.Do(MustNewHTTPRequest("POST", "http://"+cm.Host+"/index/i/field/f", strings.NewReader(`{"options":{"type": "int", "min": 0, "max": 100}}`))) + + cm.Index = "i" + cm.Field = "f" + cm.Paths = []string{file.Name()} + err = cm.Run(ctx) + if err != nil { + t.Fatalf("Import Run with keys doesn't work: %s", err) + } +} + func TestImportCommand_InvalidFile(t *testing.T) { cmd := test.MustRunCluster(t, 1)[0] diff --git a/ctl/server.go b/ctl/server.go index ad1ac5df5..90fdfb46e 100644 --- a/ctl/server.go +++ b/ctl/server.go @@ -44,7 +44,7 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) { flags.DurationVarP((*time.Duration)(&srv.Config.Cluster.LongQueryTime), "cluster.long-query-time", "", time.Minute, "Duration that will trigger log and stat messages for slow queries.") // Translation - flags.StringVarP(&srv.Config.Translation.PrimaryURL, "translation.primary-url", "", srv.Config.Translation.PrimaryURL, "URL for primary translation node for replication.") + flags.StringVarP(&srv.Config.Translation.PrimaryURL, "translation.primary-url", "", srv.Config.Translation.PrimaryURL, "DEPRECATED: URL for primary translation node for replication.") // Gossip flags.StringVarP(&srv.Config.Gossip.Port, "gossip.port", "", srv.Config.Gossip.Port, "Port to which pilosa should bind for internal state sharing.") diff --git a/encoding/proto/proto.go b/encoding/proto/proto.go index fcdde95f9..298683e92 100644 --- a/encoding/proto/proto.go +++ b/encoding/proto/proto.go @@ -890,6 +890,7 @@ func decodeRow(pr *internal.Row) *pilosa.Row { r := pilosa.NewRow() r.Attrs = decodeAttrs(pr.Attrs) + r.Keys = pr.Keys for _, v := range pr.Columns { r.SetBit(v) } diff --git a/fragment.go b/fragment.go index 2f57cccce..e4fbcbbda 100644 --- a/fragment.go +++ b/fragment.go @@ -1837,7 +1837,7 @@ func (s *fragmentSyncer) syncFragment() error { } // Retrieve remote blocks. - blocks, err := s.Cluster.InternalClient.FragmentBlocks(context.Background(), &node.URI, s.Fragment.index, s.Fragment.field, s.Fragment.shard) + blocks, err := s.Cluster.InternalClient.FragmentBlocks(context.Background(), &node.URI, s.Fragment.index, s.Fragment.field, s.Fragment.view, s.Fragment.shard) if err != nil && err != ErrFragmentNotFound { return errors.Wrap(err, "getting blocks") } @@ -1916,7 +1916,7 @@ func (s *fragmentSyncer) syncBlock(id int) error { uris = append(uris, uri) // Only sync the standard block. - rowIDs, columnIDs, err := s.Cluster.InternalClient.BlockData(context.Background(), &node.URI, f.index, f.field, f.shard, id) + rowIDs, columnIDs, err := s.Cluster.InternalClient.BlockData(context.Background(), &node.URI, f.index, f.field, f.view, f.shard, id) if err != nil { return errors.Wrap(err, "getting block") } diff --git a/holder.go b/holder.go index 3242a714d..15c8524b8 100644 --- a/holder.go +++ b/holder.go @@ -46,6 +46,10 @@ type Holder struct { // Indexes by name. indexes map[string]*Index + // Key/ID translation + translateFile *TranslateFile + NewPrimaryTranslateStore func(interface{}) TranslateStore + // opened channel is closed once Open() completes. opened chan struct{} @@ -77,6 +81,9 @@ func NewHolder() *Holder { opened: make(chan struct{}), + translateFile: NewTranslateFile(), + NewPrimaryTranslateStore: newNopTranslateStore, + broadcaster: NopBroadcaster, Stats: NopStatsClient, @@ -160,6 +167,13 @@ func (h *Holder) Close() error { return errors.Wrap(err, "closing index") } } + + if h.translateFile != nil { + if err := h.translateFile.Close(); err != nil { + return err + } + } + return nil } @@ -560,6 +574,14 @@ func (h *Holder) logStartup() error { return nil } +func (h *Holder) setPrimaryTranslateStore(node *Node) { + var nodeID string + if node != nil { + nodeID = node.ID + } + h.translateFile.SetPrimaryStore(nodeID, h.NewPrimaryTranslateStore(node)) +} + // holderSyncer is an active anti-entropy tool that compares the local holder // with a remote holder based on block checksums and resolves differences. type holderSyncer struct { @@ -638,7 +660,7 @@ func (s *holderSyncer) SyncHolder() error { // Sync fragment if own it. if err := s.syncFragment(di.Name, fi.Name, vi.Name, shard); err != nil { - return fmt.Errorf("fragment sync error: index=%s, field=%s, shard=%d, err=%s", di.Name, fi.Name, shard, err) + return fmt.Errorf("fragment sync error: index=%s, field=%s, view=%s, shard=%d, err=%s", di.Name, fi.Name, vi.Name, shard, err) } } } diff --git a/holder_test.go b/holder_test.go index d4886ff30..3920e4cc7 100644 --- a/holder_test.go +++ b/holder_test.go @@ -22,6 +22,7 @@ import ( "reflect" "strings" "testing" + "time" "github.com/pilosa/pilosa" "github.com/pilosa/pilosa/test" @@ -362,3 +363,56 @@ func TestHolderSyncer_SyncHolder(t *testing.T) { } } } + +// Ensure holder can sync time quantum views with a remote holder. +func TestHolderSyncer_TimeQuantum(t *testing.T) { + c := test.MustNewCluster(t, 2) + c[0].Config.Cluster.ReplicaN = 2 + c[0].Config.AntiEntropy.Interval = 0 + c[1].Config.Cluster.ReplicaN = 2 + c[1].Config.AntiEntropy.Interval = 0 + err := c.Start() + if err != nil { + t.Fatalf("starting cluster: %v", err) + } + defer c.Close() + + quantum := "D" + + _, err = c[0].API.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}) + if err != nil { + t.Fatalf("creating index i: %v", err) + } + _, err = c[0].API.CreateField(context.Background(), "i", "f", pilosa.OptFieldTypeTime(pilosa.TimeQuantum(quantum))) + if err != nil { + t.Fatalf("creating field f: %v", err) + } + + hldr0 := &test.Holder{Holder: c[0].Server.Holder()} + hldr1 := &test.Holder{Holder: c[1].Server.Holder()} + + // Set data on the local holder. + t1 := time.Date(2018, 8, 1, 12, 30, 0, 0, time.UTC) + t2 := time.Date(2018, 8, 2, 12, 30, 0, 0, time.UTC) + hldr0.SetBitTime("i", "f", 0, 1, &t1) + hldr0.SetBitTime("i", "f", 0, 2, &t2) + + err = c[0].Server.SyncData() + if err != nil { + t.Fatalf("syncing node 0: %v", err) + } + err = c[1].Server.SyncData() + if err != nil { + t.Fatalf("syncing node 1: %v", err) + } + + // Verify data is the same on both nodes. + for i, hldr := range []*test.Holder{hldr0, hldr1} { + if a := hldr.RowTime("i", "f", 0, t1, quantum).Columns(); !reflect.DeepEqual(a, []uint64{1}) { + t.Errorf("unexpected columns(%d/0): %+v", i, a) + } + if a := hldr.RowTime("i", "f", 0, t2, quantum).Columns(); !reflect.DeepEqual(a, []uint64{2}) { + t.Errorf("unexpected columns(%d/0): %+v", i, a) + } + } +} diff --git a/http/client.go b/http/client.go index bf5c9f5dc..d996b9e57 100644 --- a/http/client.go +++ b/http/client.go @@ -207,6 +207,37 @@ func (c *InternalClient) FragmentNodes(ctx context.Context, index string, shard return a, nil } +// Nodes returns a list of all nodes. +func (c *InternalClient) Nodes(ctx context.Context) ([]*pilosa.Node, error) { + // Execute request against the host. + u := uriPathToURL(c.defaultURI, "/internal/nodes") + + // Build request. + req, err := http.NewRequest("GET", u.String(), nil) + if err != nil { + return nil, errors.Wrap(err, "creating request") + } + + req.Header.Set("User-Agent", "pilosa/"+pilosa.Version) + req.Header.Set("Accept", "application/json") + + // Execute request. + resp, err := c.httpClient.Do(req.WithContext(ctx)) + if err != nil { + return nil, errors.Wrap(err, "executing request") + } + defer resp.Body.Close() + + var a []*pilosa.Node + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("http: status=%d", resp.StatusCode) + } else if err := json.NewDecoder(resp.Body).Decode(&a); err != nil { + return nil, fmt.Errorf("json decode: %s", err) + } + + return a, nil +} + // Query executes query against the index. func (c *InternalClient) Query(ctx context.Context, index string, queryRequest *pilosa.QueryRequest) (*pilosa.QueryResponse, error) { return c.QueryNode(ctx, c.defaultURI, index, queryRequest) @@ -291,26 +322,42 @@ func (c *InternalClient) Import(ctx context.Context, index, field string, shard return nil } +func getCoordinatorNode(nodes []*pilosa.Node) *pilosa.Node { + for _, node := range nodes { + if node.IsCoordinator { + return node + } + } + return nil +} + // ImportK bulk imports bits specified by string keys to a host. -func (c *InternalClient) ImportK(ctx context.Context, index, field string, columns []pilosa.Bit) error { +func (c *InternalClient) ImportK(ctx context.Context, index, field string, bits []pilosa.Bit) error { if index == "" { return pilosa.ErrIndexRequired } else if field == "" { return pilosa.ErrFieldRequired } - buf, err := c.marshalImportPayloadK(index, field, columns) + buf, err := c.marshalImportPayload(index, field, 0, bits) if err != nil { return fmt.Errorf("Error Creating Payload: %s", err) } - node := &pilosa.Node{ - URI: *c.defaultURI, + // Get the coordinator node; all bits are sent to the + // primary translate store (i.e. coordinator). + nodes, err := c.Nodes(ctx) + if err != nil { + return fmt.Errorf("getting nodes: %s", err) + } + coord := getCoordinatorNode(nodes) + if coord == nil { + return fmt.Errorf("could not find the coordinator node") } // Import to node. - if err := c.importNode(ctx, node, index, field, buf); err != nil { - return fmt.Errorf("import node: host=%s, err=%s", node.URI, err) + if err := c.importNode(ctx, coord, index, field, buf); err != nil { + return fmt.Errorf("import node: host=%s, err=%s", coord.URI, err) } return nil @@ -336,7 +383,9 @@ func (c *InternalClient) EnsureField(ctx context.Context, indexName string, fiel func (c *InternalClient) marshalImportPayload(index, field string, shard uint64, bits []pilosa.Bit) ([]byte, error) { // Separate row and column IDs to reduce allocations. rowIDs := Bits(bits).RowIDs() + rowKeys := Bits(bits).RowKeys() columnIDs := Bits(bits).ColumnIDs() + columnKeys := Bits(bits).ColumnKeys() timestamps := Bits(bits).Timestamps() // Marshal data to protobuf. @@ -345,27 +394,8 @@ func (c *InternalClient) marshalImportPayload(index, field string, shard uint64, Field: field, Shard: shard, RowIDs: rowIDs, - ColumnIDs: columnIDs, - Timestamps: timestamps, - }) - if err != nil { - return nil, fmt.Errorf("marshal import request: %s", err) - } - return buf, nil -} - -// marshalImportPayloadK marshalls the import parameters into a protobuf byte slice. -func (c *InternalClient) marshalImportPayloadK(index, field string, bits []pilosa.Bit) ([]byte, error) { - // Separate row and column IDs to reduce allocations. - rowKeys := Bits(bits).RowKeys() - columnKeys := Bits(bits).ColumnKeys() - timestamps := Bits(bits).Timestamps() - - // Marshal data to protobuf. - buf, err := c.serializer.Marshal(&pilosa.ImportRequest{ - Index: index, - Field: field, RowKeys: rowKeys, + ColumnIDs: columnIDs, ColumnKeys: columnKeys, Timestamps: timestamps, }) @@ -443,19 +473,53 @@ func (c *InternalClient) ImportValue(ctx context.Context, index, field string, s return nil } +// ImportValueK bulk imports keyed field values to a host. +func (c *InternalClient) ImportValueK(ctx context.Context, index, field string, vals []pilosa.FieldValue) error { + if index == "" { + return pilosa.ErrIndexRequired + } else if field == "" { + return pilosa.ErrFieldRequired + } + + buf, err := c.marshalImportValuePayload(index, field, 0, vals) + if err != nil { + return fmt.Errorf("Error Creating Payload: %s", err) + } + + // Get the coordinator node; all bits are sent to the + // primary translate store (i.e. coordinator). + nodes, err := c.Nodes(ctx) + if err != nil { + return fmt.Errorf("getting nodes: %s", err) + } + coord := getCoordinatorNode(nodes) + if coord == nil { + return fmt.Errorf("could not find the coordinator node") + } + + // Import to node. + if err := c.importNode(ctx, coord, index, field, buf); err != nil { + return fmt.Errorf("import node: host=%s, err=%s", coord.URI, err) + } + + return nil +} + // marshalImportValuePayload marshalls the import parameters into a protobuf byte slice. func (c *InternalClient) marshalImportValuePayload(index, field string, shard uint64, vals []pilosa.FieldValue) ([]byte, error) { // Separate row and column IDs to reduce allocations. columnIDs := FieldValues(vals).ColumnIDs() + columnKeys := FieldValues(vals).ColumnKeys() values := FieldValues(vals).Values() // Marshal data to protobuf. buf, err := c.serializer.Marshal(&pilosa.ImportValueRequest{ - Index: index, - Field: field, - Shard: shard, - ColumnIDs: columnIDs, - Values: values, + Index: index, + Field: field, + Shard: shard, + ColumnIDs: columnIDs, + ColumnKeys: columnKeys, + Values: values, }) if err != nil { return nil, fmt.Errorf("marshal import request: %s", err) @@ -624,7 +688,7 @@ func (c *InternalClient) CreateField(ctx context.Context, index, field string) e // 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 *pilosa.URI, index, field string, shard uint64) ([]pilosa.FragmentBlock, error) { +func (c *InternalClient) FragmentBlocks(ctx context.Context, uri *pilosa.URI, index, field, view string, shard uint64) ([]pilosa.FragmentBlock, error) { if uri == nil { uri = c.defaultURI } @@ -632,6 +696,7 @@ func (c *InternalClient) FragmentBlocks(ctx context.Context, uri *pilosa.URI, in u.RawQuery = url.Values{ "index": {index}, "field": {field}, + "view": {view}, "shard": {strconv.FormatUint(shard, 10)}, }.Encode() @@ -669,13 +734,14 @@ func (c *InternalClient) FragmentBlocks(ctx context.Context, uri *pilosa.URI, in } // BlockData returns row/column id pairs for a block. -func (c *InternalClient) BlockData(ctx context.Context, uri *pilosa.URI, index, field string, shard uint64, block int) ([]uint64, []uint64, error) { +func (c *InternalClient) BlockData(ctx context.Context, uri *pilosa.URI, index, field, view string, shard uint64, block int) ([]uint64, []uint64, error) { if uri == nil { panic("need to pass a URI to BlockData") } buf, err := c.serializer.Marshal(&pilosa.BlockDataRequest{ Index: index, Field: field, + View: view, Shard: shard, Block: uint64(block), }) @@ -858,8 +924,31 @@ func (p Bits) Less(i, j int) bool { return p[i].RowID < p[j].RowID } +// HasRowKeys returns true if any values use a row key. +func (p Bits) HasRowKeys() bool { + for i := range p { + if p[i].RowKey != "" { + return true + } + } + return false +} + +// HasColumnKeys returns true if any values use a column key. +func (p Bits) HasColumnKeys() bool { + for i := range p { + if p[i].ColumnKey != "" { + return true + } + } + return false +} + // RowIDs returns a slice of all the row IDs. func (p Bits) RowIDs() []uint64 { + if p.HasRowKeys() { + return nil + } other := make([]uint64, len(p)) for i := range p { other[i] = p[i].RowID @@ -869,6 +958,9 @@ func (p Bits) RowIDs() []uint64 { // ColumnIDs returns a slice of all the column IDs. func (p Bits) ColumnIDs() []uint64 { + if p.HasColumnKeys() { + return nil + } other := make([]uint64, len(p)) for i := range p { other[i] = p[i].ColumnID @@ -878,6 +970,9 @@ func (p Bits) ColumnIDs() []uint64 { // RowKeys returns a slice of all the row keys. func (p Bits) RowKeys() []string { + if !p.HasRowKeys() { + return nil + } other := make([]string, len(p)) for i := range p { other[i] = p[i].RowKey @@ -887,6 +982,9 @@ func (p Bits) RowKeys() []string { // ColumnKeys returns a slice of all the column keys. func (p Bits) ColumnKeys() []string { + if !p.HasColumnKeys() { + return nil + } other := make([]string, len(p)) for i := range p { other[i] = p[i].ColumnKey @@ -929,8 +1027,21 @@ func (p FieldValues) Less(i, j int) bool { return p[i].ColumnID < p[j].ColumnID } +// HasColumnKeys returns true if any values use a column key. +func (p FieldValues) HasColumnKeys() bool { + for i := range p { + if p[i].ColumnKey != "" { + return true + } + } + return false +} + // ColumnIDs returns a slice of all the column IDs. func (p FieldValues) ColumnIDs() []uint64 { + if p.HasColumnKeys() { + return nil + } other := make([]uint64, len(p)) for i := range p { other[i] = p[i].ColumnID @@ -938,6 +1049,18 @@ func (p FieldValues) ColumnIDs() []uint64 { return other } +// ColumnKeys returns a slice of all the column keys. +func (p FieldValues) ColumnKeys() []string { + if !p.HasColumnKeys() { + return nil + } + other := make([]string, len(p)) + for i := range p { + other[i] = p[i].ColumnKey + } + return other +} + // Values returns a slice of all the values. func (p FieldValues) Values() []int64 { other := make([]int64, len(p)) diff --git a/http/client_test.go b/http/client_test.go index 0d60b443e..87c433f05 100644 --- a/http/client_test.go +++ b/http/client_test.go @@ -20,6 +20,7 @@ import ( gohttp "net/http" "reflect" "testing" + "time" "github.com/davecgh/go-spew/spew" "github.com/pilosa/pilosa" @@ -128,11 +129,6 @@ func TestClient_MultiNode(t *testing.T) { Remote: false, } - _, err = client[0].Query(context.Background(), "i", queryRequest) - if err != nil { - t.Fatal(err) - } - result, err := client[0].Query(context.Background(), "i", queryRequest) if err != nil { t.Fatal(err) @@ -202,6 +198,231 @@ func TestClient_Import(t *testing.T) { } } +// Ensure client can bulk import data. +func TestClient_ImportKeys(t *testing.T) { + t.Run("SingleNode", func(t *testing.T) { + cmd := test.MustRunCluster(t, 1)[0] + host := cmd.URL() + + cmd.MustCreateIndex(t, "keyed", pilosa.IndexOptions{Keys: true}) + cmd.MustCreateIndex(t, "unkeyed", pilosa.IndexOptions{Keys: false}) + + cmd.MustCreateField(t, "keyed", "keyedf", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 1000), pilosa.OptFieldKeys()) + cmd.MustCreateField(t, "keyed", "unkeyedf", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 1000)) + cmd.MustCreateField(t, "unkeyed", "keyedf", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 1000), pilosa.OptFieldKeys()) + + // Send import request. + c := MustNewClient(host, http.GetHTTPClient(nil)) + + t.Run("Import keyed,keyed", func(t *testing.T) { + if err := c.Import(context.Background(), "keyed", "keyedf", 0, []pilosa.Bit{ + {RowKey: "green", ColumnKey: "eve"}, + {RowKey: "green", ColumnKey: "alice"}, + {RowKey: "green", ColumnKey: "bob"}, + {RowKey: "blue", ColumnKey: "eve"}, + {RowKey: "blue", ColumnKey: "alice"}, + {RowKey: "purple", ColumnKey: "eve"}, + }); err != nil { + t.Fatal(err) + } + cmd.MustRecalculateCaches(t) + resp := cmd.MustQuery(t, &pilosa.QueryRequest{ + Index: "keyed", + Query: "TopN(keyedf)", + }) + if pairs, ok := resp.Results[0].([]pilosa.Pair); !ok { + t.Fatalf("unexpected response type %T", resp.Results[0]) + } else if !reflect.DeepEqual(pairs, []pilosa.Pair{ + {Key: "green", Count: 3}, + {Key: "blue", Count: 2}, + {Key: "purple", Count: 1}, + }) { + t.Fatalf("unexpected topn result: %v", pairs) + } + }) + + t.Run("Import keyed,unkeyedf", func(t *testing.T) { + if err := c.Import(context.Background(), "keyed", "unkeyedf", 0, []pilosa.Bit{ + {RowID: 1, ColumnKey: "eve"}, + {RowID: 1, ColumnKey: "alice"}, + {RowID: 1, ColumnKey: "bob"}, + {RowID: 2, ColumnKey: "eve"}, + {RowID: 2, ColumnKey: "alice"}, + {RowID: 3, ColumnKey: "eve"}, + }); err != nil { + t.Fatal(err) + } + cmd.MustRecalculateCaches(t) + resp := cmd.MustQuery(t, &pilosa.QueryRequest{ + Index: "keyed", + Query: "TopN(unkeyedf)", + }) + if pairs, ok := resp.Results[0].([]pilosa.Pair); !ok { + t.Fatalf("unexpected response type %T", resp.Results[0]) + } else if !reflect.DeepEqual(pairs, []pilosa.Pair{ + {ID: 1, Count: 3}, + {ID: 2, Count: 2}, + {ID: 3, Count: 1}, + }) { + t.Fatalf("unexpected topn result: %v", pairs) + } + }) + + t.Run("Import unkeyed,keyed", func(t *testing.T) { + if err := c.Import(context.Background(), "unkeyed", "keyedf", 0, []pilosa.Bit{ + {RowKey: "green", ColumnID: 1}, + {RowKey: "green", ColumnID: 2}, + {RowKey: "green", ColumnID: 3}, + {RowKey: "blue", ColumnID: 1}, + {RowKey: "blue", ColumnID: 2}, + {RowKey: "purple", ColumnID: 1}, + }); err != nil { + t.Fatal(err) + } + cmd.MustRecalculateCaches(t) + resp := cmd.MustQuery(t, &pilosa.QueryRequest{ + Index: "unkeyed", + Query: "TopN(keyedf)", + }) + if pairs, ok := resp.Results[0].([]pilosa.Pair); !ok { + t.Fatalf("unexpected response type %T", resp.Results[0]) + } else if !reflect.DeepEqual(pairs, []pilosa.Pair{ + {Key: "green", Count: 3}, + {Key: "blue", Count: 2}, + {Key: "purple", Count: 1}, + }) { + t.Fatalf("unexpected topn result: %v", pairs) + } + }) + }) + + t.Run("MultiNode", func(t *testing.T) { + cluster := test.MustRunCluster(t, 2) + cmd0 := cluster[0] + cmd1 := cluster[1] + host0 := cmd0.URL() + host1 := cmd1.URL() + + cmd0.MustCreateIndex(t, "keyed", pilosa.IndexOptions{Keys: true}) + cmd0.MustCreateField(t, "keyed", "keyedf0", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 1000), pilosa.OptFieldKeys()) + 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)) + + // Import to node0. + t.Run("Import node0", func(t *testing.T) { + if err := c0.ImportK(context.Background(), "keyed", "keyedf0", []pilosa.Bit{ + {RowKey: "green", ColumnKey: "eve"}, + {RowKey: "green", ColumnKey: "alice"}, + {RowKey: "green", ColumnKey: "bob"}, + {RowKey: "blue", ColumnKey: "eve"}, + {RowKey: "blue", ColumnKey: "alice"}, + {RowKey: "purple", ColumnKey: "eve"}, + }); err != nil { + t.Fatal(err) + } + cmd0.MustRecalculateCaches(t) + resp := cmd0.MustQuery(t, &pilosa.QueryRequest{ + Index: "keyed", + Query: "TopN(keyedf0)", + }) + if pairs, ok := resp.Results[0].([]pilosa.Pair); !ok { + t.Fatalf("unexpected response type %T", resp.Results[0]) + } else if !reflect.DeepEqual(pairs, []pilosa.Pair{ + {Key: "green", Count: 3}, + {Key: "blue", Count: 2}, + {Key: "purple", Count: 1}, + }) { + t.Fatalf("unexpected topn result: %v", pairs) + } + }) + + // Import to node1 (ensure import is routed to coordinator for translation). + t.Run("Import node1", func(t *testing.T) { + if err := c1.ImportK(context.Background(), "keyed", "keyedf1", []pilosa.Bit{ + {RowKey: "green", ColumnKey: "eve"}, + {RowKey: "green", ColumnKey: "alice"}, + {RowKey: "green", ColumnKey: "bob"}, + {RowKey: "blue", ColumnKey: "eve"}, + {RowKey: "blue", ColumnKey: "alice"}, + {RowKey: "purple", ColumnKey: "eve"}, + }); err != nil { + t.Fatal(err) + } + + // Wait for translation replication. + time.Sleep(500 * time.Millisecond) + + cmd1.MustRecalculateCaches(t) + resp := cmd1.MustQuery(t, &pilosa.QueryRequest{ + Index: "keyed", + Query: "TopN(keyedf1)", + }) + if pairs, ok := resp.Results[0].([]pilosa.Pair); !ok { + t.Fatalf("unexpected response type %T", resp.Results[0]) + } else if !reflect.DeepEqual(pairs, []pilosa.Pair{ + {Key: "green", Count: 3}, + {Key: "blue", Count: 2}, + {Key: "purple", Count: 1}, + }) { + t.Fatalf("unexpected topn result: %v", pairs) + } + }) + }) + + t.Run("IntegerFieldSingleNode", func(t *testing.T) { + cmd := test.MustRunCluster(t, 1)[0] + host := cmd.URL() + holder := cmd.Server.Holder() + hldr := test.Holder{Holder: holder} + + fldName := "f" + + // Load bitmap into cache to ensure cache gets updated. + index := hldr.MustCreateIndexIfNotExists("i", pilosa.IndexOptions{Keys: true}) + field, err := index.CreateFieldIfNotExists(fldName, pilosa.OptFieldTypeInt(-100, 100)) + if err != nil { + t.Fatal(err) + } + + // Send import request. + c := MustNewClient(host, http.GetHTTPClient(nil)) + if err := c.ImportValue(context.Background(), "i", "f", 0, []pilosa.FieldValue{ + {ColumnKey: "col1", Value: -10}, + {ColumnKey: "col2", Value: 20}, + {ColumnKey: "col3", Value: 40}, + }); err != nil { + t.Fatal(err) + } + + // Verify Sum. + sum, cnt, err := field.Sum(nil, fldName) + if err != nil { + t.Fatal(err) + } + if sum != 50 || cnt != 3 { + t.Fatalf("unexpected values: got sum=%v, count=%v; expected sum=50, cnt=3", sum, cnt) + } + + // Verify Range + queryRequest := &pilosa.QueryRequest{ + Query: fmt.Sprintf(`Range(%s>10)`, fldName), + Remote: false, + } + + result, err := c.Query(context.Background(), "i", queryRequest) + if err != nil { + t.Fatal(err) + } + + if !reflect.DeepEqual(result.Results[0].(*pilosa.Row).Keys, []string{"col2", "col3"}) { + t.Fatalf("unexpected column keys: %s", spew.Sdump(result)) + } + }) +} + // Ensure client can bulk import value data. func TestClient_ImportValue(t *testing.T) { cmd := test.MustRunCluster(t, 1)[0] @@ -281,7 +502,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)) - blocks, err := c.FragmentBlocks(context.Background(), nil, "i", "f", 0) + blocks, err := c.FragmentBlocks(context.Background(), nil, "i", "f", "standard", 0) if err != nil { t.Fatal(err) } else if len(blocks) != 2 { @@ -293,7 +514,7 @@ func TestClient_FragmentBlocks(t *testing.T) { } // Verify data matches local blocks. - if a, err := cmd.API.FragmentBlocks(context.Background(), "i", "f", 0); err != nil { + if a, err := cmd.API.FragmentBlocks(context.Background(), "i", "f", "standard", 0); err != nil { t.Fatal(err) } else if !reflect.DeepEqual(a, blocks) { t.Fatalf("blocks mismatch:\n\nexp=%s\n\ngot=%s\n\n", spew.Sdump(a), spew.Sdump(blocks)) diff --git a/http/handler.go b/http/handler.go index 80d1563a0..0624cfb99 100644 --- a/http/handler.go +++ b/http/handler.go @@ -176,7 +176,7 @@ func (h *Handler) populateValidators() { h.validators["GetExport"] = queryValidationSpecRequired("index", "field", "shard") h.validators["GetFragmentData"] = queryValidationSpecRequired("index", "field", "shard") h.validators["PostFragmentData"] = queryValidationSpecRequired("index", "field", "shard") - h.validators["GetFragmentBlocks"] = queryValidationSpecRequired("index", "field", "shard") + h.validators["GetFragmentBlocks"] = queryValidationSpecRequired("index", "field", "view", "shard") } func (h *Handler) queryArgValidator(next http.Handler) http.Handler { @@ -232,6 +232,7 @@ func newRouter(handler *Handler) *mux.Router { router.HandleFunc("/internal/fragment/nodes", handler.handleGetFragmentNodes).Methods("GET").Name("GetFragmentNodes") router.HandleFunc("/internal/index/{index}/attr/diff", handler.handlePostIndexAttrDiff).Methods("POST") router.HandleFunc("/internal/index/{index}/field/{field}/attr/diff", handler.handlePostFieldAttrDiff).Methods("POST") + router.HandleFunc("/internal/nodes", handler.handleGetNodes).Methods("GET").Name("GetNodes") router.HandleFunc("/internal/shards/max", handler.handleGetShardsMax).Methods("GET") // TODO: deprecate, but it's being used by the client router.HandleFunc("/internal/translate/data", handler.handleGetTranslateData).Methods("GET") @@ -1062,6 +1063,22 @@ func (h *Handler) handleGetFragmentNodes(w http.ResponseWriter, r *http.Request) } } +// handleGetNodes handles /internal/nodes requests. +func (h *Handler) handleGetNodes(w http.ResponseWriter, r *http.Request) { + if !validHeaderAcceptJSON(r.Header) { + http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable) + return + } + + // Retrieve all nodes. + nodes := h.api.Hosts(r.Context()) + + // Write to response. + if err := json.NewEncoder(w).Encode(nodes); err != nil { + h.logger.Printf("json write error: %s", err) + } +} + // handleGetFragmentBlockData handles GET /internal/fragment/block/data requests. func (h *Handler) handleGetFragmentBlockData(w http.ResponseWriter, r *http.Request) { buf, err := h.api.FragmentBlockData(r.Context(), r.Body) @@ -1096,7 +1113,7 @@ func (h *Handler) handleGetFragmentBlocks(w http.ResponseWriter, r *http.Request return } - blocks, err := h.api.FragmentBlocks(r.Context(), q.Get("index"), q.Get("field"), shard) + blocks, err := h.api.FragmentBlocks(r.Context(), q.Get("index"), q.Get("field"), q.Get("view"), shard) if err != nil { if errors.Cause(err) == pilosa.ErrFragmentNotFound { http.Error(w, err.Error(), http.StatusNotFound) diff --git a/http/translator.go b/http/translator.go index 80c73bf5e..7ca85a33f 100644 --- a/http/translator.go +++ b/http/translator.go @@ -6,11 +6,13 @@ import ( "fmt" "io" "io/ioutil" + "log" "net/http" "net/url" "strconv" "github.com/pilosa/pilosa" + "github.com/pkg/errors" ) // Ensure implementation implements inteface. @@ -19,12 +21,31 @@ var _ pilosa.TranslateStore = (*translateStore)(nil) // translateStore represents an implementation of pilosa.TranslateStore that // communicates over HTTP. This is used with the TranslateHandler. type translateStore struct { - URL string + node *pilosa.Node } -// NewTranslateStore returns a new instance of TranslateStore. -func NewTranslateStore(rawurl string) *translateStore { - return &translateStore{URL: rawurl} +// NewTranslateStore returns a new instance of TranslateStore based on node. +// DEPRECATED: Providing a string url to this function is being deprecated. Instead, +// provide a *pilosa.Node. +func NewTranslateStore(node interface{}) pilosa.TranslateStore { + var n *pilosa.Node + switch v := node.(type) { + case string: + log.Printf("WARNING: providing a string url to NewTranslateStore() has been deprecated.") + if uri, err := pilosa.NewURIFromAddress(v); err != nil { + log.Println(errors.Wrap(err, "creating uri")) + } else { + n = &pilosa.Node{ + ID: v, + URI: *uri, + } + } + case *pilosa.Node: + n = v + default: + log.Printf("WARNING: a *pilosa.Node is the only type supported by NewTranslateStore().") + } + return &translateStore{node: n} } // TranslateColumnsToUint64 is not currently implemented. @@ -50,7 +71,7 @@ func (s *translateStore) TranslateRowToString(index, frame string, values uint64 // Reader returns a reader that can stream data from a remote store. func (s *translateStore) Reader(ctx context.Context, off int64) (io.ReadCloser, error) { // Generate remote URL. - u, err := url.Parse(s.URL) + u, err := url.Parse(s.node.URI.String()) if err != nil { return nil, err } diff --git a/server.go b/server.go index 819d1d80d..70fd1398e 100644 --- a/server.go +++ b/server.go @@ -49,7 +49,6 @@ type Server struct { // nolint: maligned // Internal holder *Holder cluster *cluster - translateFile *TranslateFile diagnostics *diagnosticsCollector executor *executor hosts []string @@ -70,8 +69,6 @@ type Server struct { // nolint: maligned isCoordinator bool syncer holderSyncer - primaryTranslateStore TranslateStore - defaultClient InternalClient dataDir string } @@ -163,9 +160,17 @@ func OptServerInternalClient(c InternalClient) ServerOption { } } +// DEPRECATED func OptServerPrimaryTranslateStore(store TranslateStore) ServerOption { return func(s *Server) error { - s.primaryTranslateStore = store + s.logger.Printf("DEPRECATED: OptServerPrimaryTranslateStore") + return nil + } +} + +func OptServerPrimaryTranslateStoreFunc(tf func(interface{}) TranslateStore) ServerOption { + return func(s *Server) error { + s.holder.NewPrimaryTranslateStore = tf return nil } } @@ -265,6 +270,7 @@ func NewServer(opts ...ServerOption) (*Server, error) { } s.holder.Path = path + s.holder.translateFile.Path = filepath.Join(path, ".keys") s.holder.Logger = s.logger s.holder.Stats.SetLogger(s.logger) @@ -272,11 +278,6 @@ func NewServer(opts ...ServerOption) (*Server, error) { s.cluster.logger = s.logger s.cluster.holder = s.holder - // Initialize translation database. - s.translateFile = NewTranslateFile() - s.translateFile.Path = filepath.Join(path, ".keys") - s.translateFile.PrimaryTranslateStore = s.primaryTranslateStore - // Get or create NodeID. s.nodeID = s.loadNodeID() if s.isCoordinator { @@ -303,7 +304,7 @@ func NewServer(opts ...ServerOption) (*Server, error) { s.executor.Holder = s.holder s.executor.Node = node s.executor.Cluster = s.cluster - s.executor.TranslateStore = s.translateFile + s.executor.TranslateStore = s.holder.translateFile s.executor.MaxWritesPerRequest = s.maxWritesPerRequest s.cluster.broadcaster = s s.cluster.maxWritesPerRequest = s.maxWritesPerRequest @@ -328,7 +329,7 @@ func (s *Server) Open() error { } // Initialize id-key storage. - if err := s.translateFile.Open(); err != nil { + if err := s.holder.translateFile.Open(); err != nil { return err } @@ -374,7 +375,6 @@ func (s *Server) Close() error { s.wg.Wait() var errh error - var errt error var errc error if s.cluster != nil { errc = s.cluster.close() @@ -382,17 +382,12 @@ func (s *Server) Close() error { if s.holder != nil { errh = s.holder.Close() } - if s.translateFile != nil { - errt = s.translateFile.Close() - } - // prefer to return holder error over translateFile error over cluster + // prefer to return holder error over cluster // error. This order is somewhat arbitrary. It would be better if we had // some way to combine all the errors, but probably not important enough to // warrant the extra complexity. if errh != nil { return errors.Wrap(errh, "closing holder") - } else if errt != nil { - return errors.Wrap(errt, "closing translateFile") } return errors.Wrap(errc, "closing cluster") } diff --git a/server/config.go b/server/config.go index 39a769ce2..8367eb835 100644 --- a/server/config.go +++ b/server/config.go @@ -71,7 +71,7 @@ type Config struct { // Gossip config is based around memberlist.Config. Gossip gossip.Config `toml:"gossip"` - // Translation config supports translation store replication. + // DEPRECATED: Translation config supports translation store replication. Translation struct { PrimaryURL string `toml:"primary-url"` } `toml:"translation"` diff --git a/server/server.go b/server/server.go index 1273a238f..4fdd711fe 100644 --- a/server/server.go +++ b/server/server.go @@ -250,10 +250,9 @@ func (m *Command) SetupServer() error { c := http.GetHTTPClient(TLSConfig) - // Setup connection to primary store if this is a replica. - var primaryTranslateStore pilosa.TranslateStore + // Primary store configuration is handled automatically now. if m.Config.Translation.PrimaryURL != "" { - primaryTranslateStore = http.NewTranslateStore(m.Config.Translation.PrimaryURL) + m.logger.Printf("DEPRECATED: The primary-url configuration option is no longer used.") } // Set Coordinator. @@ -278,7 +277,7 @@ func (m *Command) SetupServer() error { pilosa.OptServerStatsClient(statsClient), pilosa.OptServerURI(uri), pilosa.OptServerInternalClient(http.NewInternalClientFromURI(uri, c)), - pilosa.OptServerPrimaryTranslateStore(primaryTranslateStore), + pilosa.OptServerPrimaryTranslateStoreFunc(http.NewTranslateStore), pilosa.OptServerClusterDisabled(m.Config.Cluster.Disabled, m.Config.Cluster.Hosts), pilosa.OptServerSerializer(proto.Serializer{}), coordinatorOpt, diff --git a/server/server_test.go b/server/server_test.go index 1f00ebeb5..ab860dc1b 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -489,6 +489,10 @@ func TestClusteringNodesReplica2(t *testing.T) { t.Fatalf("restarting node 2: %v", err) } + if cluster[0].API.State() != pilosa.ClusterStateDegraded { + t.Fatalf("expected state to be DEGRADED, but got %s", cluster[0].API.State()) + } + // Create new main with the same config. config = cluster[1].Command.Config // config.Bind = cluster[1].API.Node().URI.HostPort() diff --git a/test/holder.go b/test/holder.go index 1d9790e2f..80778fe2a 100644 --- a/test/holder.go +++ b/test/holder.go @@ -113,14 +113,19 @@ func (h *Holder) RowTime(index, field string, rowID uint64, t time.Time, quantum return row } -// SetBit clears a bit on the given field. +// SetBit sets a bit on the given field. func (h *Holder) SetBit(index, field string, rowID, columnID uint64) { + h.SetBitTime(index, field, rowID, columnID, nil) +} + +// SetBitTime sets a bit with timestamp on the given field. +func (h *Holder) SetBitTime(index, field string, rowID, columnID uint64, t *time.Time) { idx := h.MustCreateIndexIfNotExists(index, pilosa.IndexOptions{}) f, err := idx.CreateFieldIfNotExists(field, pilosa.OptFieldTypeDefault()) if err != nil { panic(err) } - _, err = f.SetBit(rowID, columnID, nil) + _, err = f.SetBit(rowID, columnID, t) if err != nil { panic(err) } diff --git a/test/pilosa.go b/test/pilosa.go index 743592569..f6ed907ef 100644 --- a/test/pilosa.go +++ b/test/pilosa.go @@ -16,6 +16,7 @@ package test import ( "bytes" + "context" "fmt" "io" "io/ioutil" @@ -27,6 +28,7 @@ import ( "testing" "time" + "github.com/pilosa/pilosa" "github.com/pilosa/pilosa/http" "github.com/pilosa/pilosa/server" "github.com/pkg/errors" @@ -125,6 +127,45 @@ func (m *Command) Reopen() error { return m.Start() } +// MustCreateIndex uses this command's API to create an index and fails the test +// if there is an error. +func (m *Command) MustCreateIndex(t *testing.T, name string, opts pilosa.IndexOptions) *pilosa.Index { + idx, err := m.API.CreateIndex(context.Background(), name, opts) + if err != nil { + t.Fatalf("creating index: %v with options: %v, err: %v", name, opts, err) + } + return idx +} + +// MustCreateField uses this command's API to create the field. The index must +// already exist - it fails the test if there is an error. +func (m *Command) MustCreateField(t *testing.T, index, field string, opts ...pilosa.FieldOption) *pilosa.Field { + f, err := m.API.CreateField(context.Background(), index, field, opts...) + if err != nil { + t.Fatalf("creating field: %s in index: %s err: %v", field, index, err) + } + return f +} + +// MustQuery uses this command's API to execute the given query request, failing +// if Query returns a non-nil error, otherwise returning the QueryResponse. +func (m *Command) MustQuery(t *testing.T, req *pilosa.QueryRequest) pilosa.QueryResponse { + resp, err := m.API.Query(context.Background(), req) + if err != nil { + t.Fatalf("making query: %v, err: %v", req, err) + } + return resp +} + +// MustRecalculateCaches calls RecalculateCaches on the command's API, and fails +// if there is an error. +func (m *Command) MustRecalculateCaches(t *testing.T) { + err := m.API.RecalculateCaches(context.Background()) + if err != nil { + t.Fatalf("recalcluating caches: %v", err) + } +} + // URL returns the base URL string for accessing the running program. func (m *Command) URL() string { return m.API.Node().URI.String() } @@ -146,6 +187,7 @@ func (m *Command) Query(index, rawQuery, query string) (string, error) { return resp.Body, nil } +// RecalculateCaches is deprecated. Use MustRecalculateCaches. func (m *Command) RecalculateCaches() error { resp := MustDo("POST", fmt.Sprintf("%s/recalculate-caches", m.URL()), "") if resp.StatusCode != 204 { diff --git a/translate.go b/translate.go index 20696ef6c..32bcc1126 100644 --- a/translate.go +++ b/translate.go @@ -8,6 +8,7 @@ import ( "errors" "fmt" "io" + "io/ioutil" "log" "os" "path/filepath" @@ -71,6 +72,10 @@ type TranslateFile struct { // If non-nil, data is streamed from a primary and this is a read-only store. PrimaryTranslateStore TranslateStore + primaryID string // unique ID used to identify the primary store + replicationClosing chan struct{} + primaryStoreEvents chan primaryStoreEvent + repWG sync.WaitGroup // Delay after attempting to connect to a primary that the store will retry. replicationRetryInterval time.Duration @@ -86,6 +91,9 @@ func NewTranslateFile() *TranslateFile { mapSize: defaultMapSize, + replicationClosing: make(chan struct{}), + primaryStoreEvents: make(chan primaryStoreEvent), + replicationRetryInterval: defaultReplicationRetryInterval, } } @@ -109,10 +117,65 @@ func (s *TranslateFile) Open() (err error) { return err } - // Stream from primary, if available. + // Listen to primaryStoreEvents channel. + s.wg.Add(1) + go func() { defer s.wg.Done(); s.monitorPrimaryStoreEvents() }() + + return nil +} + +// primaryStoreEvent is used to set/change the primary translate store. +// It contains a TranslateStore along with an associated string ID which +// is used to determine whether the primary needs to be changed from the +// current value. +type primaryStoreEvent struct { + id string + ts TranslateStore +} + +// SetPrimaryStore sets the translate files's primary translate store. +// The id value is used to determine whether the primary needs to be changed +// from the current value (i.e. calling this multiple times with the same +// input values will no-op on all subsequent calls). +func (s *TranslateFile) SetPrimaryStore(id string, ts TranslateStore) { + go func() { + s.primaryStoreEvents <- primaryStoreEvent{ + id: id, + ts: ts, + } + }() +} + +// handlePrimaryStoreEvent changes the PrimaryTranslateStore +// used for replication by TranslateFile. +func (s *TranslateFile) handlePrimaryStoreEvent(ev primaryStoreEvent) error { + s.mu.Lock() + defer s.mu.Unlock() + + if ev.id == s.primaryID { + return nil + } + + // Stop translate store replication. + log.Printf("stop monitor replication") + close(s.replicationClosing) + s.repWG.Wait() + + // Set the primary node for translate store replication. + log.Printf("set primary translate store to %s", ev.id) + s.primaryID = ev.id + if ev.id == "" { + s.PrimaryTranslateStore = nil + } else { + s.PrimaryTranslateStore = ev.ts + } + + // Start translate store replication. Stream from primary, if available. + log.Printf("start monitor replication") if s.PrimaryTranslateStore != nil { - s.wg.Add(1) - go func() { defer s.wg.Done(); s.monitorReplication() }() + s.replicationClosing = make(chan struct{}) + s.repWG.Add(1) + go func() { defer s.repWG.Done(); s.monitorReplication() }() } return nil @@ -259,7 +322,13 @@ func (s *TranslateFile) replayEntries() error { func (s *TranslateFile) monitorReplication() { // Create context that will cancel on close. ctx, cancel := context.WithCancel(context.Background()) - go func() { <-s.closing; cancel() }() + go func() { + select { + case <-s.closing: + case <-s.replicationClosing: + } + cancel() + }() // Keep attempting to replicate until the store closes. for { @@ -270,12 +339,31 @@ func (s *TranslateFile) monitorReplication() { select { case <-s.closing: return + case <-s.replicationClosing: + return case <-time.After(s.replicationRetryInterval): log.Printf("pilosa: reconnecting to primary replica") } } } +// monitorPrimaryStoreEvents is executed in a separate goroutine and listens for changes +// to the primary store assignment. +func (s *TranslateFile) monitorPrimaryStoreEvents() { + log.Printf("monitor primary store events") + // Keep handling events until the store closes. + for { + select { + case <-s.closing: + return + case ev := <-s.primaryStoreEvents: + if err := s.handlePrimaryStoreEvent(ev); err != nil { + log.Printf("handle primary store event") + } + } + } +} + func (s *TranslateFile) replicate(ctx context.Context) error { off := s.size() @@ -989,3 +1077,38 @@ func uVarintSize(x uint64) (i int) { } return i + 1 } + +// nopTStore represents a TranslateStore that doesn't do anything. +var nopTStore TranslateStore = nopTranslateStore{} + +// newNopTranslateStore returns a translate store which does nothing. It returns a global +// object to avoid unnecessary allocations. +func newNopTranslateStore(interface{}) TranslateStore { return nopTStore } + +// nopTranslateStore represents a no-op implementation of the TranslateStore interface. +type nopTranslateStore struct{} + +// TranslateColumnsToUint64 is a no-op implementation of the TranslateStore TranslateColumnsToUint64 method. +func (s nopTranslateStore) TranslateColumnsToUint64(index string, values []string) ([]uint64, error) { + return []uint64{}, nil +} + +// TranslateColumnToString is a no-op implementation of the TranslateStore TranslateColumnToString method. +func (s nopTranslateStore) TranslateColumnToString(index string, values uint64) (string, error) { + return "", nil +} + +// TranslateRowsToUint64 is a no-op implementation of the TranslateStore TranslateRowsToUint64 method. +func (s nopTranslateStore) TranslateRowsToUint64(index, field string, values []string) ([]uint64, error) { + return []uint64{}, nil +} + +// TranslateRowToString is a no-op implementation of the TranslateStore TranslateRowToString method. +func (s nopTranslateStore) TranslateRowToString(index, field string, values uint64) (string, error) { + return "", nil +} + +// Reader is a no-op implementation of the TranslateStore Reader method. +func (s nopTranslateStore) Reader(ctx context.Context, off int64) (io.ReadCloser, error) { + return ioutil.NopCloser(bytes.NewReader(nil)), nil +} diff --git a/translate_test.go b/translate_test.go index 9856bc85c..706fe365f 100644 --- a/translate_test.go +++ b/translate_test.go @@ -383,7 +383,7 @@ func TestTranslateFile_PrimaryTranslateStore(t *testing.T) { // Create a replica that accepts writes from primary. replica := NewTranslateFile() - replica.PrimaryTranslateStore = primary + replica.SetPrimaryStore("primary", primary) if err := replica.Open(); err != nil { t.Fatal(err) } @@ -398,7 +398,7 @@ func TestTranslateFile_PrimaryTranslateStore(t *testing.T) { // Attempt to read replica until writes appear. if err := retryFor(2*time.Second, func() error { - // Verify that replica have received writes. + // Verify that replica has received writes. if value, err := replica.TranslateColumnToString("IDX0", 1); err != nil { return err } else if value != "foo" { @@ -429,7 +429,7 @@ func TestTranslateFile_PrimaryTranslateStore(t *testing.T) { t.Fatal(err) } - // Attempt to read replica until write appear. + // Attempt to read replica until writes appear. if err := retryFor(2*time.Second, func() error { if value, err := replica.TranslateColumnToString("IDX0", 2); err != nil { return err @@ -448,7 +448,7 @@ func TestTranslateFile_PrimaryTranslateStore(t *testing.T) { t.Fatal(err) } - // Attempt to read replica until write appear. + // Attempt to read replica until writes appear. if err := retryFor(2*time.Second, func() error { if value, err := replica.TranslateColumnToString("IDX0", 3); err != nil { return err @@ -461,6 +461,289 @@ func TestTranslateFile_PrimaryTranslateStore(t *testing.T) { } } +func TestTranslateFile_ReassignPrimaryTranslateStore(t *testing.T) { + t.Run("AddNode", func(t *testing.T) { + // Create a primary store that accepts writes. + primary := MustOpenTranslateFile() + defer primary.MustClose() + + // Create replica1 that accepts writes from primary. + replica1 := NewTranslateFile() + replica1.SetPrimaryStore("primary", primary) + if err := replica1.Open(); err != nil { + t.Fatal(err) + } + defer replica1.MustClose() + + // Write to the primary. + if _, err := primary.TranslateColumnsToUint64("IDX0", []string{"foo"}); err != nil { + t.Fatal(err) + } else if _, err := primary.TranslateRowsToUint64("IDX0", "FIELD0", []string{"bar", "baz"}); err != nil { + t.Fatal(err) + } + + // Attempt to read replica1 until writes appear. + if err := retryFor(2*time.Second, func() error { + // Verify that replica1 has received writes. + if value, err := replica1.TranslateColumnToString("IDX0", 1); err != nil { + return err + } else if value != "foo" { + return fmt.Errorf("unexpected column 1 value: %s", value) + } + + if value, err := replica1.TranslateRowToString("IDX0", "FIELD0", 1); err != nil { + return err + } else if value != "bar" { + return fmt.Errorf("unexpected row 1 value: %s", value) + } + + if value, err := replica1.TranslateRowToString("IDX0", "FIELD0", 2); err != nil { + return err + } else if value != "baz" { + return fmt.Errorf("unexpected row 2 value: %s", value) + } + + return nil + }); err != nil { + t.Fatal(err) + } + + // Create replica2 that accepts writes from primary, + // and change replica1's primary to be replica2. + // From: P <- R1 + // To: P <- R2 <- R1 + // Momentarily, replica1 should be ahead of replica2, so we will see log + // messages like "translate store reader past file size: sz=0 off=39" + // But eventually it should get in sync and new writes will be available + // on replica1. + replica2 := NewTranslateFile() + replica2.SetPrimaryStore("primary", primary) + replica1.SetPrimaryStore("replica2", replica2) + + if err := replica2.Open(); err != nil { + t.Fatal(err) + } + defer replica2.MustClose() + + // Attempt to read replica2 until writes appear. + if err := retryFor(2*time.Second, func() error { + // Verify that replica2 have received writes. + if value, err := replica2.TranslateColumnToString("IDX0", 1); err != nil { + return err + } else if value != "foo" { + return fmt.Errorf("unexpected column 1 value: %s", value) + } + + if value, err := replica2.TranslateRowToString("IDX0", "FIELD0", 1); err != nil { + return err + } else if value != "bar" { + return fmt.Errorf("unexpected row 1 value: %s", value) + } + + if value, err := replica2.TranslateRowToString("IDX0", "FIELD0", 2); err != nil { + return err + } else if value != "baz" { + return fmt.Errorf("unexpected row 2 value: %s", value) + } + + return nil + }); err != nil { + t.Fatal(err) + } + + // Add more data to the primary and ensure that replica1 receives the + // data (via replica2). + if _, err := primary.TranslateColumnsToUint64("IDX0", []string{"baz"}); err != nil { + t.Fatal(err) + } + + // Attempt to read replica1 until writes appear. + if err := retryFor(2*time.Second, func() error { + if value, err := replica1.TranslateColumnToString("IDX0", 2); err != nil { + return err + } else if value != "baz" { + return fmt.Errorf("unexpected column 2 value: %s", value) + } + return nil + }); err != nil { + t.Fatal(err) + } + }) + + t.Run("RemoveNode", func(t *testing.T) { + // Create a primary store that accepts writes. + primary := MustOpenTranslateFile() + defer primary.MustClose() + + // Create two replicas that accepts writes from the primary + // in a daisy-chain configuration. + // P <- R1 <- R2 + + // Create replica1. + replica1 := NewTranslateFile() + replica1.SetPrimaryStore("primary", primary) + if err := replica1.Open(); err != nil { + t.Fatal(err) + } + defer replica1.MustClose() + + // Create replica2. + replica2 := NewTranslateFile() + replica2.SetPrimaryStore("replica1", replica1) + if err := replica2.Open(); err != nil { + t.Fatal(err) + } + defer replica2.MustClose() + + // Write to the primary. + if _, err := primary.TranslateColumnsToUint64("IDX0", []string{"foo"}); err != nil { + t.Fatal(err) + } else if _, err := primary.TranslateRowsToUint64("IDX0", "FIELD0", []string{"bar", "baz"}); err != nil { + t.Fatal(err) + } + + // Attempt to read replica2 until writes appear. + if err := retryFor(2*time.Second, func() error { + // Verify that replica2 has received writes. + if value, err := replica2.TranslateColumnToString("IDX0", 1); err != nil { + return err + } else if value != "foo" { + return fmt.Errorf("unexpected column 1 value: %s", value) + } + + if value, err := replica2.TranslateRowToString("IDX0", "FIELD0", 1); err != nil { + return err + } else if value != "bar" { + return fmt.Errorf("unexpected row 1 value: %s", value) + } + + if value, err := replica2.TranslateRowToString("IDX0", "FIELD0", 2); err != nil { + return err + } else if value != "baz" { + return fmt.Errorf("unexpected row 2 value: %s", value) + } + + return nil + }); err != nil { + t.Fatal(err) + } + + // Remove replica1 from the replication chain. + // From: P <- R1 <- R2 + // To: P <- R2 + replica1.SetPrimaryStore("", nil) + + // Add more data to the primary and ensure that replica2 receives the + // data (after replica1 is removed). + if _, err := primary.TranslateColumnsToUint64("IDX0", []string{"baz"}); err != nil { + t.Fatal(err) + } + + // Set replica2 to replicate from primary. + replica2.SetPrimaryStore("primary", primary) + + // Attempt to read replica2 until writes appear. + if err := retryFor(2*time.Second, func() error { + if value, err := replica2.TranslateColumnToString("IDX0", 2); err != nil { + return err + } else if value != "baz" { + return fmt.Errorf("unexpected column 2 value: %s", value) + } + return nil + }); err != nil { + t.Fatal(err) + } + }) + + t.Run("ChangePrimaryWriter", func(t *testing.T) { + // Create a primary store that accepts writes. + primary := MustOpenTranslateFile() + defer primary.MustClose() + + // Create two replicas that accepts writes from the primary + // in a daisy-chain configuration. + // P <- R1 <- R2 + + // Create replica1. + replica1 := NewTranslateFile() + replica1.SetPrimaryStore("primary", primary) + if err := replica1.Open(); err != nil { + t.Fatal(err) + } + defer replica1.MustClose() + + // Create replica2. + replica2 := NewTranslateFile() + replica2.SetPrimaryStore("replica1", replica1) + if err := replica2.Open(); err != nil { + t.Fatal(err) + } + defer replica2.MustClose() + + // Write to the primary. + if _, err := primary.TranslateColumnsToUint64("IDX0", []string{"foo"}); err != nil { + t.Fatal(err) + } else if _, err := primary.TranslateRowsToUint64("IDX0", "FIELD0", []string{"bar", "baz"}); err != nil { + t.Fatal(err) + } + + // Attempt to read replica2 until writes appear. + if err := retryFor(2*time.Second, func() error { + // Verify that replica2 has received writes. + if value, err := replica2.TranslateColumnToString("IDX0", 1); err != nil { + return err + } else if value != "foo" { + return fmt.Errorf("unexpected column 1 value: %s", value) + } + + if value, err := replica2.TranslateRowToString("IDX0", "FIELD0", 1); err != nil { + return err + } else if value != "bar" { + return fmt.Errorf("unexpected row 1 value: %s", value) + } + + if value, err := replica2.TranslateRowToString("IDX0", "FIELD0", 2); err != nil { + return err + } else if value != "baz" { + return fmt.Errorf("unexpected row 2 value: %s", value) + } + + return nil + }); err != nil { + t.Fatal(err) + } + + // Change replica1 to be the primary writer. + // From: P <- R1 <- R2 + // To: R1 <- R2 <- P + replica1.SetPrimaryStore("", nil) + + // SetPrimaryStore is asynchronous, so we need to wait before writing new data. + time.Sleep(200 * time.Millisecond) + + // Add more data to the primary (now replica1) and ensure that primary (now a read-only replica) + // receives the data. + if _, err := replica1.TranslateColumnsToUint64("IDX0", []string{"baz"}); err != nil { + t.Fatal(err) + } + + // Set primary to replicate from replica2. + primary.SetPrimaryStore("replica2", replica2) + + // Attempt to read primary until writes appear. + if err := retryFor(2*time.Second, func() error { + if value, err := primary.TranslateColumnToString("IDX0", 2); err != nil { + return err + } else if value != "baz" { + return fmt.Errorf("unexpected column 2 value: %s", value) + } + return nil + }); err != nil { + t.Fatal(err) + } + }) +} + func BenchmarkTranslateFile_TranslateColumnsToUint64(b *testing.B) { const batchSize = 1000 @@ -567,7 +850,7 @@ func (s *TranslateFile) Reopen() error { s.TranslateFile = pilosa.NewTranslateFile() s.lock.Unlock() s.Path = prev.Path - s.PrimaryTranslateStore = prev.PrimaryTranslateStore + s.SetPrimaryStore("restored-primary", prev.PrimaryTranslateStore) return s.Open() } diff --git a/uri.go b/uri.go index 8f9df2b72..231457691 100644 --- a/uri.go +++ b/uri.go @@ -173,6 +173,9 @@ func parseAddress(address string) (uri *URI, err error) { if err != nil { return nil, errors.New("converting port string to int") } + if port > 65535 { + return nil, errors.New("port must be in range 0 - 65535") + } } uri = &URI{ Scheme: scheme, diff --git a/uri_internal_test.go b/uri_internal_test.go index 9bc6403d7..cb59c75c6 100644 --- a/uri_internal_test.go +++ b/uri_internal_test.go @@ -172,5 +172,5 @@ func validFixture() []uriItem { } func invalidFixture() []string { - return []string{"foo:bar", "http://foo:", "foo:", ":bar", "http://pilosa.com:129999999999999999999999993", "fd42:4201:f86b:7e09:216:3eff:fefa:ed80"} + return []string{"foo:bar", "http://foo:", "foo:", ":bar", "http://pilosa.com:129999999999999999999999993", "fd42:4201:f86b:7e09:216:3eff:fefa:ed80", ":65536"} }