services/keepstore/keepstore.go

   1 package main
   2
   3 import (
   4         "bufio"
   5         "bytes"
   6         "errors"
   7         "flag"
   8         "fmt"
   9         "git.curoverse.com/arvados.git/sdk/go/keepclient"
  10         "io/ioutil"
  11         "log"
  12         "net"
  13         "net/http"
  14         "os"
  15         "os/signal"
  16         "strings"
  17         "syscall"
  18         "time"
  19 )
  20
  21 // ======================
  22 // Configuration settings
  23 //
  24 // TODO(twp): make all of these configurable via command line flags
  25 // and/or configuration file settings.
  26
  27 // Default TCP address on which to listen for requests.
  28 // Initialized by the --listen flag.
  29 const DEFAULT_ADDR = ":25107"
  30
  31 // A Keep "block" is 64MB.
  32 const BLOCKSIZE = 64 * 1024 * 1024
  33
  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
  37
  38 var PROC_MOUNTS = "/proc/mounts"
  39
  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
  44
  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
  49
  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
  54
  55 // never_delete can be used to prevent the DELETE handler from
  56 // actually deleting anything.
  57 var never_delete = false
  58
  59 // ==========
  60 // Error types.
  61 //
  62 type KeepError struct {
  63         HTTPCode int
  64         ErrMsg   string
  65 }
  66
  67 var (
  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"}
  81 )
  82
  83 func (e *KeepError) Error() string {
  84         return e.ErrMsg
  85 }
  86
  87 // ========================
  88 // Internal data structures
  89 //
  90 // These global variables are used by multiple parts of the
  91 // program. They are good candidates for moving into their own
  92 // packages.
  93
  94 // The Keep VolumeManager maintains a list of available volumes.
  95 // Initialized by the --volumes flag (or by FindKeepVolumes).
  96 var KeepVM VolumeManager
  97
  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.
 101 //
 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
 105 //
 106 var pullq *WorkQueue
 107 var trashq *WorkQueue
 108
 109 var (
 110         flagSerializeIO bool
 111         flagReadonly    bool
 112 )
 113
 114 type volumeSet []Volume
 115
 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 {
 121                                 return err
 122                         }
 123                 }
 124                 return nil
 125         }
 126         if len(value) == 0 || value[0] != '/' {
 127                 return errors.New("Invalid volume: must begin with '/'.")
 128         }
 129         if _, err := os.Stat(value); err != nil {
 130                 return err
 131         }
 132         *vs = append(*vs, MakeUnixVolume(value, flagSerializeIO, flagReadonly))
 133         return nil
 134 }
 135
 136 func (vs *volumeSet) String() string {
 137         s := "["
 138         for i, v := range *vs {
 139                 if i > 0 {
 140                         s = s + " "
 141                 }
 142                 s = s + v.String()
 143         }
 144         return s + "]"
 145 }
 146
 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 {
 151         added := 0
 152         f, err := os.Open(PROC_MOUNTS)
 153         if err != nil {
 154                 log.Fatalf("opening %s: %s", PROC_MOUNTS, err)
 155         }
 156         scanner := bufio.NewScanner(f)
 157         for scanner.Scan() {
 158                 args := strings.Fields(scanner.Text())
 159                 if err := scanner.Err(); err != nil {
 160                         log.Fatalf("reading %s: %s", PROC_MOUNTS, err)
 161                 }
 162                 dev, mount := args[0], args[1]
 163                 if mount == "/" {
 164                         continue
 165                 }
 166                 if dev != "tmpfs" && !strings.HasPrefix(dev, "/dev/") {
 167                         continue
 168                 }
 169                 keepdir := mount + "/keep"
 170                 if st, err := os.Stat(keepdir); err != nil || !st.IsDir() {
 171                         continue
 172                 }
 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], ",") {
 177                         if fsopt == "ro" {
 178                                 flagReadonly = true
 179                                 break
 180                         }
 181                         if fsopt == "rw" {
 182                                 break
 183                         }
 184                 }
 185                 vs.Set(keepdir)
 186                 flagReadonly = flagReadonlyWas
 187                 added++
 188         }
 189         return added
 190 }
 191
 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).
 196
 197 func main() {
 198         log.Println("Keep started: pid", os.Getpid())
 199
 200         var (
 201                 data_manager_token_file string
 202                 listen                  string
 203                 blob_signing_key_file   string
 204                 permission_ttl_sec      int
 205                 volumes                 volumeSet
 206                 pidfile                 string
 207         )
 208         flag.StringVar(
 209                 &data_manager_token_file,
 210                 "data-manager-token-file",
 211                 "",
 212                 "File with the API token used by the Data Manager. All DELETE "+
 213                         "requests or GET /index requests must carry this token.")
 214         flag.BoolVar(
 215                 &enforce_permissions,
 216                 "enforce-permissions",
 217                 false,
 218                 "Enforce permission signatures on requests.")
 219         flag.StringVar(
 220                 &listen,
 221                 "listen",
 222                 DEFAULT_ADDR,
 223                 "Listening address, in the form \"host:port\". e.g., 10.0.1.24:8000. Omit the host part to listen on all interfaces.")
 224         flag.BoolVar(
 225                 &never_delete,
 226                 "never-delete",
 227                 false,
 228                 "If set, nothing will be deleted. HTTP 405 will be returned "+
 229                         "for valid DELETE requests.")
 230         flag.StringVar(
 231                 &blob_signing_key_file,
 232                 "permission-key-file",
 233                 "",
 234                 "Synonym for -blob-signing-key-file.")
 235         flag.StringVar(
 236                 &blob_signing_key_file,
 237                 "blob-signing-key-file",
 238                 "",
 239                 "File containing the secret key for generating and verifying "+
 240                         "blob permission signatures.")
 241         flag.IntVar(
 242                 &permission_ttl_sec,
 243                 "permission-ttl",
 244                 0,
 245                 "Synonym for -blob-signature-ttl.")
 246         flag.IntVar(
 247                 &permission_ttl_sec,
 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.")
 252         flag.BoolVar(
 253                 &flagSerializeIO,
 254                 "serialize",
 255                 false,
 256                 "Serialize read and write operations on the following volumes.")
 257         flag.BoolVar(
 258                 &flagReadonly,
 259                 "readonly",
 260                 false,
 261                 "Do not write, delete, or touch anything on the following volumes.")
 262         flag.Var(
 263                 &volumes,
 264                 "volumes",
 265                 "Deprecated synonym for -volume.")
 266         flag.Var(
 267                 &volumes,
 268                 "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.")
 270         flag.StringVar(
 271                 &pidfile,
 272                 "pid",
 273                 "",
 274                 "Path to write pid file")
 275
 276         flag.Parse()
 277
 278         if len(volumes) == 0 {
 279                 if volumes.Discover() == 0 {
 280                         log.Fatal("No volumes found.")
 281                 }
 282         }
 283
 284         for _, v := range volumes {
 285                 log.Printf("Using volume %v (writable=%v)", v, v.Writable())
 286         }
 287
 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))
 294                 } else {
 295                         log.Fatalf("reading data manager token: %s\n", err)
 296                 }
 297         }
 298         if blob_signing_key_file != "" {
 299                 if buf, err := ioutil.ReadFile(blob_signing_key_file); err == nil {
 300                         PermissionSecret = bytes.TrimSpace(buf)
 301                 } else {
 302                         log.Fatalf("reading permission key: %s\n", err)
 303                 }
 304         }
 305
 306         blob_signature_ttl = time.Duration(permission_ttl_sec) * time.Second
 307
 308         if PermissionSecret == nil {
 309                 if enforce_permissions {
 310                         log.Fatal("-enforce-permissions requires a permission key")
 311                 } else {
 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.")
 317                 }
 318         }
 319
 320         // Start a round-robin VolumeManager with the volumes we have found.
 321         KeepVM = MakeRRVolumeManager(volumes)
 322
 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)
 327         })
 328
 329         // Set up a TCP listener.
 330         listener, err := net.Listen("tcp", listen)
 331         if err != nil {
 332                 log.Fatal(err)
 333         }
 334
 335         // Initialize Pull queue and worker
 336         keepClient := &keepclient.KeepClient{
 337                 Arvados:       nil,
 338                 Want_replicas: 1,
 339                 Using_proxy:   true,
 340                 Client:        &http.Client{},
 341         }
 342
 343         // Initialize the pullq and worker
 344         pullq = NewWorkQueue()
 345         go RunPullWorker(pullq, keepClient)
 346
 347         // Initialize the trashq and worker
 348         trashq = NewWorkQueue()
 349         go RunTrashWorker(trashq)
 350
 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) {
 355                 s := <-sig
 356                 log.Println("caught signal:", s)
 357                 listener.Close()
 358         }(term)
 359         signal.Notify(term, syscall.SIGTERM)
 360
 361         if pidfile != "" {
 362                 f, err := os.Create(pidfile)
 363                 if err == nil {
 364                         fmt.Fprint(f, os.Getpid())
 365                         f.Close()
 366                 } else {
 367                         log.Printf("Error writing pid file (%s): %s", pidfile, err.Error())
 368                 }
 369         }
 370
 371         // Start listening for requests.
 372         srv := &http.Server{Addr: listen}
 373         srv.Serve(listener)
 374
 375         log.Println("shutting down")
 376
 377         if pidfile != "" {
 378                 os.Remove(pidfile)
 379         }
 380 }