9 "git.curoverse.com/arvados.git/sdk/go/keepclient"
21 // ======================
22 // Configuration settings
24 // TODO(twp): make all of these configurable via command line flags
25 // and/or configuration file settings.
27 // Default TCP address on which to listen for requests.
28 // Initialized by the --listen flag.
29 const DEFAULT_ADDR = ":25107"
31 // A Keep "block" is 64MB.
32 const BLOCKSIZE = 64 * 1024 * 1024
34 // A Keep volume must have at least MIN_FREE_KILOBYTES available
35 // in order to permit writes.
36 const MIN_FREE_KILOBYTES = BLOCKSIZE / 1024
38 var PROC_MOUNTS = "/proc/mounts"
40 // enforce_permissions controls whether permission signatures
41 // should be enforced (affecting GET and DELETE requests).
42 // Initialized by the -enforce-permissions flag.
43 var enforce_permissions bool
45 // blob_signature_ttl is the time duration for which new permission
46 // signatures (returned by PUT requests) will be valid.
47 // Initialized by the -permission-ttl flag.
48 var blob_signature_ttl time.Duration
50 // data_manager_token represents the API token used by the
51 // Data Manager, and is required on certain privileged operations.
52 // Initialized by the -data-manager-token-file flag.
53 var data_manager_token string
55 // never_delete can be used to prevent the DELETE handler from
56 // actually deleting anything.
57 var never_delete = false
62 type KeepError struct {
68 BadRequestError = &KeepError{400, "Bad Request"}
69 UnauthorizedError = &KeepError{401, "Unauthorized"}
70 CollisionError = &KeepError{500, "Collision"}
71 RequestHashError = &KeepError{422, "Hash mismatch in request"}
72 PermissionError = &KeepError{403, "Forbidden"}
73 DiskHashError = &KeepError{500, "Hash mismatch in stored data"}
74 ExpiredError = &KeepError{401, "Expired permission signature"}
75 NotFoundError = &KeepError{404, "Not Found"}
76 GenericError = &KeepError{500, "Fail"}
77 FullError = &KeepError{503, "Full"}
78 SizeRequiredError = &KeepError{411, "Missing Content-Length"}
79 TooLongError = &KeepError{413, "Block is too large"}
80 MethodDisabledError = &KeepError{405, "Method disabled"}
83 func (e *KeepError) Error() string {
87 // ========================
88 // Internal data structures
90 // These global variables are used by multiple parts of the
91 // program. They are good candidates for moving into their own
94 // The Keep VolumeManager maintains a list of available volumes.
95 // Initialized by the --volumes flag (or by FindKeepVolumes).
96 var KeepVM VolumeManager
98 // The pull list manager and trash queue are threadsafe queues which
99 // support atomic update operations. The PullHandler and TrashHandler
100 // store results from Data Manager /pull and /trash requests here.
102 // See the Keep and Data Manager design documents for more details:
103 // https://arvados.org/projects/arvados/wiki/Keep_Design_Doc
104 // https://arvados.org/projects/arvados/wiki/Data_Manager_Design_Doc
107 var trashq *WorkQueue
114 type volumeSet []Volume
116 func (vs *volumeSet) Set(value string) error {
117 if dirs := strings.Split(value, ","); len(dirs) > 1 {
118 log.Print("DEPRECATED: using comma-separated volume list.")
119 for _, dir := range dirs {
120 if err := vs.Set(dir); err != nil {
126 if len(value) == 0 || value[0] != '/' {
127 return errors.New("Invalid volume: must begin with '/'.")
129 if _, err := os.Stat(value); err != nil {
132 *vs = append(*vs, MakeUnixVolume(value, flagSerializeIO, flagReadonly))
136 func (vs *volumeSet) String() string {
138 for i, v := range *vs {
147 // Discover adds a volume for every directory named "keep" that is
148 // located at the top level of a device- or tmpfs-backed mount point
149 // other than "/". It returns the number of volumes added.
150 func (vs *volumeSet) Discover() int {
152 f, err := os.Open(PROC_MOUNTS)
154 log.Fatalf("opening %s: %s", PROC_MOUNTS, err)
156 scanner := bufio.NewScanner(f)
158 args := strings.Fields(scanner.Text())
159 if err := scanner.Err(); err != nil {
160 log.Fatalf("reading %s: %s", PROC_MOUNTS, err)
162 dev, mount := args[0], args[1]
166 if dev != "tmpfs" && !strings.HasPrefix(dev, "/dev/") {
169 keepdir := mount + "/keep"
170 if st, err := os.Stat(keepdir); err != nil || !st.IsDir() {
173 // Set the -readonly flag (but only for this volume)
174 // if the filesystem is mounted readonly.
175 flagReadonlyWas := flagReadonly
176 for _, fsopt := range strings.Split(args[3], ",") {
186 flagReadonly = flagReadonlyWas
192 // TODO(twp): continue moving as much code as possible out of main
193 // so it can be effectively tested. Esp. handling and postprocessing
194 // of command line flags (identifying Keep volumes and initializing
195 // permission arguments).
198 log.Println("Keep started: pid", os.Getpid())
201 data_manager_token_file string
203 blob_signing_key_file string
204 permission_ttl_sec int
209 &data_manager_token_file,
210 "data-manager-token-file",
212 "File with the API token used by the Data Manager. All DELETE "+
213 "requests or GET /index requests must carry this token.")
215 &enforce_permissions,
216 "enforce-permissions",
218 "Enforce permission signatures on requests.")
223 "Listening address, in the form \"host:port\". e.g., 10.0.1.24:8000. Omit the host part to listen on all interfaces.")
228 "If set, nothing will be deleted. HTTP 405 will be returned "+
229 "for valid DELETE requests.")
231 &blob_signing_key_file,
232 "permission-key-file",
234 "Synonym for -blob-signing-key-file.")
236 &blob_signing_key_file,
237 "blob-signing-key-file",
239 "File containing the secret key for generating and verifying "+
240 "blob permission signatures.")
245 "Synonym for -blob-signature-ttl.")
248 "blob-signature-ttl",
249 int(time.Duration(2*7*24*time.Hour).Seconds()),
250 "Lifetime of blob permission signatures. "+
251 "See services/api/config/application.default.yml.")
256 "Serialize read and write operations on the following volumes.")
261 "Do not write, delete, or touch anything on the following volumes.")
265 "Deprecated synonym for -volume.")
269 "Local storage directory. Can be given more than once to add multiple directories. If none are supplied, the default is to use all directories named \"keep\" that exist in the top level directory of a mount point at startup time. Can be a comma-separated list, but this is deprecated: use multiple -volume arguments instead.")
274 "Path to write pid file")
278 if len(volumes) == 0 {
279 if volumes.Discover() == 0 {
280 log.Fatal("No volumes found.")
284 for _, v := range volumes {
285 log.Printf("Using volume %v (writable=%v)", v, v.Writable())
288 // Initialize data manager token and permission key.
289 // If these tokens are specified but cannot be read,
290 // raise a fatal error.
291 if data_manager_token_file != "" {
292 if buf, err := ioutil.ReadFile(data_manager_token_file); err == nil {
293 data_manager_token = strings.TrimSpace(string(buf))
295 log.Fatalf("reading data manager token: %s\n", err)
298 if blob_signing_key_file != "" {
299 if buf, err := ioutil.ReadFile(blob_signing_key_file); err == nil {
300 PermissionSecret = bytes.TrimSpace(buf)
302 log.Fatalf("reading permission key: %s\n", err)
306 blob_signature_ttl = time.Duration(permission_ttl_sec) * time.Second
308 if PermissionSecret == nil {
309 if enforce_permissions {
310 log.Fatal("-enforce-permissions requires a permission key")
312 log.Println("Running without a PermissionSecret. Block locators " +
313 "returned by this server will not be signed, and will be rejected " +
314 "by a server that enforces permissions.")
315 log.Println("To fix this, use the -permission-key-file flag " +
316 "to specify the file containing the permission key.")
320 // Start a round-robin VolumeManager with the volumes we have found.
321 KeepVM = MakeRRVolumeManager(volumes)
323 // Tell the built-in HTTP server to direct all requests to the REST router.
324 loggingRouter := MakeLoggingRESTRouter()
325 http.HandleFunc("/", func(resp http.ResponseWriter, req *http.Request) {
326 loggingRouter.ServeHTTP(resp, req)
329 // Set up a TCP listener.
330 listener, err := net.Listen("tcp", listen)
335 // Initialize Pull queue and worker
336 keepClient := &keepclient.KeepClient{
340 Client: &http.Client{},
343 // Initialize the pullq and worker
344 pullq = NewWorkQueue()
345 go RunPullWorker(pullq, keepClient)
347 // Initialize the trashq and worker
348 trashq = NewWorkQueue()
349 go RunTrashWorker(trashq)
351 // Shut down the server gracefully (by closing the listener)
352 // if SIGTERM is received.
353 term := make(chan os.Signal, 1)
354 go func(sig <-chan os.Signal) {
356 log.Println("caught signal:", s)
359 signal.Notify(term, syscall.SIGTERM)
362 f, err := os.Create(pidfile)
364 fmt.Fprint(f, os.Getpid())
367 log.Printf("Error writing pid file (%s): %s", pidfile, err.Error())
371 // Start listening for requests.
372 srv := &http.Server{Addr: listen}
375 log.Println("shutting down")