mirrors/remark42
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'feature/partial-flush'

Browse Source
This commit is contained in:
Umputun
2018-05-25 18:10:02 -05:00
parent e8838f475e 2fa8153d19
commit a79063fa9e
16 changed files with 456 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.travis.yml
+1
View File
@@ -5,6 +5,7 @@ install:
script:
- docker build
--build-arg COVERALLS_TOKEN=$COVERALLS_TOKEN
--build-arg CODECOV_TOKEN=$CODECOV_TOKEN
--build-arg CI=$CI
--build-arg TRAVIS=$TRAVIS
--build-arg TRAVIS_BRANCH=$TRAVIS_BRANCH
Dockerfile
+7 -1
View File
@@ -1,6 +1,7 @@
FROM umputun/baseimage:buildgo-latest as build-backend
ARG COVERALLS_TOKEN
ARG CODECOV_TOKEN
ARG CI
ARG TRAVIS
ARG TRAVIS_BRANCH
@@ -27,10 +28,15 @@ RUN gometalinter --disable-all --deadline=300s --vendor --enable=vet --enable=ve
RUN mkdir -p target && /script/coverage.sh
RUN if [ "x$COVERALLS_TOKEN" = "x" ] ; then \
RUN if [ -z "$COVERALLS_TOKEN" ] ; then \
echo coverall not enabled ; \
else goveralls -coverprofile=.cover/cover.out -service=travis-ci -repotoken $COVERALLS_TOKEN; fi
RUN if [ -z "$CODECOV_TOKEN" ] ; then \
echo codecov not enabled ; \
else curl -s https://codecov.io/bash -o codecov && \
bash codecov -f .cover/cover.out -X fix; fi
RUN go build -o remark -ldflags "-X main.revision=$(git rev-parse --abbrev-ref HEAD)-$(git describe --abbrev=7 --always --tags)-$(date +%Y%m%d-%H:%M:%S) -s -w" ./app
app/main.go
+3 -3
View File
@@ -19,9 +19,9 @@ import (
"github.com/umputun/remark/app/store/service"
"github.com/umputun/remark/app/migrator"
"github.com/umputun/remark/app/rest"
"github.com/umputun/remark/app/rest/api"
"github.com/umputun/remark/app/rest/auth"
"github.com/umputun/remark/app/rest/cache"
"github.com/umputun/remark/app/rest/proxy"
)
@@ -114,8 +114,8 @@ func New(opts Opts) (*Application, error) {
MaxCommentSize: opts.MaxCommentSize,
}
cache := rest.NewLoadingCache(rest.MaxValSize(opts.MaxCachedValue), rest.MaxKeys(opts.MaxCachedItems),
rest.PostFlushFn(postFlushFn(opts.Sites, opts.Port)))
cache := cache.NewLoadingCache(cache.MaxValSize(opts.MaxCachedValue), cache.MaxKeys(opts.MaxCachedItems),
cache.PostFlushFn(postFlushFn(opts.Sites, opts.Port)))
jwtService := auth.NewJWT(opts.SecretKey, strings.HasPrefix(opts.RemarkURL, "https://"), 7*24*time.Hour)
app/rest/api/admin.go
+12 -5
View File
@@ -13,6 +13,7 @@ import (
"github.com/umputun/remark/app/migrator"
"github.com/umputun/remark/app/rest"
"github.com/umputun/remark/app/rest/cache"
"github.com/umputun/remark/app/store"
"github.com/umputun/remark/app/store/service"
)
@@ -21,7 +22,7 @@ import (
type admin struct {
dataService service.DataStore
exporter migrator.Exporter
cache rest.LoadingCache
cache cache.LoadingCache
defAvatarURL string
}
@@ -49,7 +50,7 @@ func (a *admin) deleteCommentCtrl(w http.ResponseWriter, r *http.Request) {
rest.SendErrorJSON(w, r, http.StatusInternalServerError, err, "can't delete comment")
return
}
a.cache.Flush()
a.cache.Flush(locator.SiteID, locator.URL)
render.Status(r, http.StatusOK)
render.JSON(w, r, JSON{"id": id, "locator": locator})
}
@@ -64,7 +65,7 @@ func (a *admin) setBlockCtrl(w http.ResponseWriter, r *http.Request) {
rest.SendErrorJSON(w, r, http.StatusBadRequest, err, "can't set blocking status")
return
}
a.cache.Flush()
a.cache.Flush(siteID, userID)
render.JSON(w, r, JSON{"user_id": userID, "site_id": siteID, "block": blockStatus})
}
@@ -90,7 +91,7 @@ func (a *admin) setPinCtrl(w http.ResponseWriter, r *http.Request) {
rest.SendErrorJSON(w, r, http.StatusBadRequest, err, "can't set pin status")
return
}
a.cache.Flush()
a.cache.Flush(locator.URL)
render.JSON(w, r, JSON{"id": commentID, "locator": locator, "pin": pinStatus})
}
@@ -104,7 +105,13 @@ func (a *admin) exportCtrl(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/gzip")
w.Header().Set("Content-Disposition", "attachment;filename="+exportFile)
w.WriteHeader(http.StatusOK)
writer = gzip.NewWriter(w)
gzWriter := gzip.NewWriter(w)
defer func() {
if e := gzWriter.Close(); e != nil {
log.Printf("[WARN] can't close gzip writer, %s", e)
}
}()
writer = gzWriter
}
if _, err := a.exporter.Export(writer, siteID); err != nil {
app/rest/api/admin_test.go
+35 -1
View File
@@ -1,12 +1,14 @@
package api
import (
"compress/gzip"
"encoding/json"
"fmt"
"io/ioutil"
"net/http"
"strings"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -175,7 +177,7 @@ func TestAdmin_BlockedList(t *testing.T) {
assert.Equal(t, "user2", users[1].ID)
}
func TestAdmin_Export(t *testing.T) {
func TestAdmin_ExportStream(t *testing.T) {
srv, ts := prep(t)
assert.NotNil(t, srv)
defer cleanup(ts)
@@ -194,3 +196,35 @@ func TestAdmin_Export(t *testing.T) {
assert.Equal(t, 2, strings.Count(body, "\"text\""))
t.Logf("%s", body)
}
func TestAdmin_ExportFile(t *testing.T) {
srv, ts := prep(t)
assert.NotNil(t, srv)
defer cleanup(ts)
c1 := store.Comment{Text: "test test #1",
Locator: store.Locator{SiteID: "radio-t", URL: "https://radio-t.com/blah1"}}
c2 := store.Comment{Text: "test test #2", ParentID: "p1",
Locator: store.Locator{SiteID: "radio-t", URL: "https://radio-t.com/blah2"}}
addComment(t, c1, ts)
addComment(t, c2, ts)
client := &http.Client{Timeout: 5 * time.Second}
req, err := http.NewRequest("GET", ts.URL+"/api/v1/admin/export?site=radio-t&mode=file", nil)
require.Nil(t, err)
withBasicAuth(req, "dev", "password")
resp, err := client.Do(req)
require.Nil(t, err)
assert.Equal(t, 200, resp.StatusCode)
assert.Equal(t, "application/gzip", resp.Header.Get("Content-Type"))
ungzReader, err := gzip.NewReader(resp.Body)
assert.NoError(t, err)
ungzBody, err := ioutil.ReadAll(ungzReader)
assert.NoError(t, err)
assert.Equal(t, 2, strings.Count(string(ungzBody), "\n"))
assert.Equal(t, 2, strings.Count(string(ungzBody), "\"text\""))
t.Logf("%s", string(ungzBody))
}
app/rest/api/import.go
+3 -2
View File
@@ -17,12 +17,13 @@ import (
"github.com/umputun/remark/app/migrator"
"github.com/umputun/remark/app/rest"
"github.com/umputun/remark/app/rest/cache"
)
// Import rest runs on unexposed port and available for local requests only
type Import struct {
Version string
Cache rest.LoadingCache
Cache cache.LoadingCache
NativeImporter migrator.Importer
DisqusImporter migrator.Importer
SecretKey string
@@ -92,7 +93,7 @@ func (s *Import) importCtrl(w http.ResponseWriter, r *http.Request) {
rest.SendErrorJSON(w, r, http.StatusBadRequest, err, "import failed")
return
}
s.Cache.Flush()
s.Cache.Flush(siteID)
render.Status(r, http.StatusCreated)
render.JSON(w, r, JSON{"status": "ok", "size": size})
app/rest/api/rest.go
+15 -13
View File
@@ -25,6 +25,7 @@ import (
"github.com/umputun/remark/app/migrator"
"github.com/umputun/remark/app/rest"
"github.com/umputun/remark/app/rest/auth"
"github.com/umputun/remark/app/rest/cache"
"github.com/umputun/remark/app/rest/proxy"
"github.com/umputun/remark/app/store"
"github.com/umputun/remark/app/store/service"
@@ -36,7 +37,7 @@ type Rest struct {
DataService service.DataStore
Authenticator auth.Authenticator
Exporter migrator.Exporter
Cache rest.LoadingCache
Cache cache.LoadingCache
AvatarProxy *proxy.Avatar
ImageProxy *proxy.Image
WebRoot string
@@ -212,7 +213,8 @@ func (s *Rest) createCommentCtrl(w http.ResponseWriter, r *http.Request) {
rest.SendErrorJSON(w, r, http.StatusInternalServerError, err, "can't load created comment")
return
}
s.Cache.Flush() // reset all caches
s.Cache.Flush(comment.Locator.URL, "last", comment.User.ID)
render.Status(r, http.StatusCreated)
render.JSON(w, r, &finalComment)
}
@@ -293,7 +295,7 @@ func (s *Rest) updateCommentCtrl(w http.ResponseWriter, r *http.Request) {
return
}
s.Cache.Flush() // reset all caches
s.Cache.Flush(locator.URL, "last", user.ID)
render.JSON(w, r, res)
}
@@ -307,7 +309,7 @@ func (s *Rest) findCommentsCtrl(w http.ResponseWriter, r *http.Request) {
}
log.Printf("[DEBUG] get comments for %+v, sort %s, format %s", locator, sort, r.URL.Query().Get("format"))
data, err := s.Cache.Get(rest.URLKey(r), 4*time.Hour, func() ([]byte, error) {
data, err := s.Cache.Get(cache.Key(cache.URLKey(r), locator.SiteID, locator.URL), 4*time.Hour, func() ([]byte, error) {
comments, e := s.DataService.Find(locator, sort)
if e != nil {
return nil, e
@@ -332,16 +334,16 @@ func (s *Rest) findCommentsCtrl(w http.ResponseWriter, r *http.Request) {
// GET /last/{limit}?site=siteID - last comments for the siteID, across all posts, sorted by time
func (s *Rest) lastCommentsCtrl(w http.ResponseWriter, r *http.Request) {
log.Printf("[DEBUG] get last comments for %s", r.URL.Query().Get("site"))
siteID := r.URL.Query().Get("site")
log.Printf("[DEBUG] get last comments for %s", siteID)
limit, err := strconv.Atoi(chi.URLParam(r, "limit"))
if err != nil {
limit = 0
}
data, err := s.Cache.Get(rest.URLKey(r), 4*time.Hour, func() ([]byte, error) {
comments, e := s.DataService.Last(r.URL.Query().Get("site"), limit)
data, err := s.Cache.Get(cache.Key(cache.URLKey(r), "last", siteID), 4*time.Hour, func() ([]byte, error) {
comments, e := s.DataService.Last(siteID, limit)
if e != nil {
return nil, e
}
@@ -403,7 +405,7 @@ func (s *Rest) findUserCommentsCtrl(w http.ResponseWriter, r *http.Request) {
log.Printf("[DEBUG] get comments for userID %s, %s", userID, siteID)
data, err := s.Cache.Get(rest.URLKey(r), 4*time.Hour, func() ([]byte, error) {
data, err := s.Cache.Get(cache.Key(cache.URLKey(r), userID, siteID), 4*time.Hour, func() ([]byte, error) {
comments, count, e := s.DataService.User(siteID, userID, limit)
if e != nil {
return nil, e
@@ -484,7 +486,7 @@ func (s *Rest) countMultiCtrl(w http.ResponseWriter, r *http.Request) {
}
// key could be long for multiple posts, make it sha1
key := rest.URLKey(r) + strings.Join(posts, ",")
key := cache.URLKey(r) + strings.Join(posts, ",")
hasher := sha1.New()
if _, err := hasher.Write([]byte(key)); err != nil {
rest.SendErrorJSON(w, r, http.StatusInternalServerError, err, "can't make sha1 for list of urls")
@@ -492,7 +494,7 @@ func (s *Rest) countMultiCtrl(w http.ResponseWriter, r *http.Request) {
}
sha := base64.URLEncoding.EncodeToString(hasher.Sum(nil))
data, err := s.Cache.Get(sha, 8*time.Hour, func() ([]byte, error) {
data, err := s.Cache.Get(cache.Key(sha, siteID), 8*time.Hour, func() ([]byte, error) {
counts, e := s.DataService.Counts(siteID, posts)
if e != nil {
return nil, e
@@ -520,7 +522,7 @@ func (s *Rest) listCtrl(w http.ResponseWriter, r *http.Request) {
skip = v
}
data, err := s.Cache.Get(rest.URLKey(r), 8*time.Hour, func() ([]byte, error) {
data, err := s.Cache.Get(cache.Key(cache.URLKey(r), siteID), 8*time.Hour, func() ([]byte, error) {
posts, e := s.DataService.List(siteID, limit, skip)
if e != nil {
return nil, e
@@ -554,7 +556,7 @@ func (s *Rest) voteCtrl(w http.ResponseWriter, r *http.Request) {
rest.SendErrorJSON(w, r, http.StatusBadRequest, err, "can't vote for comment")
return
}
s.Cache.Flush()
s.Cache.Flush(locator.URL)
render.JSON(w, r, JSON{"id": comment.ID, "score": comment.Score})
}
app/rest/api/rest_test.go
+16 -5
View File
@@ -18,6 +18,7 @@ import (
"github.com/stretchr/testify/require"
"github.com/umputun/remark/app/migrator"
"github.com/umputun/remark/app/rest"
"github.com/umputun/remark/app/rest/auth"
"github.com/umputun/remark/app/rest/proxy"
"github.com/umputun/remark/app/store"
@@ -160,13 +161,14 @@ func TestServer_Find(t *testing.T) {
_, code := get(t, ts.URL+"/api/v1/find?site=radio-t&url=https://radio-t.com/blah1")
assert.Equal(t, 400, code, "nothing in")
c1 := store.Comment{Text: "test test #1", ParentID: "p1",
c1 := store.Comment{Text: "test test #1", ParentID: "",
Locator: store.Locator{SiteID: "radio-t", URL: "https://radio-t.com/blah1"}}
c2 := store.Comment{Text: "test test #2", ParentID: "p1",
Locator: store.Locator{SiteID: "radio-t", URL: "https://radio-t.com/blah1"}}
id1 := addComment(t, c1, ts)
c2 := store.Comment{Text: "test test #2", ParentID: id1,
Locator: store.Locator{SiteID: "radio-t", URL: "https://radio-t.com/blah1"}}
id2 := addComment(t, c2, ts)
assert.NotEqual(t, id1, id2)
// get sorted by +time
@@ -187,6 +189,15 @@ func TestServer_Find(t *testing.T) {
assert.Equal(t, 2, len(comments), "should have 2 comments")
assert.Equal(t, id1, comments[1].ID)
assert.Equal(t, id2, comments[0].ID)
// get in tree mode
tree := rest.Tree{}
res, code = get(t, ts.URL+"/api/v1/find?site=radio-t&url=https://radio-t.com/blah1&format=tree")
assert.Equal(t, 200, code)
err = json.Unmarshal([]byte(res), &tree)
assert.Nil(t, err)
assert.Equal(t, 1, len(tree.Nodes))
assert.Equal(t, 1, len(tree.Nodes[0].Replies))
}
func TestServer_Update(t *testing.T) {
@@ -567,4 +578,4 @@ func (mc *mockCache) Get(key string, ttl time.Duration, fn func() ([]byte, error
return fn()
}
func (mc *mockCache) Flush() {}
func (mc *mockCache) Flush(scopes ...string) {}
app/rest/api/rss.go
+10 -12
View File
@@ -7,10 +7,10 @@ import (
"time"
"github.com/go-chi/chi"
"github.com/go-chi/render"
"github.com/gorilla/feeds"
"github.com/umputun/remark/app/rest"
"github.com/umputun/remark/app/rest/cache"
"github.com/umputun/remark/app/store"
)
@@ -32,7 +32,7 @@ func (s *Rest) rssPostCommentsCtrl(w http.ResponseWriter, r *http.Request) {
sort := "-time"
log.Printf("[DEBUG] get rss for post %+v", locator)
data, err := s.Cache.Get(rest.URLKey(r), 4*time.Hour, func() ([]byte, error) {
data, err := s.Cache.Get(cache.Key(cache.URLKey(r), locator.SiteID, locator.URL), 4*time.Hour, func() ([]byte, error) {
comments, e := s.DataService.Find(locator, sort)
if e != nil {
return nil, e
@@ -51,9 +51,8 @@ func (s *Rest) rssPostCommentsCtrl(w http.ResponseWriter, r *http.Request) {
}
w.Header().Set("Content-Type", "application/xml; charset=utf-8")
if status, ok := r.Context().Value(render.StatusCtxKey).(int); ok {
w.WriteHeader(status)
}
w.WriteHeader(http.StatusOK)
if _, err := w.Write(data); err != nil {
log.Printf("[WARN] failed to send response to %s, %s", r.RemoteAddr, err)
}
@@ -61,10 +60,11 @@ func (s *Rest) rssPostCommentsCtrl(w http.ResponseWriter, r *http.Request) {
// GET /rss/site?site=siteID
func (s *Rest) rssSiteCommentsCtrl(w http.ResponseWriter, r *http.Request) {
log.Printf("[DEBUG] get rss for site %s", r.URL.Query().Get("site"))
siteID := r.URL.Query().Get("site")
log.Printf("[DEBUG] get rss for site %s", siteID)
data, err := s.Cache.Get(rest.URLKey(r), 4*time.Hour, func() ([]byte, error) {
comments, e := s.DataService.Last(r.URL.Query().Get("site"), maxRssItems)
data, err := s.Cache.Get(cache.Key(cache.URLKey(r), siteID), 4*time.Hour, func() ([]byte, error) {
comments, e := s.DataService.Last(siteID, maxRssItems)
if e != nil {
return nil, e
}
@@ -78,14 +78,12 @@ func (s *Rest) rssSiteCommentsCtrl(w http.ResponseWriter, r *http.Request) {
})
if err != nil {
rest.SendErrorJSON(w, r, http.StatusInternalServerError, err, "can't get last comments")
rest.SendErrorJSON(w, r, http.StatusBadRequest, err, "can't get last comments")
return
}
w.Header().Set("Content-Type", "application/xml; charset=utf-8")
if status, ok := r.Context().Value(render.StatusCtxKey).(int); ok {
w.WriteHeader(status)
}
w.WriteHeader(http.StatusOK)
if _, err := w.Write(data); err != nil {
log.Printf("[WARN] failed to send response to %s, %s", r.RemoteAddr, err)
}
app/rest/api/rss_test.go
+6
View File
@@ -47,6 +47,9 @@ func TestServer_RssPost(t *testing.T) {
expected, res = cleanRssFormatting(expected, res)
assert.Equal(t, expected, res)
res, code = get(t, ts.URL+"/api/v1/rss/post?site=radio-t-bad&url=https://radio-t.com/blah1")
assert.Equal(t, 400, code)
}
func TestServer_RssSite(t *testing.T) {
@@ -98,6 +101,9 @@ func TestServer_RssSite(t *testing.T) {
expected, res = cleanRssFormatting(expected, res)
assert.Equal(t, expected, res)
_, code = get(t, ts.URL+"/api/v1/rss/site?site=bad-radio-t")
assert.Equal(t, 400, code)
}
func TestServer_RssWithReply(t *testing.T) {
app/rest/cache.go
-128
View File
@@ -1,128 +0,0 @@
package rest
import (
"log"
"net/http"
"strings"
"time"
"github.com/patrickmn/go-cache"
)
// LoadingCache defines interface for caching
type LoadingCache interface {
Get(key string, ttl time.Duration, fn func() ([]byte, error)) (data []byte, err error)
Flush()
}
// loadingCache implements LoadingCache interface on top of cache.Cache (go-cache)
type loadingCache struct {
bytesCache *cache.Cache
postFlushFn func()
defaultExpiration time.Duration
cleanupInterval time.Duration
maxKeys int
maxValueSize int
}
// NewLoadingCache makes loadingCache implementation
func NewLoadingCache(options ...CacheOption) LoadingCache {
res := loadingCache{
defaultExpiration: time.Hour,
cleanupInterval: 5 * time.Minute,
postFlushFn: func() {},
maxKeys: 0,
maxValueSize: 0,
}
for _, opt := range options {
if err := opt(&res); err != nil {
log.Printf("[WARN] failed to set cache option, %v", err)
}
}
res.bytesCache = cache.New(res.defaultExpiration, res.cleanupInterval)
log.Printf("[DEBUG] create cache with cleanupInterval=%s, maxKeys=%d, maxValueSize=%d",
res.cleanupInterval, res.maxKeys, res.maxValueSize)
return &res
}
// Get is loading cache method to get value by key or load via fn if not found
func (lc *loadingCache) Get(key string, ttl time.Duration, fn func() ([]byte, error)) (data []byte, err error) {
if b, ok := lc.bytesCache.Get(key); ok {
return b.([]byte), nil
}
if data, err = fn(); err != nil {
return data, err
}
if lc.allowed(data) {
lc.bytesCache.Set(key, data, ttl)
}
return data, nil
}
// Flush clears cache and calls postFlushFn async
func (lc *loadingCache) Flush() {
lc.bytesCache.Flush()
if lc.postFlushFn != nil {
go lc.postFlushFn()
}
}
func (lc *loadingCache) allowed(data []byte) bool {
if lc.maxValueSize > 0 && len(data) >= lc.maxValueSize {
return false
}
if lc.maxKeys > 0 && lc.bytesCache.ItemCount() >= lc.maxKeys {
return false
}
return true
}
// CacheOption func type
type CacheOption func(lc *loadingCache) error
// MaxValSize functional option defines the largest value's size allowed to be cached
// By default it is 0, which means unlimited.
func MaxValSize(max int) CacheOption {
return func(lc *loadingCache) error {
lc.maxValueSize = max
return nil
}
}
// MaxKeys functional option defines how many keys to keep.
// By default it is 0, which means unlimited.
func MaxKeys(max int) CacheOption {
return func(lc *loadingCache) error {
lc.maxKeys = max
return nil
}
}
// CleanupInterval functional option defines how often cleanup loop activated.
func CleanupInterval(interval time.Duration) CacheOption {
return func(lc *loadingCache) error {
lc.cleanupInterval = interval
return nil
}
}
// PostFlushFn functional option defines how callback function called after each Flush.
func PostFlushFn(postFlushFn func()) CacheOption {
return func(lc *loadingCache) error {
lc.postFlushFn = postFlushFn
return nil
}
}
// URLKey gets url from request to use it as cache key
// admins will have different keys in order to prevent leak of admin-only data to regular users
func URLKey(r *http.Request) string {
adminPrefix := "admin!!"
key := strings.TrimPrefix(r.URL.String(), adminPrefix) // prevents attach with fake url to get admin view
if user, err := GetUserInfo(r); err == nil && user.Admin { // make separate cache key for admins
key = adminPrefix + key
}
return key
}
app/rest/cache/cache.go
Vendored
+153
View File
@@ -0,0 +1,153 @@
package cache
import (
"log"
"strings"
"sync"
"time"
"github.com/patrickmn/go-cache"
"github.com/pkg/errors"
)
// LoadingCache defines interface for caching
type LoadingCache interface {
Get(key string, ttl time.Duration, fn func() ([]byte, error)) (data []byte, err error)
Flush(scopes ...string)
}
// Key makes full key from primary key and scopes
func Key(key string, scopes ...string) string {
return strings.Join(scopes, "$$") + "@@" + key
}
func parseKey(fullKey string) (key string, scopes []string, err error) {
elems := strings.Split(fullKey, "@@")
if len(elems) != 2 {
return "", nil, errors.Errorf("can't parse cache key %s", key)
}
scopes = strings.Split(elems[0], "$$")
if len(scopes) == 1 && scopes[0] == "" {
scopes = []string{}
}
key = elems[1]
return key, scopes, nil
}
// loadingCache implements LoadingCache interface on top of cache.Cache (go-cache)
type loadingCache struct {
bytesCache *cache.Cache
postFlushFn func()
defaultExpiration time.Duration
cleanupInterval time.Duration
maxKeys int
maxValueSize int
activeKeys map[string]struct{} // keep all current cached keys
lock sync.Mutex
}
// NewLoadingCache makes loadingCache implementation
func NewLoadingCache(options ...Option) LoadingCache {
res := loadingCache{
defaultExpiration: time.Hour,
cleanupInterval: 5 * time.Minute,
postFlushFn: func() {},
maxKeys: 0,
maxValueSize: 0,
activeKeys: map[string]struct{}{},
}
for _, opt := range options {
if err := opt(&res); err != nil {
log.Printf("[WARN] failed to set cache option, %v", err)
}
}
res.bytesCache = cache.New(res.defaultExpiration, res.cleanupInterval)
// OnEvicted called automatically for expired and manually deleted
res.bytesCache.OnEvicted(func(key string, _ interface{}) {
res.withLock(func() { delete(res.activeKeys, key) })
})
log.Printf("[DEBUG] create cache with cleanupInterval=%s, maxKeys=%d, maxValueSize=%d",
res.cleanupInterval, res.maxKeys, res.maxValueSize)
return &res
}
// Get is loading cache method to get value by key or load via fn if not found
func (lc *loadingCache) Get(key string, ttl time.Duration, fn func() ([]byte, error)) (data []byte, err error) {
if b, ok := lc.bytesCache.Get(key); ok {
return b.([]byte), nil
}
if data, err = fn(); err != nil {
return data, err
}
if lc.allowed(data) {
lc.bytesCache.Set(key, data, ttl)
lc.withLock(func() { lc.activeKeys[key] = struct{}{} })
}
return data, nil
}
func (lc *loadingCache) withLock(fn func()) {
lc.lock.Lock()
fn()
lc.lock.Unlock()
}
// Flush clears cache and calls postFlushFn async
func (lc *loadingCache) Flush(scopes ...string) {
if len(scopes) == 0 {
lc.bytesCache.Flush()
lc.withLock(func() { lc.activeKeys = map[string]struct{}{} })
go lc.postFlushFn()
return
}
// check if fullKey has matching scopes
inScope := func(fullKey string) bool {
for _, s := range scopes {
_, keyScopes, err := parseKey(fullKey)
if err != nil {
return false
}
for _, ks := range keyScopes {
if ks == s {
return true
}
}
}
return false
}
// all matchedKeys should be collected first
// we can't delete it from locked section, it will lock on eviction callback
matchedKeys := []string{}
lc.withLock(func() {
for k := range lc.activeKeys {
if inScope(k) {
matchedKeys = append(matchedKeys, k)
}
}
})
for _, mkey := range matchedKeys {
lc.bytesCache.Delete(mkey)
}
if lc.postFlushFn != nil {
go lc.postFlushFn()
}
}
func (lc *loadingCache) allowed(data []byte) bool {
if lc.maxValueSize > 0 && len(data) >= lc.maxValueSize {
return false
}
if lc.maxKeys > 0 && lc.bytesCache.ItemCount() >= lc.maxKeys {
return false
}
return true
}
app/rest/cache_test.go → app/rest/cache/cache_test.go
Vendored
+107 -2
View File
@@ -1,4 +1,4 @@
package rest
package cache
import (
"fmt"
@@ -10,6 +10,7 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/umputun/remark/app/rest"
"github.com/umputun/remark/app/store"
)
@@ -139,7 +140,7 @@ func TestLoadingCache_URLKey(t *testing.T) {
assert.Equal(t, "http://blah/123?key=v&k2=v2", key)
user := store.User{Admin: true}
r = SetUserInfo(r, user)
r = rest.SetUserInfo(r, user)
key = URLKey(r)
assert.Equal(t, "admin!!http://blah/123?key=v&k2=v2", key)
}
@@ -171,3 +172,107 @@ func TestLoadingCache_Parallel(t *testing.T) {
wg.Wait()
assert.Equal(t, int32(0), atomic.LoadInt32(&coldCalls))
}
func TestLoadingCache_Scopes(t *testing.T) {
lc := NewLoadingCache(CleanupInterval(time.Second))
res, err := lc.Get(Key("key", "s1", "s2"), time.Minute, func() ([]byte, error) {
return []byte("value"), nil
})
assert.Nil(t, err)
assert.Equal(t, "value", string(res))
res, err = lc.Get(Key("key2", "s2"), time.Minute, func() ([]byte, error) {
return []byte("value2"), nil
})
assert.Nil(t, err)
assert.Equal(t, "value2", string(res))
assert.Equal(t, 2, len(lc.(*loadingCache).activeKeys))
lc.Flush("s1")
assert.Equal(t, 1, len(lc.(*loadingCache).activeKeys))
lc.Get(Key("key2", "s2"), time.Minute, func() ([]byte, error) {
assert.Fail(t, "should stay")
return nil, nil
})
res, err = lc.Get(Key("key", "s1", "s2"), time.Minute, func() ([]byte, error) {
return []byte("value-upd"), nil
})
assert.Equal(t, "value-upd", string(res), "was deleted, update")
}
func TestLoadingCache_Flush(t *testing.T) {
lc := NewLoadingCache(CleanupInterval(time.Second))
addToCache := func(key string, scopes ...string) {
res, err := lc.Get(key, time.Minute, func() ([]byte, error) {
return []byte("value" + key), nil
})
require.Nil(t, err)
require.Equal(t, "value"+key, string(res))
}
init := func() {
lc.Flush()
addToCache(Key("key1", "s1", "s2"))
addToCache(Key("key2", "s1", "s2", "s3"))
addToCache(Key("key3", "s1", "s2", "s3"))
addToCache(Key("key4", "s2", "s3"))
addToCache(Key("key5", "s2"))
addToCache(Key("key6"))
addToCache(Key("key7", "s4", "s3"))
require.Equal(t, 7, len(lc.(*loadingCache).activeKeys), "cache init")
}
tbl := []struct {
scopes []string
left int
msg string
}{
{[]string{}, 0, "full flush, no scopes"},
{[]string{"s0"}, 7, "flush wrong scope"},
{[]string{"s1"}, 4, "flush s1 scope"},
{[]string{"s2", "s1"}, 2, "flush s2+s1 scope"},
{[]string{"s1", "s2"}, 2, "flush s1+s2 scope"},
{[]string{"s1", "s2", "s4"}, 1, "flush s1+s2+s4 scope"},
{[]string{"s1", "s2", "s3"}, 1, "flush s1+s2+s3 scope"},
{[]string{"s1", "s2", "ss"}, 2, "flush s1+s2+wrong scope"},
}
for i, tt := range tbl {
init()
lc.Flush(tt.scopes...)
assert.Equal(t, tt.left, len(lc.(*loadingCache).activeKeys), "keys size, %s #%d", tt.msg, i)
assert.Equal(t, tt.left, len(lc.(*loadingCache).bytesCache.Items()), "items size, %s #%d", tt.msg, i)
}
}
func TestLoadingCache_Keys(t *testing.T) {
tbl := []struct {
key string
scopes []string
full string
}{
{"key1", []string{"s1"}, "s1@@key1"},
{"key2", []string{"s11", "s2"}, "s11$$s2@@key2"},
{"key3", []string{}, "@@key3"},
}
for n, tt := range tbl {
full := Key(tt.key, tt.scopes...)
assert.Equal(t, tt.full, full, "making key, #%d", n)
k, s, e := parseKey(full)
assert.Nil(t, e)
assert.Equal(t, tt.scopes, s)
assert.Equal(t, tt.key, k)
}
_, _, err := parseKey("abc")
assert.Error(t, err)
_, _, err = parseKey("")
assert.Error(t, err)
}
app/rest/cache/options.go
Vendored
+57
View File
@@ -0,0 +1,57 @@
package cache
import (
"net/http"
"strings"
"time"
"github.com/umputun/remark/app/rest"
)
// Option func type
type Option func(lc *loadingCache) error
// MaxValSize functional option defines the largest value's size allowed to be cached
// By default it is 0, which means unlimited.
func MaxValSize(max int) Option {
return func(lc *loadingCache) error {
lc.maxValueSize = max
return nil
}
}
// MaxKeys functional option defines how many keys to keep.
// By default it is 0, which means unlimited.
func MaxKeys(max int) Option {
return func(lc *loadingCache) error {
lc.maxKeys = max
return nil
}
}
// CleanupInterval functional option defines how often cleanup loop activated.
func CleanupInterval(interval time.Duration) Option {
return func(lc *loadingCache) error {
lc.cleanupInterval = interval
return nil
}
}
// PostFlushFn functional option defines how callback function called after each Flush.
func PostFlushFn(postFlushFn func()) Option {
return func(lc *loadingCache) error {
lc.postFlushFn = postFlushFn
return nil
}
}
// URLKey gets url from request to use it as cache key
// admins will have different keys in order to prevent leak of admin-only data to regular users
func URLKey(r *http.Request) string {
adminPrefix := "admin!!"
key := strings.TrimPrefix(r.URL.String(), adminPrefix) // prevents attach with fake url to get admin view
if user, err := rest.GetUserInfo(r); err == nil && user.Admin { // make separate cache key for admins
key = adminPrefix + key
}
return key
}
app/rest/proxy/avatar.go
+5 -6
View File
@@ -14,7 +14,6 @@ import (
"time"
"github.com/go-chi/chi"
"github.com/go-chi/render"
"github.com/pkg/errors"
"github.com/umputun/remark/app/rest"
@@ -54,6 +53,10 @@ func (p *Avatar) Put(u store.User) (avatarURL string, err error) {
}
}()
if resp.StatusCode != http.StatusOK {
return "", errors.Errorf("failed to get avatar from the orig, status %s", resp.Status)
}
// get ID and location of locally cached avatar
encID := store.EncodeID(u.ID)
location := p.location(encID) // location adds partition to path
@@ -122,11 +125,7 @@ func (p *Avatar) Routes(middlewares ...func(http.Handler) http.Handler) (string,
if fi, e := fh.Stat(); e == nil {
w.Header().Set("Content-Length", strconv.Itoa(int(fi.Size())))
}
// write all headers
if status, ok := r.Context().Value(render.StatusCtxKey).(int); ok {
w.WriteHeader(status)
}
w.WriteHeader(http.StatusOK)
if _, err = io.Copy(w, fh); err != nil {
log.Printf("[WARN] can't send response to %s, %s", r.RemoteAddr, err)
}
app/rest/proxy/avatar_test.go
+26 -5
View File
@@ -4,16 +4,18 @@ import (
"bytes"
"fmt"
"io"
"log"
"net/http"
"net/http/httptest"
"os"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/umputun/remark/app/store"
)
func TestPut(t *testing.T) {
func TestAvatar_Put(t *testing.T) {
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path == "/pic.png" {
@@ -46,18 +48,36 @@ func TestPut(t *testing.T) {
assert.Equal(t, int64(21), fi.Size())
}
func TestPutNoAvatar(t *testing.T) {
func TestAvatar_PutFailed(t *testing.T) {
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
log.Print("request: ", r.URL.Path)
w.WriteHeader(http.StatusNotFound)
}))
defer ts.Close()
p := Avatar{StorePath: "/tmp/avatars.test", RoutePath: "/avatar"}
u := store.User{ID: "user1", Name: "user1 name"}
_, err := p.Put(u)
assert.Error(t, err)
assert.EqualError(t, err, "no picture for user1")
u = store.User{ID: "user1", Name: "user1 name", Picture: "http://127.0.0.1:12345/avater/pic"}
_, err = p.Put(u)
require.Error(t, err)
assert.Contains(t, err.Error(), "connect: connection refused")
u = store.User{ID: "user1", Name: "user1 name", Picture: ts.URL + "/avatar/pic"}
_, err = p.Put(u)
require.Error(t, err)
assert.Contains(t, err.Error(), "failed to get avatar from the orig")
}
func TestRoutes(t *testing.T) {
func TestAvatar_Routes(t *testing.T) {
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path == "/pic.png" {
w.Header().Set("Content-Type", "image/*")
w.Header().Set("Custom-Header", "xyz")
fmt.Fprint(w, "some picture bin data")
return
}
@@ -87,6 +107,7 @@ func TestRoutes(t *testing.T) {
assert.Equal(t, []string{"image/*"}, rr.HeaderMap["Content-Type"])
assert.Equal(t, []string{"21"}, rr.HeaderMap["Content-Length"])
assert.Equal(t, []string(nil), rr.HeaderMap["Custom-Header"], "strip all custom headers")
assert.NotNil(t, rr.HeaderMap["Etag"])
bb := bytes.Buffer{}
@@ -96,7 +117,7 @@ func TestRoutes(t *testing.T) {
assert.Equal(t, "some picture bin data", bb.String())
}
func TestLocation(t *testing.T) {
func TestAvatar_Location(t *testing.T) {
p := Avatar{StorePath: "/tmp/avatars.test"}
tbl := []struct {