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

Remove etcd cache

Browse source
This commit is contained in:
Kuba Podgórski 2021-02-25 11:30:09 +01:00
parent 1fc3d37134
commit 0f4b273d3a
3 changed files with 10 additions and 102 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

94
etcd/cache.go
View file

@ -1,94 +0,0 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package etcd
import (
"context"
"sync"
"time"
"github.com/pilosa/pilosa/v2/topology"
)
// EtcdWithCache is a wrapper around the Etcd type which will return a
// cached value when the number of requests come in below a configured
// frequency. It also breaks the cache after a configured TTL.
type EtcdWithCache struct {
*Etcd
peerMetadataMu sync.RWMutex
peerMetadata map[string][]byte
peersMu sync.Mutex // peer-list cache updates
nodes []*topology.Node // unmarshalled Node data
nodesTTL int // seconds
nodesLastRequest time.Time // last time requested
}
// NewEtcdWithCache returns a new instance of Cache.
func NewEtcdWithCache(opt Options, replicas int) *EtcdWithCache {
return &EtcdWithCache{
Etcd: NewEtcd(opt, replicas),
nodesTTL: 6,
peerMetadata: make(map[string][]byte),
}
}
// Metadata is a cache wrapper around the Metadator.Metadata method.
func (c *EtcdWithCache) Metadata(ctx context.Context, peerID string) ([]byte, error) {
c.peerMetadataMu.RLock()
v, ok := c.peerMetadata[peerID]
c.peerMetadataMu.RUnlock()
if ok {
return v, nil
}
v, err := c.Etcd.Metadata(ctx, peerID)
if err == nil {
c.peerMetadataMu.Lock()
c.peerMetadata[peerID] = v
c.peerMetadataMu.Unlock()
}
return v, err
}
// Nodes caches the result of the underlying implementation's node list.
func (c *EtcdWithCache) Nodes() []*topology.Node {
c.peersMu.Lock()
defer c.peersMu.Unlock()
now := time.Now()
if now.Sub(c.nodesLastRequest) > (time.Duration(c.nodesTTL) * time.Second) {
c.nodes = c.Etcd.Nodes()
c.nodesLastRequest = now
}
return c.nodes
}
// SetNodes implements the Noder interface as NOP
// (because we can't force to set nodes for etcd).
func (c *EtcdWithCache) SetNodes(nodes []*topology.Node) {}
// AppendNode implements the Noder interface as NOP
// (because resizer is responsible for adding new nodes).
func (c *EtcdWithCache) AppendNode(node *topology.Node) {}
// RemoveNode implements the Noder interface as NOP
// (because resizer is responsible for removing existing nodes)
func (c *EtcdWithCache) RemoveNode(nodeID string) bool {
return false
}

16
etcd/embed.go
View file

@ -106,6 +106,12 @@ func NewEtcd(opt Options, replicas int) *Etcd {
func (e *Etcd) Close() error {
_ = testhook.Closed(pilosa.NewAuditor(), e, nil)
if e.cli != nil {
if err := e.cli.Close(); err != nil {
log.Printf("Error closing etcd client: %v", err)
}
}
if e.e != nil {
if e.resizeCancel != nil {
e.resizeCancel()
@ -117,10 +123,6 @@ func (e *Etcd) Close() error {
<-e.e.Server.StopNotify()
}
if e.cli != nil {
return e.cli.Close()
}
return nil
}
@ -250,15 +252,15 @@ func (e *Etcd) nodeState(ctx context.Context, peerID string) (disco.NodeState, e
return disco.NodeStateUnknown, err
}
if len(resp.KVs) > 1 {
if len(resp.Kvs) > 1 {
return disco.NodeStateUnknown, disco.ErrTooManyResults
}
if len(resp.KVs) == 0 {
if len(resp.Kvs) == 0 {
return disco.NodeStateUnknown, disco.ErrNoResults
}
return disco.NodeState(resp.KVs[0].Value), nil
return disco.NodeState(resp.Kvs[0].Value), nil
}
func (e *Etcd) NodeStates(ctx context.Context) (map[string]disco.NodeState, error) {

2
server/server.go
View file

@ -390,7 +390,7 @@ func (m *Command) SetupServer() error {
m.Config.Etcd.Dir = filepath.Join(path, pilosa.DefaultDiscoDir)
}
e := petcd.NewEtcdWithCache(m.Config.Etcd, m.Config.Cluster.ReplicaN)
e := petcd.NewEtcd(m.Config.Etcd, m.Config.Cluster.ReplicaN)
discoOpt := pilosa.OptServerDisCo(e, e, e, e, e, e, e)
serverOptions := []pilosa.ServerOption{