-
Notifications
You must be signed in to change notification settings - Fork 105
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
cmd/telemeter-server,pkg/cache: cache auth resps
This commit adds new functionality to Telemeter server to enable it to cache the responses to authentication requests made while handling remote-write requests. The only currently implemented backend is Memcached, because it is so simple, but any shared k-v would serve our purposes. Signed-off-by: Lucas Servén Marín <lserven@gmail.com>
- Loading branch information
Showing
5 changed files
with
212 additions
and
10 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,104 @@ | ||
| package cache | ||
|
|
||
| import ( | ||
| "bufio" | ||
| "bytes" | ||
| "net/http" | ||
| "net/http/httputil" | ||
|
|
||
| "github.com/go-kit/kit/log" | ||
| "github.com/go-kit/kit/log/level" | ||
| "github.com/pkg/errors" | ||
| "github.com/prometheus/client_golang/prometheus" | ||
| ) | ||
|
|
||
| // Cacher is able to get and set key value pairs. | ||
| type Cacher interface { | ||
| Get(string) ([]byte, bool, error) | ||
| Set(string, []byte) error | ||
| } | ||
|
|
||
| // KeyFunc generates a cache key from a http.Request. | ||
| type KeyFunc func(*http.Request) (string, error) | ||
|
|
||
| // RoundTripper is a http.RoundTripper than can get and set responses from a cache. | ||
| type RoundTripper struct { | ||
| c Cacher | ||
| key KeyFunc | ||
| next http.RoundTripper | ||
|
|
||
| l log.Logger | ||
|
|
||
| // Metrics. | ||
| cacheReadsTotal *prometheus.CounterVec | ||
| cacheWritesTotal *prometheus.CounterVec | ||
| } | ||
|
|
||
| // RoundTrip implements the RoundTripper interface. | ||
| func (r *RoundTripper) RoundTrip(req *http.Request) (*http.Response, error) { | ||
| key, err := r.key(req) | ||
| if err != nil { | ||
| return nil, errors.Wrap(err, "failed to generate key from request") | ||
| } | ||
|
|
||
| raw, ok, err := r.c.Get(key) | ||
| if err != nil { | ||
| r.cacheReadsTotal.WithLabelValues("error").Inc() | ||
| return nil, errors.Wrap(err, "failed to retrieve value from cache") | ||
| } | ||
|
|
||
| if !ok { | ||
| r.cacheReadsTotal.WithLabelValues("miss").Inc() | ||
| resp, err := r.next.RoundTrip(req) | ||
| if err == nil && resp.StatusCode/200 == 1 { | ||
| // Try to cache the response but don't block. | ||
| defer func() { | ||
| raw, err := httputil.DumpResponse(resp, true) | ||
| if err != nil { | ||
| level.Error(r.l).Log("msg", "failed to dump response", "err", err) | ||
| return | ||
| } | ||
| if err := r.c.Set(key, raw); err != nil { | ||
| r.cacheWritesTotal.WithLabelValues("error").Inc() | ||
| level.Error(r.l).Log("msg", "failed to set value in cache", "err", err) | ||
| return | ||
| } | ||
| r.cacheWritesTotal.WithLabelValues("success").Inc() | ||
| }() | ||
| } | ||
| return resp, err | ||
| } | ||
|
|
||
| r.cacheReadsTotal.WithLabelValues("hit").Inc() | ||
| resp, err := http.ReadResponse(bufio.NewReader(bytes.NewBuffer(raw)), req) | ||
| return resp, errors.Wrap(err, "failed to read response") | ||
| } | ||
|
|
||
| // NewRoundTripper creates a new http.RoundTripper that returns http.Responses | ||
| // from a cache. | ||
| func NewRoundTripper(c Cacher, key KeyFunc, next http.RoundTripper, l log.Logger, reg prometheus.Registerer) http.RoundTripper { | ||
| rt := &RoundTripper{ | ||
| c: c, | ||
| key: key, | ||
| next: next, | ||
| l: l, | ||
| cacheReadsTotal: prometheus.NewCounterVec( | ||
| prometheus.CounterOpts{ | ||
| Name: "cache_reads_total", | ||
| Help: "The number of read requests made to the cache.", | ||
| }, []string{"result"}, | ||
| ), | ||
| cacheWritesTotal: prometheus.NewCounterVec( | ||
| prometheus.CounterOpts{ | ||
| Name: "cache_writes_total", | ||
| Help: "The number of write requests made to the cache.", | ||
| }, []string{"result"}, | ||
| ), | ||
| } | ||
|
|
||
| if reg != nil { | ||
| reg.MustRegister(rt.cacheReadsTotal, rt.cacheWritesTotal) | ||
| } | ||
|
|
||
| return rt | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,67 @@ | ||
| package memcached | ||
|
|
||
| import ( | ||
| "crypto/sha256" | ||
| "fmt" | ||
|
|
||
| "github.com/bradfitz/gomemcache/memcache" | ||
| "github.com/pkg/errors" | ||
|
|
||
| tcache "github.com/openshift/telemeter/pkg/cache" | ||
| ) | ||
|
|
||
| // cache is a Cacher implemented on top of Memcached. | ||
| type cache struct { | ||
| *memcache.Client | ||
| expiration int32 | ||
| } | ||
|
|
||
| // New creates a new Cache from a list of Memcached servers | ||
| // and key expiration time given in seconds. | ||
| func New(expiration int32, servers ...string) tcache.Cacher { | ||
| return &cache{ | ||
| memcache.New(servers...), | ||
| expiration, | ||
| } | ||
| } | ||
|
|
||
| // Get returns a value from Memcached. | ||
| func (c *cache) Get(key string) ([]byte, bool, error) { | ||
| key, err := hash(key) | ||
| if err != nil { | ||
| return nil, false, err | ||
| } | ||
| i, err := c.Client.Get(key) | ||
| if err != nil { | ||
| if err == memcache.ErrCacheMiss { | ||
| return nil, false, nil | ||
| } | ||
| return nil, false, err | ||
| } | ||
|
|
||
| return i.Value, true, nil | ||
| } | ||
|
|
||
| // Set sets a value in Memcached. | ||
| func (c *cache) Set(key string, value []byte) error { | ||
| key, err := hash(key) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| i := memcache.Item{ | ||
| Key: key, | ||
| Value: value, | ||
| Expiration: c.expiration, | ||
| } | ||
| return c.Client.Set(&i) | ||
| } | ||
|
|
||
| // hashKey hashes the given key to ensure that it is less than 250 bytes, | ||
| // as Memcached cannot handler longer keys. | ||
| func hash(key string) (string, error) { | ||
| h := sha256.New() | ||
| if _, err := h.Write([]byte(key)); err != nil { | ||
| return "", errors.Wrap(err, "failed to hash key") | ||
| } | ||
| return fmt.Sprintf("%x", (h.Sum(nil))), nil | ||
| } |