services/keepstore/keepstore.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package main
   6
   7 import (
   8         "flag"
   9         "fmt"
  10         "net"
  11         "os"
  12         "os/signal"
  13         "syscall"
  14         "time"
  15
  16         "git.curoverse.com/arvados.git/sdk/go/arvados"
  17         "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
  18         "git.curoverse.com/arvados.git/sdk/go/config"
  19         "git.curoverse.com/arvados.git/sdk/go/keepclient"
  20         "github.com/coreos/go-systemd/daemon"
  21         "github.com/prometheus/client_golang/prometheus"
  22 )
  23
  24 var version = "dev"
  25
  26 // A Keep "block" is 64MB.
  27 const BlockSize = 64 * 1024 * 1024
  28
  29 // A Keep volume must have at least MinFreeKilobytes available
  30 // in order to permit writes.
  31 const MinFreeKilobytes = BlockSize / 1024
  32
  33 // ProcMounts /proc/mounts
  34 var ProcMounts = "/proc/mounts"
  35
  36 var bufs *bufferPool
  37
  38 // KeepError types.
  39 //
  40 type KeepError struct {
  41         HTTPCode int
  42         ErrMsg   string
  43 }
  44
  45 var (
  46         BadRequestError     = &KeepError{400, "Bad Request"}
  47         UnauthorizedError   = &KeepError{401, "Unauthorized"}
  48         CollisionError      = &KeepError{500, "Collision"}
  49         RequestHashError    = &KeepError{422, "Hash mismatch in request"}
  50         PermissionError     = &KeepError{403, "Forbidden"}
  51         DiskHashError       = &KeepError{500, "Hash mismatch in stored data"}
  52         ExpiredError        = &KeepError{401, "Expired permission signature"}
  53         NotFoundError       = &KeepError{404, "Not Found"}
  54         GenericError        = &KeepError{500, "Fail"}
  55         FullError           = &KeepError{503, "Full"}
  56         SizeRequiredError   = &KeepError{411, "Missing Content-Length"}
  57         TooLongError        = &KeepError{413, "Block is too large"}
  58         MethodDisabledError = &KeepError{405, "Method disabled"}
  59         ErrNotImplemented   = &KeepError{500, "Unsupported configuration"}
  60         ErrClientDisconnect = &KeepError{503, "Client disconnected"}
  61 )
  62
  63 func (e *KeepError) Error() string {
  64         return e.ErrMsg
  65 }
  66
  67 // ========================
  68 // Internal data structures
  69 //
  70 // These global variables are used by multiple parts of the
  71 // program. They are good candidates for moving into their own
  72 // packages.
  73
  74 // The Keep VolumeManager maintains a list of available volumes.
  75 // Initialized by the --volumes flag (or by FindKeepVolumes).
  76 var KeepVM VolumeManager
  77
  78 // The pull list manager and trash queue are threadsafe queues which
  79 // support atomic update operations. The PullHandler and TrashHandler
  80 // store results from Data Manager /pull and /trash requests here.
  81 //
  82 // See the Keep and Data Manager design documents for more details:
  83 // https://arvados.org/projects/arvados/wiki/Keep_Design_Doc
  84 // https://arvados.org/projects/arvados/wiki/Data_Manager_Design_Doc
  85 //
  86 var pullq *WorkQueue
  87 var trashq *WorkQueue
  88
  89 func main() {
  90         deprecated.beforeFlagParse(theConfig)
  91
  92         dumpConfig := flag.Bool("dump-config", false, "write current configuration to stdout and exit (useful for migrating from command line flags to config file)")
  93         getVersion := flag.Bool("version", false, "Print version information and exit.")
  94
  95         defaultConfigPath := "/etc/arvados/keepstore/keepstore.yml"
  96         var configPath string
  97         flag.StringVar(
  98                 &configPath,
  99                 "config",
 100                 defaultConfigPath,
 101                 "YAML or JSON configuration file `path`")
 102         flag.Usage = usage
 103         flag.Parse()
 104
 105         // Print version information if requested
 106         if *getVersion {
 107                 fmt.Printf("keepstore %s\n", version)
 108                 return
 109         }
 110
 111         deprecated.afterFlagParse(theConfig)
 112
 113         err := config.LoadFile(theConfig, configPath)
 114         if err != nil && (!os.IsNotExist(err) || configPath != defaultConfigPath) {
 115                 log.Fatal(err)
 116         }
 117
 118         if *dumpConfig {
 119                 log.Fatal(config.DumpAndExit(theConfig))
 120         }
 121
 122         log.Printf("keepstore %s started", version)
 123
 124         metricsRegistry := prometheus.NewRegistry()
 125
 126         err = theConfig.Start(metricsRegistry)
 127         if err != nil {
 128                 log.Fatal(err)
 129         }
 130
 131         if pidfile := theConfig.PIDFile; pidfile != "" {
 132                 f, err := os.OpenFile(pidfile, os.O_RDWR|os.O_CREATE, 0777)
 133                 if err != nil {
 134                         log.Fatalf("open pidfile (%s): %s", pidfile, err)
 135                 }
 136                 defer f.Close()
 137                 err = syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)
 138                 if err != nil {
 139                         log.Fatalf("flock pidfile (%s): %s", pidfile, err)
 140                 }
 141                 defer os.Remove(pidfile)
 142                 err = f.Truncate(0)
 143                 if err != nil {
 144                         log.Fatalf("truncate pidfile (%s): %s", pidfile, err)
 145                 }
 146                 _, err = fmt.Fprint(f, os.Getpid())
 147                 if err != nil {
 148                         log.Fatalf("write pidfile (%s): %s", pidfile, err)
 149                 }
 150                 err = f.Sync()
 151                 if err != nil {
 152                         log.Fatalf("sync pidfile (%s): %s", pidfile, err)
 153                 }
 154         }
 155
 156         var cluster *arvados.Cluster
 157         cfg, err := arvados.GetConfig(arvados.DefaultConfigFile)
 158         if err != nil && os.IsNotExist(err) {
 159                 log.Warnf("DEPRECATED: proceeding without cluster configuration file %q (%s)", arvados.DefaultConfigFile, err)
 160                 cluster = &arvados.Cluster{
 161                         ClusterID: "xxxxx",
 162                 }
 163         } else if err != nil {
 164                 log.Fatalf("load config %q: %s", arvados.DefaultConfigFile, err)
 165         } else {
 166                 cluster, err = cfg.GetCluster("")
 167                 if err != nil {
 168                         log.Fatalf("config error in %q: %s", arvados.DefaultConfigFile, err)
 169                 }
 170         }
 171
 172         log.Println("keepstore starting, pid", os.Getpid())
 173         defer log.Println("keepstore exiting, pid", os.Getpid())
 174
 175         // Start a round-robin VolumeManager with the volumes we have found.
 176         KeepVM = MakeRRVolumeManager(theConfig.Volumes)
 177
 178         // Middleware/handler stack
 179         router := MakeRESTRouter(cluster, metricsRegistry)
 180
 181         // Set up a TCP listener.
 182         listener, err := net.Listen("tcp", theConfig.Listen)
 183         if err != nil {
 184                 log.Fatal(err)
 185         }
 186
 187         // Initialize keepclient for pull workers
 188         keepClient := &keepclient.KeepClient{
 189                 Arvados:       &arvadosclient.ArvadosClient{},
 190                 Want_replicas: 1,
 191         }
 192
 193         // Initialize the pullq and workers
 194         pullq = NewWorkQueue()
 195         for i := 0; i < 1 || i < theConfig.PullWorkers; i++ {
 196                 go RunPullWorker(pullq, keepClient)
 197         }
 198
 199         // Initialize the trashq and workers
 200         trashq = NewWorkQueue()
 201         for i := 0; i < 1 || i < theConfig.TrashWorkers; i++ {
 202                 go RunTrashWorker(trashq)
 203         }
 204
 205         // Start emptyTrash goroutine
 206         doneEmptyingTrash := make(chan bool)
 207         go emptyTrash(doneEmptyingTrash, theConfig.TrashCheckInterval.Duration())
 208
 209         // Shut down the server gracefully (by closing the listener)
 210         // if SIGTERM is received.
 211         term := make(chan os.Signal, 1)
 212         go func(sig <-chan os.Signal) {
 213                 s := <-sig
 214                 log.Println("caught signal:", s)
 215                 doneEmptyingTrash <- true
 216                 listener.Close()
 217         }(term)
 218         signal.Notify(term, syscall.SIGTERM)
 219         signal.Notify(term, syscall.SIGINT)
 220
 221         if _, err := daemon.SdNotify(false, "READY=1"); err != nil {
 222                 log.Printf("Error notifying init daemon: %v", err)
 223         }
 224         log.Println("listening at", listener.Addr())
 225         srv := &server{}
 226         srv.Handler = router
 227         srv.Serve(listener)
 228 }
 229
 230 // Periodically (once per interval) invoke EmptyTrash on all volumes.
 231 func emptyTrash(done <-chan bool, interval time.Duration) {
 232         ticker := time.NewTicker(interval)
 233
 234         for {
 235                 select {
 236                 case <-ticker.C:
 237                         for _, v := range theConfig.Volumes {
 238                                 if v.Writable() {
 239                                         v.EmptyTrash()
 240                                 }
 241                         }
 242                 case <-done:
 243                         ticker.Stop()
 244                         return
 245                 }
 246         }
 247 }