X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/c0f9c128aabb366435d751a3ea1a63b76c177f5b..1e27ecf1368de5932c4af00c6cff9595c501f5f6:/services/keepstore/trash_worker.go diff --git a/services/keepstore/trash_worker.go b/services/keepstore/trash_worker.go index 8f78658c3a..cbb831ebc0 100644 --- a/services/keepstore/trash_worker.go +++ b/services/keepstore/trash_worker.go @@ -1,19 +1,22 @@ +// Copyright (C) The Arvados Authors. All rights reserved. +// +// SPDX-License-Identifier: AGPL-3.0 + package main import ( "errors" - "log" "time" -) -/* - Keepstore initiates trash worker channel goroutine. - The channel will process trash list. - For each (next) trash request: - Delete the block indicated by the trash request Locator - Repeat -*/ + "git.curoverse.com/arvados.git/sdk/go/arvados" +) +// RunTrashWorker is used by Keepstore to initiate trash worker channel goroutine. +// The channel will process trash list. +// For each (next) trash request: +// Delete the block indicated by the trash request Locator +// Repeat +// func RunTrashWorker(trashq *WorkQueue) { for item := range trashq.NextItem { trashRequest := item.(TrashRequest) @@ -24,32 +27,42 @@ func RunTrashWorker(trashq *WorkQueue) { // TrashItem deletes the indicated block from every writable volume. func TrashItem(trashRequest TrashRequest) { - reqMtime := time.Unix(trashRequest.BlockMtime, 0) - if time.Since(reqMtime) < blob_signature_ttl { - log.Printf("WARNING: data manager asked to delete a %v old block %v (BlockMtime %d = %v), but my blob_signature_ttl is %v! Skipping.", - time.Since(reqMtime), + reqMtime := time.Unix(0, trashRequest.BlockMtime) + if time.Since(reqMtime) < theConfig.BlobSignatureTTL.Duration() { + log.Printf("WARNING: data manager asked to delete a %v old block %v (BlockMtime %d = %v), but my blobSignatureTTL is %v! Skipping.", + arvados.Duration(time.Since(reqMtime)), trashRequest.Locator, trashRequest.BlockMtime, reqMtime, - blob_signature_ttl) + theConfig.BlobSignatureTTL) + return + } + + var volumes []Volume + if uuid := trashRequest.MountUUID; uuid == "" { + volumes = KeepVM.AllWritable() + } else if v := KeepVM.Lookup(uuid, true); v == nil { + log.Printf("warning: trash request for nonexistent mount: %v", trashRequest) return + } else { + volumes = []Volume{v} } - for _, volume := range KeepVM.AllWritable() { + for _, volume := range volumes { mtime, err := volume.Mtime(trashRequest.Locator) if err != nil { log.Printf("%v Delete(%v): %v", volume, trashRequest.Locator, err) continue } - if trashRequest.BlockMtime != mtime.Unix() { - log.Printf("%v Delete(%v): mtime on volume is %v does not match trash list value %v", volume, trashRequest.Locator, mtime.Unix(), trashRequest.BlockMtime) + if trashRequest.BlockMtime != mtime.UnixNano() { + log.Printf("%v Delete(%v): stored mtime %v does not match trash list value %v", volume, trashRequest.Locator, mtime.UnixNano(), trashRequest.BlockMtime) continue } - if never_delete { - err = errors.New("did not delete block because never_delete is true") + if !theConfig.EnableDelete { + err = errors.New("did not delete block because EnableDelete is false") } else { - err = volume.Delete(trashRequest.Locator) + err = volume.Trash(trashRequest.Locator) } if err != nil {