-
Notifications
You must be signed in to change notification settings - Fork 402
/
server.go
205 lines (173 loc) · 4.27 KB
/
server.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
// Copyright (C) 2019 Storj Labs, Inc.
// See LICENSE for copying information.
// Package testredis is package for starting a redis test server
package testredis
import (
"bufio"
"context"
"errors"
"fmt"
"io"
"io/ioutil"
"log"
"net"
"os"
"os/exec"
"path/filepath"
"runtime/pprof"
"strconv"
"strings"
"time"
"github.com/alicebob/miniredis/v2"
"github.com/go-redis/redis/v8"
"storj.io/common/processgroup"
)
const (
fallbackAddr = "localhost:6379"
fallbackPort = 6379
)
// Server represents a redis server.
type Server interface {
Addr() string
Close() error
// FastForward is a function for enforce the TTL of keys in
// implementations what they have not exercise the expiration by themselves
// (e.g. Minitredis). This method is a no-op in implementations which support
// the expiration as usual.
//
// All the keys whose TTL minus d become <= 0 will be removed.
FastForward(d time.Duration)
}
func freeport() (addr string, port int) {
listener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
return fallbackAddr, fallbackPort
}
netaddr := listener.Addr().(*net.TCPAddr)
addr = netaddr.String()
port = netaddr.Port
_ = listener.Close()
time.Sleep(time.Second)
return addr, port
}
// Start starts a redis-server when available, otherwise falls back to miniredis.
func Start(ctx context.Context) (Server, error) {
server, err := Process(ctx)
if err != nil {
log.Println("failed to start redis-server: ", err)
return Mini(ctx)
}
return server, err
}
// Process starts a redis-server test process.
func Process(ctx context.Context) (Server, error) {
tmpdir, err := ioutil.TempDir("", "storj-redis")
if err != nil {
return nil, err
}
// find a suitable port for listening
addr, port := freeport()
// write a configuration file, because redis doesn't support flags
confpath := filepath.Join(tmpdir, "test.conf")
arguments := []string{
"daemonize no",
"bind 127.0.0.1",
"port " + strconv.Itoa(port),
"timeout 0",
"databases 2",
"dbfilename dump.rdb",
"dir " + tmpdir,
}
conf := strings.Join(arguments, "\n") + "\n"
err = ioutil.WriteFile(confpath, []byte(conf), 0755)
if err != nil {
return nil, err
}
// start the process
cmd := exec.Command("redis-server", confpath)
processgroup.Setup(cmd)
read, write, err := os.Pipe()
if err != nil {
return nil, err
}
cmd.Stdout = write
if err := cmd.Start(); err != nil {
return nil, err
}
cleanup := func() {
processgroup.Kill(cmd)
_ = os.RemoveAll(tmpdir)
}
// wait for redis to become ready
waitForReady := make(chan error, 1)
go func() {
// wait for the message that looks like
// v3 "The server is now ready to accept connections on port 6379"
// v4 "Ready to accept connections"
scanner := bufio.NewScanner(read)
for scanner.Scan() {
line := scanner.Text()
if strings.Contains(line, "to accept") {
break
}
}
waitForReady <- scanner.Err()
_, _ = io.Copy(ioutil.Discard, read)
}()
select {
case err := <-waitForReady:
if err != nil {
cleanup()
return nil, err
}
case <-time.After(3 * time.Second):
cleanup()
return nil, errors.New("redis timeout")
}
// test whether we can actually connect
if err := pingServer(ctx, addr); err != nil {
cleanup()
return nil, fmt.Errorf("unable to ping: %w", err)
}
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 (process *process) FastForward(_ time.Duration) {}
func pingServer(ctx context.Context, addr string) error {
client := redis.NewClient(&redis.Options{Addr: addr, DB: 1})
defer func() { _ = client.Close() }()
return client.Ping(ctx).Err()
}
// Mini starts miniredis server.
func Mini(ctx context.Context) (Server, error) {
var server *miniredis.Miniredis
var err error
pprof.Do(ctx, pprof.Labels("db", "miniredis"), func(ctx context.Context) {
server, err = miniredis.Run()
})
if err != nil {
return nil, err
}
return &miniserver{server}, nil
}
type miniserver struct {
*miniredis.Miniredis
}
// Close closes the underlying miniredis server.
func (s *miniserver) Close() error {
s.Miniredis.Close()
return nil
}
func (s *miniserver) FastForward(d time.Duration) {
s.Miniredis.FastForward(d)
}