featurebase/etcd/cache.go

// 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"
	"encoding/json"
	"log"
	"sort"
	"sync"
	"time"

	"github.com/pilosa/pilosa/v2/disco"
	"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

	stateMu sync.Mutex

	nodeStates         map[string]nodeState
	nodeStateTTL       int // seconds
	nodeStateFrequency int // max requests per second allowed before using the cache

	clusterStateVal         disco.ClusterState
	clusterStateTTL         int // seconds
	clusterStateFrequency   int // max requests per second allowed before using the cache
	clusterStateLastRequest time.Time
	clusterStateLastCache   time.Time
}

type nodeState struct {
	val         disco.NodeState
	lastRequest time.Time
	lastCache   time.Time
}

// NewEtcdWithCache returns a new instance of Cache.
func NewEtcdWithCache(opt Options, replicas int) *EtcdWithCache {
	return &EtcdWithCache{
		Etcd: NewEtcd(opt, replicas),

		nodeStateTTL:          6,
		nodeStateFrequency:    1,
		clusterStateTTL:       6,
		clusterStateFrequency: 1,

		peerMetadata: make(map[string][]byte),
		nodeStates:   make(map[string]nodeState),
	}
}

// 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
}

// ClusterState is a cache wrapper around the Stator.ClusterState method.
func (c *EtcdWithCache) ClusterState(ctx context.Context) (disco.ClusterState, error) {
	c.stateMu.Lock()
	defer c.stateMu.Unlock()

	now := time.Now()
	if now.Sub(c.clusterStateLastCache) > (time.Duration(c.clusterStateTTL)*time.Second) ||
		now.Sub(c.clusterStateLastRequest) > (time.Second/time.Duration(c.clusterStateFrequency)) {
		v, err := c.Etcd.ClusterState(ctx)
		if err == nil {
			// In order to avoid NodeState() returning a cached value after
			// cluster state has changed, we reset the node state caches to
			// ensure that the next call to NodeState() returns the latest
			// value. And we only need to do this if the cluster state value has
			// actually changed.
			if c.clusterStateVal != v {
				for k, ns := range c.nodeStates {
					ns.lastCache = time.Time{}
					c.nodeStates[k] = ns
				}
			}

			c.clusterStateVal = v
			c.clusterStateLastCache = now
			c.clusterStateLastRequest = now
		}
		return v, err
	}
	c.clusterStateLastRequest = now
	return c.clusterStateVal, nil
}

// NodeState is a cache wrapper around the Stator.NodeState method.
func (c *EtcdWithCache) NodeState(ctx context.Context, peerID string) (disco.NodeState, error) {
	c.stateMu.Lock()
	defer c.stateMu.Unlock()

	ns := c.nodeStates[peerID]

	now := time.Now()
	if now.Sub(ns.lastCache) > (time.Duration(c.nodeStateTTL)*time.Second) ||
		now.Sub(ns.lastRequest) > (time.Second/time.Duration(c.nodeStateFrequency)) {
		v, err := c.Etcd.NodeState(ctx, peerID)
		if err == nil {
			// In order to avoid ClusterState() returning a cached value after a
			// node state has changed, we reset the cluster state cache to
			// ensure that the next call to ClusterState() returns the latest
			// value. And we only need to do this if the node state value has
			// actually changed.
			if ns.val != v {
				c.clusterStateLastCache = time.Time{}
			}

			ns.val = v
			ns.lastCache = now
			ns.lastRequest = now
			c.nodeStates[peerID] = ns
		}
		return v, err
	}
	ns.lastRequest = now
	c.nodeStates[peerID] = ns
	return ns.val, nil
}

// Nodes implements the Noder interface.
func (c *EtcdWithCache) Nodes() []*topology.Node {
	peers := c.Peers()
	nodes := make([]*topology.Node, len(peers))
	for i, peer := range peers {
		node := &topology.Node{}
		if meta, err := c.Metadata(context.Background(), peer.ID); err != nil {
			log.Println(err, "getting metadata") // TODO: handle this with a logger
		} else if err := json.Unmarshal(meta, node); err != nil {
			log.Println(err, "unmarshaling json metadata")
		}

		node.ID = peer.ID

		nodes[i] = node
	}

	// Nodes must be sorted.
	sort.Sort(topology.ByID(nodes))

	return 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
}