Add configurable groupcache capacity (#6678)

* Add configurable groupcache capacity

Signed-off-by: Danny Kopping <danny.kopping@grafana.com>

* Improving the verbiage for clarity - thanks Travis!

Changed the default size to 100MB

Signed-off-by: Danny Kopping <danny.kopping@grafana.com>

* Adding config docs

Signed-off-by: Danny Kopping <danny.kopping@grafana.com>

* Removed test - it was pretty redundant, plus groupcache panics if you try init its http server more than once

Signed-off-by: Danny Kopping <danny.kopping@grafana.com>

docs/sources/configuration/_index.md

@@ -2660,6 +2660,16 @@ This way, one doesn't have to replicate configuration in multiple places.
 # Address and port number where the compactor API is served.
 # CLI flag: -common.compactor-address
 [compactor_address: <string> | default = ""]
+
+# Groupcache is an in-process, distributed cache that behaves similarly to memcached but is built-in to Loki
+groupcache:
+  # Enable groupcache
+  # CLI flag: -common.groupcache.enabled
+  [enabled: <boolean>: default = false]
+  # Set the maximum available memory to use for each groupcache group
+  # NOTE: there are 3 caches (result, chunk, and index query), so the maximum used memory will be *triple* the value specified here.
+  # CLI flag: -common.groupcache.capacity-per-cache-mb
+  [capacity_per_cache_mb: <int>: default = 100]
 ```
 
 ## analytics

pkg/loki/modules.go

@@ -167,15 +167,15 @@ func (t *Loki) initGroupcache() (_ services.Service, err error) {
 	}
 
 	t.Cfg.Common.GroupCacheConfig.Ring.ListenPort = t.Cfg.Server.HTTPListenPort
-	rm, err := cache.NewgGroupcacheRingManager(t.Cfg.Common.GroupCacheConfig, util_log.Logger, prometheus.DefaultRegisterer)
+	rm, err := cache.NewGroupcacheRingManager(t.Cfg.Common.GroupCacheConfig, util_log.Logger, prometheus.DefaultRegisterer)
 	if err != nil {
 		return nil, gerrors.Wrap(err, "new index gateway ring manager")
 	}
 
 	t.groupcacheRingManager = rm
 	t.Server.HTTP.Path("/groupcache/ring").Methods("GET", "POST").Handler(t.groupcacheRingManager)
 
-	gc, err := cache.NewGroupCache(rm, t.Server, util_log.Logger, prometheus.DefaultRegisterer)
+	gc, err := cache.NewGroupCache(rm, t.Cfg.Common.GroupCacheConfig, t.Server, util_log.Logger, prometheus.DefaultRegisterer)
 	if err != nil {
 		return nil, err
 	}

pkg/storage/chunk/cache/groupcache.go

@@ -45,6 +45,7 @@ type GroupCache struct {
 	wg                   sync.WaitGroup
 	reg                  prometheus.Registerer
 	startWaitingForClose context.CancelFunc
+	capacity             int64
 }
 
 // RingCfg is a wrapper for the Groupcache ring configuration plus the replication factor.
@@ -53,8 +54,9 @@ type RingCfg struct {
 }
 
 type GroupCacheConfig struct {
-	Enabled bool    `yaml:"enabled,omitempty"`
-	Ring    RingCfg `yaml:"ring,omitempty"`
+	Enabled    bool    `yaml:"enabled,omitempty"`
+	Ring       RingCfg `yaml:"ring,omitempty"`
+	CapacityMB int64   `yaml:"capacity_per_cache_mb,omitempty"`
 
 	Cache Cache `yaml:"-"`
 }
@@ -64,14 +66,16 @@ func (cfg *GroupCacheConfig) RegisterFlagsWithPrefix(prefix, _ string, f *flag.F
 	cfg.Ring.RegisterFlagsWithPrefix(prefix, "", f)
 
 	f.BoolVar(&cfg.Enabled, prefix+".enabled", false, "Whether or not groupcache is enabled")
+	f.Int64Var(&cfg.CapacityMB, prefix+".capacity-per-cache-mb", 100, "Capacity of each groupcache group in MB (default: 100). "+
+		"NOTE: there are 3 caches (result, chunk, and index query), so the maximum used memory will be *triple* the value specified here.")
 }
 
 type ringManager interface {
 	Addr() string
 	Ring() ring.ReadRing
 }
 
-func NewGroupCache(rm ringManager, server *server.Server, logger log.Logger, reg prometheus.Registerer) (*GroupCache, error) {
+func NewGroupCache(rm ringManager, config GroupCacheConfig, server *server.Server, logger log.Logger, reg prometheus.Registerer) (*GroupCache, error) {
 	addr := fmt.Sprintf("http://%s", rm.Addr())
 	level.Info(logger).Log("msg", "groupcache local address set to", "addr", addr)
 
@@ -88,6 +92,7 @@ func NewGroupCache(rm ringManager, server *server.Server, logger log.Logger, reg
 		wg:                   sync.WaitGroup{},
 		startWaitingForClose: cancel,
 		reg:                  reg,
+		capacity:             config.CapacityMB * 1e6, // MB => B
 	}
 
 	go func() {
@@ -176,7 +181,7 @@ func (c *GroupCache) NewGroup(name string, ct stats.CacheType) Cache {
 	}, []string{"operation"})
 
 	g := &group{
-		cache:         groupcache.NewGroup(name, 1<<30, missGetter),
+		cache:         groupcache.NewGroup(name, c.capacity, missGetter),
 		logger:        c.logger,
 		wg:            &c.wg,
 		cacheType:     ct,

pkg/storage/chunk/cache/groupcache_ringmanager.go

@@ -43,7 +43,7 @@ type GroupcacheRingManager struct {
 // NewRingManager is the recommended way of instantiating a GroupcacheRingManager.
 //
 // The other functions will assume the GroupcacheRingManager was instantiated through this function.
-func NewgGroupcacheRingManager(cfg GroupCacheConfig, log log.Logger, registerer prometheus.Registerer) (*GroupcacheRingManager, error) {
+func NewGroupcacheRingManager(cfg GroupCacheConfig, log log.Logger, registerer prometheus.Registerer) (*GroupcacheRingManager, error) {
 	rm := &GroupcacheRingManager{
 		cfg: cfg, log: log,
 	}

pkg/storage/chunk/cache/groupcache_test.go

@@ -47,12 +47,10 @@ func TestGroupCache(t *testing.T) {
 }
 
 func setupGroupCache() (*GroupCache, error) {
-	return NewGroupCache(
-		&mockRingManager{},
-		&server.Server{HTTP: mux.NewRouter()},
-		log.NewNopLogger(),
-		nil,
-	)
+	return NewGroupCache(&mockRingManager{}, GroupCacheConfig{
+		Enabled:    true,
+		CapacityMB: 1,
+	}, &server.Server{HTTP: mux.NewRouter()}, log.NewNopLogger(), nil)
 }
 
 type mockRingManager struct{}