cache: add TTL support

Adds TTL support to galaxycache as per this pull request:
thanos-community/galaxycache#1

Signed-off-by: Giedrius Statkevičius <giedrius.statkevicius@vinted.com>

go.mod

@@ -103,6 +103,8 @@ replace (
 	// Make sure Prometheus version is pinned as Prometheus semver does not include Go APIs.
 	github.com/prometheus/prometheus => github.com/prometheus/prometheus v1.8.2-0.20211119115433-692a54649ed7
 	github.com/sercand/kuberesolver => github.com/sercand/kuberesolver v2.4.0+incompatible
+
+	github.com/vimeo/galaxycache => github.com/thanos-community/galaxycache v0.0.0-20211122094458-3a32041a1f1e
 	google.golang.org/grpc => google.golang.org/grpc v1.40.0
 
 	// Overriding to use latest commit

go.sum

@@ -1651,6 +1651,8 @@ github.com/tencentcloud/tencentcloud-sdk-go/tencentcloud/common v1.0.194/go.mod
 github.com/tencentcloud/tencentcloud-sdk-go/tencentcloud/kms v1.0.194/go.mod h1:yrBKWhChnDqNz1xuXdSbWXG56XawEq0G5j1lg4VwBD4=
 github.com/tencentyun/cos-go-sdk-v5 v0.7.31 h1:NujkkOKMJ3IFs1+trCwXOKRCIPQ8qI5Lxul9JkhTg6M=
 github.com/tencentyun/cos-go-sdk-v5 v0.7.31/go.mod h1:4E4+bQ2gBVJcgEC9Cufwylio4mXOct2iu05WjgEBx1o=
+github.com/thanos-community/galaxycache v0.0.0-20211122094458-3a32041a1f1e h1:f1Zsv7OAU9iQhZwigp50Yl38W10g/vd5NC8Rdk1Jzng=
+github.com/thanos-community/galaxycache v0.0.0-20211122094458-3a32041a1f1e/go.mod h1:jXcofnrSln/cLI6/dhlBxPQZEEQHVPCcFaH75M+nSzM=
 github.com/thanos-io/thanos v0.8.1-0.20200109203923-552ffa4c1a0d/go.mod h1:usT/TxtJQ7DzinTt+G9kinDQmRS5sxwu0unVKZ9vdcw=
 github.com/thanos-io/thanos v0.13.1-0.20200731083140-69b87607decf/go.mod h1:G8caR6G7pSDreRDvFm9wFuyjEBztmr8Ag3kBYpa/fEc=
 github.com/thanos-io/thanos v0.13.1-0.20200807203500-9b578afb4763/go.mod h1:KyW0a93tsh7v4hXAwo2CVAIRYuZT1Kkf4e04gisQjAg=
@@ -1698,8 +1700,6 @@ github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyC
 github.com/valyala/fasttemplate v1.0.1/go.mod h1:UQGH1tvbgY+Nz5t2n7tXsz52dQxojPUpymEIMZ47gx8=
 github.com/valyala/fasttemplate v1.2.1/go.mod h1:KHLXt3tVN2HBp8eijSv/kGJopbvo7S+qRAEEKiv+SiQ=
 github.com/vektah/gqlparser v1.1.2/go.mod h1:1ycwN7Ij5njmMkPPAOaRFY4rET2Enx7IkVv3vaXspKw=
-github.com/vimeo/galaxycache v0.0.0-20210323154928-b7e5d71c067a h1:5rsX36sN7yTddsMiWRv1bkPF52m6U0DvBdJDKJus7f0=
-github.com/vimeo/galaxycache v0.0.0-20210323154928-b7e5d71c067a/go.mod h1:rGzH8dZnpycpywZjY5k/G1k8q5bzVUiIUM8EsTni2NU=
 github.com/vishvananda/netlink v0.0.0-20181108222139-023a6dafdcdf/go.mod h1:+SR5DhBJrl6ZM7CoCKvpw5BKroDKQ+PJqOg65H/2ktk=
 github.com/vishvananda/netlink v1.1.0/go.mod h1:cTgwzPIzzgDAYoQrMm0EdrjRUBkTqKYppBueQtXaqoE=
 github.com/vishvananda/netlink v1.1.1-0.20201029203352-d40f9887b852/go.mod h1:twkDnbuQxJYemMlGd4JFIcuhgX83tXhKS2B/PRMpOho=

pkg/cache/groupcache.go

@@ -78,17 +78,21 @@ func parseGroupcacheConfig(conf []byte) (GroupcacheConfig, error) {
 }
 
 // NewGroupcache creates a new Groupcache instance.
-func NewGroupcache(logger log.Logger, reg prometheus.Registerer, conf []byte, basepath string, r *route.Router, bucket objstore.Bucket) (*Groupcache, error) {
+func NewGroupcache(logger log.Logger, reg prometheus.Registerer, conf []byte, basepath string, r *route.Router, bucket objstore.Bucket,
+	isTSDBChunkFile, isMetaFile, isBlocksRootDir func(path string) bool,
+	MetaFileExistsTTL, MetafileDoesntExistTTL, MetafileContentTTL, ChunkObjectAttrsTTL, ChunkSubrangeTTL, BlocksIterTTL time.Duration) (*Groupcache, error) {
 	config, err := parseGroupcacheConfig(conf)
 	if err != nil {
 		return nil, err
 	}
 
-	return NewGroupcacheWithConfig(logger, reg, config, basepath, r, bucket)
+	return NewGroupcacheWithConfig(logger, reg, config, basepath, r, bucket, isTSDBChunkFile, isMetaFile, isBlocksRootDir, MetaFileExistsTTL, MetafileDoesntExistTTL, MetafileContentTTL, ChunkObjectAttrsTTL, ChunkSubrangeTTL, BlocksIterTTL)
 }
 
 // NewGroupcacheWithConfig creates a new Groupcache instance with the given config.
-func NewGroupcacheWithConfig(logger log.Logger, reg prometheus.Registerer, conf GroupcacheConfig, basepath string, r *route.Router, bucket objstore.Bucket) (*Groupcache, error) {
+func NewGroupcacheWithConfig(logger log.Logger, reg prometheus.Registerer, conf GroupcacheConfig, basepath string, r *route.Router, bucket objstore.Bucket,
+	isTSDBChunkFile, isMetaFile, isBlocksRootDir func(path string) bool,
+	MetaFileExistsTTL, MetafileDoesntExistTTL, MetafileContentTTL, ChunkObjectAttrsTTL, ChunkSubrangeTTL, BlocksIterTTL time.Duration) (*Groupcache, error) {
 	httpProto := galaxyhttp.NewHTTPFetchProtocol(&galaxyhttp.HTTPOptions{
 		BasePath: basepath,
 	})
@@ -140,7 +144,11 @@ func NewGroupcacheWithConfig(logger log.Logger, reg prometheus.Registerer, conf
 				if err != nil {
 					return err
 				}
-				return dest.UnmarshalBinary(finalAttrs)
+
+				if isTSDBChunkFile(parsedData.Name) {
+					return dest.UnmarshalBinary(finalAttrs, time.Now().Add(ChunkObjectAttrsTTL))
+				}
+				panic("caching bucket layer must not call on unconfigured paths")
 			case cachekey.IterVerb:
 				var list []string
 				if err := bucket.Iter(ctx, parsedData.Name, func(s string) error {
@@ -154,7 +162,11 @@ func NewGroupcacheWithConfig(logger log.Logger, reg prometheus.Registerer, conf
 				if err != nil {
 					return err
 				}
-				return dest.UnmarshalBinary(encodedList)
+
+				if isBlocksRootDir(parsedData.Name) {
+					return dest.UnmarshalBinary(encodedList, time.Now().Add(BlocksIterTTL))
+				}
+				panic("caching bucket layer must not call on unconfigured paths")
 			case cachekey.ContentVerb:
 				rc, err := bucket.Get(ctx, parsedData.Name)
 				if err != nil {
@@ -167,14 +179,26 @@ func NewGroupcacheWithConfig(logger log.Logger, reg prometheus.Registerer, conf
 					return err
 				}
 
-				return dest.UnmarshalBinary(b)
+				if isMetaFile(parsedData.Name) {
+					return dest.UnmarshalBinary(b, time.Now().Add(MetafileContentTTL))
+				}
+				panic("caching bucket layer must not call on unconfigured paths")
+
 			case cachekey.ExistsVerb:
 				exists, err := bucket.Exists(ctx, parsedData.Name)
 				if err != nil {
 					return err
 				}
 
-				return dest.UnmarshalBinary([]byte(strconv.FormatBool(exists)))
+				if isMetaFile(parsedData.Name) {
+					if exists {
+						return dest.UnmarshalBinary([]byte(strconv.FormatBool(exists)), time.Now().Add(MetaFileExistsTTL))
+					} else {
+						return dest.UnmarshalBinary([]byte(strconv.FormatBool(exists)), time.Now().Add(MetafileDoesntExistTTL))
+					}
+				}
+				panic("caching bucket layer must not call on unconfigured paths")
+
 			case cachekey.SubrangeVerb:
 				rc, err := bucket.GetRange(ctx, parsedData.Name, parsedData.Start, parsedData.End-parsedData.Start)
 				if err != nil {
@@ -187,7 +211,11 @@ func NewGroupcacheWithConfig(logger log.Logger, reg prometheus.Registerer, conf
 					return err
 				}
 
-				return dest.UnmarshalBinary(b)
+				if isTSDBChunkFile(parsedData.Name) {
+					return dest.UnmarshalBinary(b, time.Now().Add(ChunkSubrangeTTL))
+				}
+				panic("caching bucket layer must not call on unconfigured paths")
+
 			}
 
 			return nil
@@ -218,8 +246,14 @@ func (c *Groupcache) Fetch(ctx context.Context, keys []string) map[string][]byte
 			continue
 		}
 
-		if len(codec) > 0 {
-			data[k] = codec
+		retrievedData, _, err := codec.MarshalBinary()
+		if err != nil {
+			level.Error(c.logger).Log("msg", "failed retrieving data", "err", err, "key", k)
+			continue
+		}
+
+		if len(retrievedData) > 0 {
+			data[k] = retrievedData
 		}
 	}
 

pkg/store/cache/caching_bucket_factory.go

@@ -43,6 +43,8 @@ type CachingWithBackendConfig struct {
 	// Maximum number of GetRange requests issued by this bucket for single GetRange call. Zero or negative value = unlimited.
 	MaxChunksGetRangeRequests int `yaml:"max_chunks_get_range_requests"`
 
+	MetafileMaxSize model.Bytes `yaml:"metafile_max_size"`
+
 	// TTLs for various cache items.
 	ChunkObjectAttrsTTL time.Duration `yaml:"chunk_object_attrs_ttl"`
 	ChunkSubrangeTTL    time.Duration `yaml:"chunk_subrange_ttl"`
@@ -54,7 +56,6 @@ type CachingWithBackendConfig struct {
 	MetafileExistsTTL      time.Duration `yaml:"metafile_exists_ttl"`
 	MetafileDoesntExistTTL time.Duration `yaml:"metafile_doesnt_exist_ttl"`
 	MetafileContentTTL     time.Duration `yaml:"metafile_content_ttl"`
-	MetafileMaxSize        model.Bytes   `yaml:"metafile_max_size"`
 }
 
 func (cfg *CachingWithBackendConfig) Defaults() {
@@ -103,7 +104,10 @@ func NewCachingBucketFromYaml(yamlContent []byte, bucket objstore.Bucket, logger
 	case string(GroupcacheBucketCacheProvider):
 		const basePath = "/_galaxycache/"
 
-		c, err = cache.NewGroupcache(logger, reg, backendConfig, basePath, r, bucket)
+		c, err = cache.NewGroupcache(logger, reg, backendConfig, basePath, r, bucket,
+			isTSDBChunkFile, isMetaFile, isBlocksRootDir,
+			config.MetafileExistsTTL, config.MetafileDoesntExistTTL, config.MetafileContentTTL,
+			config.ChunkObjectAttrsTTL, config.ChunkSubrangeTTL, config.BlocksIterTTL)
 		if err != nil {
 			return nil, errors.Wrap(err, "failed to create groupcache")
 		}