package main
import (
+ "encoding/json"
+ "fmt"
"git.curoverse.com/arvados.git/sdk/go/arvadosclient"
"git.curoverse.com/arvados.git/sdk/go/arvadostest"
"git.curoverse.com/arvados.git/sdk/go/keepclient"
- "git.curoverse.com/arvados.git/services/datamanager/keep"
"io/ioutil"
- "log"
"net/http"
"os"
"os/exec"
"time"
)
+const (
+ ActiveUserToken = "3kg6k6lzmp9kj5cpkcoxie963cmvjahbt2fod9zru30k1jqdmi"
+ AdminToken = "4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h"
+)
+
var arv arvadosclient.ArvadosClient
var keepClient *keepclient.KeepClient
var keepServers []string
// start api and keep servers
arvadostest.ResetEnv()
arvadostest.StartAPI()
- arvadostest.StartKeep()
+ arvadostest.StartKeep(2, false)
- // make arvadosclient
- var err error
- arv, err = arvadosclient.MakeArvadosClient()
- if err != nil {
- t.Fatal("Error creating arv")
- }
+ arv = makeArvadosClient()
// keep client
keepClient = &keepclient.KeepClient{
Arvados: &arv,
- Want_replicas: 1,
+ Want_replicas: 2,
Using_proxy: true,
Client: &http.Client{},
}
// discover keep services
if err := keepClient.DiscoverKeepServers(); err != nil {
- t.Fatal("Error discovering keep services")
+ t.Fatalf("Error discovering keep services: %s", err)
}
+ keepServers = []string{}
for _, host := range keepClient.LocalRoots() {
keepServers = append(keepServers, host)
}
}
func TearDownDataManagerTest(t *testing.T) {
- arvadostest.StopKeep()
+ arvadostest.StopKeep(2)
arvadostest.StopAPI()
}
-func PutBlock(t *testing.T, data string) string {
+func putBlock(t *testing.T, data string) string {
locator, _, err := keepClient.PutB([]byte(data))
if err != nil {
t.Fatalf("Error putting test data for %s %s %v", data, locator, err)
t.Fatalf("No locator found after putting test data")
}
- return locator
+ splits := strings.Split(locator, "+")
+ return splits[0] + "+" + splits[1]
}
-func GetBlock(t *testing.T, locator string, data string) {
+func getBlock(t *testing.T, locator string, data string) {
reader, blocklen, _, err := keepClient.Get(locator)
if err != nil {
t.Fatalf("Error getting test data in setup for %s %s %v", data, locator, err)
}
// Create a collection using arv-put
-func CreateCollection(t *testing.T, data string) string {
+func createCollection(t *testing.T, data string) string {
tempfile, err := ioutil.TempFile(os.TempDir(), "temp-test-file")
defer os.Remove(tempfile.Name())
t.Fatalf("Error running arv-put %s", err)
}
- collection_uuid := string(output[0:27]) // trim terminating char
- return collection_uuid
+ uuid := string(output[0:27]) // trim terminating char
+ return uuid
}
-// Get collection using arv-get
-var locatorMatcher = regexp.MustCompile("^([0-9a-f]{32})([+](.*))?$")
+// Get collection locator
+var locatorMatcher = regexp.MustCompile(`^([0-9a-f]{32})\+(\d*)(.*)$`)
+
+func getFirstLocatorFromCollection(t *testing.T, uuid string) string {
+ manifest := getCollection(t, uuid)["manifest_text"].(string)
+
+ locator := strings.Split(manifest, " ")[1]
+ match := locatorMatcher.FindStringSubmatch(locator)
+ if match == nil {
+ t.Fatalf("No locator found in collection manifest %s", manifest)
+ }
-func GetCollection(t *testing.T, collection_uuid string) string {
- // get collection
- output, err := exec.Command("arv-get", collection_uuid).Output()
+ return match[1] + "+" + match[2]
+}
+func getCollection(t *testing.T, uuid string) Dict {
+ getback := make(Dict)
+ err := arv.Get("collections", uuid, nil, &getback)
if err != nil {
- t.Fatalf("Error during arv-get %s", err)
+ t.Fatalf("Error getting collection %s", err)
+ }
+ if getback["uuid"] != uuid {
+ t.Fatalf("Get collection uuid did not match original: $s, result: $s", uuid, getback["uuid"])
}
- locator := strings.Split(string(output), " ")[1]
- match := locatorMatcher.FindStringSubmatch(locator)
- if match == nil {
- t.Fatalf("No locator found in collection manifest %s", string(output))
+ return getback
+}
+
+func updateCollection(t *testing.T, uuid string, paramName string, paramValue string) {
+ err := arv.Update("collections", uuid, arvadosclient.Dict{
+ "collection": arvadosclient.Dict{
+ paramName: paramValue,
+ },
+ }, &arvadosclient.Dict{})
+
+ if err != nil {
+ t.Fatalf("Error updating collection %s", err)
}
- return match[1]
}
type Dict map[string]interface{}
-func DeleteCollection(t *testing.T, collection_uuid string) {
+func deleteCollection(t *testing.T, uuid string) {
getback := make(Dict)
- err := arv.Delete("collections", collection_uuid, nil, &getback)
+ err := arv.Delete("collections", uuid, nil, &getback)
if err != nil {
t.Fatalf("Error deleting collection %s", err)
}
- if getback["uuid"] != collection_uuid {
- t.Fatalf("Delete collection uuid did not match original: $s, result: $s", collection_uuid, getback["uuid"])
+ if getback["uuid"] != uuid {
+ t.Fatalf("Delete collection uuid did not match original: $s, result: $s", uuid, getback["uuid"])
}
}
-func DataManagerSingleRun(t *testing.T) {
- err := singlerun()
+func dataManagerSingleRun(t *testing.T) {
+ err := singlerun(arv)
if err != nil {
t.Fatalf("Error during singlerun %s", err)
}
}
-func MakeRequest(t *testing.T, path string) string {
+func getBlockIndexesForServer(t *testing.T, i int) []string {
+ var indexes []string
+
+ path := keepServers[i] + "/index"
client := http.Client{}
- req, err := http.NewRequest("GET", path, strings.NewReader("resp"))
- req.Header.Add("Authorization", "OAuth2 "+keep.GetDataManagerToken(nil))
+ req, err := http.NewRequest("GET", path, nil)
+ req.Header.Add("Authorization", "OAuth2 "+AdminToken)
req.Header.Add("Content-Type", "application/octet-stream")
- // resp, err := client.Do(req)
resp, err := client.Do(req)
+ defer resp.Body.Close()
+
if err != nil {
t.Fatalf("Error during %s %s", path, err)
}
t.Fatalf("Error reading response from %s %s", path, err)
}
- return string(body)
+ lines := strings.Split(string(body), "\n")
+ for _, line := range lines {
+ indexes = append(indexes, strings.Split(line, " ")...)
+ }
+
+ return indexes
}
-func GetBlockIndexes(t *testing.T) []string {
- var indexes []string
+func getBlockIndexes(t *testing.T) [][]string {
+ var indexes [][]string
for i := 0; i < len(keepServers); i++ {
- resp := MakeRequest(t, keepServers[i]+"/index")
- lines := strings.Split(resp, "\n")
- for _, line := range lines {
- indexes = append(indexes, strings.Split(line, " ")...)
- }
+ indexes = append(indexes, getBlockIndexesForServer(t, i))
}
-
return indexes
}
-func VerifyBlocks(t *testing.T, not_expected []string, expected []string) {
- blocks := GetBlockIndexes(t)
- for _, block := range not_expected {
- exists := ValueInArray(block, blocks)
- if exists {
- t.Fatalf("Found unexpected block in index %s", block)
+func verifyBlocks(t *testing.T, notExpected []string, expected []string, minReplication int) {
+ blocks := getBlockIndexes(t)
+
+ for _, block := range notExpected {
+ for _, idx := range blocks {
+ if valueInArray(block, idx) {
+ t.Fatalf("Found unexpected block %s", block)
+ }
}
}
+
for _, block := range expected {
- exists := ValueInArray(block, blocks)
- if !exists {
- t.Fatalf("Did not find expected block in index %s", block)
+ nFound := 0
+ for _, idx := range blocks {
+ if valueInArray(block, idx) {
+ nFound++
+ }
+ }
+ if nFound < minReplication {
+ t.Fatalf("Found %d replicas of block %s, expected >= %d", nFound, block, minReplication)
}
}
}
-func ValueInArray(value string, list []string) bool {
+func valueInArray(value string, list []string) bool {
for _, v := range list {
- if strings.HasPrefix(v, value) {
+ if value == v {
return true
}
}
return false
}
+/*
+Test env uses two keep volumes. The volume names can be found by reading the files
+ ARVADOS_HOME/tmp/keep0.volume and ARVADOS_HOME/tmp/keep1.volume
+
+The keep volumes are of the dir structure:
+ volumeN/subdir/locator
+*/
+func backdateBlocks(t *testing.T, oldUnusedBlockLocators []string) {
+ // First get rid of any size hints in the locators
+ var trimmedBlockLocators []string
+ for _, block := range oldUnusedBlockLocators {
+ trimmedBlockLocators = append(trimmedBlockLocators, strings.Split(block, "+")[0])
+ }
+
+ // Get the working dir so that we can read keep{n}.volume files
+ wd, err := os.Getwd()
+ if err != nil {
+ t.Fatalf("Error getting working dir %s", err)
+ }
+
+ // Now cycle through the two keep volumes
+ oldTime := time.Now().AddDate(0, -2, 0)
+ for i := 0; i < 2; i++ {
+ filename := fmt.Sprintf("%s/../../tmp/keep%d.volume", wd, i)
+ volumeDir, err := ioutil.ReadFile(filename)
+ if err != nil {
+ t.Fatalf("Error reading keep volume file %s %s", filename, err)
+ }
+
+ // Read the keep volume dir structure
+ volumeContents, err := ioutil.ReadDir(string(volumeDir))
+ if err != nil {
+ t.Fatalf("Error reading keep dir %s %s", string(volumeDir), err)
+ }
+
+ // Read each subdir for each of the keep volume dir
+ for _, subdir := range volumeContents {
+ subdirName := fmt.Sprintf("%s/%s", volumeDir, subdir.Name())
+ subdirContents, err := ioutil.ReadDir(string(subdirName))
+ if err != nil {
+ t.Fatalf("Error reading keep dir %s %s", string(subdirName), err)
+ }
+
+ // Now we got to the files. The files are names are the block locators
+ for _, fileInfo := range subdirContents {
+ blockName := fileInfo.Name()
+ myname := fmt.Sprintf("%s/%s", subdirName, blockName)
+ if valueInArray(blockName, trimmedBlockLocators) {
+ err = os.Chtimes(myname, oldTime, oldTime)
+ }
+ }
+ }
+ }
+}
+
+func getStatus(t *testing.T, path string) interface{} {
+ client := http.Client{}
+ req, err := http.NewRequest("GET", path, nil)
+ req.Header.Add("Authorization", "OAuth2 "+AdminToken)
+ req.Header.Add("Content-Type", "application/octet-stream")
+ resp, err := client.Do(req)
+ if err != nil {
+ t.Fatalf("Error during %s %s", path, err)
+ }
+ defer resp.Body.Close()
+
+ var s interface{}
+ json.NewDecoder(resp.Body).Decode(&s)
+
+ return s
+}
+
+// Wait until PullQueue and TrashQueue are empty on all keepServers.
+func waitUntilQueuesFinishWork(t *testing.T) {
+ for _, ks := range keepServers {
+ for done := false; !done; {
+ time.Sleep(100 * time.Millisecond)
+ s := getStatus(t, ks+"/status.json")
+ for _, qName := range []string{"PullQueue", "TrashQueue"} {
+ qStatus := s.(map[string]interface{})[qName].(map[string]interface{})
+ if qStatus["Queued"].(float64)+qStatus["InProgress"].(float64) == 0 {
+ done = true
+ }
+ }
+ }
+ }
+}
+
+/*
+Create some blocks and backdate some of them.
+Also create some collections and delete some of them.
+Verify block indexes.
+*/
func TestPutAndGetBlocks(t *testing.T) {
- log.Print("TestPutAndGetBlocks start")
defer TearDownDataManagerTest(t)
SetupDataManagerTest(t)
- // Put some blocks and change their mtime to old
- var oldBlockLocators []string
- oldBlockData := "this block will have older mtime"
- for i := 0; i < 2; i++ {
- oldBlockLocators = append(oldBlockLocators, PutBlock(t, oldBlockData+string(i)))
+ // Put some blocks which will be backdated later on
+ // The first one will also be used in a collection and hence should not be deleted when datamanager runs.
+ // The rest will be old and unreferenced and hence should be deleted when datamanager runs.
+ var oldUnusedBlockLocators []string
+ oldUnusedBlockData := "this block will have older mtime"
+ for i := 0; i < 5; i++ {
+ oldUnusedBlockLocators = append(oldUnusedBlockLocators, putBlock(t, fmt.Sprintf("%s%d", oldUnusedBlockData, i)))
}
- for i := 0; i < 2; i++ {
- GetBlock(t, oldBlockLocators[i], oldBlockData+string(i))
+ for i := 0; i < 5; i++ {
+ getBlock(t, oldUnusedBlockLocators[i], fmt.Sprintf("%s%d", oldUnusedBlockData, i))
}
- // Put some more new blocks
+ // The rest will be old and unreferenced and hence should be deleted when datamanager runs.
+ oldUsedBlockData := "this collection block will have older mtime"
+ oldUsedBlockLocator := putBlock(t, oldUsedBlockData)
+ getBlock(t, oldUsedBlockLocator, oldUsedBlockData)
+
+ // Put some more blocks which will not be backdated; hence they are still new, but not in any collection.
+ // Hence, even though unreferenced, these should not be deleted when datamanager runs.
var newBlockLocators []string
newBlockData := "this block is newer"
- for i := 0; i < 1; i++ {
- newBlockLocators = append(newBlockLocators, PutBlock(t, newBlockData+string(i)))
+ for i := 0; i < 5; i++ {
+ newBlockLocators = append(newBlockLocators, putBlock(t, fmt.Sprintf("%s%d", newBlockData, i)))
}
- for i := 0; i < 1; i++ {
- GetBlock(t, newBlockLocators[i], newBlockData+string(i))
+ for i := 0; i < 5; i++ {
+ getBlock(t, newBlockLocators[i], fmt.Sprintf("%s%d", newBlockData, i))
}
- // Create a collection
- collection_uuid := CreateCollection(t, "some data for collection creation")
-
- collection_locator := GetCollection(t, collection_uuid)
+ // Create a collection that would be deleted later on
+ toBeDeletedCollectionUUID := createCollection(t, "some data for collection creation")
+ toBeDeletedCollectionLocator := getFirstLocatorFromCollection(t, toBeDeletedCollectionUUID)
- /*
- // Invoking datamanager singlerun or /index several times is resulting in errors
- // Hence, for now just invoke once at the end of test
- var expected []string
- expected = append(expected, oldBlockLocators...)
- expected = append(expected, newBlockLocators...)
- expected = append(expected, collection_locator)
+ // Create another collection that has the same data as the one of the old blocks
+ oldUsedBlockCollectionUUID := createCollection(t, oldUsedBlockData)
+ oldUsedBlockCollectionLocator := getFirstLocatorFromCollection(t, oldUsedBlockCollectionUUID)
+ if oldUsedBlockCollectionLocator != oldUsedBlockLocator {
+ t.Fatalf("Locator of the collection with the same data as old block is different %s", oldUsedBlockCollectionLocator)
+ }
- VerifyBlocks(t, nil, expected)
+ // Create another collection whose replication level will be changed
+ replicationCollectionUUID := createCollection(t, "replication level on this collection will be reduced")
+ replicationCollectionLocator := getFirstLocatorFromCollection(t, replicationCollectionUUID)
+
+ // Create two collections with same data; one will be deleted later on
+ dataForTwoCollections := "one of these collections will be deleted"
+ oneOfTwoWithSameDataUUID := createCollection(t, dataForTwoCollections)
+ oneOfTwoWithSameDataLocator := getFirstLocatorFromCollection(t, oneOfTwoWithSameDataUUID)
+ secondOfTwoWithSameDataUUID := createCollection(t, dataForTwoCollections)
+ secondOfTwoWithSameDataLocator := getFirstLocatorFromCollection(t, secondOfTwoWithSameDataUUID)
+ if oneOfTwoWithSameDataLocator != secondOfTwoWithSameDataLocator {
+ t.Fatalf("Locators for both these collections expected to be same: %s %s", oneOfTwoWithSameDataLocator, secondOfTwoWithSameDataLocator)
+ }
- // Run datamanager in singlerun mode
- DataManagerSingleRun(t)
- */
+ // Verify blocks before doing any backdating / deleting.
+ var expected []string
+ expected = append(expected, oldUnusedBlockLocators...)
+ expected = append(expected, newBlockLocators...)
+ expected = append(expected, toBeDeletedCollectionLocator)
+ expected = append(expected, replicationCollectionLocator)
+ expected = append(expected, oneOfTwoWithSameDataLocator)
+ expected = append(expected, secondOfTwoWithSameDataLocator)
+
+ verifyBlocks(t, nil, expected, 2)
+
+ // Run datamanager in singlerun mode
+ dataManagerSingleRun(t)
+ waitUntilQueuesFinishWork(t)
+
+ verifyBlocks(t, nil, expected, 2)
+
+ // Backdate the to-be old blocks and delete the collections
+ backdateBlocks(t, oldUnusedBlockLocators)
+ deleteCollection(t, toBeDeletedCollectionUUID)
+ deleteCollection(t, secondOfTwoWithSameDataUUID)
+
+ // Run data manager again
+ dataManagerSingleRun(t)
+ waitUntilQueuesFinishWork(t)
+
+ // Get block indexes and verify that all backdated blocks except the first one used in collection are not included.
+ expected = expected[:0]
+ expected = append(expected, oldUsedBlockLocator)
+ expected = append(expected, newBlockLocators...)
+ expected = append(expected, toBeDeletedCollectionLocator)
+ expected = append(expected, oneOfTwoWithSameDataLocator)
+ expected = append(expected, secondOfTwoWithSameDataLocator)
+
+ verifyBlocks(t, oldUnusedBlockLocators, expected, 2)
+
+ // Reduce desired replication on replicationCollectionUUID
+ // collection, and verify that Data Manager does not reduce
+ // actual replication any further than that. (It might not
+ // reduce actual replication at all; that's OK for this test.)
+
+ // Reduce desired replication level.
+ updateCollection(t, replicationCollectionUUID, "replication_desired", "1")
+ collection := getCollection(t, replicationCollectionUUID)
+ if collection["replication_desired"].(interface{}) != float64(1) {
+ t.Fatalf("After update replication_desired is not 1; instead it is %v", collection["replication_desired"])
+ }
- // Change mtime on old blocks and delete the collection
- DeleteCollection(t, collection_uuid)
+ // Verify data is currently overreplicated.
+ verifyBlocks(t, nil, []string{replicationCollectionLocator}, 2)
- time.Sleep(1 * time.Second)
- DataManagerSingleRun(t)
+ // Run data manager again
+ dataManagerSingleRun(t)
+ waitUntilQueuesFinishWork(t)
- // Give some time for pull worker and trash worker to finish
- time.Sleep(10 * time.Second)
+ // Verify data is not underreplicated.
+ verifyBlocks(t, nil, []string{replicationCollectionLocator}, 1)
- // Get block indexes and verify that the deleted collection block is no longer found
- var not_expected []string
- not_expected = append(not_expected, oldBlockLocators...)
- not_expected = append(not_expected, collection_locator)
- //VerifyBlocks(t, not_expected, newBlockLocators)
- VerifyBlocks(t, nil, newBlockLocators)
+ // Verify *other* collections' data is not underreplicated.
+ verifyBlocks(t, oldUnusedBlockLocators, expected, 2)
}
-// Invoking datamanager singlerun several times results in errors.
-// Until that issue is resolved, don't run this test in the meantime.
-func x_TestInvokeDatamanagerSingleRunRepeatedly(t *testing.T) {
+func TestDatamanagerSingleRunRepeatedly(t *testing.T) {
defer TearDownDataManagerTest(t)
SetupDataManagerTest(t)
for i := 0; i < 10; i++ {
- err := singlerun()
+ err := singlerun(arv)
if err != nil {
t.Fatalf("Got an error during datamanager singlerun: %v", err)
}
- time.Sleep(1 * time.Second)
+ }
+}
+
+func TestGetStatusRepeatedly(t *testing.T) {
+ defer TearDownDataManagerTest(t)
+ SetupDataManagerTest(t)
+
+ for i := 0; i < 10; i++ {
+ for j := 0; j < 2; j++ {
+ s := getStatus(t, keepServers[j]+"/status.json")
+
+ var pullQueueStatus interface{}
+ pullQueueStatus = s.(map[string]interface{})["PullQueue"]
+ var trashQueueStatus interface{}
+ trashQueueStatus = s.(map[string]interface{})["TrashQueue"]
+
+ if pullQueueStatus.(map[string]interface{})["Queued"] == nil ||
+ pullQueueStatus.(map[string]interface{})["InProgress"] == nil ||
+ trashQueueStatus.(map[string]interface{})["Queued"] == nil ||
+ trashQueueStatus.(map[string]interface{})["InProgress"] == nil {
+ t.Fatalf("PullQueue and TrashQueue status not found")
+ }
+
+ time.Sleep(100 * time.Millisecond)
+ }
+ }
+}
+
+func TestRunDatamanagerWithBogusServer(t *testing.T) {
+ defer TearDownDataManagerTest(t)
+ SetupDataManagerTest(t)
+
+ arv.ApiServer = "bogus-server"
+
+ err := singlerun(arv)
+ if err == nil {
+ t.Fatalf("Expected error during singlerun with bogus server")
+ }
+}
+
+func TestRunDatamanagerAsNonAdminUser(t *testing.T) {
+ defer TearDownDataManagerTest(t)
+ SetupDataManagerTest(t)
+
+ arv.ApiToken = ActiveUserToken
+
+ err := singlerun(arv)
+ if err == nil {
+ t.Fatalf("Expected error during singlerun as non-admin user")
}
}
projects / arvados.git / blobdiff