pico

feat(pgs): lru cache for object info and special files rss

created pr with 59.1 on 2025-04-06T03:01:44Z · by c8ef7d19

added 59.2 on 2025-04-06T03:03:49Z · by c8ef7d19


                      1:  2cf56f0 ! 1:  26daea4 feat(pgs): lru cache for object info and special files


                      2:  caace51 ! 2:  b004b64 chore(pgs): use http cache clear event to rm lru cache for special files

added 59.3 on 2025-04-06T19:08:31Z · by c8ef7d19


                      1:  26daea4 = 1:  26daea4 feat(pgs): lru cache for object info and special files


                      2:  b004b64 = 2:  b004b64 chore(pgs): use http cache clear event to rm lru cache for special files


                      -:  ------- > 3:  59f5618 refactor(pgs): store lru cache on web router

added 59.4 on 2025-04-06T19:41:38Z · by c8ef7d19


                      1:  26daea4 = 1:  26daea4 feat(pgs): lru cache for object info and special files


                      2:  b004b64 = 2:  b004b64 chore(pgs): use http cache clear event to rm lru cache for special files


                      3:  59f5618 = 3:  59f5618 refactor(pgs): store lru cache on web router


                      -:  ------- > 4:  ee12290 refactor(pgs): update minio lru and remove object info cache

cmds

checkout latest patchset:

ssh pr.pico.sh print 59 | git am -3

checkout any patchset in a patch request:

ssh pr.pico.sh print 59.[rev] | git am -3

add changes to patch request:

git format-patch main --stdout | ssh pr.pico.sh pr add 59

Patchset `59.4` on 2025-04-06T19:41:38Z · commit ee12290

feat(pgs): lru cache for object info and special files

Eric Bower 2025-04-06T02:55:19Z

chore(pgs): use http cache clear event to rm lru cache for special files

Eric Bower 2025-04-06T03:03:27Z

refactor(pgs): store lru cache on web router

Eric Bower 2025-04-06T19:07:36Z

→ refactor(pgs): update minio lru and remove object info cache

Eric Bower 2025-04-06T19:41:14Z

refactor(pgs): update minio lru and remove object info cache

Eric Bower 2025-04-06T19:41:14Z

Semantic diff summary

1 added, 9 modified, 1 signature changed, 0 removed across 5 analyzed files

pkg/apps/pgs/web.go

method_declaration initRouters added
function_declaration ServeAsset modified

pkg/apps/pgs/web_asset_handler.go

function_declaration ServeHTTP modified

pkg/pobj/storage/minio.go

chunk lines 5-11 modified
type_declaration StorageMinio modified
function_declaration NewStorageMinio signature changed
function_declaration GetBucket modified
function_declaration GetObject modified

pkg/pobj/util.go

function_declaration EnvDriverDetector modified

pkg/shared/storage/minio.go

type_declaration StorageMinio modified
function_declaration NewStorageMinio modified

+11 -9 pkg/apps/pgs/web.go #

......@@ -109,9 +109,20 @@ func NewWebRouter(cfg *PgsConfig) *WebRouter {
		HeadersCache:   expirable.NewLRU[string, []*HeaderRule](2048, nil, cache.CacheTimeout),
	}
	router.initRouters()
	go router.watchCacheClear()
	return router
}

func (web *WebRouter) watchCacheClear() {
	for key := range web.Cfg.CacheClearingQueue {
		web.Cfg.Logger.Info("lru cache clear request", "key", key)
		rKey := filepath.Join(key, "_redirects")
		web.RedirectsCache.Remove(rKey)
		hKey := filepath.Join(key, "_headers")
		web.HeadersCache.Remove(hKey)
	}
}

func (web *WebRouter) initRouters() {
	// ensure legacy router is disabled
	// GODEBUG=httpmuxgo121=0

......@@ -516,15 +527,6 @@ func (web *WebRouter) ServeAsset(fname string, opts *storage.ImgProcessOpts, fro
		}
	}

	go func() {
		for key := range web.Cfg.CacheClearingQueue {
			rKey := filepath.Join(key, "_redirects")
			web.RedirectsCache.Remove(rKey)
			hKey := filepath.Join(key, "_headers")
			web.HeadersCache.Remove(hKey)
		}
	}()

	asset := &ApiAssetHandler{
		WebRouter: web,
		Logger:    logger,

+6 -1 pkg/apps/pgs/web_asset_handler.go #

......@@ -43,10 +43,12 @@ func (h *ApiAssetHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	var redirects []*RedirectRule

	redirectsCacheKey := filepath.Join(getSurrogateKey(h.UserID, h.ProjectDir), "_redirects")
	logger.Info("looking for _redirects in lru cache", "key", redirectsCacheKey)
	if cachedRedirects, found := h.RedirectsCache.Get(redirectsCacheKey); found {
		fmt.Println(cachedRedirects)
		logger.Info("_redirects found in lru cache", "key", redirectsCacheKey)
		redirects = cachedRedirects
	} else {
		logger.Info("_redirects not found in lru cache", "key", redirectsCacheKey)
		redirectFp, redirectInfo, err := h.Cfg.Storage.GetObject(h.Bucket, filepath.Join(h.ProjectDir, "_redirects"))
		if err == nil {
			defer redirectFp.Close()

......@@ -174,9 +176,12 @@ func (h *ApiAssetHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	var headers []*HeaderRule

	headersCacheKey := filepath.Join(getSurrogateKey(h.UserID, h.ProjectDir), "_headers")
	logger.Info("looking for _headers in lru cache", "key", headersCacheKey)
	if cachedHeaders, found := h.HeadersCache.Get(headersCacheKey); found {
		logger.Info("_headers found in lru", "key", headersCacheKey)
		headers = cachedHeaders
	} else {
		logger.Info("_headers not found in lru cache", "key", headersCacheKey)
		headersFp, headersInfo, err := h.Cfg.Storage.GetObject(h.Bucket, filepath.Join(h.ProjectDir, "_headers"))
		if err == nil {
			defer headersFp.Close()

+29 -36 pkg/pobj/storage/minio.go #

......@@ -5,6 +5,7 @@ import (
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net/url"
	"os"
	"path/filepath"

......@@ -21,8 +22,10 @@ import (
)

type StorageMinio struct {
	Client *minio.Client
	Admin  *madmin.AdminClient
	Client      *minio.Client
	Admin       *madmin.AdminClient
	BucketCache *expirable.LRU[string, CachedBucket]
	Logger      *slog.Logger
}

type CachedBucket struct {

......@@ -38,12 +41,9 @@ type CachedObjectInfo struct {
var (
	_ ObjectStorage = &StorageMinio{}
	_ ObjectStorage = (*StorageMinio)(nil)

	bucketCache     = expirable.NewLRU[string, CachedBucket](2048, nil, cache.CacheTimeout)
	objectInfoCache = expirable.NewLRU[string, CachedObjectInfo](2048, nil, cache.CacheTimeout)
)

func NewStorageMinio(address, user, pass string) (*StorageMinio, error) {
func NewStorageMinio(logger *slog.Logger, address, user, pass string) (*StorageMinio, error) {
	endpoint, err := url.Parse(address)
	if err != nil {
		return nil, err

......@@ -70,16 +70,20 @@ func NewStorageMinio(address, user, pass string) (*StorageMinio, error) {
	}

	mini := &StorageMinio{
		Client: mClient,
		Admin:  aClient,
		Client:      mClient,
		Admin:       aClient,
		BucketCache: expirable.NewLRU[string, CachedBucket](2048, nil, cache.CacheTimeout),
		Logger:      logger,
	}
	return mini, err
}

func (s *StorageMinio) GetBucket(name string) (Bucket, error) {
	if cachedBucket, found := bucketCache.Get(name); found {
	if cachedBucket, found := s.BucketCache.Get(name); found {
		s.Logger.Info("bucket found in lru cache", "name", name)
		return cachedBucket.Bucket, cachedBucket.Error
	}
	s.Logger.Info("bucket not found in lru cache", "name", name)

	bucket := Bucket{
		Name: name,

......@@ -91,11 +95,11 @@ func (s *StorageMinio) GetBucket(name string) (Bucket, error) {
			err = errors.New("bucket does not exist")
		}

		bucketCache.Add(name, CachedBucket{bucket, err})
		s.BucketCache.Add(name, CachedBucket{bucket, err})
		return bucket, err
	}

	bucketCache.Add(name, CachedBucket{bucket, nil})
	s.BucketCache.Add(name, CachedBucket{bucket, nil})

	return bucket, nil
}

......@@ -188,34 +192,23 @@ func (s *StorageMinio) GetObject(bucket Bucket, fpath string) (utils.ReadAndRead

	cacheKey := filepath.Join(bucket.Name, fpath)

	cachedInfo, found := objectInfoCache.Get(cacheKey)
	if found {
		objInfo = cachedInfo.ObjectInfo

		if cachedInfo.Error != nil {
			return nil, objInfo, cachedInfo.Error
		}
	} else {
		info, err := s.Client.StatObject(context.Background(), bucket.Name, fpath, minio.StatObjectOptions{})
		if err != nil {
			objectInfoCache.Add(cacheKey, CachedObjectInfo{objInfo, err})
			return nil, objInfo, err
		}
	s.Logger.Info("object info not found in lru cache", "key", cacheKey)
	info, err := s.Client.StatObject(context.Background(), bucket.Name, fpath, minio.StatObjectOptions{})
	if err != nil {
		return nil, objInfo, err
	}

		objInfo.LastModified = info.LastModified
		objInfo.ETag = info.ETag
		objInfo.Metadata = info.Metadata
		objInfo.UserMetadata = info.UserMetadata
		objInfo.Size = info.Size
	objInfo.LastModified = info.LastModified
	objInfo.ETag = info.ETag
	objInfo.Metadata = info.Metadata
	objInfo.UserMetadata = info.UserMetadata
	objInfo.Size = info.Size

		if mtime, ok := info.UserMetadata["Mtime"]; ok {
			mtimeUnix, err := strconv.Atoi(mtime)
			if err == nil {
				objInfo.LastModified = time.Unix(int64(mtimeUnix), 0)
			}
	if mtime, ok := info.UserMetadata["Mtime"]; ok {
		mtimeUnix, err := strconv.Atoi(mtime)
		if err == nil {
			objInfo.LastModified = time.Unix(int64(mtimeUnix), 0)
		}

		objectInfoCache.Add(cacheKey, CachedObjectInfo{objInfo, nil})
	}

	obj, err := s.Client.GetObject(context.Background(), bucket.Name, fpath, minio.GetObjectOptions{})

+1 -1 pkg/pobj/util.go #

......@@ -29,7 +29,7 @@ func EnvDriverDetector(logger *slog.Logger) (storage.ObjectStorage, error) {
			"url", url,
			"user", user,
		)
		return storage.NewStorageMinio(url, user, pass)
		return storage.NewStorageMinio(logger, url, user, pass)
	}

	// implied driver == "fs"

+2 -3 pkg/shared/storage/minio.go #

......@@ -13,15 +13,14 @@ import (

type StorageMinio struct {
	*sst.StorageMinio
	Logger *slog.Logger
}

func NewStorageMinio(logger *slog.Logger, address, user, pass string) (*StorageMinio, error) {
	st, err := sst.NewStorageMinio(address, user, pass)
	st, err := sst.NewStorageMinio(logger, address, user, pass)
	if err != nil {
		return nil, err
	}
	return &StorageMinio{st, logger}, nil
	return &StorageMinio{st}, nil
}

func (s *StorageMinio) ServeObject(bucket sst.Bucket, fpath string, opts *ImgProcessOpts) (io.ReadCloser, *sst.ObjectInfo, error) {

...	...		@@ -109,9 +109,20 @@ func NewWebRouter(cfg PgsConfig) WebRouter {
109	109		HeadersCache: expirable.NewLRU[string, []*HeaderRule](2048, nil, cache.CacheTimeout),
110	110		}
111	111		router.initRouters()
	112	+	go router.watchCacheClear()
112	113		return router
113	114		}
114	115
	116	+	func (web *WebRouter) watchCacheClear() {
	117	+	for key := range web.Cfg.CacheClearingQueue {
	118	+	web.Cfg.Logger.Info("lru cache clear request", "key", key)
	119	+	rKey := filepath.Join(key, "_redirects")
	120	+	web.RedirectsCache.Remove(rKey)
	121	+	hKey := filepath.Join(key, "_headers")
	122	+	web.HeadersCache.Remove(hKey)
	123	+	}
	124	+	}
	125	+
115	126		func (web *WebRouter) initRouters() {
116	127		// ensure legacy router is disabled
117	128		// GODEBUG=httpmuxgo121=0

...	...		@@ -516,15 +527,6 @@ func (web WebRouter) ServeAsset(fname string, opts storage.ImgProcessOpts, fro
516	527		}
517	528		}
518	529
519		-	go func() {
520		-	for key := range web.Cfg.CacheClearingQueue {
521		-	rKey := filepath.Join(key, "_redirects")
522		-	web.RedirectsCache.Remove(rKey)
523		-	hKey := filepath.Join(key, "_headers")
524		-	web.HeadersCache.Remove(hKey)
525		-	}
526		-	}()
527		-
528	530		asset := &ApiAssetHandler{
529	531		WebRouter: web,
530	532		Logger: logger,

...	...		@@ -174,9 +176,12 @@ func (h ApiAssetHandler) ServeHTTP(w http.ResponseWriter, r http.Request) {
174	176		var headers []*HeaderRule
175	177
176	178		headersCacheKey := filepath.Join(getSurrogateKey(h.UserID, h.ProjectDir), "_headers")
	179	+	logger.Info("looking for _headers in lru cache", "key", headersCacheKey)
177	180		if cachedHeaders, found := h.HeadersCache.Get(headersCacheKey); found {
	181	+	logger.Info("_headers found in lru", "key", headersCacheKey)
178	182		headers = cachedHeaders
179	183		} else {
	184	+	logger.Info("_headers not found in lru cache", "key", headersCacheKey)
180	185		headersFp, headersInfo, err := h.Cfg.Storage.GetObject(h.Bucket, filepath.Join(h.ProjectDir, "_headers"))
181	186		if err == nil {
182	187		defer headersFp.Close()

...	...		@@ -70,16 +70,20 @@ func NewStorageMinio(address, user, pass string) (*StorageMinio, error) {
70	70		}
71	71
72	72		mini := &StorageMinio{
73		-	Client: mClient,
74		-	Admin: aClient,
	73	+	Client: mClient,
	74	+	Admin: aClient,
	75	+	BucketCache: expirable.NewLRU[string, CachedBucket](2048, nil, cache.CacheTimeout),
	76	+	Logger: logger,
75	77		}
76	78		return mini, err
77	79		}
78	80
79	81		func (s *StorageMinio) GetBucket(name string) (Bucket, error) {
80		-	if cachedBucket, found := bucketCache.Get(name); found {
	82	+	if cachedBucket, found := s.BucketCache.Get(name); found {
	83	+	s.Logger.Info("bucket found in lru cache", "name", name)
81	84		return cachedBucket.Bucket, cachedBucket.Error
82	85		}
	86	+	s.Logger.Info("bucket not found in lru cache", "name", name)
83	87
84	88		bucket := Bucket{
85	89		Name: name,

...	...		@@ -91,11 +95,11 @@ func (s *StorageMinio) GetBucket(name string) (Bucket, error) {
91	95		err = errors.New("bucket does not exist")
92	96		}
93	97
94		-	bucketCache.Add(name, CachedBucket{bucket, err})
	98	+	s.BucketCache.Add(name, CachedBucket{bucket, err})
95	99		return bucket, err
96	100		}
97	101
98		-	bucketCache.Add(name, CachedBucket{bucket, nil})
	102	+	s.BucketCache.Add(name, CachedBucket{bucket, nil})
99	103
100	104		return bucket, nil
101	105		}

...	...		@@ -43,10 +43,12 @@ func (h ApiAssetHandler) ServeHTTP(w http.ResponseWriter, r http.Request) {
43	43		var redirects []*RedirectRule
44	44
45	45		redirectsCacheKey := filepath.Join(getSurrogateKey(h.UserID, h.ProjectDir), "_redirects")
	46	+	logger.Info("looking for _redirects in lru cache", "key", redirectsCacheKey)
46	47		if cachedRedirects, found := h.RedirectsCache.Get(redirectsCacheKey); found {
47		-	fmt.Println(cachedRedirects)
	48	+	logger.Info("_redirects found in lru cache", "key", redirectsCacheKey)
48	49		redirects = cachedRedirects
49	50		} else {
	51	+	logger.Info("_redirects not found in lru cache", "key", redirectsCacheKey)
50	52		redirectFp, redirectInfo, err := h.Cfg.Storage.GetObject(h.Bucket, filepath.Join(h.ProjectDir, "_redirects"))
51	53		if err == nil {
52	54		defer redirectFp.Close()

...	...		@@ -5,6 +5,7 @@ import (
5	5		"errors"
6	6		"fmt"
7	7		"io"
	8	+	"log/slog"
8	9		"net/url"
9	10		"os"
10	11		"path/filepath"

...	...		@@ -21,8 +22,10 @@ import (
21	22		)
22	23
23	24		type StorageMinio struct {
24		-	Client *minio.Client
25		-	Admin *madmin.AdminClient
	25	+	Client *minio.Client
	26	+	Admin *madmin.AdminClient
	27	+	BucketCache *expirable.LRU[string, CachedBucket]
	28	+	Logger *slog.Logger
26	29		}
27	30
28	31		type CachedBucket struct {

...	...		@@ -38,12 +41,9 @@ type CachedObjectInfo struct {
38	41		var (
39	42		_ ObjectStorage = &StorageMinio{}
40	43		_ ObjectStorage = (*StorageMinio)(nil)
41		-
42		-	bucketCache = expirable.NewLRU[string, CachedBucket](2048, nil, cache.CacheTimeout)
43		-	objectInfoCache = expirable.NewLRU[string, CachedObjectInfo](2048, nil, cache.CacheTimeout)
44	44		)
45	45
46		-	func NewStorageMinio(address, user, pass string) (*StorageMinio, error) {
	46	+	func NewStorageMinio(logger slog.Logger, address, user, pass string) (StorageMinio, error) {
47	47		endpoint, err := url.Parse(address)
48	48		if err != nil {
49	49		return nil, err

...	...		@@ -188,34 +192,23 @@ func (s *StorageMinio) GetObject(bucket Bucket, fpath string) (utils.ReadAndRead
188	192
189	193		cacheKey := filepath.Join(bucket.Name, fpath)
190	194
191		-	cachedInfo, found := objectInfoCache.Get(cacheKey)
192		-	if found {
193		-	objInfo = cachedInfo.ObjectInfo
194		-
195		-	if cachedInfo.Error != nil {
196		-	return nil, objInfo, cachedInfo.Error
197		-	}
198		-	} else {
199		-	info, err := s.Client.StatObject(context.Background(), bucket.Name, fpath, minio.StatObjectOptions{})
200		-	if err != nil {
201		-	objectInfoCache.Add(cacheKey, CachedObjectInfo{objInfo, err})
202		-	return nil, objInfo, err
203		-	}
	195	+	s.Logger.Info("object info not found in lru cache", "key", cacheKey)
	196	+	info, err := s.Client.StatObject(context.Background(), bucket.Name, fpath, minio.StatObjectOptions{})
	197	+	if err != nil {
	198	+	return nil, objInfo, err
	199	+	}
204	200
205		-	objInfo.LastModified = info.LastModified
206		-	objInfo.ETag = info.ETag
207		-	objInfo.Metadata = info.Metadata
208		-	objInfo.UserMetadata = info.UserMetadata
209		-	objInfo.Size = info.Size
	201	+	objInfo.LastModified = info.LastModified
	202	+	objInfo.ETag = info.ETag
	203	+	objInfo.Metadata = info.Metadata
	204	+	objInfo.UserMetadata = info.UserMetadata
	205	+	objInfo.Size = info.Size
210	206
211		-	if mtime, ok := info.UserMetadata["Mtime"]; ok {
212		-	mtimeUnix, err := strconv.Atoi(mtime)
213		-	if err == nil {
214		-	objInfo.LastModified = time.Unix(int64(mtimeUnix), 0)
215		-	}
	207	+	if mtime, ok := info.UserMetadata["Mtime"]; ok {
	208	+	mtimeUnix, err := strconv.Atoi(mtime)
	209	+	if err == nil {
	210	+	objInfo.LastModified = time.Unix(int64(mtimeUnix), 0)
216	211		}
217		-
218		-	objectInfoCache.Add(cacheKey, CachedObjectInfo{objInfo, nil})
219	212		}
220	213
221	214		obj, err := s.Client.GetObject(context.Background(), bucket.Name, fpath, minio.GetObjectOptions{})

...	...		@@ -29,7 +29,7 @@ func EnvDriverDetector(logger *slog.Logger) (storage.ObjectStorage, error) {
29	29		"url", url,
30	30		"user", user,
31	31		)
32		-	return storage.NewStorageMinio(url, user, pass)
	32	+	return storage.NewStorageMinio(logger, url, user, pass)
33	33		}
34	34
35	35		// implied driver == "fs"

...	...		@@ -13,15 +13,14 @@ import (
13	13
14	14		type StorageMinio struct {
15	15		*sst.StorageMinio
16		-	Logger *slog.Logger
17	16		}
18	17
19	18		func NewStorageMinio(logger slog.Logger, address, user, pass string) (StorageMinio, error) {
20		-	st, err := sst.NewStorageMinio(address, user, pass)
	19	+	st, err := sst.NewStorageMinio(logger, address, user, pass)
21	20		if err != nil {
22	21		return nil, err
23	22		}
24		-	return &StorageMinio{st, logger}, nil
	23	+	return &StorageMinio{st}, nil
25	24		}
26	25
27	26		func (s StorageMinio) ServeObject(bucket sst.Bucket, fpath string, opts ImgProcessOpts) (io.ReadCloser, *sst.ObjectInfo, error) {

pico

Patchset 59.4 on 2025-04-06T19:41:38Z · commit ee12290

Patchset `59.4` on 2025-04-06T19:41:38Z · commit ee12290