7 "git.curoverse.com/arvados.git/sdk/go/keepclient"
19 // ======================
20 // Configuration settings
22 // TODO(twp): make all of these configurable via command line flags
23 // and/or configuration file settings.
25 // Default TCP address on which to listen for requests.
26 // Initialized by the --listen flag.
27 const DefaultAddr = ":25107"
29 // A Keep "block" is 64MB.
30 const BlockSize = 64 * 1024 * 1024
32 // A Keep volume must have at least MinFreeKilobytes available
33 // in order to permit writes.
34 const MinFreeKilobytes = BlockSize / 1024
36 // ProcMounts /proc/mounts
37 var ProcMounts = "/proc/mounts"
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
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
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
54 // neverDelete can be used to prevent the DELETE handler from
55 // actually deleting anything.
56 var neverDelete = true
63 type KeepError struct {
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"}
84 func (e *KeepError) Error() string {
88 // ========================
89 // Internal data structures
91 // These global variables are used by multiple parts of the
92 // program. They are good candidates for moving into their own
95 // The Keep VolumeManager maintains a list of available volumes.
96 // Initialized by the --volumes flag (or by FindKeepVolumes).
97 var KeepVM VolumeManager
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.
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
108 var trashq *WorkQueue
110 type volumeSet []Volume
118 func (vs *volumeSet) String() string {
119 return fmt.Sprintf("%+v", (*vs)[:])
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).
128 log.Println("keepstore starting, pid", os.Getpid())
129 defer log.Println("keepstore exiting, pid", os.Getpid())
132 dataManagerTokenFile string
134 blobSigningKeyFile string
139 &dataManagerTokenFile,
140 "data-manager-token-file",
142 "File with the API token used by the Data Manager. All DELETE "+
143 "requests or GET /index requests must carry this token.")
146 "enforce-permissions",
148 "Enforce permission signatures on requests.")
153 "Listening address, in the form \"host:port\". e.g., 10.0.1.24:8000. Omit the host part to listen on all interfaces.")
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.")
163 "permission-key-file",
165 "Synonym for -blob-signing-key-file.")
168 "blob-signing-key-file",
170 "File containing the secret key for generating and verifying "+
171 "blob permission signatures.")
176 "Synonym for -blob-signature-ttl.")
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.")
187 "Serialize read and write operations on the following volumes.")
192 "Do not write, delete, or touch anything on the following volumes.")
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.")
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))
207 log.Fatal("-max-buffers must be greater than zero.")
209 bufs = newBufferPool(maxBuffers, BlockSize)
212 f, err := os.OpenFile(pidfile, os.O_RDWR|os.O_CREATE, 0777)
214 log.Fatalf("open pidfile (%s): %s", pidfile, err)
216 err = syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)
218 log.Fatalf("flock pidfile (%s): %s", pidfile, err)
222 log.Fatalf("truncate pidfile (%s): %s", pidfile, err)
224 _, err = fmt.Fprint(f, os.Getpid())
226 log.Fatalf("write pidfile (%s): %s", pidfile, err)
230 log.Fatalf("sync pidfile (%s): %s", pidfile, err)
233 defer os.Remove(pidfile)
236 if len(volumes) == 0 {
237 if (&unixVolumeAdder{&volumes}).Discover() == 0 {
238 log.Fatal("No volumes found.")
242 for _, v := range volumes {
243 log.Printf("Using volume %v (writable=%v)", v, v.Writable())
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))
253 log.Fatalf("reading data manager token: %s\n", err)
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.")
262 if blobSigningKeyFile != "" {
263 if buf, err := ioutil.ReadFile(blobSigningKeyFile); err == nil {
264 PermissionSecret = bytes.TrimSpace(buf)
266 log.Fatalf("reading permission key: %s\n", err)
270 blobSignatureTTL = time.Duration(permissionTTLSec) * time.Second
272 if PermissionSecret == nil {
273 if enforcePermissions {
274 log.Fatal("-enforce-permissions requires a permission key")
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.")
284 // Start a round-robin VolumeManager with the volumes we have found.
285 KeepVM = MakeRRVolumeManager(volumes)
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)
293 // Set up a TCP listener.
294 listener, err := net.Listen("tcp", listen)
299 // Initialize Pull queue and worker
300 keepClient := &keepclient.KeepClient{
303 Client: &http.Client{},
306 // Initialize the pullq and worker
307 pullq = NewWorkQueue()
308 go RunPullWorker(pullq, keepClient)
310 // Initialize the trashq and worker
311 trashq = NewWorkQueue()
312 go RunTrashWorker(trashq)
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) {
319 log.Println("caught signal:", s)
322 signal.Notify(term, syscall.SIGTERM)
323 signal.Notify(term, syscall.SIGINT)
325 log.Println("listening at", listen)
326 srv := &http.Server{Addr: listen}