services/keepstore/keepstore.go

   1 package main
   2
   3 import (
   4         "bytes"
   5         "flag"
   6         "fmt"
   7         "git.curoverse.com/arvados.git/services/keepstore/pull_list"
   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 DEFAULT_ADDR = ":25107"
  28
  29 // A Keep "block" is 64MB.
  30 const BLOCKSIZE = 64 * 1024 * 1024
  31
  32 // A Keep volume must have at least MIN_FREE_KILOBYTES available
  33 // in order to permit writes.
  34 const MIN_FREE_KILOBYTES = BLOCKSIZE / 1024
  35
  36 var PROC_MOUNTS = "/proc/mounts"
  37
  38 // enforce_permissions controls whether permission signatures
  39 // should be enforced (affecting GET and DELETE requests).
  40 // Initialized by the --enforce-permissions flag.
  41 var enforce_permissions bool
  42
  43 // permission_ttl is the time duration for which new permission
  44 // signatures (returned by PUT requests) will be valid.
  45 // Initialized by the --permission-ttl flag.
  46 var permission_ttl time.Duration
  47
  48 // data_manager_token represents the API token used by the
  49 // Data Manager, and is required on certain privileged operations.
  50 // Initialized by the --data-manager-token-file flag.
  51 var data_manager_token string
  52
  53 // never_delete can be used to prevent the DELETE handler from
  54 // actually deleting anything.
  55 var never_delete = false
  56
  57 // ==========
  58 // Error types.
  59 //
  60 type KeepError struct {
  61         HTTPCode int
  62         ErrMsg   string
  63 }
  64
  65 var (
  66         BadRequestError     = &KeepError{400, "Bad Request"}
  67         UnauthorizedError   = &KeepError{401, "Unauthorized"}
  68         CollisionError      = &KeepError{500, "Collision"}
  69         RequestHashError    = &KeepError{422, "Hash mismatch in request"}
  70         PermissionError     = &KeepError{403, "Forbidden"}
  71         DiskHashError       = &KeepError{500, "Hash mismatch in stored data"}
  72         ExpiredError        = &KeepError{401, "Expired permission signature"}
  73         NotFoundError       = &KeepError{404, "Not Found"}
  74         GenericError        = &KeepError{500, "Fail"}
  75         FullError           = &KeepError{503, "Full"}
  76         TooLongError        = &KeepError{504, "Timeout"}
  77         MethodDisabledError = &KeepError{405, "Method disabled"}
  78 )
  79
  80 func (e *KeepError) Error() string {
  81         return e.ErrMsg
  82 }
  83
  84 // ========================
  85 // Internal data structures
  86 //
  87 // These global variables are used by multiple parts of the
  88 // program. They are good candidates for moving into their own
  89 // packages.
  90
  91 // The Keep VolumeManager maintains a list of available volumes.
  92 // Initialized by the --volumes flag (or by FindKeepVolumes).
  93 var KeepVM VolumeManager
  94
  95 // The pull list manager is a singleton pull list (a list of blocks
  96 // that the current keepstore process should be pulling from remote
  97 // keepstore servers in order to increase data replication) with
  98 // atomic update methods that are safe to use from multiple
  99 // goroutines.
 100 var pullmgr *pull_list.Manager
 101
 102 // TODO(twp): continue moving as much code as possible out of main
 103 // so it can be effectively tested. Esp. handling and postprocessing
 104 // of command line flags (identifying Keep volumes and initializing
 105 // permission arguments).
 106
 107 func main() {
 108         log.Println("Keep started: pid", os.Getpid())
 109
 110         // Parse command-line flags:
 111         //
 112         // -listen=ipaddr:port
 113         //    Interface on which to listen for requests. Use :port without
 114         //    an ipaddr to listen on all network interfaces.
 115         //    Examples:
 116         //      -listen=127.0.0.1:4949
 117         //      -listen=10.0.1.24:8000
 118         //      -listen=:25107 (to listen to port 25107 on all interfaces)
 119         //
 120         // -volumes
 121         //    A comma-separated list of directories to use as Keep volumes.
 122         //    Example:
 123         //      -volumes=/var/keep01,/var/keep02,/var/keep03/subdir
 124         //
 125         //    If -volumes is empty or is not present, Keep will select volumes
 126         //    by looking at currently mounted filesystems for /keep top-level
 127         //    directories.
 128
 129         var (
 130                 data_manager_token_file string
 131                 listen                  string
 132                 permission_key_file     string
 133                 permission_ttl_sec      int
 134                 serialize_io            bool
 135                 volumearg               string
 136                 pidfile                 string
 137         )
 138         flag.StringVar(
 139                 &data_manager_token_file,
 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                 &enforce_permissions,
 146                 "enforce-permissions",
 147                 false,
 148                 "Enforce permission signatures on requests.")
 149         flag.StringVar(
 150                 &listen,
 151                 "listen",
 152                 DEFAULT_ADDR,
 153                 "Interface on which to listen for requests, in the format "+
 154                         "ipaddr:port. e.g. -listen=10.0.1.24:8000. Use -listen=:port "+
 155                         "to listen on all network interfaces.")
 156         flag.BoolVar(
 157                 &never_delete,
 158                 "never-delete",
 159                 false,
 160                 "If set, nothing will be deleted. HTTP 405 will be returned "+
 161                         "for valid DELETE requests.")
 162         flag.StringVar(
 163                 &permission_key_file,
 164                 "permission-key-file",
 165                 "",
 166                 "File containing the secret key for generating and verifying "+
 167                         "permission signatures.")
 168         flag.IntVar(
 169                 &permission_ttl_sec,
 170                 "permission-ttl",
 171                 1209600,
 172                 "Expiration time (in seconds) for newly generated permission "+
 173                         "signatures.")
 174         flag.BoolVar(
 175                 &serialize_io,
 176                 "serialize",
 177                 false,
 178                 "If set, all read and write operations on local Keep volumes will "+
 179                         "be serialized.")
 180         flag.StringVar(
 181                 &volumearg,
 182                 "volumes",
 183                 "",
 184                 "Comma-separated list of directories to use for Keep volumes, "+
 185                         "e.g. -volumes=/var/keep1,/var/keep2. If empty or not "+
 186                         "supplied, Keep will scan mounted filesystems for volumes "+
 187                         "with a /keep top-level directory.")
 188
 189         flag.StringVar(
 190                 &pidfile,
 191                 "pid",
 192                 "",
 193                 "Path to write pid file")
 194
 195         flag.Parse()
 196
 197         // Look for local keep volumes.
 198         var keepvols []string
 199         if volumearg == "" {
 200                 // TODO(twp): decide whether this is desirable default behavior.
 201                 // In production we may want to require the admin to specify
 202                 // Keep volumes explicitly.
 203                 keepvols = FindKeepVolumes()
 204         } else {
 205                 keepvols = strings.Split(volumearg, ",")
 206         }
 207
 208         // Check that the specified volumes actually exist.
 209         var goodvols []Volume = nil
 210         for _, v := range keepvols {
 211                 if _, err := os.Stat(v); err == nil {
 212                         log.Println("adding Keep volume:", v)
 213                         newvol := MakeUnixVolume(v, serialize_io)
 214                         goodvols = append(goodvols, &newvol)
 215                 } else {
 216                         log.Printf("bad Keep volume: %s\n", err)
 217                 }
 218         }
 219
 220         if len(goodvols) == 0 {
 221                 log.Fatal("could not find any keep volumes")
 222         }
 223
 224         // Initialize data manager token and permission key.
 225         // If these tokens are specified but cannot be read,
 226         // raise a fatal error.
 227         if data_manager_token_file != "" {
 228                 if buf, err := ioutil.ReadFile(data_manager_token_file); err == nil {
 229                         data_manager_token = strings.TrimSpace(string(buf))
 230                 } else {
 231                         log.Fatalf("reading data manager token: %s\n", err)
 232                 }
 233         }
 234         if permission_key_file != "" {
 235                 if buf, err := ioutil.ReadFile(permission_key_file); err == nil {
 236                         PermissionSecret = bytes.TrimSpace(buf)
 237                 } else {
 238                         log.Fatalf("reading permission key: %s\n", err)
 239                 }
 240         }
 241
 242         // Initialize permission TTL
 243         permission_ttl = time.Duration(permission_ttl_sec) * time.Second
 244
 245         // If --enforce-permissions is true, we must have a permission key
 246         // to continue.
 247         if PermissionSecret == nil {
 248                 if enforce_permissions {
 249                         log.Fatal("--enforce-permissions requires a permission key")
 250                 } else {
 251                         log.Println("Running without a PermissionSecret. Block locators " +
 252                                 "returned by this server will not be signed, and will be rejected " +
 253                                 "by a server that enforces permissions.")
 254                         log.Println("To fix this, run Keep with --permission-key-file=<path> " +
 255                                 "to define the location of a file containing the permission key.")
 256                 }
 257         }
 258
 259         // Start a round-robin VolumeManager with the volumes we have found.
 260         KeepVM = MakeRRVolumeManager(goodvols)
 261
 262         // Tell the built-in HTTP server to direct all requests to the REST
 263         // router.
 264         http.Handle("/", MakeRESTRouter())
 265
 266         // Set up a TCP listener.
 267         listener, err := net.Listen("tcp", listen)
 268         if err != nil {
 269                 log.Fatal(err)
 270         }
 271
 272         // Shut down the server gracefully (by closing the listener)
 273         // if SIGTERM is received.
 274         term := make(chan os.Signal, 1)
 275         go func(sig <-chan os.Signal) {
 276                 s := <-sig
 277                 log.Println("caught signal:", s)
 278                 listener.Close()
 279         }(term)
 280         signal.Notify(term, syscall.SIGTERM)
 281
 282         if pidfile != "" {
 283                 f, err := os.Create(pidfile)
 284                 if err == nil {
 285                         fmt.Fprint(f, os.Getpid())
 286                         f.Close()
 287                 } else {
 288                         log.Printf("Error writing pid file (%s): %s", pidfile, err.Error())
 289                 }
 290         }
 291
 292         // Start listening for requests.
 293         srv := &http.Server{Addr: listen}
 294         srv.Serve(listener)
 295
 296         log.Println("shutting down")
 297
 298         if pidfile != "" {
 299                 os.Remove(pidfile)
 300         }
 301 }