certusone
/
wormhole-explorer
mirror of https://github.com/certusone/wormhole-explorer.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Add cache to protocols stats endpoints (/protocols/stats) (#1149) 

* add cache to protocols stats endpoint

 add mock for cache

add unit test case for cache miss

 change from to current

* add configs

* add missing config for api-service.yaml

* add cache ttl for staging-mainnet
This commit is contained in:
Mariano 2024-03-04 15:55:53 -03:00 committed by GitHub
parent 54cf6c7e54
commit 6bbfa7bf23
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 188 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

58
api/handlers/protocols/service.go
View File

@ -2,15 +2,22 @@ package protocols
import (
"context"
"encoding/json"
"github.com/wormhole-foundation/wormhole-explorer/common/client/cache"
"go.uber.org/zap"
"strconv"
"strings"
"sync"
"time"
)
type Service struct {
Protocols []string
repo *Repository
logger *zap.Logger
Protocols []string
repo *Repository
logger *zap.Logger
cache cache.Cache
cacheKeyPrefix string
cacheTTL int
}
type ProtocolTotalValuesDTO struct {
@ -24,11 +31,14 @@ type ProtocolTotalValuesDTO struct {
Error string `json:"error,omitempty"`
}
func NewService(protocols []string, repo *Repository, logger *zap.Logger) *Service {
func NewService(protocols []string, repo *Repository, logger *zap.Logger, cache cache.Cache, cacheKeyPrefix string, cacheTTL int) *Service {
return &Service{
Protocols: protocols,
repo: repo,
logger: logger,
Protocols: protocols,
repo: repo,
logger: logger,
cache: cache,
cacheKeyPrefix: cacheKeyPrefix,
cacheTTL: cacheTTL,
}
}
@ -51,36 +61,48 @@ func (s *Service) GetProtocolsTotalValues(ctx context.Context) []ProtocolTotalVa
return resultsSlice
}
func (s *Service) getProtocolTotalValues(ctx context.Context, wg *sync.WaitGroup, contributor string, results chan<- ProtocolTotalValuesDTO) {
func (s *Service) getProtocolTotalValues(ctx context.Context, wg *sync.WaitGroup, protocol string, results chan<- ProtocolTotalValuesDTO) {
defer wg.Done()
cacheKey := s.cacheKeyPrefix + ":" + strings.ToUpper(protocol)
cachedValue, errCache := s.cache.Get(ctx, cacheKey)
if errCache == nil {
var val ProtocolTotalValuesDTO
errCacheUnmarshall := json.Unmarshal([]byte(cachedValue), &val)
if errCacheUnmarshall == nil {
results <- val
return
}
s.logger.Error("error unmarshalling cache value", zap.Error(errCacheUnmarshall), zap.String("cache_key", cacheKey))
}
type statsResult struct {
result stats
Err error
}
statsRes := make(chan statsResult, 1)
go func() {
rowStats, errStats := s.repo.getProtocolStats(ctx, contributor)
rowStats, errStats := s.repo.getProtocolStats(ctx, protocol)
statsRes <- statsResult{result: rowStats, Err: errStats}
close(statsRes)
}()
activity, err := s.repo.getProtocolActivity(ctx, contributor)
activity, err := s.repo.getProtocolActivity(ctx, protocol)
if err != nil {
s.logger.Error("error fetching protocol activity", zap.Error(err), zap.String("protocol", contributor))
results <- ProtocolTotalValuesDTO{Protocol: contributor, Error: err.Error()}
s.logger.Error("error fetching protocol activity", zap.Error(err), zap.String("protocol", protocol))
results <- ProtocolTotalValuesDTO{Protocol: protocol, Error: err.Error()}
return
}
rStats := <-statsRes
if rStats.Err != nil {
s.logger.Error("error fetching protocol stats", zap.Error(rStats.Err), zap.String("protocol", contributor))
results <- ProtocolTotalValuesDTO{Protocol: contributor, Error: rStats.Err.Error()}
s.logger.Error("error fetching protocol stats", zap.Error(rStats.Err), zap.String("protocol", protocol))
results <- ProtocolTotalValuesDTO{Protocol: protocol, Error: rStats.Err.Error()}
return
}
dto := ProtocolTotalValuesDTO{
Protocol: contributor,
Protocol: protocol,
TotalValueLocked: rStats.result.Latest.TotalValueLocked,
TotalMessages: rStats.result.Latest.TotalMessages,
TotalValueTransferred: activity.TotalValueTransferred,
@ -95,5 +117,11 @@ func (s *Service) getProtocolTotalValues(ctx context.Context, wg *sync.WaitGroup
dto.LastDayDiffPercentage = strconv.FormatFloat(float64(last24HrMessages)/float64(totalMessagesAsFromLast24hr)*100, 'f', 2, 64) + "%"
}
dtoJson, _ := json.Marshal(dto) // don't handle error since the full lifecycle of the dto is under this scope
errCache = s.cache.Set(ctx, cacheKey, string(dtoJson), time.Duration(s.cacheTTL)*time.Minute)
if errCache != nil {
s.logger.Error("error setting cache", zap.Error(errCache), zap.String("cache_key", cacheKey))
}
results <- dto
}

95
api/handlers/protocols/service_test.go
View File

@ -9,9 +9,12 @@ import (
"github.com/stretchr/testify/assert"
"github.com/test-go/testify/mock"
"github.com/wormhole-foundation/wormhole-explorer/api/handlers/protocols"
"github.com/wormhole-foundation/wormhole-explorer/common/client/cache"
cacheMock "github.com/wormhole-foundation/wormhole-explorer/common/client/cache/mock"
"github.com/wormhole-foundation/wormhole-explorer/common/dbconsts"
"go.uber.org/zap"
"testing"
"time"
)
func TestService_GetProtocolsTotalValues(t *testing.T) {
@ -56,7 +59,7 @@ func TestService_GetProtocolsTotalValues(t *testing.T) {
queryAPI.On("Query", ctx, activityQuery).Return(respActivityLast, nil)
repository := protocols.NewRepository(queryAPI, "protocols_bucket", "protocols_bucket", "v1", "v1", zap.NewNop())
service := protocols.NewService([]string{"protocol1"}, repository, zap.NewNop())
service := protocols.NewService([]string{"protocol1"}, repository, zap.NewNop(), cache.NewDummyCacheClient(), "WORMSCAN:PROTOCOLS", 0)
values := service.GetProtocolsTotalValues(ctx)
assert.Equal(t, 1, len(values))
@ -101,7 +104,7 @@ func TestService_GetProtocolsTotalValues_FailedFetchingActivity(t *testing.T) {
queryAPI.On("Query", ctx, activityQuery).Return(&api.QueryTableResult{}, errors.New("mocked_fetching_activity_error"))
repository := protocols.NewRepository(queryAPI, "protocols_bucket", "protocols_bucket", "v1", "v1", zap.NewNop())
service := protocols.NewService([]string{"protocol1"}, repository, zap.NewNop())
service := protocols.NewService([]string{"protocol1"}, repository, zap.NewNop(), cache.NewDummyCacheClient(), "WORMSCAN:PROTOCOLS", 0)
values := service.GetProtocolsTotalValues(ctx)
assert.Equal(t, 1, len(values))
@ -143,7 +146,7 @@ func TestService_GetProtocolsTotalValues_FailedFetchingStats(t *testing.T) {
queryAPI.On("Query", ctx, activityQuery).Return(respActivityLast, errNil)
repository := protocols.NewRepository(queryAPI, "protocols_bucket", "protocols_bucket", "v1", "v1", zap.NewNop())
service := protocols.NewService([]string{"protocol1"}, repository, zap.NewNop())
service := protocols.NewService([]string{"protocol1"}, repository, zap.NewNop(), cache.NewDummyCacheClient(), "WORMSCAN:PROTOCOLS", 0)
values := service.GetProtocolsTotalValues(ctx)
assert.Equal(t, 1, len(values))
@ -152,6 +155,92 @@ func TestService_GetProtocolsTotalValues_FailedFetchingStats(t *testing.T) {
assert.Equal(t, "mocked_fetching_stats_error", values[0].Error)
}
func TestService_GetProtocolsTotalValues_CacheHit(t *testing.T) {
ctx := context.Background()
mockCache := &cacheMock.CacheMock{}
var cacheErr error
cacheErr = nil
mockCache.On("Get", ctx, "WORMSCAN:PROTOCOLS:PROTOCOL1").Return(`{"protocol":"protocol1","total_messages":7,"total_value_locked":5,"total_value_secured":9,"total_value_transferred":7,"last_day_messages":4,"last_day_diff_percentage":"75.00%"}`, cacheErr)
service := protocols.NewService([]string{"protocol1"}, nil, zap.NewNop(), mockCache, "WORMSCAN:PROTOCOLS", 0)
values := service.GetProtocolsTotalValues(ctx)
assert.Equal(t, 1, len(values))
assert.Equal(t, "protocol1", values[0].Protocol)
assert.Equal(t, 5.00, values[0].TotalValueLocked)
assert.Equal(t, uint64(7), values[0].TotalMessages)
assert.Equal(t, 9.00, values[0].TotalValueSecured)
assert.Equal(t, 7.00, values[0].TotalValueTransferred)
assert.Equal(t, uint64(4), values[0].LastDayMessages)
assert.Equal(t, "75.00%", values[0].LastDayDiffPercentage)
}
func TestService_GetProtocolsTotalValues_CacheMiss_FetchAndUpdate(t *testing.T) {
ctx := context.Background()
mockCache := &cacheMock.CacheMock{}
mockCache.On("Get", ctx, "WORMSCAN:PROTOCOLS:PROTOCOL1").Return("", cache.ErrNotFound) // mock cache miss
// mock cache update, validate it's called once.
mockCache.On("Set",
ctx,
"WORMSCAN:PROTOCOLS:PROTOCOL1",
`{"protocol":"protocol1","total_messages":7,"total_value_locked":5,"total_value_secured":9,"total_value_transferred":7,"last_day_messages":3,"last_day_diff_percentage":"75.00%"}`,
time.Duration(60)*time.Minute).
Return(nil).
Times(1)
var errNil error
respStatsLatest := &mockQueryTableResult{}
respStatsLatest.On("Next").Return(true)
respStatsLatest.On("Err").Return(errNil)
respStatsLatest.On("Close").Return(errNil)
respStatsLatest.On("Record").Return(query.NewFluxRecord(1, map[string]interface{}{
"protocol": "protocol1",
"total_messages": uint64(7),
"total_value_locked": float64(5),
}))
respStatsLastDay := &mockQueryTableResult{}
respStatsLastDay.On("Next").Return(true)
respStatsLastDay.On("Err").Return(errNil)
respStatsLastDay.On("Close").Return(errNil)
respStatsLastDay.On("Record").Return(query.NewFluxRecord(1, map[string]interface{}{
"protocol": "protocol1",
"total_messages": uint64(4),
"total_value_locked": float64(5),
}))
respActivityLast := &mockQueryTableResult{}
respActivityLast.On("Next").Return(true)
respActivityLast.On("Err").Return(errNil)
respActivityLast.On("Close").Return(errNil)
respActivityLast.On("Record").Return(query.NewFluxRecord(1, map[string]interface{}{
"protocol": "protocol1",
"total_messages": uint64(15),
"total_value_transferred": float64(7),
"total_value_secure": float64(9),
}))
queryAPI := &mockQueryAPI{}
queryAPI.On("Query", ctx, fmt.Sprintf(protocols.QueryTemplateLatestPoint, "protocols_bucket", dbconsts.ProtocolsStatsMeasurement, "protocol1", "v1")).Return(respStatsLatest, nil)
queryAPI.On("Query", ctx, fmt.Sprintf(protocols.QueryTemplateLast24Point, "protocols_bucket", dbconsts.ProtocolsStatsMeasurement, "protocol1", "v1")).Return(respStatsLastDay, nil)
activityQuery := fmt.Sprintf(protocols.QueryTemplateActivityLatestPoint, "protocols_bucket", dbconsts.ProtocolsActivityMeasurement, "protocol1", "v1")
queryAPI.On("Query", ctx, activityQuery).Return(respActivityLast, nil)
repository := protocols.NewRepository(queryAPI, "protocols_bucket", "protocols_bucket", "v1", "v1", zap.NewNop())
service := protocols.NewService([]string{"protocol1"}, repository, zap.NewNop(), mockCache, "WORMSCAN:PROTOCOLS", 60)
values := service.GetProtocolsTotalValues(ctx)
assert.Equal(t, 1, len(values))
assert.Equal(t, "protocol1", values[0].Protocol)
assert.Equal(t, 5.00, values[0].TotalValueLocked)
assert.Equal(t, uint64(7), values[0].TotalMessages)
assert.Equal(t, 9.00, values[0].TotalValueSecured)
assert.Equal(t, 7.00, values[0].TotalValueTransferred)
assert.Equal(t, uint64(3), values[0].LastDayMessages)
assert.Equal(t, "75.00%", values[0].LastDayDiffPercentage)
}
type mockQueryAPI struct {
mock.Mock
}

28
api/internal/config/config.go
View File

@ -32,12 +32,14 @@ type AppConfig struct {
Name string
}
Cache struct {
URL string
TvlKey string
TvlExpiration int
Enabled bool
MetricExpiration int
Prefix string
URL string
TvlKey string
TvlExpiration int
Enabled bool
MetricExpiration int
Prefix string
ProtocolsStatsKey string
ProtocolsStatsExpiration int
}
PORT int
LogLevel string
@ -80,12 +82,14 @@ func (cfg *AppConfig) GetLogLevel() (ipfslog.LogLevel, error) {
func defaulConfig() *AppConfig {
return &AppConfig{
Cache: struct {
URL string
TvlKey string
TvlExpiration int
Enabled bool
MetricExpiration int
Prefix string
URL string
TvlKey string
TvlExpiration int
Enabled bool
MetricExpiration int
Prefix string
ProtocolsStatsKey string
ProtocolsStatsExpiration int
}{
MetricExpiration: 10,
},

2
api/main.go
View File

@ -179,7 +179,7 @@ func main() {
relaysService := relays.NewService(relaysRepo, rootLogger)
operationsService := operations.NewService(operationsRepo, rootLogger)
statsService := stats.NewService(statsRepo, cache, expirationTime, metrics, rootLogger)
protocolsService := protocols.NewService(cfg.Protocols, protocolsRepo, rootLogger)
protocolsService := protocols.NewService(cfg.Protocols, protocolsRepo, rootLogger, cache, cfg.Cache.ProtocolsStatsKey, cfg.Cache.ProtocolsStatsExpiration)
// Set up a custom error handler
response.SetEnableStackTrace(*cfg)

26
common/client/cache/mock/mock.go vendored Normal file
View File

@ -0,0 +1,26 @@
package mock
import (
"context"
"github.com/test-go/testify/mock"
"time"
)
// CacheMock exported type to provide mock for cache.Cache interface
type CacheMock struct {
mock.Mock
}
func (c *CacheMock) Get(ctx context.Context, key string) (string, error) {
args := c.Called(ctx, key)
return args.String(0), args.Error(1)
}
func (c *CacheMock) Close() error {
return nil
}
func (c *CacheMock) Set(ctx context.Context, key string, value interface{}, expirations time.Duration) error {
args := c.Called(ctx, key, value, expirations)
return args.Error(0)
}

4
deploy/api/api-service.yaml
View File

@ -151,6 +151,10 @@ spec:
key: protocols-activity-version
- name: WORMSCAN_PROTOCOLS
value: {{ .WORMSCAN_PROTOCOLS }}
- name: WORMSCAN_CACHE_PROTOCOLSSTATSEXPIRATION
value: "{{ .WORMSCAN_CACHE_PROTOCOLSSTATSEXPIRATION }}"
- name: WORMSCAN_CACHE_PROTOCOLSSTATSKEY
value: "WORMSCAN:PROTOCOLS_STATS"
resources:
limits:
memory: {{ .RESOURCES_LIMITS_MEMORY }}

1
deploy/api/env/production-mainnet.env vendored
View File

@ -21,3 +21,4 @@ WORMSCAN_VAAPAYLOADPARSER_URL=
WORMSCAN_VAAPAYLOADPARSER_TIMEOUT=10
WORMSCAN_VAAPAYLOADPARSER_ENABLED=true
WORMSCAN_PROTOCOLS=
WORMSCAN_CACHE_PROTOCOLSSTATSEXPIRATION=60

1
deploy/api/env/production-testnet.env vendored
View File

@ -21,3 +21,4 @@ WORMSCAN_VAAPAYLOADPARSER_URL=
WORMSCAN_VAAPAYLOADPARSER_TIMEOUT=10
WORMSCAN_VAAPAYLOADPARSER_ENABLED=true
WORMSCAN_PROTOCOLS=
WORMSCAN_CACHE_PROTOCOLSSTATSEXPIRATION=60

1
deploy/api/env/staging-mainnet.env vendored
View File

@ -21,3 +21,4 @@ WORMSCAN_VAAPAYLOADPARSER_URL=
WORMSCAN_VAAPAYLOADPARSER_TIMEOUT=10
WORMSCAN_VAAPAYLOADPARSER_ENABLED=true
WORMSCAN_PROTOCOLS=allbridge,mayan
WORMSCAN_CACHE_PROTOCOLSSTATSEXPIRATION=60

1
deploy/api/env/staging-testnet.env vendored
View File

@ -21,3 +21,4 @@ WORMSCAN_VAAPAYLOADPARSER_URL=
WORMSCAN_VAAPAYLOADPARSER_TIMEOUT=10
WORMSCAN_VAAPAYLOADPARSER_ENABLED=true
WORMSCAN_PROTOCOLS=
WORMSCAN_CACHE_PROTOCOLSSTATSEXPIRATION=60

4
jobs/jobs/protocols/activity/activity.go
View File

@ -30,7 +30,7 @@ func (m *ProtocolsActivityJob) Run(ctx context.Context) error {
wg.Add(clientsQty)
errs := make(chan error, clientsQty)
ts := time.Now().UTC().Truncate(time.Hour) // make minutes and seconds zero, so we only work with date and hour
from := ts.Add(-1 * time.Hour)
from := time.Unix(0, 0).UTC()
m.logger.Info("running protocols activity job ", zap.Time("from", from), zap.Time("to", ts))
for _, cs := range m.activityFetchers {
go func(c ClientActivity) {
@ -40,7 +40,7 @@ func (m *ProtocolsActivityJob) Run(ctx context.Context) error {
errs <- err
return
}
errs <- m.updateActivity(ctx, c.ProtocolName(), m.version, activity, from)
errs <- m.updateActivity(ctx, c.ProtocolName(), m.version, activity, ts)
}(cs)
}