JakeHillion/storj
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

storage/redis/redisserver: simplify redisserver creation

Browse Source 
Change-Id: I881576a7881db671b5abeeca7120a022987cc47f
This commit is contained in:
Cameron Ayer 2020-01-31 13:28:42 -05:00 committed by Cameron
parent b22bf16b35
commit 33d696b096
7 changed files with 75 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
private/testplanet/satellite.go
View File

@ -256,12 +256,12 @@ func (planet *Planet) newSatellites(count int) ([]*SatelliteSystem, error) {
}
planet.databases = append(planet.databases, pointerDB)
liveAccountingServer := redisserver.NewMini()
addr, _, err := liveAccountingServer.Run()
redis, err := redisserver.Mini()
if err != nil {
return xs, errs.Wrap(err)
return nil, err
}
planet.databases = append(planet.databases, liveAccountingServer)
planet.databases = append(planet.databases, redis)
config := satellite.Config{
Server: server.Config{
@ -382,7 +382,7 @@ func (planet *Planet) newSatellites(count int) ([]*SatelliteSystem, error) {
Interval: defaultInterval,
},
LiveAccounting: live.Config{
StorageBackend: "redis://" + addr + "?db=0",
StorageBackend: "redis://" + redis.Addr() + "?db=0",
},
Mail: mailservice.Config{
SMTPServerAddress: "smtp.mail.test:587",
@ -541,6 +541,8 @@ func createNewSystem(log *zap.Logger, peer *satellite.Core, api *satellite.API,
system.Accounting.ProjectUsage = peer.Accounting.ProjectUsage
system.Accounting.ReportedRollup = peer.Accounting.ReportedRollupChore
system.LiveAccounting = peer.LiveAccounting
system.Marketing.Listener = api.Marketing.Listener
system.Marketing.Endpoint = api.Marketing.Endpoint

6
private/testrevocation/db.go
View File

@ -21,12 +21,12 @@ func RunDBs(t *testing.T, test func(*testing.T, extensions.RevocationDB, storage
ctx := testcontext.New(t)
defer ctx.Cleanup()
addr, cleanup, err := redisserver.Start()
redis, err := redisserver.Mini()
require.NoError(t, err)
defer cleanup()
defer ctx.Check(redis.Close)
// Test using redis-backed revocation DB
dbURL := "redis://" + addr + "?db=0"
dbURL := "redis://" + redis.Addr() + "?db=0"
db, err := revocation.NewDB(dbURL)
require.NoError(t, err)
defer ctx.Check(db.Close)

22
satellite/accounting/live/live_test.go
View File

@ -32,15 +32,15 @@ func TestLiveAccountingCache(t *testing.T) {
ctx := testcontext.New(t)
defer ctx.Cleanup()
address, cleanup, err := redisserver.Start()
redis, err := redisserver.Mini()
require.NoError(t, err)
defer cleanup()
defer ctx.Check(redis.Close)
for _, tt := range tests {
var config live.Config
if tt.backend == "redis" {
config = live.Config{
StorageBackend: "redis://" + address + "?db=0",
StorageBackend: "redis://" + redis.Addr() + "?db=0",
}
}
@ -75,12 +75,12 @@ func TestRedisCacheConcurrency(t *testing.T) {
ctx := testcontext.New(t)
defer ctx.Cleanup()
address, cleanup, err := redisserver.Start()
redis, err := redisserver.Mini()
require.NoError(t, err)
defer cleanup()
defer ctx.Check(redis.Close)
config := live.Config{
StorageBackend: "redis://" + address + "?db=0",
StorageBackend: "redis://" + redis.Addr() + "?db=0",
}
cache, err := live.NewCache(zaptest.NewLogger(t).Named("live-accounting"), config)
require.NoError(t, err)
@ -109,9 +109,9 @@ func TestRedisCacheConcurrency(t *testing.T) {
func populateCache(ctx context.Context, cache accounting.Cache) (projectIDs []uuid.UUID, sum int64, _ error) {
const (
valuesListSize = 1000
valuesListSize = 10
valueMultiplier = 4096
numProjects = 200
numProjects = 100
)
// make a largish list of varying values
someValues := make([]int64, valuesListSize)
@ -162,15 +162,15 @@ func TestGetAllProjectTotals(t *testing.T) {
ctx := testcontext.New(t)
defer ctx.Cleanup()
address, cleanup, err := redisserver.Start()
redis, err := redisserver.Mini()
require.NoError(t, err)
defer cleanup()
defer ctx.Check(redis.Close)
for _, tt := range tests {
var config live.Config
if tt.backend == "redis" {
config = live.Config{
StorageBackend: "redis://" + address + "?db=0",
StorageBackend: "redis://" + redis.Addr() + "?db=0",
}
}

7
satellite/console/consoleweb/consoleql/mutation_test.go
View File

@ -67,12 +67,11 @@ func TestGrapqhlMutation(t *testing.T) {
)
require.NoError(t, err)
miniredis := redisserver.NewMini()
addr, cleanup, err := miniredis.Run()
redis, err := redisserver.Mini()
require.NoError(t, err)
defer cleanup()
defer ctx.Check(redis.Close)
cache, err := live.NewCache(log.Named("cache"), live.Config{StorageBackend: "redis://" + addr + "?db=0"})
cache, err := live.NewCache(log.Named("cache"), live.Config{StorageBackend: "redis://" + redis.Addr() + "?db=0"})
require.NoError(t, err)
projectUsage := accounting.NewService(db.ProjectAccounting(), cache, 0)

7
satellite/console/consoleweb/consoleql/query_test.go
View File

@ -52,12 +52,11 @@ func TestGraphqlQuery(t *testing.T) {
)
require.NoError(t, err)
miniredis := redisserver.NewMini()
addr, cleanup, err := miniredis.Run()
redis, err := redisserver.Mini()
require.NoError(t, err)
defer cleanup()
defer ctx.Check(redis.Close)
cache, err := live.NewCache(log.Named("cache"), live.Config{StorageBackend: "redis://" + addr + "?db=0"})
cache, err := live.NewCache(log.Named("cache"), live.Config{StorageBackend: "redis://" + redis.Addr() + "?db=0"})
require.NoError(t, err)
projectUsage := accounting.NewService(db.ProjectAccounting(), cache, 0)

14
storage/redis/client_test.go
View File

@ -6,18 +6,20 @@ package redis
import (
"testing"
"github.com/stretchr/testify/require"
"storj.io/storj/storage/redis/redisserver"
"storj.io/storj/storage/testsuite"
)
func TestSuite(t *testing.T) {
addr, cleanup, err := redisserver.Start()
redis, err := redisserver.Start()
if err != nil {
t.Fatal(err)
}
defer cleanup()
defer func() { require.NoError(t, redis.Close()) }()
client, err := NewClient(addr, "", 1)
client, err := NewClient(redis.Addr(), "", 1)
if err != nil {
t.Fatal(err)
}
@ -34,13 +36,13 @@ func TestInvalidConnection(t *testing.T) {
}
func BenchmarkSuite(b *testing.B) {
addr, cleanup, err := redisserver.Start()
redis, err := redisserver.Start()
if err != nil {
b.Fatal(err)
}
defer cleanup()
defer func() { require.NoError(b, redis.Close()) }()
client, err := NewClient(addr, "", 1)
client, err := NewClient(redis.Addr(), "", 1)
if err != nil {
b.Fatal(err)
}

77
storage/redis/redisserver/server.go
View File

@ -30,9 +30,10 @@ const (
fallbackPort = 6379
)
// Mini is a wrapper for *miniredis.MiniRedis which implements the io.Closer interface.
type Mini struct {
server *miniredis.Miniredis
// Server represents a redis server.
type Server interface {
Addr() string
Close() error
}
func freeport() (addr string, port int) {
@ -50,26 +51,24 @@ func freeport() (addr string, port int) {
}
// Start starts a redis-server when available, otherwise falls back to miniredis
func Start() (addr string, cleanup func(), err error) {
addr, cleanup, err = Process()
func Start() (Server, error) {
server, err := Process()
if err != nil {
log.Println("failed to start redis-server: ", err)
mini := NewMini()
return mini.Run()
return Mini()
}
return addr, cleanup, err
return server, err
}
// Process starts a redis-server test process
func Process() (addr string, cleanup func(), err error) {
func Process() (Server, error) {
tmpdir, err := ioutil.TempDir("", "storj-redis")
if err != nil {
return "", nil, err
return nil, err
}
// find a suitable port for listening
var port int
addr, port = freeport()
addr, port := freeport()
// write a configuration file, because redis doesn't support flags
confpath := filepath.Join(tmpdir, "test.conf")
@ -86,7 +85,7 @@ func Process() (addr string, cleanup func(), err error) {
conf := strings.Join(arguments, "\n") + "\n"
err = ioutil.WriteFile(confpath, []byte(conf), 0755)
if err != nil {
return "", nil, err
return nil, err
}
// start the process
@ -95,15 +94,15 @@ func Process() (addr string, cleanup func(), err error) {
read, write, err := os.Pipe()
if err != nil {
return "", nil, err
return nil, err
}
cmd.Stdout = write
if err := cmd.Start(); err != nil {
return "", nil, err
return nil, err
}
cleanup = func() {
cleanup := func() {
processgroup.Kill(cmd)
_ = os.RemoveAll(tmpdir)
}
@ -129,48 +128,52 @@ func Process() (addr string, cleanup func(), err error) {
case err := <-waitForReady:
if err != nil {
cleanup()
return "", nil, err
return nil, err
}
case <-time.After(3 * time.Second):
cleanup()
return "", nil, errors.New("redis timeout")
return nil, errors.New("redis timeout")
}
// test whether we can actually connect
if err := pingServer(addr); err != nil {
cleanup()
return "", nil, fmt.Errorf("unable to ping: %v", err)
return nil, fmt.Errorf("unable to ping: %v", err)
}
return addr, cleanup, nil
return &process{addr, cleanup}, nil
}
type process struct {
addr string
close func()
}
func (process *process) Addr() string { return process.addr }
func (process *process) Close() error { process.close(); return nil }
func pingServer(addr string) error {
client := redis.NewClient(&redis.Options{Addr: addr, DB: 1})
defer func() { _ = client.Close() }()
return client.Ping().Err()
}
// NewMini creates a new Mini.
func NewMini() *Mini {
return &Mini{
server: miniredis.NewMiniRedis(),
}
}
// Run starts the miniredis server.
func (mini *Mini) Run() (addr string, cleanup func(), err error) {
err = mini.server.Start()
// Mini starts miniredis server
func Mini() (Server, error) {
server, err := miniredis.Run()
if err != nil {
return "", nil, err
return nil, err
}
return mini.server.Addr(), func() {
mini.server.Close()
}, nil
return &miniserver{server}, nil
}
// Close closes the miniredis server.
func (mini *Mini) Close() error {
mini.server.Close()
type miniserver struct {
*miniredis.Miniredis
}
// Close closes the underlying miniredis server
func (s *miniserver) Close() error {
s.Miniredis.Close()
return nil
}