embedded-cache: Bring fifocache and groupcache into single tent.

CHANGELOG.md

@@ -5,6 +5,7 @@
 #### Loki
 
 ##### Enhancements
+* [6821](https://github.com/grafana/loki/pull/6821) **kavirajk**: Introduce new cache type `embedded-cache` which is an in-process cache system that runs loki without the need for an external cache (like memcached, redis, etc). It can be run in two modes `distributed: false` (default, and same as old `fifocache`) and `distributed: true` which runs cache in distributed fashion sharding keys across peers if Loki is run in microservices or SSD mode.
 * [6691](https://github.com/grafana/loki/pull/6691) **dannykopping**: Update production-ready Loki cluster in docker-compose
 * [6317](https://github.com/grafana/loki/pull/6317) **dannykoping**: General: add cache usage statistics
 * [6444](https://github.com/grafana/loki/pull/6444) **aminesnow** Add TLS config to query frontend.

cmd/loki/loki-local-config.yaml

@@ -16,6 +16,14 @@ common:
     kvstore:
       store: inmemory
 
+query_range:
+  results_cache:
+    cache:
+      embedded_cache:
+        enabled: true
+        distributed: true
+        max_size_mb: 100
+
 schema_config:
   configs:
     - from: 2020-10-24

docs/sources/upgrading/_index.md

@@ -33,6 +33,12 @@ The output is incredibly verbose as it shows the entire internal config struct u
 
 ### Loki
 
+#### Fifocache is deprecated
+
+We introduced a new cache called `embedded-cache` which is an in-process cache system that make it possible to run Loki without the need for an external cache (like Memcached, Redis, etc). It can be run in two modes `distributed: false` (default, and same as old `fifocache`) and `distributed: true` which runs cache in distributed fashion sharding keys across peers if Loki is run in microservices or SSD mode.
+
+Currently `embedded-cache` with `distributed: true` can be enabled only for results cache.
+
 #### Evenly spread queriers across kubernetes nodes
 
 We now evenly spread queriers across the available kubernetes nodes, but allowing more than one querier to be scheduled into the same node.
@@ -87,7 +93,7 @@ limits_config:
   split_queries_by_interval: 10m
 ```
 
-In 2.5.0 it can only be defined in the `limits_config` section, **Loki will fail to start if you do not remove the `split_queries_by_interval` config from the `query_range` section.** 
+In 2.5.0 it can only be defined in the `limits_config` section, **Loki will fail to start if you do not remove the `split_queries_by_interval` config from the `query_range` section.**
 
 Additionally, it has a new default value of `30m` rather than `0`.
 
@@ -130,7 +136,7 @@ Meanwhile, the legacy format is a string in the following format:
 * `parallelise_shardable_queries` under the `query_range` config now defaults to `true`.
 * `split_queries_by_interval` under the `limits_config` config now defaults to `30m`, it was `0s`.
 * `max_chunk_age` in the `ingester` config now defaults to `2h` previously it was `1h`.
-* `query_ingesters_within` under the `querier` config now defaults to `3h`, previously it was `0s`. Any query (or subquery) that has an end time more than `3h` ago will not be sent to the ingesters, this saves work on the ingesters for data they normally don't contain. If you regularly write old data to Loki you may need to return this value to `0s` to always query ingesters. 
+* `query_ingesters_within` under the `querier` config now defaults to `3h`, previously it was `0s`. Any query (or subquery) that has an end time more than `3h` ago will not be sent to the ingesters, this saves work on the ingesters for data they normally don't contain. If you regularly write old data to Loki you may need to return this value to `0s` to always query ingesters.
 * `max_concurrent` under the `querier` config now defaults to `10` instead of `20`.
 * `match_max_concurrent` under the `frontend_worker` config now defaults to true, this supersedes the `parallelism` setting which can now be removed from your config. Controlling query parallelism of a single process can now be done with the `querier` `max_concurrent` setting.
 * `flush_op_timeout` under the `ingester` configuration block now defaults to `10m`, increased from `10s`. This can help when replaying a large WAL on Loki startup, and avoid `msg="failed to flush" ... context deadline exceeded` errors.
@@ -378,7 +384,7 @@ server:
   grpc_server_ping_without_stream_allowed: true
 ```
 
-[This issue](https://github.com/grafana/loki/issues/4375) has some more information on the change. 
+[This issue](https://github.com/grafana/loki/issues/4375) has some more information on the change.
 
 #### Some metric prefixes have changed from `cortex_` to `loki_`
 

pkg/loki/common/common.go

@@ -3,11 +3,10 @@ package common
 import (
 	"flag"
 
-	"github.com/grafana/loki/pkg/storage/chunk/cache"
-
 	"github.com/grafana/dskit/flagext"
 	"github.com/grafana/dskit/netutil"
 
+	"github.com/grafana/loki/pkg/storage/chunk/cache"
 	"github.com/grafana/loki/pkg/storage/chunk/client/aws"
 	"github.com/grafana/loki/pkg/storage/chunk/client/azure"
 	"github.com/grafana/loki/pkg/storage/chunk/client/baidubce"
@@ -46,10 +45,8 @@ type Config struct {
 	// CompactorAddress is the http address of the compactor in the form http://host:port
 	CompactorAddress string `yaml:"compactor_address"`
 
-	// GroupCacheConfig is the configuration to use when groupcache is enabled.
-	//
-	// This is a common config because, when enabled, it is used across all caches
-	GroupCacheConfig cache.GroupCacheConfig `yaml:"groupcache"`
+	// Global embedded-cache config. Independent of what type of cache, we need some singleton configs like Ring configuration when running in distributed fashion.
+	EmbeddedCacheConfig cache.EmbeddedCacheSingletonConfig `yaml:"embedded_cache"`
 }
 
 func (c *Config) RegisterFlags(f *flag.FlagSet) {
@@ -63,10 +60,9 @@ func (c *Config) RegisterFlags(f *flag.FlagSet) {
 	throwaway.StringVar(&c.InstanceAddr, "common.instance-addr", "", "Default advertised address to be used by Loki components.")
 	throwaway.Var((*flagext.StringSlice)(&c.InstanceInterfaceNames), "common.instance-interface-names", "List of network interfaces to read address from.")
 
-	// flags that only live in common
-	c.GroupCacheConfig.RegisterFlagsWithPrefix("common.groupcache", "", f)
-
 	f.StringVar(&c.CompactorAddress, "common.compactor-address", "", "the http address of the compactor in the form http://host:port")
+
+	c.EmbeddedCacheConfig.RegisterFlagsWithPrefix("common.embedded-cache", "", f)
 }
 
 type Storage struct {

pkg/loki/config_wrapper.go

@@ -96,8 +96,6 @@ func (c *ConfigWrapper) ApplyDynamicConfig() cfg.Source {
 
 		applyInstanceConfigs(r, &defaults)
 
-		applyCommonCacheConfigs(r, &defaults)
-
 		applyCommonReplicationFactor(r, &defaults)
 
 		applyDynamicRingConfigs(r, &defaults)
@@ -166,7 +164,9 @@ func applyInstanceConfigs(r, defaults *ConfigWrapper) {
 		}
 		r.Frontend.FrontendV2.Addr = r.Common.InstanceAddr
 		r.IndexGateway.Ring.InstanceAddr = r.Common.InstanceAddr
-		r.Common.GroupCacheConfig.Ring.InstanceAddr = r.Common.InstanceAddr
+		if r.QueryRange.CacheConfig.EmbeddedCache.IsEnabledWithDistributed() {
+			r.Common.EmbeddedCacheConfig.Ring.InstanceAddr = r.Common.InstanceAddr
+		}
 	}
 
 	if !reflect.DeepEqual(r.Common.InstanceInterfaceNames, defaults.Common.InstanceInterfaceNames) {
@@ -175,18 +175,9 @@ func applyInstanceConfigs(r, defaults *ConfigWrapper) {
 		}
 		r.Frontend.FrontendV2.InfNames = r.Common.InstanceInterfaceNames
 		r.IndexGateway.Ring.InstanceInterfaceNames = r.Common.InstanceInterfaceNames
-		r.Common.GroupCacheConfig.Ring.InstanceInterfaceNames = r.Common.InstanceInterfaceNames
-	}
-}
-
-// applyCommonCacheConfigs applies to Loki components the cache-related configurations under the common config section
-// NOTE: only used for GroupCache at the moment
-// TODO: apply to other caches as well
-func applyCommonCacheConfigs(r, _ *ConfigWrapper) {
-	if r.Config.Common.GroupCacheConfig.Enabled {
-		r.Config.ChunkStoreConfig.ChunkCacheConfig.EnableGroupCache = true
-		r.Config.QueryRange.ResultsCacheConfig.CacheConfig.EnableGroupCache = true
-		r.Config.StorageConfig.IndexQueriesCacheConfig.EnableGroupCache = true
+		if r.QueryRange.CacheConfig.EmbeddedCache.IsEnabledWithDistributed() {
+			r.Common.EmbeddedCacheConfig.Ring.InstanceInterfaceNames = r.Common.InstanceInterfaceNames
+		}
 	}
 }
 
@@ -315,17 +306,18 @@ func applyConfigToRings(r, defaults *ConfigWrapper, rc util.RingConfig, mergeWit
 		r.IndexGateway.Ring.KVStore = rc.KVStore
 	}
 
-	// GroupCacheRing
-	if mergeWithExisting || reflect.DeepEqual(r.Common.GroupCacheConfig.Ring, defaults.Common.GroupCacheConfig.Ring) {
-		r.Common.GroupCacheConfig.Ring.HeartbeatTimeout = rc.HeartbeatTimeout
-		r.Common.GroupCacheConfig.Ring.HeartbeatPeriod = rc.HeartbeatPeriod
-		r.Common.GroupCacheConfig.Ring.InstancePort = rc.InstancePort
-		r.Common.GroupCacheConfig.Ring.InstanceAddr = rc.InstanceAddr
-		r.Common.GroupCacheConfig.Ring.InstanceID = rc.InstanceID
-		r.Common.GroupCacheConfig.Ring.InstanceInterfaceNames = rc.InstanceInterfaceNames
-		r.Common.GroupCacheConfig.Ring.InstanceZone = rc.InstanceZone
-		r.Common.GroupCacheConfig.Ring.ZoneAwarenessEnabled = rc.ZoneAwarenessEnabled
-		r.Common.GroupCacheConfig.Ring.KVStore = rc.KVStore
+	// EmbeddedCache distributed ring.
+	if r.QueryRange.CacheConfig.EmbeddedCache.IsEnabledWithDistributed() &&
+		(mergeWithExisting || reflect.DeepEqual(r.Common.EmbeddedCacheConfig.Ring, defaults.Common.EmbeddedCacheConfig.Ring)) {
+		r.Common.EmbeddedCacheConfig.Ring.HeartbeatTimeout = rc.HeartbeatTimeout
+		r.Common.EmbeddedCacheConfig.Ring.HeartbeatPeriod = rc.HeartbeatPeriod
+		r.Common.EmbeddedCacheConfig.Ring.InstancePort = rc.InstancePort
+		r.Common.EmbeddedCacheConfig.Ring.InstanceAddr = rc.InstanceAddr
+		r.Common.EmbeddedCacheConfig.Ring.InstanceID = rc.InstanceID
+		r.Common.EmbeddedCacheConfig.Ring.InstanceInterfaceNames = rc.InstanceInterfaceNames
+		r.Common.EmbeddedCacheConfig.Ring.InstanceZone = rc.InstanceZone
+		r.Common.EmbeddedCacheConfig.Ring.ZoneAwarenessEnabled = rc.ZoneAwarenessEnabled
+		r.Common.EmbeddedCacheConfig.Ring.KVStore = rc.KVStore
 	}
 }
 
@@ -361,7 +353,7 @@ func applyTokensFilePath(cfg *ConfigWrapper) error {
 	if err != nil {
 		return err
 	}
-	cfg.Common.GroupCacheConfig.Ring.TokensFilePath = f
+	cfg.Common.EmbeddedCacheConfig.Ring.TokensFilePath = f
 	return nil
 }
 
@@ -440,8 +432,8 @@ func appendLoopbackInterface(cfg, defaults *ConfigWrapper) {
 		cfg.IndexGateway.Ring.InstanceInterfaceNames = append(cfg.IndexGateway.Ring.InstanceInterfaceNames, loopbackIface)
 	}
 
-	if reflect.DeepEqual(cfg.Common.GroupCacheConfig.Ring.InstanceInterfaceNames, defaults.Common.GroupCacheConfig.Ring.InstanceInterfaceNames) {
-		cfg.Common.GroupCacheConfig.Ring.InstanceInterfaceNames = append(cfg.Common.GroupCacheConfig.Ring.InstanceInterfaceNames, loopbackIface)
+	if reflect.DeepEqual(cfg.Common.EmbeddedCacheConfig.Ring.InstanceInterfaceNames, defaults.Common.EmbeddedCacheConfig.Ring.InstanceInterfaceNames) {
+		cfg.Common.EmbeddedCacheConfig.Ring.InstanceInterfaceNames = append(cfg.Common.EmbeddedCacheConfig.Ring.InstanceInterfaceNames, loopbackIface)
 	}
 }
 
@@ -456,7 +448,7 @@ func applyMemberlistConfig(r *ConfigWrapper) {
 	r.QueryScheduler.SchedulerRing.KVStore.Store = memberlistStr
 	r.CompactorConfig.CompactorRing.KVStore.Store = memberlistStr
 	r.IndexGateway.Ring.KVStore.Store = memberlistStr
-	r.Common.GroupCacheConfig.Ring.KVStore.Store = memberlistStr
+	r.Common.EmbeddedCacheConfig.Ring.KVStore.Store = memberlistStr
 }
 
 var ErrTooManyStorageConfigs = errors.New("too many storage configs provided in the common config, please only define one storage backend")

pkg/loki/config_wrapper_test.go

@@ -821,23 +821,24 @@ ingester:
 		})
 	})
 
-	t.Run("common groupcache setting is applied to chunk, index, and result caches", func(t *testing.T) {
+	t.Run("embedded-cache setting is applied to result caches", func(t *testing.T) {
 		// ensure they are all false by default
 		config, _, _ := configWrapperFromYAML(t, minimalConfig, nil)
-		assert.False(t, config.ChunkStoreConfig.ChunkCacheConfig.EnableGroupCache)
-		assert.False(t, config.StorageConfig.IndexQueriesCacheConfig.EnableGroupCache)
-		assert.False(t, config.QueryRange.ResultsCacheConfig.CacheConfig.EnableGroupCache)
+		assert.False(t, config.QueryRange.ResultsCacheConfig.CacheConfig.EmbeddedCache.Enabled)
+		assert.False(t, config.QueryRange.ResultsCacheConfig.CacheConfig.EmbeddedCache.Distributed)
 
 		configFileString := `---
-common:
-  groupcache:
-    enabled: true`
+query_range:
+  results_cache:
+    cache:
+      embedded_cache:
+        enabled: true
+        distributed: true`
 
 		config, _ = testContext(configFileString, nil)
 
-		assert.True(t, config.ChunkStoreConfig.ChunkCacheConfig.EnableGroupCache)
-		assert.True(t, config.StorageConfig.IndexQueriesCacheConfig.EnableGroupCache)
-		assert.True(t, config.QueryRange.ResultsCacheConfig.CacheConfig.EnableGroupCache)
+		assert.True(t, config.QueryRange.ResultsCacheConfig.CacheConfig.EmbeddedCache.Enabled)
+		assert.True(t, config.QueryRange.ResultsCacheConfig.CacheConfig.EmbeddedCache.Distributed)
 	})
 }
 
@@ -867,16 +868,18 @@ chunk_store_config:
 			assert.False(t, config.ChunkStoreConfig.ChunkCacheConfig.EnableFifoCache)
 		})
 
-		t.Run("no FIFO cache enabled by default if GroupCache is set", func(t *testing.T) {
+		t.Run("if distributed cache is set for results cache, FIFO cache should be disabled.", func(t *testing.T) {
 			configFileString := `---
-common:
-  groupcache:
-    enabled: true`
+query_range:
+  results_cache:
+    cache:
+      embedded_cache:
+        enabled: true
+        distributed: true`
 
 			config, _, _ := configWrapperFromYAML(t, configFileString, nil)
-			assert.False(t, config.ChunkStoreConfig.ChunkCacheConfig.EnableFifoCache)
-			assert.False(t, config.QueryRange.ResultsCacheConfig.CacheConfig.EnableFifoCache)
-			assert.True(t, config.ChunkStoreConfig.ChunkCacheConfig.EnableGroupCache)
+			assert.True(t, config.QueryRange.CacheConfig.EmbeddedCache.IsEnabledWithDistributed())
+			assert.False(t, config.QueryRange.CacheConfig.EnableFifoCache)
 		})
 
 		t.Run("FIFO cache is enabled by default if no other cache is set", func(t *testing.T) {

pkg/loki/loki.go

@@ -255,7 +255,7 @@ type Loki struct {
 	queryScheduler           *scheduler.Scheduler
 	usageReport              *usagestats.Reporter
 	indexGatewayRingManager  *indexgateway.RingManager
-	groupcacheRingManager    *cache.GroupcacheRingManager
+	embeddedcacheRingManager *cache.GroupcacheRingManager
 
 	clientMetrics       storage.ClientMetrics
 	deleteClientMetrics *deletion.DeleteRequestClientMetrics
@@ -476,7 +476,7 @@ func (t *Loki) setupModuleManager() error {
 	mm.RegisterModule(RuntimeConfig, t.initRuntimeConfig, modules.UserInvisibleModule)
 	mm.RegisterModule(MemberlistKV, t.initMemberlistKV, modules.UserInvisibleModule)
 	mm.RegisterModule(Ring, t.initRing, modules.UserInvisibleModule)
-	mm.RegisterModule(GroupCache, t.initGroupcache, modules.UserInvisibleModule)
+	mm.RegisterModule(Embededcache, t.initEmbeddedCache, modules.UserInvisibleModule)
 	mm.RegisterModule(Overrides, t.initOverrides, modules.UserInvisibleModule)
 	mm.RegisterModule(OverridesExporter, t.initOverridesExporter)
 	mm.RegisterModule(TenantConfigs, t.initTenantConfigs, modules.UserInvisibleModule)
@@ -503,16 +503,16 @@ func (t *Loki) setupModuleManager() error {
 	// Add dependencies
 	deps := map[string][]string{
 		Ring:                     {RuntimeConfig, Server, MemberlistKV},
-		GroupCache:               {RuntimeConfig, Server, MemberlistKV},
+		Embededcache:             {RuntimeConfig, Server, MemberlistKV},
 		UsageReport:              {},
 		Overrides:                {RuntimeConfig},
 		OverridesExporter:        {Overrides, Server},
 		TenantConfigs:            {RuntimeConfig},
 		Distributor:              {Ring, Server, Overrides, TenantConfigs, UsageReport},
-		Store:                    {Overrides, GroupCache, IndexGatewayRing},
+		Store:                    {Overrides, Embededcache, IndexGatewayRing},
 		Ingester:                 {Store, Server, MemberlistKV, TenantConfigs, UsageReport},
 		Querier:                  {Store, Ring, Server, IngesterQuerier, TenantConfigs, UsageReport},
-		QueryFrontendTripperware: {Server, GroupCache, Overrides, TenantConfigs},
+		QueryFrontendTripperware: {Server, Embededcache, Overrides, TenantConfigs},
 		QueryFrontend:            {QueryFrontendTripperware, UsageReport},
 		QueryScheduler:           {Server, Overrides, MemberlistKV, UsageReport},
 		Ruler:                    {Ring, Server, Store, RulerStorage, IngesterQuerier, Overrides, TenantConfigs, UsageReport},