diff --git a/api.go b/api.go index ed0a925a6..c745b2a0b 100644 --- a/api.go +++ b/api.go @@ -697,7 +697,7 @@ func (api *API) ImportValue(ctx context.Context, req internal.ImportValueRequest // MaxShards returns the maximum shard number for each index in a map. func (api *API) MaxShards(ctx context.Context) map[string]uint64 { - return api.holder.MaxShards() + return api.holder.maxShards() } // StatsWithTags returns an instance of whatever implementation of StatsClient diff --git a/cluster.go b/cluster.go index a5108a2f9..a3f5aa775 100644 --- a/cluster.go +++ b/cluster.go @@ -1150,7 +1150,7 @@ func (c *cluster) generateResizeJobByAction(nodeAction nodeAction) (*resizeJob, Node: EncodeNode(toCluster.unprotectedNodeByID(id)), Coordinator: EncodeNode(c.coordinatorNode()), Sources: sources, - Schema: c.holder.EncodeSchema(), // Include the schema to ensure it's in sync on the receiving node. + Schema: c.holder.encodeSchema(), // Include the schema to ensure it's in sync on the receiving node. ClusterStatus: c.Status(), } j.Instructions = append(j.Instructions, instr) @@ -1205,7 +1205,7 @@ func (c *cluster) followResizeInstruction(instr *internal.ResizeInstruction) err // Sync the schema received in the resize instruction. c.logger.Printf("Holder ApplySchema") - if err := c.holder.ApplySchema(instr.Schema); err != nil { + if err := c.holder.applySchema(instr.Schema); err != nil { return errors.Wrap(err, "applying schema") } diff --git a/holder.go b/holder.go index bad001787..8cf830a84 100644 --- a/holder.go +++ b/holder.go @@ -36,8 +36,8 @@ const ( // defaultCacheFlushInterval is the default value for Fragment.CacheFlushInterval. defaultCacheFlushInterval = 1 * time.Minute - // FileLimit is the maximum open file limit (ulimit -n) to automatically set. - FileLimit = 262144 // (512^2) + // fileLimit is the maximum open file limit (ulimit -n) to automatically set. + fileLimit = 262144 // (512^2) ) // Holder represents a container for indexes. @@ -50,7 +50,7 @@ type Holder struct { // opened channel is closed once Open() completes. opened chan struct{} - Broadcaster broadcaster + broadcaster broadcaster NewAttrStore func(string) AttrStore @@ -65,7 +65,7 @@ type Holder struct { Path string // The interval at which the cached row ids are persisted to disk. - CacheFlushInterval time.Duration + cacheFlushInterval time.Duration Logger Logger } @@ -78,12 +78,12 @@ func NewHolder() *Holder { opened: make(chan struct{}), - Broadcaster: NopBroadcaster, + broadcaster: NopBroadcaster, Stats: NopStatsClient, NewAttrStore: newNopAttrStore, - CacheFlushInterval: defaultCacheFlushInterval, + cacheFlushInterval: defaultCacheFlushInterval, Logger: NopLogger, } @@ -200,8 +200,8 @@ func (h *Holder) HasData() (bool, error) { return false, nil } -// MaxShards returns MaxShard map for all indexes. -func (h *Holder) MaxShards() map[string]uint64 { +// maxShards returns MaxShard map for all indexes. +func (h *Holder) maxShards() map[string]uint64 { a := make(map[string]uint64) for _, index := range h.Indexes() { a[index.Name()] = index.MaxShard() @@ -229,8 +229,8 @@ func (h *Holder) Schema() []*IndexInfo { return a } -// ApplySchema applies an internal Schema to Holder. -func (h *Holder) ApplySchema(schema *internal.Schema) error { +// applySchema applies an internal Schema to Holder. +func (h *Holder) applySchema(schema *internal.Schema) error { // Create indexes that don't exist. for _, index := range schema.Indexes { opt := IndexOptions{} @@ -257,15 +257,15 @@ func (h *Holder) ApplySchema(schema *internal.Schema) error { return nil } -// EncodeMaxShards creates and internal representation of max shards. -func (h *Holder) EncodeMaxShards() *internal.MaxShards { +// encodeMaxShards creates and internal representation of max shards. +func (h *Holder) encodeMaxShards() *internal.MaxShards { return &internal.MaxShards{ - Standard: h.MaxShards(), + Standard: h.maxShards(), } } -// EncodeSchema creates an internal representation of schema. -func (h *Holder) EncodeSchema() *internal.Schema { +// encodeSchema creates an internal representation of schema. +func (h *Holder) encodeSchema() *internal.Schema { return &internal.Schema{ Indexes: EncodeIndexes(h.Indexes()), } @@ -360,7 +360,7 @@ func (h *Holder) newIndex(path, name string) (*Index, error) { } index.Logger = h.Logger index.Stats = h.Stats.WithTags(fmt.Sprintf("index:%s", index.Name())) - index.broadcaster = h.Broadcaster + index.broadcaster = h.broadcaster index.NewAttrStore = h.NewAttrStore index.columnAttrStore = h.NewAttrStore(filepath.Join(index.path, ".data")) return index, nil @@ -423,7 +423,7 @@ func (h *Holder) fragment(index, field, view string, shard uint64) *Fragment { // monitorCacheFlush periodically flushes all fragment caches sequentially. // This is run in a goroutine. func (h *Holder) monitorCacheFlush() { - ticker := time.NewTicker(h.CacheFlushInterval) + ticker := time.NewTicker(h.cacheFlushInterval) defer ticker.Stop() for { @@ -476,11 +476,11 @@ func (h *Holder) setFileLimit() { return } // If the soft limit is lower than the FileLimit constant, we will try to change it. - if oldLimit.Cur < FileLimit { - newLimit.Cur = FileLimit + if oldLimit.Cur < fileLimit { + newLimit.Cur = fileLimit // If the hard limit is not high enough, we will try to change it too. - if oldLimit.Max < FileLimit { - newLimit.Max = FileLimit + if oldLimit.Max < fileLimit { + newLimit.Max = fileLimit } else { newLimit.Max = oldLimit.Max } @@ -508,8 +508,8 @@ func (h *Holder) setFileLimit() { if err := syscall.Getrlimit(syscall.RLIMIT_NOFILE, oldLimit); err != nil { 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) + 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) } } } diff --git a/server.go b/server.go index 0e03365ae..93a5dac36 100644 --- a/server.go +++ b/server.go @@ -298,7 +298,7 @@ func NewServer(opts ...ServerOption) (*Server, error) { s.executor.MaxWritesPerRequest = s.maxWritesPerRequest s.cluster.broadcaster = s s.cluster.maxWritesPerRequest = s.maxWritesPerRequest - s.holder.Broadcaster = s + s.holder.broadcaster = s err = s.cluster.setup() if err != nil { @@ -572,8 +572,8 @@ func (s *Server) LocalStatus() (proto.Message, error) { ns := internal.NodeStatus{ Node: EncodeNode(s.cluster.Node), - MaxShards: s.holder.EncodeMaxShards(), - Schema: s.holder.EncodeSchema(), + MaxShards: s.holder.encodeMaxShards(), + Schema: s.holder.encodeSchema(), } return &ns, nil @@ -606,12 +606,12 @@ func (s *Server) mergeRemoteStatus(ns *internal.NodeStatus) error { } // Sync schema. - if err := s.holder.ApplySchema(ns.Schema); err != nil { + if err := s.holder.applySchema(ns.Schema); err != nil { return errors.Wrap(err, "applying schema") } // Sync maxShards. - oldmaxshards := s.holder.MaxShards() + oldmaxshards := s.holder.maxShards() for index, newMax := range ns.MaxShards.Standard { localIndex := s.holder.Index(index) // if we don't know about an index locally, log an error because diff --git a/utils_internal_test.go b/utils_internal_test.go index f7309961c..ca7bd1fa7 100644 --- a/utils_internal_test.go +++ b/utils_internal_test.go @@ -360,7 +360,7 @@ func (t *ClusterCluster) FollowResizeInstruction(instr *internal.ResizeInstructi destCluster := t.clusterByID(instrNode.ID) // Sync the schema received in the resize instruction. - if err := destCluster.holder.ApplySchema(instr.Schema); err != nil { + if err := destCluster.holder.applySchema(instr.Schema); err != nil { return err }