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

Merge branch 'master' into translate-keys-endpoint

Browse source
This commit is contained in:
Yuce Tekol 2018-11-21 16:38:23 +03:00
commit 2c91bf69b3
No known key found for this signature in database
GPG key ID: CB59E46D2FB90573
12 changed files with 98 additions and 63 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
api.go
View file

@ -957,7 +957,7 @@ func (api *API) validateShardOwnership(indexName string, shard uint64) error {
}
func (api *API) indexField(indexName string, fieldName string, shard uint64) (*Index, *Field, error) {
api.server.logger.Printf("importing: %v %v %v", indexName, fieldName, shard)
api.server.logger.Debugf("importing: %v %v %v", indexName, fieldName, shard)
// Find the Index.
index := api.holder.Index(indexName)

21
cluster.go
View file

@ -347,8 +347,6 @@ func (c *cluster) unprotectedUpdateCoordinator(n *Node) bool {
// addNode adds a node to the Cluster and updates and saves the
// new topology. unprotected.
func (c *cluster) addNode(node *Node) error {
c.logger.Printf("add node %s to cluster on %s", node, c.Node)
// If the node being added is the coordinator, set it for this node.
if node.IsCoordinator {
c.Coordinator = node.ID
@ -481,7 +479,7 @@ func (c *cluster) setNodeState(state string) error { // nolint: unparam
State: state,
}
c.logger.Printf("Sending State %s (%s)", state, c.Coordinator)
c.logger.Printf("sending state %s (%s)", state, c.Coordinator)
if err := c.sendTo(c.coordinatorNode(), ns); err != nil {
return fmt.Errorf("sending node state error: err=%s", err)
}
@ -970,7 +968,6 @@ func (c *cluster) close() error {
}
func (c *cluster) markAsJoined() {
c.logger.Printf("mark node as joined (received coordinator update)")
if !c.joined {
c.joined = true
close(c.joining)
@ -1069,7 +1066,6 @@ func (c *cluster) unprotectedSetStateAndBroadcast(state string) error {
}
// Broadcast cluster status changes to the cluster.
status := c.unprotectedStatus()
c.logger.Printf("broadcasting ClusterStatus: %s", status)
return c.broadcaster.SendSync(status) // TODO fix c.Status
}
@ -1246,14 +1242,14 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error {
return errors.Wrap(err, "merging cluster status")
}
c.logger.Printf("MergeClusterStatus done, start goroutine")
c.logger.Printf("done MergeClusterStatus, start goroutine")
// The actual resizing runs in a goroutine because we don't want to block
// the distribution of other ResizeInstructions to the rest of the cluster.
go func() {
// Make sure the holder has opened.
<-c.holder.opened
c.holder.opened.Recv()
// Prepare the return message.
complete := &ResizeInstructionComplete{
@ -1266,7 +1262,7 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error {
if err := func() error {
// Sync the schema received in the resize instruction.
c.logger.Printf("Holder ApplySchema")
c.logger.Debugf("holder applySchema")
if err := c.holder.applySchema(instr.Schema); err != nil {
return errors.Wrap(err, "applying schema")
}
@ -1651,17 +1647,17 @@ func (c *cluster) ReceiveEvent(e *NodeEvent) (err error) {
switch e.Event {
case NodeJoin:
c.logger.Printf("nodeJoin of %s on %s", e.Node.URI, c.Node.URI)
c.logger.Debugf("nodeJoin of %s on %s", e.Node.URI, c.Node.URI)
// Ignore the event if this is not the coordinator.
if !c.isCoordinator() {
return nil
}
return c.nodeJoin(e.Node)
case NodeLeave:
c.logger.Printf("received node leave on %s: %s, uri: %v", c.Node, e.Node, e.Node.URI)
c.mu.Lock()
defer c.mu.Unlock()
if c.unprotectedIsCoordinator() {
c.logger.Printf("received node leave: %v", e.Node)
// if removeNodeBasicSorted succeeds, that means that the node was
// not already removed by a removeNode request. We treat this as the
// host being temporarily unavailable, and expect it to come back
@ -1673,7 +1669,6 @@ func (c *cluster) ReceiveEvent(e *NodeEvent) (err error) {
err = c.unprotectedSetStateAndBroadcast(c.determineClusterState())
}
}
c.logger.Printf("finished node leave on %s: %s, uri: %v", c.Node, e.Node, e.Node.URI)
case NodeUpdate:
c.logger.Printf("received node update event: id: %v, string: %v, uri: %v", e.Node.ID, e.Node.String(), e.Node.URI)
// NodeUpdate is intentionally not implemented.
@ -1686,7 +1681,7 @@ func (c *cluster) ReceiveEvent(e *NodeEvent) (err error) {
func (c *cluster) nodeJoin(node *Node) error {
c.mu.Lock()
defer c.mu.Unlock()
c.logger.Printf("NodeJoin event on coordinator, node: %s, id: %s", node.URI, node.ID)
c.logger.Printf("node join event on coordinator, node: %s, id: %s", node.URI, node.ID)
if c.needTopologyAgreement() {
// A host that is not part of the topology can't be added to the STARTING cluster.
if !c.Topology.ContainsID(node.ID) {
@ -1726,7 +1721,7 @@ func (c *cluster) nodeJoin(node *Node) error {
// the cluster.
if cnode := c.unprotectedNodeByID(node.ID); cnode != nil {
if cnode.URI != node.URI {
c.logger.Printf("Node: %v changed URI from %s to %s", cnode.ID, cnode.URI, node.URI)
c.logger.Printf("node: %v changed URI from %s to %s", cnode.ID, cnode.URI, node.URI)
cnode.URI = node.URI
}
return c.unprotectedSetStateAndBroadcast(c.determineClusterState())

5
cmd/import.go
View file

@ -19,10 +19,8 @@ import (
"io"
"github.com/pilosa/pilosa"
"github.com/spf13/cobra"
"github.com/pilosa/pilosa/ctl"
"github.com/spf13/cobra"
)
var Importer *ctl.ImportCommand
@ -55,6 +53,7 @@ omitted. If it is present then its format should be YYYY-MM-DDTHH:MM.
flags.StringVarP(&Importer.Field, "field", "f", "", "Field to import into.")
flags.BoolVar(&Importer.IndexOptions.Keys, "index-keys", false, "Specify keys=true when creating an index")
flags.BoolVar(&Importer.FieldOptions.Keys, "field-keys", false, "Specify keys=true when creating a field")
flags.StringVar(&Importer.FieldOptions.Type, "field-type", "", "Specify the field type when creating a field. One of: set, int, time, bool, mutex")
flags.Int64Var(&Importer.FieldOptions.Min, "field-min", 0, "Specify the minimum for an int field on creation")
flags.Int64Var(&Importer.FieldOptions.Max, "field-max", 0, "Specify the maximum for an int field on creation")
flags.StringVar(&Importer.FieldOptions.CacheType, "field-cache-type", pilosa.CacheTypeRanked, "Specify the cache type for a set field on creation. One of: none, lru, ranked")

16
ctl/import.go
View file

@ -99,13 +99,15 @@ func (cmd *ImportCommand) Run(ctx context.Context) error {
cmd.client = client
if cmd.CreateSchema {
// set the correct type for the field
if cmd.FieldOptions.TimeQuantum != "" {
cmd.FieldOptions.Type = "time"
} else if cmd.FieldOptions.Min != 0 || cmd.FieldOptions.Max != 0 {
cmd.FieldOptions.Type = "int"
} else {
cmd.FieldOptions.Type = "set"
if cmd.FieldOptions.Type == "" {
// set the correct type for the field
if cmd.FieldOptions.TimeQuantum != "" {
cmd.FieldOptions.Type = "time"
} else if cmd.FieldOptions.Min != 0 || cmd.FieldOptions.Max != 0 {
cmd.FieldOptions.Type = "int"
} else {
cmd.FieldOptions.Type = "set"
}
}
err := cmd.ensureSchema(ctx)
if err != nil {

4
executor.go
View file

@ -1664,7 +1664,7 @@ func (e *executor) executeSetRowShard(ctx context.Context, index string, c *pql.
if err != nil {
return false, errors.Wrap(err, "creating view")
}
fragment, err = view.createFragmentIfNotExists(shard)
fragment, err = view.CreateFragmentIfNotExists(shard)
if err != nil {
return false, errors.Wrapf(err, "creating fragment: %d", shard)
}
@ -2055,7 +2055,7 @@ func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64,
if !opt.Remote {
nodes = Nodes(e.Cluster.nodes).Clone()
} else {
nodes = []*Node{e.Cluster.unprotectedNodeByID(e.Node.ID)}
nodes = []*Node{e.Cluster.nodeByID(e.Node.ID)}
}
// Start mapping across all primary owners.

7
fragment.go
View file

@ -1492,9 +1492,6 @@ func (f *fragment) bulkImportStandard(rowIDs, columnIDs []uint64, options *Impor
lastRowID = rowID
rowSet[rowID] = struct{}{}
}
// Invalidate block checksum.
delete(f.checksums, int(rowID/HashBlockSize))
}
f.mu.Lock()
@ -1518,6 +1515,9 @@ func (f *fragment) bulkImportStandard(rowIDs, columnIDs []uint64, options *Impor
// Update cache counts for all affected rows.
for rowID := range rowSet {
// Invalidate block checksum.
delete(f.checksums, int(rowID/HashBlockSize))
n := results.CountRange(rowID*ShardWidth, (rowID+1)*ShardWidth)
f.cache.BulkAdd(rowID, n)
}
@ -1734,7 +1734,6 @@ func (f *fragment) snapshot() error {
// f.mu must be locked when calling it.
func unprotectedWriteToFragment(f *fragment, bm *roaring.Bitmap) error { // nolint: interfacer
f.Logger.Printf("fragment: snapshotting %s/%s/%s/%d", f.index, f.field, f.view, f.shard)
completeMessage := fmt.Sprintf("fragment: snapshot complete %s/%s/%s/%d", f.index, f.field, f.view, f.shard)
start := time.Now()
defer track(start, completeMessage, f.stats, f.Logger)

17
fragment_internal_test.go
View file

@ -25,6 +25,8 @@ import (
"testing"
"testing/quick"
"golang.org/x/sync/errgroup"
"github.com/davecgh/go-spew/spew"
"github.com/pilosa/pilosa/pql"
"github.com/pilosa/pilosa/roaring"
@ -1399,6 +1401,21 @@ func TestFragment_ImportSet(t *testing.T) {
}
}
func TestFragment_ConcurrentImport(t *testing.T) {
t.Run("bulkImportStandard", func(t *testing.T) {
f := mustOpenFragment("i", "f", viewStandard, 0, "")
defer f.Close()
eg := errgroup.Group{}
eg.Go(func() error { return f.bulkImportStandard([]uint64{1, 2}, []uint64{1, 2}, &ImportOptions{}) })
eg.Go(func() error { return f.bulkImportStandard([]uint64{3, 4}, []uint64{3, 4}, &ImportOptions{}) })
err := eg.Wait()
if err != nil {
t.Fatalf("importing data to fragment: %v", err)
}
})
}
// Ensure a fragment can import mutually exclusive values.
func TestFragment_ImportMutex(t *testing.T) {
tests := []struct {

40
holder.go
View file

@ -57,7 +57,7 @@ type Holder struct {
NewPrimaryTranslateStore func(interface{}) TranslateStore
// opened channel is closed once Open() completes.
opened chan struct{}
opened lockedChan
broadcaster broadcaster
@ -79,13 +79,39 @@ type Holder struct {
Logger logger.Logger
}
// lockedChan looks a little ridiculous admittedly, but exists for good reason.
// The channel within is used (for example) to signal to other goroutines when
// the Holder has finished opening (via closing the channel). However, it is
// possible for the holder to be closed and then reopened, but a channel which
// is closed cannot be re-opened. We must create a new channel - this creates a
// data race with any goroutine which might be accessing the channel. To ensure
// that there is no data race on the value of the channel itself, we wrap any
// operation on it with an RWMutex so that we can guarantee that nothing is
// trying to listen on it when it gets swapped.
type lockedChan struct {
ch chan struct{}
mu sync.RWMutex
}
func (lc *lockedChan) Close() {
lc.mu.RLock()
close(lc.ch)
lc.mu.RUnlock()
}
func (lc *lockedChan) Recv() {
lc.mu.RLock()
<-lc.ch
lc.mu.RUnlock()
}
// NewHolder returns a new instance of Holder.
func NewHolder() *Holder {
return &Holder{
indexes: make(map[string]*Index),
closing: make(chan struct{}),
opened: make(chan struct{}),
opened: lockedChan{ch: make(chan struct{})},
translateFile: NewTranslateFile(),
NewPrimaryTranslateStore: newNopTranslateStore,
@ -159,7 +185,7 @@ func (h *Holder) Open() error {
h.Stats.Open()
close(h.opened)
h.opened.Close()
return nil
}
@ -184,7 +210,9 @@ func (h *Holder) Close() error {
}
// Reset opened in case Holder needs to be reopened.
h.opened = make(chan struct{})
h.opened.mu.Lock()
h.opened.ch = make(chan struct{})
h.opened.mu.Unlock()
return nil
}
@ -474,7 +502,7 @@ func (h *Holder) flushCaches() {
}
if err := fragment.FlushCache(); err != nil {
h.Logger.Printf("error flushing cache: err=%s, path=%s", err, fragment.cachePath())
h.Logger.Printf("ERROR flushing cache: err=%s, path=%s", err, fragment.cachePath())
}
}
}
@ -535,7 +563,7 @@ func (h *Holder) setFileLimit() {
h.Logger.Printf("ERROR checking open file limit: %s", err)
} else {
if oldLimit.Cur < fileLimit {
h.Logger.Printf("WARNING: Tried to set open file limit to %d, but it is %d. You may consider running \"sudo ulimit -n %d\" before starting Pilosa to avoid \"too many open files\" error. See https://www.pilosa.com/docs/administration/#open-file-limits for more information.", fileLimit, oldLimit.Cur, fileLimit)
h.Logger.Printf("WARNING: Tried to set open file limit to %d, but it is %d. You may consider running \"sudo ulimit -n %d\" before starting Pilosa to avoid \"too many open files\" error. See https://www.pilosa.com/docs/latest/administration/#open-file-limits for more information.", fileLimit, oldLimit.Cur, fileLimit)
}
}
}

18
http/client.go
View file

@ -91,9 +91,7 @@ func (c *InternalClient) maxShardByIndex(ctx context.Context) (map[string]uint64
defer resp.Body.Close()
var rsp getShardsMaxResponse
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("http: status=%d", resp.StatusCode)
} else if err := json.NewDecoder(resp.Body).Decode(&rsp); err != nil {
if err := json.NewDecoder(resp.Body).Decode(&rsp); err != nil {
return nil, fmt.Errorf("json decode: %s", err)
}
@ -152,7 +150,7 @@ func (c *InternalClient) CreateIndex(ctx context.Context, index string, opt pilo
// Execute request against the host.
resp, err := c.executeRequest(req.WithContext(ctx))
if err != nil {
if resp.StatusCode == http.StatusConflict {
if resp != nil && resp.StatusCode == http.StatusConflict {
return pilosa.ErrIndexExists
}
return err
@ -258,8 +256,6 @@ func (c *InternalClient) QueryNode(ctx context.Context, uri *pilosa.URI, index s
body, err := ioutil.ReadAll(resp.Body)
if err != nil {
return nil, errors.Wrap(err, "reading")
} else if resp.StatusCode != http.StatusOK {
return nil, errors.New(string(body))
}
qresp := &pilosa.QueryResponse{}
@ -689,7 +685,7 @@ func (c *InternalClient) backupShardNode(ctx context.Context, index, field strin
// Execute request.
resp, err := c.executeRequest(req.WithContext(ctx))
if err != nil {
if resp.StatusCode == http.StatusNotFound {
if resp != nil && resp.StatusCode == http.StatusNotFound {
return nil, pilosa.ErrFragmentNotFound
}
return nil, err
@ -746,7 +742,7 @@ func (c *InternalClient) CreateFieldWithOptions(ctx context.Context, index, fiel
// Execute request against the host.
resp, err := c.executeRequest(req.WithContext(ctx))
if err != nil {
if resp.StatusCode == http.StatusConflict {
if resp != nil && resp.StatusCode == http.StatusConflict {
return pilosa.ErrFieldExists
}
return err
@ -782,7 +778,7 @@ func (c *InternalClient) FragmentBlocks(ctx context.Context, uri *pilosa.URI, in
resp, err := c.executeRequest(req.WithContext(ctx))
if err != nil {
// Return the appropriate error.
if resp.StatusCode == http.StatusNotFound {
if resp != nil && resp.StatusCode == http.StatusNotFound {
return nil, pilosa.ErrFragmentNotFound
}
return nil, err
@ -825,7 +821,7 @@ func (c *InternalClient) BlockData(ctx context.Context, uri *pilosa.URI, index,
resp, err := c.executeRequest(req.WithContext(ctx))
if err != nil {
if resp.StatusCode == http.StatusNotFound {
if resp != nil && resp.StatusCode == http.StatusNotFound {
return nil, nil, nil
}
return nil, nil, err
@ -904,7 +900,7 @@ func (c *InternalClient) RowAttrDiff(ctx context.Context, uri *pilosa.URI, index
// Execute request.
resp, err := c.executeRequest(req.WithContext(ctx))
if err != nil {
if resp.StatusCode == http.StatusNotFound {
if resp != nil && resp.StatusCode == http.StatusNotFound {
return nil, pilosa.ErrFieldNotFound
}
return nil, err

8
server.go
View file

@ -585,7 +585,6 @@ func (s *Server) SendSync(m Message) error {
msg = append([]byte{getMessageType(m)}, msg...)
for _, node := range s.cluster.nodes {
node := node
s.logger.Printf("SendSync to: %s", node.URI)
// Don't forward the message to ourselves.
if s.uri == node.URI {
continue
@ -606,7 +605,6 @@ func (s *Server) SendAsync(m Message) error {
// SendTo represents an implementation of Broadcaster.
func (s *Server) SendTo(to *Node, m Message) error {
s.logger.Printf("SendTo: %s", to.URI)
msg, err := s.serializer.Marshal(m)
if err != nil {
return fmt.Errorf("marshaling message: %v", err)
@ -630,7 +628,7 @@ func (s *Server) handleRemoteStatus(pb Message) {
go func() {
// Make sure the holder has opened.
<-s.holder.opened
s.holder.opened.Recv()
err := s.mergeRemoteStatus(pb.(*NodeStatus))
if err != nil {
@ -658,7 +656,7 @@ func (s *Server) mergeRemoteStatus(ns *NodeStatus) error {
// if we don't know about a field locally, log an error because
// fields should be created and synced prior to shard creation
if f == nil {
s.logger.Printf("Local Field not found: %s/%s", is.Name, fs.Name)
s.logger.Printf("local field not found: %s/%s", is.Name, fs.Name)
continue
}
if err := f.AddRemoteAvailableShards(fs.AvailableShards); err != nil {
@ -703,7 +701,7 @@ func (s *Server) monitorDiagnostics() {
s.diagnostics.CheckVersion()
err = s.diagnostics.Flush()
if err != nil {
s.logger.Printf("Diagnostics error: %s", err)
s.logger.Printf("diagnostics error: %s", err)
}
}

16
server/server.go
View file

@ -142,7 +142,7 @@ func (m *Command) Start() (err error) {
go func() {
err := m.Handler.Serve()
if err != nil {
m.logger.Printf("Handler serve error: %v", err)
m.logger.Printf("handler serve error: %v", err)
}
}()
@ -151,7 +151,7 @@ func (m *Command) Start() (err error) {
return errors.Wrap(err, "opening server")
}
m.logger.Printf("Listening as %s\n", m.API.Node().URI)
m.logger.Printf("listening as %s\n", m.API.Node().URI)
return nil
}
@ -163,27 +163,31 @@ func (m *Command) Wait() error {
signal.Notify(c, os.Interrupt, syscall.SIGTERM)
select {
case sig := <-c:
m.logger.Printf("Received %s; gracefully shutting down...\n", sig.String())
m.logger.Printf("received signal '%s', gracefully shutting down...\n", sig.String())
// Second signal causes a hard shutdown.
go func() { <-c; os.Exit(1) }()
return errors.Wrap(m.Close(), "closing command")
case <-m.done:
m.logger.Printf("Server closed externally")
m.logger.Printf("server closed externally")
return nil
}
}
// setupLogger sets up the logger based on the configuration.
func (m *Command) setupLogger() error {
var err error
if m.Config.LogPath == "" {
m.logOutput = m.Stderr
} else {
m.logOutput, err = os.OpenFile(m.Config.LogPath, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0600)
f, err := os.OpenFile(m.Config.LogPath, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0600)
if err != nil {
return errors.Wrap(err, "opening file")
}
m.logOutput = f
err = syscall.Dup2(int(f.Fd()), int(os.Stderr.Fd()))
if err != nil {
return errors.Wrap(err, "dup2ing stderr onto logfile")
}
}
if m.Config.Verbose {

7
translate.go
View file

@ -195,12 +195,11 @@ func (s *TranslateFile) handlePrimaryStoreEvent(ev primaryStoreEvent) error {
}
// Stop translate store replication.
s.logger.Printf("stop monitor replication")
close(s.replicationClosing)
s.repWG.Wait()
// Set the primary node for translate store replication.
s.logger.Printf("set primary translate store to %s", ev.id)
s.logger.Debugf("set primary translate store to %s", ev.id)
s.primaryID = ev.id
if ev.id == "" {
s.PrimaryTranslateStore = nil
@ -209,7 +208,6 @@ func (s *TranslateFile) handlePrimaryStoreEvent(ev primaryStoreEvent) error {
}
// Start translate store replication. Stream from primary, if available.
s.logger.Printf("start monitor replication")
if s.PrimaryTranslateStore != nil {
s.replicationClosing = make(chan struct{})
s.repWG.Add(1)
@ -386,7 +384,6 @@ func (s *TranslateFile) monitorReplication() {
// monitorPrimaryStoreEvents is executed in a separate goroutine and listens for changes
// to the primary store assignment.
func (s *TranslateFile) monitorPrimaryStoreEvents() {
s.logger.Printf("monitor primary store events")
// Keep handling events until the store closes.
for {
select {
@ -404,7 +401,7 @@ func (s *TranslateFile) replicate(ctx context.Context) error {
off := s.size()
// Connect to remote primary.
s.logger.Printf("pilosa: replicating from offset %d", off)
s.logger.Debugf("pilosa: replicating from offset %d", off)
rc, err := s.PrimaryTranslateStore.Reader(ctx, off)
if err != nil {
return err