diff --git a/.travis.yml b/.travis.yml index 1c64345a..01016812 100644 --- a/.travis.yml +++ b/.travis.yml @@ -3,6 +3,9 @@ install: - docker-compose --version script: + - docker run -d --name=mongo mongo:3.6 && sleep 3 + - export MONGO_TEST=$(docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' mongo) + - echo "running mongo on $MONGO_TEST" - docker build --build-arg COVERALLS_TOKEN=$COVERALLS_TOKEN --build-arg CI=$CI @@ -16,5 +19,7 @@ script: --build-arg TRAVIS_PULL_REQUEST_SHA=$TRAVIS_PULL_REQUEST_SHA --build-arg TRAVIS_REPO_SLUG=$TRAVIS_REPO_SLUG --build-arg TRAVIS_TAG=$TRAVIS_TAG - --build-arg MONGO_TEST=$MONGO_REMARK_TEST + --build-arg MONGO_TEST=$MONGO_TEST . + - docker rm -f mongo + \ No newline at end of file diff --git a/Dockerfile b/Dockerfile index 9f039463..2f4a1a24 100644 --- a/Dockerfile +++ b/Dockerfile @@ -37,7 +37,7 @@ RUN echo "mongo=${MONGO_TEST}" >> /etc/hosts RUN if [ -z "$SKIP_BACKEND_TEST" ] ; then \ if [ -f .mongo ] ; then export MONGO_TEST=$(cat .mongo) ; fi && \ gometalinter --disable-all --deadline=300s --vendor --enable=vet --enable=vetshadow --enable=golint \ - --enable=staticcheck --enable=ineffassign --enable=goconst --enable=errcheck --enable=unconvert \ + --enable=staticcheck --enable=ineffassign --enable=errcheck --enable=unconvert \ --enable=deadcode --enable=gosimple --enable=gas --exclude=test --exclude=mock --exclude=vendor ./... ; \ else echo "skip backend linters" ; fi diff --git a/backend/Gopkg.lock b/backend/Gopkg.lock index e512a56e..6c9c2cf2 100644 --- a/backend/Gopkg.lock +++ b/backend/Gopkg.lock @@ -87,6 +87,15 @@ revision = "9a09a574c336c6ae2338a65bbebed2baab2a713c" version = "v1.0.0" +[[projects]] + name = "github.com/go-pkgz/repeater" + packages = [ + ".", + "strategy" + ] + revision = "f2a67dcf050cab24d57132a7d8b45553ceab817b" + version = "v1.0.0" + [[projects]] name = "github.com/golang/protobuf" packages = ["proto"] @@ -256,6 +265,6 @@ [solve-meta] analyzer-name = "dep" analyzer-version = 1 - inputs-digest = "e933250b2582d1ff7d2b165c4a8a934762677a2ebaaea743fb3c485868c6d793" + inputs-digest = "da9242a76135f9ee04287bf437ffa42708fc36ec54f39c74127f41a3473f24ac" solver-name = "gps-cdcl" solver-version = 1 diff --git a/backend/app/main.go b/backend/app/main.go index 343a4dba..0e1d1aa3 100644 --- a/backend/app/main.go +++ b/backend/app/main.go @@ -28,7 +28,6 @@ import ( ) // Opts with command line flags and env -// nolint:maligned type Opts struct { SecretKey string `long:"secret" env:"SECRET" required:"true" description:"secret key"` RemarkURL string `long:"url" env:"REMARK_URL" required:"true" description:"url to remark"` @@ -36,6 +35,7 @@ type Opts struct { Store StoreGroup `group:"store" namespace:"store" env-namespace:"STORE"` Avatar AvatarGroup `group:"avatar" namespace:"avatar" env-namespace:"AVATAR"` Cache CacheGroup `group:"cache" namespace:"cache" env-namespace:"CACHE"` + Mongo MongoGroup `group:"mongo" namespace:"mongo" env-namespace:"MONGO"` Sites []string `long:"site" env:"SITE" default:"remark" description:"site names" env-delim:","` Admins []string `long:"admin" env:"ADMIN" description:"admin(s) names" env-delim:","` @@ -79,7 +79,6 @@ type StoreGroup struct { Path string `long:"path" env:"PATH" default:"./var" description:"parent dir for bolt files"` Timeout time.Duration `long:"timeout" env:"TIMEOUT" default:"30s" description:"bolt timeout"` } `group:"bolt" namespace:"bolt" env-namespace:"BOLT"` - Mongo MongoOpts `group:"mongo" namespace:"mongo" env-namespace:"MONGO"` } // AvatarGroup defines options group for avatar params @@ -88,13 +87,12 @@ type AvatarGroup struct { FS struct { Path string `long:"path" env:"PATH" default:"./var/avatars" description:"avatars location"` } `group:"fs" namespace:"fs" env-namespace:"FS"` - Mongo MongoOpts `group:"mongo" namespace:"mongo" env-namespace:"MONGO"` - RszLmt int `long:"rsz-lmt" env:"RESIZE" default:"0" description:"max image size for resizing avatars on save"` + RszLmt int `long:"rsz-lmt" env:"RESIZE" default:"0" description:"max image size for resizing avatars on save"` } // CacheGroup defines options group for cache params type CacheGroup struct { - Type string `long:"type" env:"TYPE" description:"type of cache" choice:"mem" choice:"redis" default:"mem"` + Type string `long:"type" env:"TYPE" description:"type of cache" choice:"mem" choice:"mongo" default:"mem"` Max struct { Items int `long:"items" env:"ITEMS" default:"1000" description:"max cached items"` Value int `long:"value" env:"VALUE" default:"65536" description:"max size of cached value"` @@ -102,8 +100,8 @@ type CacheGroup struct { } `group:"max" namespace:"max" env-namespace:"MAX"` } -// MongoOpts holds all mongo params -type MongoOpts struct { +// MongoGroup holds all mongo params, used by store, avatar and cache +type MongoGroup struct { URL string `long:"url" env:"URL" description:"mongo url"` DB string `long:"db" env:"DB" default:"remark42" description:"mongo database"` } @@ -163,7 +161,7 @@ func New(opts Opts) (*Application, error) { return nil, errors.Errorf("invalid remark42 url %s", opts.RemarkURL) } - storeEngine, err := makeDataStore(opts.Store, opts.Sites) + storeEngine, err := makeDataStore(opts.Store, opts.Mongo, opts.Sites) if err != nil { return nil, err } @@ -176,8 +174,7 @@ func New(opts Opts) (*Application, error) { Admins: opts.Admins, } - loadingCache, err := cache.NewMemoryCache(cache.MaxCacheSize(opts.Cache.Max.Size), cache.MaxValSize(opts.Cache.Max.Value), - cache.MaxKeys(opts.Cache.Max.Items)) + loadingCache, err := makeCache(opts.Cache, opts.Mongo) if err != nil { return nil, err } @@ -186,7 +183,7 @@ func New(opts Opts) (*Application, error) { jwtService := auth.NewJWT(opts.SecretKey, strings.HasPrefix(opts.RemarkURL, "https://"), opts.Auth.TTL.JWT, opts.Auth.TTL.Cookie) - avatarStore, err := makeAvatarStore(opts.Avatar) + avatarStore, err := makeAvatarStore(opts.Avatar, opts.Mongo) if err != nil { return nil, errors.Wrap(err, "failed to make avatar store") } @@ -296,7 +293,7 @@ func (a *Application) activateBackup(ctx context.Context) { } // makeDataStore creates store for all sites -func makeDataStore(group StoreGroup, siteNames []string) (result engine.Interface, err error) { +func makeDataStore(group StoreGroup, mg MongoGroup, siteNames []string) (result engine.Interface, err error) { switch group.Type { case "bolt": if err = makeDirs(group.Bolt.Path); err != nil { @@ -308,11 +305,11 @@ func makeDataStore(group StoreGroup, siteNames []string) (result engine.Interfac } result, err = engine.NewBoltDB(bolt.Options{Timeout: group.Bolt.Timeout}, sites...) case "mongo": - mgServer, e := makeMongo(group.Mongo) + mgServer, e := makeMongo(mg) if e != nil { return result, errors.Wrap(e, "failed to create mongo server") } - conn := mongo.NewConnection(mgServer, group.Mongo.DB, "") + conn := mongo.NewConnection(mgServer, mg.DB, "") result, err = engine.NewMongo(conn, 500, 100*time.Millisecond) default: return nil, errors.Errorf("unsupported store type %s", group.Type) @@ -320,7 +317,7 @@ func makeDataStore(group StoreGroup, siteNames []string) (result engine.Interfac return result, errors.Wrap(err, "can't initialize data store") } -func makeAvatarStore(group AvatarGroup) (avatar.Store, error) { +func makeAvatarStore(group AvatarGroup, mg MongoGroup) (avatar.Store, error) { switch group.Type { case "fs": if err := makeDirs(group.FS.Path); err != nil { @@ -328,16 +325,33 @@ func makeAvatarStore(group AvatarGroup) (avatar.Store, error) { } return avatar.NewLocalFS(group.FS.Path, group.RszLmt), nil case "mongo": - mgServer, err := makeMongo(group.Mongo) + mgServer, err := makeMongo(mg) if err != nil { return nil, errors.Wrap(err, "failed to create mongo server") } - conn := mongo.NewConnection(mgServer, group.Mongo.DB, "") + conn := mongo.NewConnection(mgServer, mg.DB, "") return avatar.NewGridFS(conn, group.RszLmt), nil } return nil, errors.Errorf("unsupported avatar store type %s", group.Type) } +func makeCache(group CacheGroup, mg MongoGroup) (cache.LoadingCache, error) { + switch group.Type { + case "mem": + return cache.NewMemoryCache(cache.MaxCacheSize(group.Max.Size), cache.MaxValSize(group.Max.Value), + cache.MaxKeys(group.Max.Items)) + case "mongo": + mgServer, err := makeMongo(mg) + if err != nil { + return nil, errors.Wrap(err, "failed to create mongo server") + } + conn := mongo.NewConnection(mgServer, mg.DB, "cache") + return cache.NewMongoCache(conn, cache.MaxCacheSize(group.Max.Size), cache.MaxValSize(group.Max.Value), + cache.MaxKeys(group.Max.Items)) + } + return nil, errors.Errorf("unsupported cache type %s", group.Type) +} + // mkdir -p for all dirs func makeDirs(dirs ...string) error { @@ -367,11 +381,11 @@ func makeDirs(dirs ...string) error { return nil } -func makeMongo(mopts MongoOpts) (result *mongo.Server, err error) { - if mopts.URL == "" { +func makeMongo(mg MongoGroup) (result *mongo.Server, err error) { + if mg.URL == "" { return nil, errors.New("no mongo URL provided") } - return mongo.NewServerWithURL(mopts.URL, 10*time.Second) + return mongo.NewServerWithURL(mg.URL, 10*time.Second) } func makeAuthProviders(jwtService *auth.JWT, avatarProxy *proxy.Avatar, ds *service.DataStore, opts Opts) []auth.Provider { diff --git a/backend/app/rest/api/admin.go b/backend/app/rest/api/admin.go index 04baee07..c1fd52e8 100644 --- a/backend/app/rest/api/admin.go +++ b/backend/app/rest/api/admin.go @@ -57,7 +57,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(locator.SiteID, locator.URL) + a.cache.Flush(cache.Flusher(locator.SiteID).Scopes(locator.URL, "last")) render.Status(r, http.StatusOK) render.JSON(w, r, JSON{"id": id, "locator": locator}) } @@ -73,7 +73,7 @@ func (a *admin) deleteUserCtrl(w http.ResponseWriter, r *http.Request) { rest.SendErrorJSON(w, r, http.StatusInternalServerError, err, "can't delete user") return } - a.cache.Flush(siteID, userID) + a.cache.Flush(cache.Flusher(siteID).Scopes(userID, siteID)) render.Status(r, http.StatusOK) render.JSON(w, r, JSON{"user_id": userID, "site_id": siteID}) } @@ -118,7 +118,7 @@ func (a *admin) deleteMeRequestCtrl(w http.ResponseWriter, r *http.Request) { rest.SendErrorJSON(w, r, http.StatusBadRequest, err, "can't delete user") return } - a.cache.Flush(claims.SiteID, claims.User.ID) + a.cache.Flush(cache.Flusher(claims.SiteID).Scopes(claims.SiteID, claims.User.ID, "last")) render.Status(r, http.StatusOK) render.JSON(w, r, JSON{"user_id": claims.User.ID, "site_id": claims.SiteID}) } @@ -140,7 +140,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(siteID, userID) + a.cache.Flush(cache.Flusher(siteID).Scopes(userID, siteID)) render.JSON(w, r, JSON{"user_id": userID, "site_id": siteID, "block": blockStatus}) } @@ -177,7 +177,7 @@ func (a *admin) setReadOnlyCtrl(w http.ResponseWriter, r *http.Request) { rest.SendErrorJSON(w, r, http.StatusBadRequest, err, "can't set readonly status") return } - a.cache.Flush(locator.SiteID) + a.cache.Flush(cache.Flusher(locator.SiteID).Scopes(locator.URL, locator.SiteID)) render.JSON(w, r, JSON{"locator": locator, "read-only": roStatus}) } @@ -191,7 +191,7 @@ func (a *admin) setVerifyCtrl(w http.ResponseWriter, r *http.Request) { rest.SendErrorJSON(w, r, http.StatusBadRequest, err, "can't set verify status") return } - a.cache.Flush(siteID, userID) + a.cache.Flush(cache.Flusher(siteID).Scopes(siteID, userID)) render.JSON(w, r, JSON{"user": userID, "verified": verifyStatus}) } @@ -206,7 +206,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(locator.URL) + a.cache.Flush(cache.Flusher(locator.SiteID).Scopes(locator.URL)) render.JSON(w, r, JSON{"id": commentID, "locator": locator, "pin": pinStatus}) } diff --git a/backend/app/rest/api/migrator.go b/backend/app/rest/api/migrator.go index 8e17c9ac..c8bba0da 100644 --- a/backend/app/rest/api/migrator.go +++ b/backend/app/rest/api/migrator.go @@ -98,7 +98,7 @@ func (m *Migrator) importCtrl(w http.ResponseWriter, r *http.Request) { rest.SendErrorJSON(w, r, http.StatusBadRequest, err, "import failed") return } - m.Cache.Flush(siteID) + m.Cache.Flush(cache.Flusher(siteID).Scopes(siteID)) render.Status(r, http.StatusCreated) render.JSON(w, r, JSON{"status": "ok", "size": size}) diff --git a/backend/app/rest/api/migrator_test.go b/backend/app/rest/api/migrator_test.go index d141dae1..98a34908 100644 --- a/backend/app/rest/api/migrator_test.go +++ b/backend/app/rest/api/migrator_test.go @@ -15,6 +15,7 @@ import ( "github.com/stretchr/testify/require" "github.com/umputun/remark/backend/app/migrator" + "github.com/umputun/remark/backend/app/rest/cache" "github.com/umputun/remark/backend/app/store/engine" "github.com/umputun/remark/backend/app/store/service" ) @@ -111,7 +112,7 @@ func prepImportSrv(t *testing.T) (svc *Migrator, ts *httptest.Server) { DisqusImporter: &migrator.Disqus{DataStore: dataStore}, NativeImporter: &migrator.Remark{DataStore: dataStore}, NativeExported: &migrator.Remark{DataStore: dataStore}, - Cache: &mockCache{}, + Cache: &cache.Nop{}, SecretKey: "123456", } diff --git a/backend/app/rest/api/rest_private.go b/backend/app/rest/api/rest_private.go index 5435f570..62a2163a 100644 --- a/backend/app/rest/api/rest_private.go +++ b/backend/app/rest/api/rest_private.go @@ -18,6 +18,7 @@ import ( "github.com/umputun/remark/backend/app/rest" "github.com/umputun/remark/backend/app/rest/auth" + "github.com/umputun/remark/backend/app/rest/cache" "github.com/umputun/remark/backend/app/store" "github.com/umputun/remark/backend/app/store/service" ) @@ -70,7 +71,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(comment.Locator.URL, "last", comment.User.ID, comment.Locator.SiteID) + s.Cache.Flush(cache.Flusher(comment.Locator.SiteID). + Scopes(comment.Locator.URL, "last", comment.User.ID, comment.Locator.SiteID)) render.Status(r, http.StatusCreated) render.JSON(w, r, &finalComment) @@ -121,7 +123,7 @@ func (s *Rest) updateCommentCtrl(w http.ResponseWriter, r *http.Request) { return } - s.Cache.Flush(locator.URL, "last", user.ID) + s.Cache.Flush(cache.Flusher(locator.SiteID).Scopes(locator.URL, "last", user.ID)) render.JSON(w, r, res) } @@ -155,7 +157,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(locator.URL, comment.User.ID) + s.Cache.Flush(cache.Flusher(locator.SiteID).Scopes(locator.URL, comment.User.ID)) render.JSON(w, r, JSON{"id": comment.ID, "score": comment.Score}) } diff --git a/backend/app/rest/api/rest_public.go b/backend/app/rest/api/rest_public.go index b4fd0eda..bf397ab0 100644 --- a/backend/app/rest/api/rest_public.go +++ b/backend/app/rest/api/rest_public.go @@ -27,7 +27,8 @@ 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(cache.Key(cache.URLKey(r), locator.SiteID, locator.URL), func() ([]byte, error) { + key := cache.NewKey(locator.SiteID).ID(cache.URLKey(r)).Scopes(locator.SiteID, locator.URL) + data, err := s.Cache.Get(key, func() ([]byte, error) { comments, e := s.DataService.Find(locator, sort) if e != nil { return nil, e @@ -90,7 +91,8 @@ func (s *Rest) previewCommentCtrl(w http.ResponseWriter, r *http.Request) { func (s *Rest) infoCtrl(w http.ResponseWriter, r *http.Request) { locator := store.Locator{SiteID: r.URL.Query().Get("site"), URL: r.URL.Query().Get("url")} - data, err := s.Cache.Get(cache.Key(cache.URLKey(r), locator.SiteID, locator.URL), func() ([]byte, error) { + key := cache.NewKey(locator.SiteID).ID(cache.URLKey(r)).Scopes(locator.SiteID, locator.URL) + data, err := s.Cache.Get(key, func() ([]byte, error) { info, e := s.DataService.Info(locator, s.ReadOnlyAge) if e != nil { return nil, e @@ -116,7 +118,8 @@ func (s *Rest) lastCommentsCtrl(w http.ResponseWriter, r *http.Request) { limit = 0 } - data, err := s.Cache.Get(cache.Key(cache.URLKey(r), "last", siteID), func() ([]byte, error) { + key := cache.NewKey(siteID).ID(cache.URLKey(r)).Scopes("last") + data, err := s.Cache.Get(key, func() ([]byte, error) { comments, e := s.DataService.Last(siteID, limit) if e != nil { return nil, e @@ -179,7 +182,8 @@ 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(cache.Key(cache.URLKey(r), userID, siteID), func() ([]byte, error) { + key := cache.NewKey(siteID).ID(cache.URLKey(r)).Scopes(userID, siteID) + data, err := s.Cache.Get(key, func() ([]byte, error) { comments, e := s.DataService.User(siteID, userID, limit, 0) if e != nil { return nil, e @@ -259,15 +263,15 @@ func (s *Rest) countMultiCtrl(w http.ResponseWriter, r *http.Request) { } // key could be long for multiple posts, make it sha1 - key := cache.URLKey(r) + strings.Join(posts, ",") + k := cache.URLKey(r) + strings.Join(posts, ",") hasher := sha1.New() - if _, err := hasher.Write([]byte(key)); err != nil { + if _, err := hasher.Write([]byte(k)); err != nil { rest.SendErrorJSON(w, r, http.StatusInternalServerError, err, "can't make sha1 for list of urls") return } sha := base64.URLEncoding.EncodeToString(hasher.Sum(nil)) - - data, err := s.Cache.Get(cache.Key(sha, siteID), func() ([]byte, error) { + key := cache.NewKey(siteID).ID(sha).Scopes(siteID) + data, err := s.Cache.Get(key, func() ([]byte, error) { counts, e := s.DataService.Counts(siteID, posts) if e != nil { return nil, e @@ -295,7 +299,8 @@ func (s *Rest) listCtrl(w http.ResponseWriter, r *http.Request) { skip = v } - data, err := s.Cache.Get(cache.Key(cache.URLKey(r), siteID), func() ([]byte, error) { + key := cache.NewKey(siteID).ID(cache.URLKey(r)).Scopes(siteID) + data, err := s.Cache.Get(key, func() ([]byte, error) { posts, e := s.DataService.List(siteID, limit, skip) if e != nil { return nil, e diff --git a/backend/app/rest/api/rest_test.go b/backend/app/rest/api/rest_test.go index 6dd65230..936e8df2 100644 --- a/backend/app/rest/api/rest_test.go +++ b/backend/app/rest/api/rest_test.go @@ -17,6 +17,7 @@ import ( "github.com/umputun/remark/backend/app/migrator" "github.com/umputun/remark/backend/app/rest/auth" + "github.com/umputun/remark/backend/app/rest/cache" "github.com/umputun/remark/backend/app/rest/proxy" "github.com/umputun/remark/backend/app/store" "github.com/umputun/remark/backend/app/store/avatar" @@ -90,7 +91,7 @@ func prep(t *testing.T) (srv *Rest, ts *httptest.Server) { JWTService: auth.NewJWT("12345", false, time.Minute, time.Hour), }, Exporter: &migrator.Remark{DataStore: dataStore}, - Cache: &mockCache{}, + Cache: &cache.Nop{}, WebRoot: "/tmp", RemarkURL: "https://demo.remark42.com", AvatarProxy: &proxy.Avatar{Store: avatar.NewLocalFS("/tmp", 300), RoutePath: "/api/v1/avatar"}, @@ -163,11 +164,3 @@ func cleanup(ts *httptest.Server, srv *Rest) { os.Remove(testDb) os.Remove(testHTML) } - -type mockCache struct{} - -func (mc *mockCache) Get(key string, fn func() ([]byte, error)) (data []byte, err error) { - return fn() -} - -func (mc *mockCache) Flush(scopes ...string) {} diff --git a/backend/app/rest/api/rss.go b/backend/app/rest/api/rss.go index 12245848..39ef0fb6 100644 --- a/backend/app/rest/api/rss.go +++ b/backend/app/rest/api/rss.go @@ -35,7 +35,8 @@ func (s *Rest) rssPostCommentsCtrl(w http.ResponseWriter, r *http.Request) { locator := store.Locator{SiteID: r.URL.Query().Get("site"), URL: r.URL.Query().Get("url")} log.Printf("[DEBUG] get rss for post %+v", locator) - data, err := s.Cache.Get(cache.Key(cache.URLKey(r), locator.SiteID, locator.URL), func() ([]byte, error) { + key := cache.NewKey(locator.SiteID).ID(cache.URLKey(r)).Scopes(locator.SiteID, locator.URL) + data, err := s.Cache.Get(key, func() ([]byte, error) { comments, e := s.DataService.Find(locator, "-time") if e != nil { return nil, e @@ -66,7 +67,8 @@ func (s *Rest) rssSiteCommentsCtrl(w http.ResponseWriter, r *http.Request) { siteID := r.URL.Query().Get("site") log.Printf("[DEBUG] get rss for site %s", siteID) - data, err := s.Cache.Get(cache.Key(cache.URLKey(r), siteID, "last"), func() ([]byte, error) { + key := cache.NewKey(siteID).ID(cache.URLKey(r)).Scopes(siteID, "last") + data, err := s.Cache.Get(key, func() ([]byte, error) { comments, e := s.DataService.Last(siteID, maxRssItems) if e != nil { return nil, e @@ -98,7 +100,8 @@ func (s *Rest) rssRepliesCtrl(w http.ResponseWriter, r *http.Request) { siteID := r.URL.Query().Get("site") log.Printf("[DEBUG] get rss replies to user %s for site %s", userID, siteID) - data, err := s.Cache.Get(cache.Key(cache.URLKey(r), siteID, "last"), func() (res []byte, e error) { + key := cache.NewKey(siteID).ID(cache.URLKey(r)).Scopes(siteID, "last") + data, err := s.Cache.Get(key, func() (res []byte, e error) { comments, e := s.DataService.Last(siteID, maxLastCommentsReply) if e != nil { return nil, errors.Wrap(e, "can't get last comments") diff --git a/backend/app/rest/cache/cache.go b/backend/app/rest/cache/cache.go index 10badd95..b0c4d62c 100644 --- a/backend/app/rest/cache/cache.go +++ b/backend/app/rest/cache/cache.go @@ -5,32 +5,88 @@ import ( "strings" "github.com/pkg/errors" + "github.com/umputun/remark/backend/app/rest" ) // LoadingCache defines interface for caching type LoadingCache interface { - Get(key string, fn func() ([]byte, error)) (data []byte, err error) - Flush(scopes ...string) + Get(key Key, fn func() ([]byte, error)) (data []byte, err error) + Flush(req FlusherRequest) } -// Key makes full key from primary key and scopes -func Key(key string, scopes ...string) string { - return strings.Join(scopes, "$$") + "@@" + key +type cacheWithOpts interface { + LoadingCache + setMaxValSize(max int) error + setMaxKeys(max int) error + setMaxCacheSize(max int64) error + setPostFlushFn(postFlushFn func()) error +} + +// Key for cache +type Key struct { + id string + siteID string + scopes []string +} + +// NewKey makes keys for site +func NewKey(site string) Key { + res := Key{siteID: site} + return res +} + +// ID sets key id +func (k Key) ID(id string) Key { + k.id = id + return k +} + +// Scopes of the key +func (k Key) Scopes(scopes ...string) Key { + k.scopes = scopes + return k +} + +// Merge makes full string key from primary key and scopes +func (k Key) Merge() string { + return strings.Join(k.scopes, "$$") + "@@" + k.id + "@@" + k.siteID } // ParseKey gets compound key created by Key func and split it to the actual key and scopes -func ParseKey(fullKey string) (key string, scopes []string, err error) { +func ParseKey(fullKey string) (Key, error) { elems := strings.Split(fullKey, "@@") - if len(elems) != 2 { - return "", nil, errors.Errorf("can't parse cache key %s", key) + if len(elems) != 3 { + return Key{}, errors.Errorf("can't parse cache key %s", fullKey) } - scopes = strings.Split(elems[0], "$$") + scopes := strings.Split(elems[0], "$$") if len(scopes) == 1 && scopes[0] == "" { scopes = []string{} } - key = elems[1] - return key, scopes, nil + key := Key{ + scopes: scopes, + id: elems[1], + siteID: elems[2], + } + return key, nil +} + +// FlusherRequest used as input for cache.Flush +type FlusherRequest struct { + siteID string + scopes []string +} + +// Flusher makes new FlusherRequest with empty scopes +func Flusher(siteID string) FlusherRequest { + res := FlusherRequest{siteID: siteID} + return res +} + +// Scopes adds scopes to FlusherRequest +func (f FlusherRequest) Scopes(scopes ...string) FlusherRequest { + f.scopes = scopes + return f } // URLKey gets url from request to use it as cache key @@ -43,3 +99,14 @@ func URLKey(r *http.Request) string { } return key } + +// Nop does nothing for caching, passing fn call only +type Nop struct{} + +// Get calls fn, no actual caching +func (n *Nop) Get(key Key, fn func() ([]byte, error)) (data []byte, err error) { + return fn() +} + +// Flush does nothing for NoopCache +func (n *Nop) Flush(req FlusherRequest) {} diff --git a/backend/app/rest/cache/cache_test.go b/backend/app/rest/cache/cache_test.go index 31d934c5..8b0e0645 100644 --- a/backend/app/rest/cache/cache_test.go +++ b/backend/app/rest/cache/cache_test.go @@ -15,24 +15,25 @@ func TestCache_Keys(t *testing.T) { scopes []string full string }{ - {"key1", []string{"s1"}, "s1@@key1"}, - {"key2", []string{"s11", "s2"}, "s11$$s2@@key2"}, - {"key3", []string{}, "@@key3"}, + {"key1", []string{"s1"}, "s1@@key1@@site"}, + {"key2", []string{"s11", "s2"}, "s11$$s2@@key2@@site"}, + {"key3", []string{}, "@@key3@@site"}, } for n, tt := range tbl { - full := Key(tt.key, tt.scopes...) + k := NewKey("site").ID(tt.key).Scopes(tt.scopes...) + full := k.Merge() assert.Equal(t, tt.full, full, "making key, #%d", n) - k, s, e := ParseKey(full) + k, e := ParseKey(full) assert.Nil(t, e) - assert.Equal(t, tt.scopes, s) - assert.Equal(t, tt.key, k) + assert.Equal(t, tt.scopes, k.scopes) + assert.Equal(t, tt.key, k.id) } - _, _, err := ParseKey("abc") + _, err := ParseKey("abc") assert.Error(t, err) - _, _, err = ParseKey("") + _, err = ParseKey("") assert.Error(t, err) } diff --git a/backend/app/rest/cache/memory.go b/backend/app/rest/cache/memory.go index 1426f999..a40153b0 100644 --- a/backend/app/rest/cache/memory.go +++ b/backend/app/rest/cache/memory.go @@ -20,6 +20,8 @@ type memoryCache struct { // NewMemoryCache makes memoryCache implementation func NewMemoryCache(options ...Option) (LoadingCache, error) { + log.Print("[INFO] make memory cache") + res := memoryCache{ postFlushFn: func() {}, maxKeys: 1000, @@ -48,8 +50,9 @@ func NewMemoryCache(options ...Option) (LoadingCache, error) { } // Get is loading cache method to get value by key or load via fn if not found -func (m *memoryCache) Get(key string, fn func() ([]byte, error)) (data []byte, err error) { - if b, ok := m.bytesCache.Get(key); ok { +func (m *memoryCache) Get(key Key, fn func() ([]byte, error)) (data []byte, err error) { + mkey := key.Merge() + if b, ok := m.bytesCache.Get(mkey); ok { return b.([]byte), nil } @@ -57,7 +60,7 @@ func (m *memoryCache) Get(key string, fn func() ([]byte, error)) (data []byte, e return data, err } if m.allowed(data) { - m.bytesCache.Add(key, data) + m.bytesCache.Add(mkey, data) atomic.AddInt64(&m.currentSize, int64(len(data))) if m.maxCacheSize > 0 && atomic.LoadInt64(&m.currentSize) > m.maxCacheSize { @@ -70,9 +73,9 @@ func (m *memoryCache) Get(key string, fn func() ([]byte, error)) (data []byte, e } // Flush clears cache and calls postFlushFn async -func (m *memoryCache) Flush(scopes ...string) { +func (m *memoryCache) Flush(req FlusherRequest) { - if len(scopes) == 0 { + if len(req.scopes) == 0 { m.bytesCache.Purge() go m.postFlushFn() return @@ -80,12 +83,12 @@ func (m *memoryCache) Flush(scopes ...string) { // check if fullKey has matching scopes inScope := func(fullKey string) bool { - _, keyScopes, err := ParseKey(fullKey) + key, err := ParseKey(fullKey) if err != nil { return false } - for _, s := range scopes { - for _, ks := range keyScopes { + for _, s := range req.scopes { + for _, ks := range key.scopes { if ks == s { return true } @@ -112,3 +115,32 @@ func (m *memoryCache) allowed(data []byte) bool { } return true } + +func (m *memoryCache) setMaxValSize(max int) error { + m.maxValueSize = max + if max <= 0 { + return errors.Errorf("negative size for MaxValSize, %d", max) + } + return nil +} + +func (m *memoryCache) setMaxKeys(max int) error { + m.maxKeys = max + if max <= 0 { + return errors.Errorf("negative size for MaxKeys, %d", max) + } + return nil +} + +func (m *memoryCache) setMaxCacheSize(max int64) error { + m.maxCacheSize = max + if max <= 0 { + return errors.Errorf("negative size or MaxCacheSize, %d", max) + } + return nil +} + +func (m *memoryCache) setPostFlushFn(postFlushFn func()) error { + m.postFlushFn = postFlushFn + return nil +} diff --git a/backend/app/rest/cache/memory_test.go b/backend/app/rest/cache/memory_test.go index eb03bad5..35fe6b6a 100644 --- a/backend/app/rest/cache/memory_test.go +++ b/backend/app/rest/cache/memory_test.go @@ -17,7 +17,7 @@ func TestMemoryCache_Get(t *testing.T) { var postFnCall, coldCalls int32 lc, err := NewMemoryCache(PostFlushFn(func() { atomic.AddInt32(&postFnCall, 1) })) require.Nil(t, err) - res, err := lc.Get("key", func() ([]byte, error) { + res, err := lc.Get(NewKey("site").ID("key"), func() ([]byte, error) { atomic.AddInt32(&coldCalls, 1) return []byte("result"), nil }) @@ -26,7 +26,7 @@ func TestMemoryCache_Get(t *testing.T) { assert.Equal(t, int32(1), atomic.LoadInt32(&coldCalls)) assert.Equal(t, int32(0), atomic.LoadInt32(&postFnCall)) - res, err = lc.Get("key", func() ([]byte, error) { + res, err = lc.Get(NewKey("site").ID("key"), func() ([]byte, error) { atomic.AddInt32(&coldCalls, 1) return []byte("result"), nil }) @@ -35,11 +35,11 @@ func TestMemoryCache_Get(t *testing.T) { assert.Equal(t, int32(1), atomic.LoadInt32(&coldCalls)) assert.Equal(t, int32(0), atomic.LoadInt32(&postFnCall)) - lc.Flush() + lc.Flush(Flusher("site")) time.Sleep(100 * time.Millisecond) // let postFn to do its thing assert.Equal(t, int32(1), atomic.LoadInt32(&postFnCall)) - _, err = lc.Get("key", func() ([]byte, error) { + _, err = lc.Get(NewKey("site").ID("key"), func() ([]byte, error) { return nil, errors.New("err") }) assert.NotNil(t, err) @@ -53,7 +53,7 @@ func TestMemoryCache_MaxKeys(t *testing.T) { // put 5 keys to cache for i := 0; i < 5; i++ { - res, e := lc.Get(fmt.Sprintf("key-%d", i), func() ([]byte, error) { + res, e := lc.Get(NewKey("site").ID(fmt.Sprintf("key-%d", i)), func() ([]byte, error) { atomic.AddInt32(&coldCalls, 1) return []byte(fmt.Sprintf("result-%d", i)), nil }) @@ -64,14 +64,14 @@ func TestMemoryCache_MaxKeys(t *testing.T) { } // check if really cached - res, err := lc.Get("key-3", func() ([]byte, error) { + res, err := lc.Get(NewKey("site").ID("key-3"), func() ([]byte, error) { return []byte("result-blah"), nil }) assert.Nil(t, err) assert.Equal(t, "result-3", string(res), "should be cached") // try to cache after maxKeys reached - res, err = lc.Get("key-X", func() ([]byte, error) { + res, err = lc.Get(NewKey("site").ID("key-X"), func() ([]byte, error) { return []byte("result-X"), nil }) assert.Nil(t, err) @@ -80,13 +80,13 @@ func TestMemoryCache_MaxKeys(t *testing.T) { assert.Equal(t, 5, lc.(*memoryCache).bytesCache.Len()) // put to cache and make sure it cached - res, err = lc.Get("key-Z", func() ([]byte, error) { + res, err = lc.Get(NewKey("site").ID("key-Z"), func() ([]byte, error) { return []byte("result-Z"), nil }) assert.Nil(t, err) assert.Equal(t, "result-Z", string(res)) - res, err = lc.Get("key-Z", func() ([]byte, error) { + res, err = lc.Get(NewKey("site").ID("key-Z"), func() ([]byte, error) { return []byte("result-Zzzz"), nil }) assert.Nil(t, err) @@ -98,26 +98,26 @@ func TestMemoryCache_MaxValueSize(t *testing.T) { lc, err := NewMemoryCache(MaxKeys(5), MaxValSize(10)) require.Nil(t, err) // put good size value to cache and make sure it cached - res, err := lc.Get("key-Z", func() ([]byte, error) { + res, err := lc.Get(NewKey("site").ID("key-Z"), func() ([]byte, error) { return []byte("result-Z"), nil }) assert.Nil(t, err) assert.Equal(t, "result-Z", string(res)) - res, err = lc.Get("key-Z", func() ([]byte, error) { + res, err = lc.Get(NewKey("site").ID("key-Z"), func() ([]byte, error) { return []byte("result-Zzzz"), nil }) assert.Nil(t, err) assert.Equal(t, "result-Z", string(res), "got cached value") // put too big value to cache and make sure it is not cached - res, err = lc.Get("key-Big", func() ([]byte, error) { + res, err = lc.Get(NewKey("site").ID("key-Big"), func() ([]byte, error) { return []byte("1234567890"), nil }) assert.Nil(t, err) assert.Equal(t, "1234567890", string(res)) - res, err = lc.Get("key-Big", func() ([]byte, error) { + res, err = lc.Get(NewKey("site").ID("key-Big"), func() ([]byte, error) { return []byte("result-big"), nil }) assert.Nil(t, err) @@ -129,21 +129,21 @@ func TestMemoryCache_MaxCacheSize(t *testing.T) { require.Nil(t, err) // put good size value to cache and make sure it cached - res, err := lc.Get("key-Z", func() ([]byte, error) { + res, err := lc.Get(NewKey("site").ID("key-Z"), func() ([]byte, error) { return []byte("result-Z"), nil }) assert.Nil(t, err) assert.Equal(t, "result-Z", string(res)) assert.Equal(t, int64(8), lc.(*memoryCache).currentSize) - _, err = lc.Get("key-Z2", func() ([]byte, error) { + _, err = lc.Get(NewKey("site").ID("key-Z2"), func() ([]byte, error) { return []byte("result-Z"), nil }) assert.Nil(t, err) assert.Equal(t, int64(16), lc.(*memoryCache).currentSize) // this will cause removal - _, err = lc.Get("key-Z3", func() ([]byte, error) { + _, err = lc.Get(NewKey("site").ID("key-Z3"), func() ([]byte, error) { return []byte("result-Z"), nil }) assert.Nil(t, err) @@ -163,7 +163,7 @@ func TestMemoryCache_MaxCacheSizeParallel(t *testing.T) { go func() { time.Sleep(time.Duration(rand.Intn(100)) * time.Nanosecond) defer wg.Done() - res, err := lc.Get(fmt.Sprintf("key-%d", i), func() ([]byte, error) { + res, err := lc.Get(NewKey("site").ID(fmt.Sprintf("key-%d", i)), func() ([]byte, error) { return []byte(fmt.Sprintf("result-%d", i)), nil }) require.Nil(t, err) @@ -182,7 +182,7 @@ func TestMemoryCache_Parallel(t *testing.T) { lc, err := NewMemoryCache() require.Nil(t, err) - res, err := lc.Get("key", func() ([]byte, error) { + res, err := lc.Get(NewKey("site").ID("key"), func() ([]byte, error) { return []byte("value"), nil }) assert.Nil(t, err) @@ -194,7 +194,7 @@ func TestMemoryCache_Parallel(t *testing.T) { i := i go func() { defer wg.Done() - res, err := lc.Get("key", func() ([]byte, error) { + res, err := lc.Get(NewKey("site").ID("key"), func() ([]byte, error) { atomic.AddInt32(&coldCalls, 1) return []byte(fmt.Sprintf("result-%d", i)), nil }) @@ -210,28 +210,28 @@ func TestMemoryCache_Scopes(t *testing.T) { lc, err := NewMemoryCache() require.Nil(t, err) - res, err := lc.Get(Key("key", "s1", "s2"), func() ([]byte, error) { + res, err := lc.Get(NewKey("site").ID("key").Scopes("s1", "s2"), func() ([]byte, error) { return []byte("value"), nil }) assert.Nil(t, err) assert.Equal(t, "value", string(res)) - res, err = lc.Get(Key("key2", "s2"), func() ([]byte, error) { + res, err = lc.Get(NewKey("site").ID("key2").Scopes("s2"), func() ([]byte, error) { return []byte("value2"), nil }) assert.Nil(t, err) assert.Equal(t, "value2", string(res)) assert.Equal(t, 2, lc.(*memoryCache).bytesCache.Len()) - lc.Flush("s1") + lc.Flush(Flusher("site").Scopes("s1")) assert.Equal(t, 1, lc.(*memoryCache).bytesCache.Len()) - _, err = lc.Get(Key("key2", "s2"), func() ([]byte, error) { + _, err = lc.Get(NewKey("site").ID("key2").Scopes("s2"), func() ([]byte, error) { assert.Fail(t, "should stay") return nil, nil }) assert.Nil(t, err) - res, err = lc.Get(Key("key", "s1", "s2"), func() ([]byte, error) { + res, err = lc.Get(NewKey("site").ID("key").Scopes("s1", "s2"), func() ([]byte, error) { return []byte("value-upd"), nil }) assert.Nil(t, err) @@ -242,23 +242,23 @@ func TestMemoryCache_Flush(t *testing.T) { lc, err := NewMemoryCache() require.Nil(t, err) - addToCache := func(key string, scopes ...string) { - res, err := lc.Get(key, func() ([]byte, error) { - return []byte("value" + key), nil + addToCache := func(id string, scopes ...string) { + res, err := lc.Get(NewKey("site").ID(id).Scopes(scopes...), func() ([]byte, error) { + return []byte("value" + id), nil }) require.Nil(t, err) - require.Equal(t, "value"+key, string(res)) + require.Equal(t, "value"+id, 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")) + lc.Flush(Flusher("site")) + addToCache("key1", "s1", "s2") + addToCache("key2", "s1", "s2", "s3") + addToCache("key3", "s1", "s2", "s3") + addToCache("key4", "s2", "s3") + addToCache("key5", "s2") + addToCache("key6") + addToCache("key7", "s4", "s3") require.Equal(t, 7, lc.(*memoryCache).bytesCache.Len(), "cache init") } @@ -279,7 +279,7 @@ func TestMemoryCache_Flush(t *testing.T) { for i, tt := range tbl { init() - lc.Flush(tt.scopes...) + lc.Flush(Flusher("site").Scopes(tt.scopes...)) assert.Equal(t, tt.left, lc.(*memoryCache).bytesCache.Len(), "keys size, %s #%d", tt.msg, i) } } @@ -287,14 +287,14 @@ func TestMemoryCache_Flush(t *testing.T) { func TestMemoryCache_FlushFailed(t *testing.T) { lc, err := NewMemoryCache() require.Nil(t, err) - val, err := lc.Get("invalid-composite", func() ([]byte, error) { + val, err := lc.Get(NewKey("site").ID("invalid-composite"), func() ([]byte, error) { return []byte("value"), nil }) assert.Nil(t, err) assert.Equal(t, "value", string(val)) assert.Equal(t, 1, lc.(*memoryCache).bytesCache.Len()) - lc.Flush("invalid-composite") + lc.Flush(Flusher("site").Scopes("invalid-composite")) assert.Equal(t, 1, lc.(*memoryCache).bytesCache.Len()) } diff --git a/backend/app/rest/cache/mongo.go b/backend/app/rest/cache/mongo.go new file mode 100644 index 00000000..8f0c8e99 --- /dev/null +++ b/backend/app/rest/cache/mongo.go @@ -0,0 +1,188 @@ +package cache + +import ( + "log" + "time" + + "github.com/go-pkgz/repeater" + + "github.com/globalsign/mgo" + "github.com/globalsign/mgo/bson" + "github.com/go-pkgz/mongo" + multierror "github.com/hashicorp/go-multierror" + "github.com/pkg/errors" +) + +type mongoCache struct { + connection *mongo.Connection + postFlushFn func() + maxKeys int + maxValueSize int + maxCacheSize int64 +} + +const cacheCollection = "cache" + +type mongoDoc struct { + SiteID string `bson:"site"` + Key string `bson:"key"` + Scopes []string `bson:"scopes,omitempty"` + Data []byte `bson:"data"` +} + +// NewMongoCache makes mongoCache implementation +func NewMongoCache(connection *mongo.Connection, options ...Option) (LoadingCache, error) { + log.Printf("[INFO] make mongo cache with %s", connection) + res := &mongoCache{ + connection: connection, + postFlushFn: func() {}, + maxKeys: 1000, + maxValueSize: 0, + } + for _, opt := range options { + if err := opt(res); err != nil { + return nil, errors.Wrap(err, "failed to set cache option") + } + } + if err := res.prepare(); err != nil { + return nil, err + } + return res, nil +} + +// Get is loading cache method to get value by key or load via fn if not found +func (m *mongoCache) Get(key Key, fn func() ([]byte, error)) (data []byte, err error) { + + d := mongoDoc{} + + // repeat find from cache with small delay to avoid mgo random error + rep := repeater.NewDefault(5, 10*time.Millisecond) + mgErr := rep.Do(func() error { + return m.connection.WithCustomCollection(cacheCollection, func(coll *mgo.Collection) error { + return coll.Find(bson.M{"site": key.siteID, "key": key.id}).One(&d) + }) + }, mgo.ErrNotFound) + if mgErr == nil { // cached result found + return d.Data, nil + } + + if data, err = fn(); err != nil { + return data, err + } + + if mgErr != mgo.ErrNotFound { // some other error in mgo query, don't try to update cache + log.Printf("[WARN] unexpected mgo error %+v", mgErr) + return data, err + } + + if !m.allowed(data) { + return data, nil + } + + d = mongoDoc{ + SiteID: key.siteID, + Key: key.id, + Data: data, + Scopes: key.scopes, + } + err = m.connection.WithCustomCollection(cacheCollection, func(coll *mgo.Collection) error { + _, e := coll.Upsert(bson.M{"site": key.siteID, "key": key.id}, bson.M{"$set": d}) + return e + }) + if err != nil { + return nil, errors.Wrapf(err, "can't set cached value for %+v", key) + } + + if m.maxKeys > 0 { + err = m.cleanup(key.siteID) + } + + return data, errors.Wrap(err, "failed to cleanup cached records") +} + +func (m *mongoCache) cleanup(siteID string) (err error) { + ids := []struct { + ID bson.ObjectId `bson:"_id"` + }{} + + err = m.connection.WithCustomCollection(cacheCollection, func(coll *mgo.Collection) error { + n, countErr := coll.Find(bson.M{"site": siteID}).Count() + if countErr != nil { + return countErr + } + if countErr == nil && n > m.maxKeys { + if findErr := coll.Find(bson.M{"site": siteID}).Sort("+id").Limit(n - m.maxKeys).All(&ids); findErr == nil { + bsonIDs := []bson.ObjectId{} + for _, id := range ids { + bsonIDs = append(bsonIDs, id.ID) + } + _, removalErr := coll.RemoveAll(bson.M{"_id": bson.M{"$in": bsonIDs}}) + return removalErr + } + } + return nil + }) + return err +} + +// Flush clears cache and calls postFlushFn async +func (m *mongoCache) Flush(req FlusherRequest) { + err := m.connection.WithCustomCollection(cacheCollection, func(coll *mgo.Collection) error { + q := bson.M{"site": req.siteID} + if len(req.scopes) > 0 { + q["scopes"] = bson.M{"$in": req.scopes} + } + _, e := coll.RemoveAll(q) + return e + }) + + if err == nil && m.postFlushFn != nil { + m.postFlushFn() + } +} + +// prepare collections with all indexes +func (m *mongoCache) prepare() error { + errs := new(multierror.Error) + return m.connection.WithCustomCollection(cacheCollection, func(coll *mgo.Collection) error { + errs = multierror.Append(errs, coll.EnsureIndexKey("site", "key")) + errs = multierror.Append(errs, coll.EnsureIndexKey("site", "scopes")) + return errors.Wrapf(errs.ErrorOrNil(), "can't create index for %s", cacheCollection) + }) +} + +func (m *mongoCache) allowed(data []byte) bool { + if m.maxValueSize > 0 && len(data) >= m.maxValueSize { + return false + } + return true +} + +func (m *mongoCache) setMaxValSize(max int) error { + m.maxValueSize = max + if max <= 0 { + return errors.Errorf("negative size for MaxValSize, %d", max) + } + return nil +} + +func (m *mongoCache) setMaxKeys(max int) error { + m.maxKeys = max + if max <= 0 { + return errors.Errorf("negative size for MaxKeys, %d", max) + } + return nil +} + +func (m *mongoCache) setMaxCacheSize(max int64) error { + m.maxCacheSize = max + if max <= 0 { + return errors.Errorf("negative size or MaxCacheSize, %d", max) + } + return nil +} + +func (m *mongoCache) setPostFlushFn(postFlushFn func()) error { + m.postFlushFn = postFlushFn + return nil +} diff --git a/backend/app/rest/cache/mongo_test.go b/backend/app/rest/cache/mongo_test.go new file mode 100644 index 00000000..be2a7e4c --- /dev/null +++ b/backend/app/rest/cache/mongo_test.go @@ -0,0 +1,244 @@ +package cache + +import ( + "fmt" + "log" + "os" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/globalsign/mgo" + "github.com/globalsign/mgo/bson" + "github.com/go-pkgz/mongo" + "github.com/pkg/errors" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestMongoCache_Get(t *testing.T) { + conn, err := mongo.MakeTestConnection(t) + assert.NoError(t, err) + defer mongo.RemoveTestCollections(t, conn, "cache") + + var postFnCall, coldCalls int32 + lc, err := NewMongoCache(conn, PostFlushFn(func() { atomic.AddInt32(&postFnCall, 1) })) + require.Nil(t, err) + res, err := lc.Get(NewKey("site").ID("key"), func() ([]byte, error) { + atomic.AddInt32(&coldCalls, 1) + return []byte("result"), nil + }) + assert.Nil(t, err) + assert.Equal(t, "result", string(res)) + assert.Equal(t, int32(1), atomic.LoadInt32(&coldCalls)) + assert.Equal(t, int32(0), atomic.LoadInt32(&postFnCall)) + + res, err = lc.Get(NewKey("site").ID("key"), func() ([]byte, error) { + atomic.AddInt32(&coldCalls, 1) + return []byte("result"), nil + }) + assert.Nil(t, err) + assert.Equal(t, "result", string(res)) + assert.Equal(t, int32(1), atomic.LoadInt32(&coldCalls)) + assert.Equal(t, int32(0), atomic.LoadInt32(&postFnCall)) + + lc.Flush(Flusher("site")) + time.Sleep(100 * time.Millisecond) // let postFn to do its thing + assert.Equal(t, int32(1), atomic.LoadInt32(&postFnCall)) + + _, err = lc.Get(NewKey("site").ID("key"), func() ([]byte, error) { + return nil, errors.New("err") + }) + assert.NotNil(t, err) +} + +func TestMongoCache_MaxKeys(t *testing.T) { + var postFnCall, coldCalls int32 + conn, err := mongo.MakeTestConnection(t) + assert.NoError(t, err) + defer mongo.RemoveTestCollections(t, conn, "cache") + + lc, err := NewMongoCache(conn, PostFlushFn(func() { atomic.AddInt32(&postFnCall, 1) }), + MaxKeys(5), MaxValSize(10)) + require.Nil(t, err) + + // put 5 keys to cache + for i := 0; i < 5; i++ { + res, e := lc.Get(NewKey("site").ID(fmt.Sprintf("key-%d", i)), func() ([]byte, error) { + atomic.AddInt32(&coldCalls, 1) + return []byte(fmt.Sprintf("result-%d", i)), nil + }) + assert.Nil(t, e) + assert.Equal(t, fmt.Sprintf("result-%d", i), string(res)) + assert.Equal(t, int32(i+1), atomic.LoadInt32(&coldCalls)) + assert.Equal(t, int32(0), atomic.LoadInt32(&postFnCall)) + } + + // check if really cached + res, err := lc.Get(NewKey("site").ID("key-3"), func() ([]byte, error) { + return []byte("result-blah"), nil + }) + assert.Nil(t, err) + assert.Equal(t, "result-3", string(res), "should be cached") + + // try to cache after maxKeys reached + res, err = lc.Get(NewKey("site").ID("key-X"), func() ([]byte, error) { + return []byte("result-X"), nil + }) + assert.Nil(t, err) + assert.Equal(t, "result-X", string(res)) + + conn.WithCustomCollection("cache", func(coll *mgo.Collection) error { + n, e := coll.Find(bson.M{"site": "site"}).Count() + require.NoError(t, e) + require.Equal(t, 5, n) + r := mongoDoc{} + require.NoError(t, coll.Find(bson.M{"site": "site"}).Sort("+_id").One(&r)) + assert.Equal(t, "key-1", r.Key) + return nil + }) + + // put to cache and make sure it cached + res, err = lc.Get(NewKey("site").ID("key-Z"), func() ([]byte, error) { + return []byte("result-Z"), nil + }) + assert.Nil(t, err) + assert.Equal(t, "result-Z", string(res)) + + res, err = lc.Get(NewKey("site").ID("key-Z"), func() ([]byte, error) { + return []byte("result-Zzzz"), nil + }) + assert.Nil(t, err) + assert.Equal(t, "result-Z", string(res), "got cached value") + + conn.WithCustomCollection("cache", func(coll *mgo.Collection) error { + n, e := coll.Find(bson.M{"site": "site"}).Count() + require.NoError(t, e) + require.Equal(t, 5, n) + r := mongoDoc{} + require.NoError(t, coll.Find(bson.M{"site": "site"}).Sort("+_id").One(&r)) + assert.Equal(t, "key-2", r.Key) + return nil + }) +} + +func TestMongoCache_Parallel(t *testing.T) { + var coldCalls int32 + conn, err := mongo.MakeTestConnection(t) + assert.NoError(t, err) + defer mongo.RemoveTestCollections(t, conn, "cache") + lc, err := NewMongoCache(conn) + require.Nil(t, err) + + res, err := lc.Get(NewKey("site").ID("key").Scopes("s1", "s2"), func() ([]byte, error) { + return []byte("value"), nil + }) + assert.Nil(t, err) + assert.Equal(t, "value", string(res)) + wg := sync.WaitGroup{} + for i := 0; i < 100; i++ { + wg.Add(1) + i := i + go func() { + defer wg.Done() + r, err := lc.Get(NewKey("site").ID("key").Scopes("s1", "s2"), func() ([]byte, error) { + atomic.AddInt32(&coldCalls, 1) + return []byte(fmt.Sprintf("result-%d", i)), nil + }) + require.Nil(t, err) + v := string(r) + assert.Equal(t, "value", v, "th=%d", i) + }() + } + wg.Wait() + assert.Equal(t, int32(0), atomic.LoadInt32(&coldCalls)) +} + +func TestMongoCache_Flush(t *testing.T) { + conn, err := mongo.MakeTestConnection(t) + assert.NoError(t, err) + defer mongo.RemoveTestCollections(t, conn, "cache") + lc, err := NewMongoCache(conn) + require.Nil(t, err) + + addToCache := func(id string, scopes ...string) { + res, err := lc.Get(NewKey("site").ID(id).Scopes(scopes...), func() ([]byte, error) { + return []byte("value" + id), nil + }) + require.Nil(t, err) + require.Equal(t, "value"+id, string(res)) + } + + cacheSize := func() (count int) { + conn.WithCustomCollection("cache", func(coll *mgo.Collection) (e error) { + count, e = coll.Find(bson.M{"site": "site"}).Count() + require.NoError(t, e) + return e + }) + return count + } + + init := func() { + lc.Flush(Flusher("site")) + addToCache("key1", "s1", "s2") + addToCache("key2", "s1", "s2", "s3") + addToCache("key3", "s1", "s2", "s3") + addToCache("key4", "s2", "s3") + addToCache("key5", "s2") + addToCache("key6") + addToCache("key7", "s4", "s3") + require.Equal(t, 7, cacheSize(), "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(Flusher("site").Scopes(tt.scopes...)) + assert.Equal(t, tt.left, cacheSize(), "keys size, %s #%d", tt.msg, i) + } +} + +func BenchmarkMongoCache(b *testing.B) { + log.Print("[DEBUG] connect to mongo test instance") + srv, err := mongo.NewServerWithURL(os.Getenv("MONGO_TEST"), 10*time.Second) + assert.Nil(b, err, "failed to dial") + collName := fmt.Sprintf("test_%d", time.Now().Nanosecond()) + conn := mongo.NewConnection(srv, "test", collName) + + data := "" + for i := 0; i < 1000; i++ { + data += "x" + } + lc, err := NewMongoCache(conn) + require.Nil(b, err) + res, err := lc.Get(NewKey("site").ID("key").Scopes("s1", "s2"), func() ([]byte, error) { + return []byte(data), nil + }) + require.Nil(b, err) + require.True(b, strings.HasPrefix(string(res), "xxxx"), string(res)) + + key := NewKey("site").ID("key").Scopes("s1", "s2") + b.ResetTimer() + for i := 0; i < b.N; i++ { + lc.Get(key, func() ([]byte, error) { + b.Fail() + return nil, nil + }) + } +} diff --git a/backend/app/rest/cache/options.go b/backend/app/rest/cache/options.go index 89a9c5a3..f0911eb2 100644 --- a/backend/app/rest/cache/options.go +++ b/backend/app/rest/cache/options.go @@ -1,50 +1,35 @@ package cache -import "github.com/pkg/errors" - // Option func type -type Option func(lc *memoryCache) error +type Option func(lc cacheWithOpts) 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 *memoryCache) error { - lc.maxValueSize = max - if max <= 0 { - return errors.Errorf("negative size for MaxValSize, %d", max) - } - return nil + return func(lc cacheWithOpts) error { + return lc.setMaxValSize(max) } } // 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 *memoryCache) error { - lc.maxKeys = max - if max <= 0 { - return errors.Errorf("negative size for MaxKeys, %d", max) - } - return nil + return func(lc cacheWithOpts) error { + return lc.setMaxKeys(max) } } // MaxCacheSize functional option defines the total size of cached data. // By default it is 0, which means unlimited. func MaxCacheSize(max int64) Option { - return func(lc *memoryCache) error { - lc.maxCacheSize = max - if max <= 0 { - return errors.Errorf("negative size or MaxCacheSize, %d", max) - } - return nil + return func(lc cacheWithOpts) error { + return lc.setMaxCacheSize(max) } } // PostFlushFn functional option defines how callback function called after each Flush. func PostFlushFn(postFlushFn func()) Option { - return func(lc *memoryCache) error { - lc.postFlushFn = postFlushFn - return nil + return func(lc cacheWithOpts) error { + return lc.setPostFlushFn(postFlushFn) } } diff --git a/backend/remark.rest b/backend/remark.rest index 9c8f26dd..6d0e0f97 100644 --- a/backend/remark.rest +++ b/backend/remark.rest @@ -86,9 +86,6 @@ Content-Type: application/json ### list commented posts GET {{host}}/api/v1/list?site={{site}}&limit=10&skip=5 -### get config -GET {{host}}/api/v1/config - ### block user PUT {{host}}/api/v1/admin/user/disqus_grigorybakunov?site={{site}}&block=1 diff --git a/backend/vendor/github.com/go-pkgz/repeater/.gitignore b/backend/vendor/github.com/go-pkgz/repeater/.gitignore new file mode 100644 index 00000000..f1c181ec --- /dev/null +++ b/backend/vendor/github.com/go-pkgz/repeater/.gitignore @@ -0,0 +1,12 @@ +# Binaries for programs and plugins +*.exe +*.exe~ +*.dll +*.so +*.dylib + +# Test binary, build with `go test -c` +*.test + +# Output of the go coverage tool, specifically when used with LiteIDE +*.out diff --git a/backend/vendor/github.com/go-pkgz/repeater/.travis.yml b/backend/vendor/github.com/go-pkgz/repeater/.travis.yml new file mode 100644 index 00000000..baf01e00 --- /dev/null +++ b/backend/vendor/github.com/go-pkgz/repeater/.travis.yml @@ -0,0 +1,18 @@ +language: go + +go: + - "1.10.x" + +go_import_path: github.com/go-pkgz/repeater + +services: mongodb + +before_install: + - go get github.com/mattn/goveralls + - go get gopkg.in/alecthomas/gometalinter.v2 + - $GOPATH/bin/gometalinter.v2 --install + +script: + - go test ./... + - $GOPATH/bin/gometalinter.v2 --exclude=test --exclude=mock --exclude=vendor ./... + - $GOPATH/bin/goveralls -service=travis-ci diff --git a/backend/vendor/github.com/go-pkgz/repeater/LICENSE b/backend/vendor/github.com/go-pkgz/repeater/LICENSE new file mode 100644 index 00000000..ca125214 --- /dev/null +++ b/backend/vendor/github.com/go-pkgz/repeater/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2018 Umputun + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/backend/vendor/github.com/go-pkgz/repeater/README.md b/backend/vendor/github.com/go-pkgz/repeater/README.md new file mode 100644 index 00000000..8b04fc4e --- /dev/null +++ b/backend/vendor/github.com/go-pkgz/repeater/README.md @@ -0,0 +1,36 @@ +# Repeater [![Build Status](https://travis-ci.org/go-pkgz/repeater.svg?branch=master)](https://travis-ci.org/go-pkgz/repeater) [![Go Report Card](https://goreportcard.com/badge/github.com/go-pkgz/repeater)](https://goreportcard.com/report/github.com/go-pkgz/repeater) [![Coverage Status](https://coveralls.io/repos/github/go-pkgz/repeater/badge.svg?branch=master)](https://coveralls.io/github/go-pkgz/repeater?branch=master) + +Repeater calls a function until it returns no error, up to some number of iterations and delays defined by strategy. It terminates immediately on err from the provided (optional) list of critical errors. + +## Install and update + +`go get -u github.com/go-pkgz/repeater` + +## How to use + +New Repeater created by `New(strtg strategy.Interface)` or shortcut for defaults - `NewDefault(repeats int, delay time.Duration) *Repeater`. + +To activate invoke `Do` method. `Do` repeats func until no error returned. Predefined (optional) errors terminates the loop immediately. + +`func (r Repeater) Do(fun func() error, errors ...error) (err error)` + +### Repeating strategy + +User can provide his own strategy implementing the interface: + +```go +type Interface interface { + Start(ctx context.Context) chan struct{} +} +``` + +Returned channels used as "ticks," i.e., for each repeat or initial operation one read from this channel needed. Closing this channel indicates "done with retries." It is pretty much the same idea as `time.Timer` or `time.Tick` implements. Note - the first (technically not-repeated-yet) call won't happen **until something sent to the channel**. For this reason, the typical strategy sends first "tick" before the first wait/sleep. + +Three most common strategies provided by package and ready to use: +1. **Fixed delay**, up to max number of attempts - `NewFixedDelay(repeats int, delay time.Duration)`. +It is the default strategy used by `repeater.NewDefault` constructor +2. **BackOff** with jitter provides exponential backoff. It starts from 100ms interval and goes in steps with `last * math.Pow(factor, attempt)`. Optional jitter randomizes intervals a little bit. The strategy created by `NewBackoff(repeats int, factor float64, jitter bool)`. _Factor = 1 effectively makes this strategy fixed with 100ms delay._ +3. **Once** strategy does not do any repeats and mainly used for tests/mocks - `NewOnce()` + + + diff --git a/backend/vendor/github.com/go-pkgz/repeater/repeater.go b/backend/vendor/github.com/go-pkgz/repeater/repeater.go new file mode 100644 index 00000000..c1721cb3 --- /dev/null +++ b/backend/vendor/github.com/go-pkgz/repeater/repeater.go @@ -0,0 +1,60 @@ +// Package repeater call fun till it returns no error, up to repeat some number of iterations and delays defined by strategy. +// Repeats number and delays defined by strategy.Interface. Terminates immediately on err from +// provided, optional list of critical errors +package repeater + +import ( + "context" + "time" + + "github.com/go-pkgz/repeater/strategy" +) + +// Repeater is the main object, should be made by New or NewDefault, embeds strategy +type Repeater struct { + strategy.Interface +} + +// New repeater with a given strategy. If strategy=nil initializes with FixedDelay 5sec, 10 times. +func New(strtg strategy.Interface) *Repeater { + if strtg == nil { + strtg = strategy.NewFixedDelay(10, time.Second*5) + } + result := Repeater{Interface: strtg} + return &result +} + +// NewDefault makes repeater with FixedDelay strategy +func NewDefault(repeats int, delay time.Duration) *Repeater { + return New(strategy.NewFixedDelay(repeats, delay)) +} + +// Do repeats fun till no error. Predefined (optional) errors terminate immediately +func (r Repeater) Do(fun func() error, errors ...error) (err error) { + + ctx, cancelFunc := context.WithCancel(context.Background()) + defer cancelFunc() // ensure strategy's channel termination + + inErrors := func(err error) bool { + for _, e := range errors { + if e == err { + return true + } + } + return false + } + + ch := r.Start(ctx) // channel of ticks-like events provided by strategy + + // closed channel indicates completion or early termination, set by strategy + for range ch { + + if err = fun(); err == nil { + return nil + } + if err != nil && inErrors(err) { //terminate on critical error from provided list + return err + } + } + return err +} diff --git a/backend/vendor/github.com/go-pkgz/repeater/strategy/backoff.go b/backend/vendor/github.com/go-pkgz/repeater/strategy/backoff.go new file mode 100644 index 00000000..0f5b8013 --- /dev/null +++ b/backend/vendor/github.com/go-pkgz/repeater/strategy/backoff.go @@ -0,0 +1,55 @@ +package strategy + +import ( + "context" + "math" + "math/rand" + "time" +) + +// Backoff implements strategy.Interface for exponential-backoff +// it starts from 100ms and goes in steps with last * math.Pow(factor, attempt) +// optional jitter randomize intervals a little bit. +type Backoff struct { + repeats int + factor float64 + jitter bool +} + +// NewBackoff makes Backoff strategy with given factor and optional jitter +func NewBackoff(repeats int, factor float64, jitter bool) Interface { + if repeats == 0 { + repeats = 1 + } + if factor <= 0 { + factor = 1 + } + result := Backoff{repeats: repeats, factor: factor, jitter: jitter} + return &result +} + +// Start returns channel, similar to time.Timer +// then publishing signals to channel ch for retries attempt. Closed ch indicates "done" event +// consumer (repeater) should stop it explicitly after completion +func (b *Backoff) Start(ctx context.Context) (ch chan struct{}) { + ch = make(chan struct{}) + go func() { + defer close(ch) + rnd := rand.New(rand.NewSource(int64(time.Now().Nanosecond()))) + minDelay := 100 * time.Millisecond // starts 100ms + for i := 0; i < b.repeats; i++ { + select { + case <-ctx.Done(): + return + default: + ch <- struct{}{} + delay := float64(minDelay) * math.Pow(b.factor, float64(i)) + if b.jitter { + delay = rnd.Float64()*(float64(2*minDelay)) + (delay - float64(minDelay)) + } + time.Sleep(time.Duration(delay)) + } + } + }() + return ch +} diff --git a/backend/vendor/github.com/go-pkgz/repeater/strategy/fixed.go b/backend/vendor/github.com/go-pkgz/repeater/strategy/fixed.go new file mode 100644 index 00000000..adcef013 --- /dev/null +++ b/backend/vendor/github.com/go-pkgz/repeater/strategy/fixed.go @@ -0,0 +1,41 @@ +package strategy + +import ( + "context" + "time" +) + +// FixedDelay implements strategy.Interface for fixed intervals up to max repeats +type FixedDelay struct { + repeats int + delay time.Duration +} + +// NewFixedDelay makes a Interface +func NewFixedDelay(repeats int, delay time.Duration) Interface { + if repeats == 0 { + repeats = 1 + } + result := FixedDelay{repeats: repeats, delay: delay} + return &result +} + +// Start returns channel, similar to time.Timer +// then publishing signals to channel ch for retries attempt. +// can be terminated (canceled) via context. +func (s *FixedDelay) Start(ctx context.Context) (ch chan struct{}) { + ch = make(chan struct{}) + go func() { + defer close(ch) + for i := 0; i < s.repeats; i++ { + select { + case <-ctx.Done(): + return + default: + ch <- struct{}{} + time.Sleep(s.delay) + } + } + }() + return ch +} diff --git a/backend/vendor/github.com/go-pkgz/repeater/strategy/strategy.go b/backend/vendor/github.com/go-pkgz/repeater/strategy/strategy.go new file mode 100644 index 00000000..5ebb5ebb --- /dev/null +++ b/backend/vendor/github.com/go-pkgz/repeater/strategy/strategy.go @@ -0,0 +1,28 @@ +// Package strategy defines repeater's strategy and implements some. +// Strategy result is a channel acting like time.Timer ot time.Tick +package strategy + +import "context" + +// Interface for repeater strategy. Returns channel with ticks +type Interface interface { + Start(ctx context.Context) chan struct{} +} + +// Once strategy eliminate repeats and makes a single try only +type Once struct{} + +// NewOnce makes no-repeat strategy +func NewOnce() Interface { + return &Once{} +} + +// Start returns closed channel with a single element to prevent any repeats +func (s *Once) Start(ctx context.Context) (ch chan struct{}) { + ch = make(chan struct{}) + go func() { + ch <- struct{}{} + close(ch) + }() + return ch +}