1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: AGPL-3.0
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"
25 // A Keep "block" is 64MB.
26 const BlockSize = 64 * 1024 * 1024
28 // A Keep volume must have at least MinFreeKilobytes available
29 // in order to permit writes.
30 const MinFreeKilobytes = BlockSize / 1024
32 // ProcMounts /proc/mounts
33 var ProcMounts = "/proc/mounts"
39 type KeepError struct {
45 BadRequestError = &KeepError{400, "Bad Request"}
46 UnauthorizedError = &KeepError{401, "Unauthorized"}
47 CollisionError = &KeepError{500, "Collision"}
48 RequestHashError = &KeepError{422, "Hash mismatch in request"}
49 PermissionError = &KeepError{403, "Forbidden"}
50 DiskHashError = &KeepError{500, "Hash mismatch in stored data"}
51 ExpiredError = &KeepError{401, "Expired permission signature"}
52 NotFoundError = &KeepError{404, "Not Found"}
53 VolumeBusyError = &KeepError{503, "Volume backend busy"}
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"}
63 func (e *KeepError) Error() string {
67 // ========================
68 // Internal data structures
70 // These global variables are used by multiple parts of the
71 // program. They are good candidates for moving into their own
74 // The Keep VolumeManager maintains a list of available volumes.
75 // Initialized by the --volumes flag (or by FindKeepVolumes).
76 var KeepVM VolumeManager
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.
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
90 deprecated.beforeFlagParse(theConfig)
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.")
95 defaultConfigPath := "/etc/arvados/keepstore/keepstore.yml"
101 "YAML or JSON configuration file `path`")
105 // Print version information if requested
107 fmt.Printf("keepstore %s\n", version)
111 deprecated.afterFlagParse(theConfig)
113 err := config.LoadFile(theConfig, configPath)
114 if err != nil && (!os.IsNotExist(err) || configPath != defaultConfigPath) {
119 log.Fatal(config.DumpAndExit(theConfig))
122 log.Printf("keepstore %s started", version)
124 err = theConfig.Start()
129 if pidfile := theConfig.PIDFile; pidfile != "" {
130 f, err := os.OpenFile(pidfile, os.O_RDWR|os.O_CREATE, 0777)
132 log.Fatalf("open pidfile (%s): %s", pidfile, err)
135 err = syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)
137 log.Fatalf("flock pidfile (%s): %s", pidfile, err)
139 defer os.Remove(pidfile)
142 log.Fatalf("truncate pidfile (%s): %s", pidfile, err)
144 _, err = fmt.Fprint(f, os.Getpid())
146 log.Fatalf("write pidfile (%s): %s", pidfile, err)
150 log.Fatalf("sync pidfile (%s): %s", pidfile, err)
154 var cluster *arvados.Cluster
155 cfg, err := arvados.GetConfig(arvados.DefaultConfigFile)
156 if err != nil && os.IsNotExist(err) {
157 log.Warnf("DEPRECATED: proceeding without cluster configuration file %q (%s)", arvados.DefaultConfigFile, err)
158 cluster = &arvados.Cluster{
161 } else if err != nil {
162 log.Fatalf("load config %q: %s", arvados.DefaultConfigFile, err)
164 cluster, err = cfg.GetCluster("")
166 log.Fatalf("config error in %q: %s", arvados.DefaultConfigFile, err)
170 log.Println("keepstore starting, pid", os.Getpid())
171 defer log.Println("keepstore exiting, pid", os.Getpid())
173 // Start a round-robin VolumeManager with the volumes we have found.
174 KeepVM = MakeRRVolumeManager(theConfig.Volumes)
176 // Middleware/handler stack
177 router := MakeRESTRouter(cluster)
179 // Set up a TCP listener.
180 listener, err := net.Listen("tcp", theConfig.Listen)
185 // Initialize keepclient for pull workers
186 keepClient := &keepclient.KeepClient{
187 Arvados: &arvadosclient.ArvadosClient{},
191 // Initialize the pullq and workers
192 pullq = NewWorkQueue()
193 for i := 0; i < 1 || i < theConfig.PullWorkers; i++ {
194 go RunPullWorker(pullq, keepClient)
197 // Initialize the trashq and workers
198 trashq = NewWorkQueue()
199 for i := 0; i < 1 || i < theConfig.TrashWorkers; i++ {
200 go RunTrashWorker(trashq)
203 // Start emptyTrash goroutine
204 doneEmptyingTrash := make(chan bool)
205 go emptyTrash(doneEmptyingTrash, theConfig.TrashCheckInterval.Duration())
207 // Shut down the server gracefully (by closing the listener)
208 // if SIGTERM is received.
209 term := make(chan os.Signal, 1)
210 go func(sig <-chan os.Signal) {
212 log.Println("caught signal:", s)
213 doneEmptyingTrash <- true
216 signal.Notify(term, syscall.SIGTERM)
217 signal.Notify(term, syscall.SIGINT)
219 if _, err := daemon.SdNotify(false, "READY=1"); err != nil {
220 log.Printf("Error notifying init daemon: %v", err)
222 log.Println("listening at", listener.Addr())
228 // Periodically (once per interval) invoke EmptyTrash on all volumes.
229 func emptyTrash(done <-chan bool, interval time.Duration) {
230 ticker := time.NewTicker(interval)
235 for _, v := range theConfig.Volumes {