7816: CopyReaderToLog renamed to ReadWriteLines. Use Writer instead of Logger
[arvados.git] / services / keepstore / keepstore.go
1 package main
2
3 import (
4         "bytes"
5         "flag"
6         "fmt"
7         "git.curoverse.com/arvados.git/sdk/go/keepclient"
8         "io/ioutil"
9         "log"
10         "net"
11         "net/http"
12         "os"
13         "os/signal"
14         "strings"
15         "syscall"
16         "time"
17 )
18
19 // ======================
20 // Configuration settings
21 //
22 // TODO(twp): make all of these configurable via command line flags
23 // and/or configuration file settings.
24
25 // Default TCP address on which to listen for requests.
26 // Initialized by the --listen flag.
27 const DefaultAddr = ":25107"
28
29 // A Keep "block" is 64MB.
30 const BlockSize = 64 * 1024 * 1024
31
32 // A Keep volume must have at least MinFreeKilobytes available
33 // in order to permit writes.
34 const MinFreeKilobytes = BlockSize / 1024
35
36 // ProcMounts /proc/mounts
37 var ProcMounts = "/proc/mounts"
38
39 // enforcePermissions controls whether permission signatures
40 // should be enforced (affecting GET and DELETE requests).
41 // Initialized by the -enforce-permissions flag.
42 var enforcePermissions bool
43
44 // blobSignatureTTL is the time duration for which new permission
45 // signatures (returned by PUT requests) will be valid.
46 // Initialized by the -permission-ttl flag.
47 var blobSignatureTTL time.Duration
48
49 // dataManagerToken represents the API token used by the
50 // Data Manager, and is required on certain privileged operations.
51 // Initialized by the -data-manager-token-file flag.
52 var dataManagerToken string
53
54 // neverDelete can be used to prevent the DELETE handler from
55 // actually deleting anything.
56 var neverDelete = true
57
58 var maxBuffers = 128
59 var bufs *bufferPool
60
61 // KeepError types.
62 //
63 type KeepError struct {
64         HTTPCode int
65         ErrMsg   string
66 }
67
68 var (
69         BadRequestError     = &KeepError{400, "Bad Request"}
70         UnauthorizedError   = &KeepError{401, "Unauthorized"}
71         CollisionError      = &KeepError{500, "Collision"}
72         RequestHashError    = &KeepError{422, "Hash mismatch in request"}
73         PermissionError     = &KeepError{403, "Forbidden"}
74         DiskHashError       = &KeepError{500, "Hash mismatch in stored data"}
75         ExpiredError        = &KeepError{401, "Expired permission signature"}
76         NotFoundError       = &KeepError{404, "Not Found"}
77         GenericError        = &KeepError{500, "Fail"}
78         FullError           = &KeepError{503, "Full"}
79         SizeRequiredError   = &KeepError{411, "Missing Content-Length"}
80         TooLongError        = &KeepError{413, "Block is too large"}
81         MethodDisabledError = &KeepError{405, "Method disabled"}
82 )
83
84 func (e *KeepError) Error() string {
85         return e.ErrMsg
86 }
87
88 // ========================
89 // Internal data structures
90 //
91 // These global variables are used by multiple parts of the
92 // program. They are good candidates for moving into their own
93 // packages.
94
95 // The Keep VolumeManager maintains a list of available volumes.
96 // Initialized by the --volumes flag (or by FindKeepVolumes).
97 var KeepVM VolumeManager
98
99 // The pull list manager and trash queue are threadsafe queues which
100 // support atomic update operations. The PullHandler and TrashHandler
101 // store results from Data Manager /pull and /trash requests here.
102 //
103 // See the Keep and Data Manager design documents for more details:
104 // https://arvados.org/projects/arvados/wiki/Keep_Design_Doc
105 // https://arvados.org/projects/arvados/wiki/Data_Manager_Design_Doc
106 //
107 var pullq *WorkQueue
108 var trashq *WorkQueue
109
110 type volumeSet []Volume
111
112 var (
113         flagSerializeIO bool
114         flagReadonly    bool
115         volumes         volumeSet
116 )
117
118 func (vs *volumeSet) String() string {
119         return fmt.Sprintf("%+v", (*vs)[:])
120 }
121
122 // TODO(twp): continue moving as much code as possible out of main
123 // so it can be effectively tested. Esp. handling and postprocessing
124 // of command line flags (identifying Keep volumes and initializing
125 // permission arguments).
126
127 func main() {
128         log.Println("keepstore starting, pid", os.Getpid())
129         defer log.Println("keepstore exiting, pid", os.Getpid())
130
131         var (
132                 dataManagerTokenFile string
133                 listen               string
134                 blobSigningKeyFile   string
135                 permissionTTLSec     int
136                 pidfile              string
137         )
138         flag.StringVar(
139                 &dataManagerTokenFile,
140                 "data-manager-token-file",
141                 "",
142                 "File with the API token used by the Data Manager. All DELETE "+
143                         "requests or GET /index requests must carry this token.")
144         flag.BoolVar(
145                 &enforcePermissions,
146                 "enforce-permissions",
147                 false,
148                 "Enforce permission signatures on requests.")
149         flag.StringVar(
150                 &listen,
151                 "listen",
152                 DefaultAddr,
153                 "Listening address, in the form \"host:port\". e.g., 10.0.1.24:8000. Omit the host part to listen on all interfaces.")
154         flag.BoolVar(
155                 &neverDelete,
156                 "never-delete",
157                 true,
158                 "If true, nothing will be deleted. "+
159                         "Warning: the relevant features in keepstore and data manager have not been extensively tested. "+
160                         "You should leave this option alone unless you can afford to lose data.")
161         flag.StringVar(
162                 &blobSigningKeyFile,
163                 "permission-key-file",
164                 "",
165                 "Synonym for -blob-signing-key-file.")
166         flag.StringVar(
167                 &blobSigningKeyFile,
168                 "blob-signing-key-file",
169                 "",
170                 "File containing the secret key for generating and verifying "+
171                         "blob permission signatures.")
172         flag.IntVar(
173                 &permissionTTLSec,
174                 "permission-ttl",
175                 0,
176                 "Synonym for -blob-signature-ttl.")
177         flag.IntVar(
178                 &permissionTTLSec,
179                 "blob-signature-ttl",
180                 int(time.Duration(2*7*24*time.Hour).Seconds()),
181                 "Lifetime of blob permission signatures. "+
182                         "See services/api/config/application.default.yml.")
183         flag.BoolVar(
184                 &flagSerializeIO,
185                 "serialize",
186                 false,
187                 "Serialize read and write operations on the following volumes.")
188         flag.BoolVar(
189                 &flagReadonly,
190                 "readonly",
191                 false,
192                 "Do not write, delete, or touch anything on the following volumes.")
193         flag.StringVar(
194                 &pidfile,
195                 "pid",
196                 "",
197                 "Path to write pid file during startup. This file is kept open and locked with LOCK_EX until keepstore exits, so `fuser -k pidfile` is one way to shut down. Exit immediately if there is an error opening, locking, or writing the pid file.")
198         flag.IntVar(
199                 &maxBuffers,
200                 "max-buffers",
201                 maxBuffers,
202                 fmt.Sprintf("Maximum RAM to use for data buffers, given in multiples of block size (%d MiB). When this limit is reached, HTTP requests requiring buffers (like GET and PUT) will wait for buffer space to be released.", BlockSize>>20))
203
204         flag.Parse()
205
206         if maxBuffers < 0 {
207                 log.Fatal("-max-buffers must be greater than zero.")
208         }
209         bufs = newBufferPool(maxBuffers, BlockSize)
210
211         if pidfile != "" {
212                 f, err := os.OpenFile(pidfile, os.O_RDWR|os.O_CREATE, 0777)
213                 if err != nil {
214                         log.Fatalf("open pidfile (%s): %s", pidfile, err)
215                 }
216                 err = syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)
217                 if err != nil {
218                         log.Fatalf("flock pidfile (%s): %s", pidfile, err)
219                 }
220                 err = f.Truncate(0)
221                 if err != nil {
222                         log.Fatalf("truncate pidfile (%s): %s", pidfile, err)
223                 }
224                 _, err = fmt.Fprint(f, os.Getpid())
225                 if err != nil {
226                         log.Fatalf("write pidfile (%s): %s", pidfile, err)
227                 }
228                 err = f.Sync()
229                 if err != nil {
230                         log.Fatalf("sync pidfile (%s): %s", pidfile, err)
231                 }
232                 defer f.Close()
233                 defer os.Remove(pidfile)
234         }
235
236         if len(volumes) == 0 {
237                 if (&unixVolumeAdder{&volumes}).Discover() == 0 {
238                         log.Fatal("No volumes found.")
239                 }
240         }
241
242         for _, v := range volumes {
243                 log.Printf("Using volume %v (writable=%v)", v, v.Writable())
244         }
245
246         // Initialize data manager token and permission key.
247         // If these tokens are specified but cannot be read,
248         // raise a fatal error.
249         if dataManagerTokenFile != "" {
250                 if buf, err := ioutil.ReadFile(dataManagerTokenFile); err == nil {
251                         dataManagerToken = strings.TrimSpace(string(buf))
252                 } else {
253                         log.Fatalf("reading data manager token: %s\n", err)
254                 }
255         }
256
257         if neverDelete != true {
258                 log.Print("never-delete is not set. Warning: the relevant features in keepstore and data manager have not " +
259                         "been extensively tested. You should leave this option alone unless you can afford to lose data.")
260         }
261
262         if blobSigningKeyFile != "" {
263                 if buf, err := ioutil.ReadFile(blobSigningKeyFile); err == nil {
264                         PermissionSecret = bytes.TrimSpace(buf)
265                 } else {
266                         log.Fatalf("reading permission key: %s\n", err)
267                 }
268         }
269
270         blobSignatureTTL = time.Duration(permissionTTLSec) * time.Second
271
272         if PermissionSecret == nil {
273                 if enforcePermissions {
274                         log.Fatal("-enforce-permissions requires a permission key")
275                 } else {
276                         log.Println("Running without a PermissionSecret. Block locators " +
277                                 "returned by this server will not be signed, and will be rejected " +
278                                 "by a server that enforces permissions.")
279                         log.Println("To fix this, use the -blob-signing-key-file flag " +
280                                 "to specify the file containing the permission key.")
281                 }
282         }
283
284         // Start a round-robin VolumeManager with the volumes we have found.
285         KeepVM = MakeRRVolumeManager(volumes)
286
287         // Tell the built-in HTTP server to direct all requests to the REST router.
288         loggingRouter := MakeLoggingRESTRouter()
289         http.HandleFunc("/", func(resp http.ResponseWriter, req *http.Request) {
290                 loggingRouter.ServeHTTP(resp, req)
291         })
292
293         // Set up a TCP listener.
294         listener, err := net.Listen("tcp", listen)
295         if err != nil {
296                 log.Fatal(err)
297         }
298
299         // Initialize Pull queue and worker
300         keepClient := &keepclient.KeepClient{
301                 Arvados:       nil,
302                 Want_replicas: 1,
303                 Client:        &http.Client{},
304         }
305
306         // Initialize the pullq and worker
307         pullq = NewWorkQueue()
308         go RunPullWorker(pullq, keepClient)
309
310         // Initialize the trashq and worker
311         trashq = NewWorkQueue()
312         go RunTrashWorker(trashq)
313
314         // Shut down the server gracefully (by closing the listener)
315         // if SIGTERM is received.
316         term := make(chan os.Signal, 1)
317         go func(sig <-chan os.Signal) {
318                 s := <-sig
319                 log.Println("caught signal:", s)
320                 listener.Close()
321         }(term)
322         signal.Notify(term, syscall.SIGTERM)
323         signal.Notify(term, syscall.SIGINT)
324
325         log.Println("listening at", listen)
326         srv := &http.Server{Addr: listen}
327         srv.Serve(listener)
328 }