X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/505be2eb9ac1f36a5e31b374451193bc1ab7e90e..f9ee2fb124ae78696a0a0bb2ac0f8228f7446505:/services/datamanager/datamanager_test.go diff --git a/services/datamanager/datamanager_test.go b/services/datamanager/datamanager_test.go index 4f9b149123..4dc940e5a2 100644 --- a/services/datamanager/datamanager_test.go +++ b/services/datamanager/datamanager_test.go @@ -1,6 +1,7 @@ package main import ( + "encoding/json" "fmt" "git.curoverse.com/arvados.git/sdk/go/arvadosclient" "git.curoverse.com/arvados.git/sdk/go/arvadostest" @@ -67,7 +68,8 @@ func PutBlock(t *testing.T, data string) string { 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) { @@ -111,7 +113,7 @@ func CreateCollection(t *testing.T, data string) string { // Get collection using arv-get var locatorMatcher = regexp.MustCompile(`^([0-9a-f]{32})\+(\d*)(.*)$`) -func GetCollection(t *testing.T, uuid string) string { +func GetCollectionManifest(t *testing.T, uuid string) string { output, err := exec.Command("arv-get", uuid).Output() if err != nil { t.Fatalf("Error during arv-get %s", err) @@ -126,6 +128,31 @@ func GetCollection(t *testing.T, uuid string) string { 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 getting collection %s", err) + } + if getback["uuid"] != uuid { + t.Fatalf("Get collection uuid did not match original: $s, result: $s", uuid, getback["uuid"]) + } + + 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) + } +} + type Dict map[string]interface{} func DeleteCollection(t *testing.T, uuid string) { @@ -146,12 +173,17 @@ func DataManagerSingleRun(t *testing.T) { } } -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, err := http.NewRequest("GET", path, nil) req.Header.Add("Authorization", "OAuth2 "+keep.GetDataManagerToken(nil)) req.Header.Add("Content-Type", "application/octet-stream") resp, err := client.Do(req) + defer resp.Body.Close() + if err != nil { t.Fatalf("Error during %s %s", path, err) } @@ -161,26 +193,26 @@ func MakeRequest(t *testing.T, path string) string { 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 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) { +func VerifyBlocks(t *testing.T, notExpected []string, expected []string) { blocks := GetBlockIndexes(t) - for _, block := range not_expected { + for _, block := range notExpected { exists := ValueInArray(block, blocks) if exists { t.Fatalf("Found unexpected block in index %s", block) @@ -224,7 +256,7 @@ func BackdateBlocks(t *testing.T, oldBlockLocators []string) { } // Now cycle through the two keep volumes - oldTime := time.Now().AddDate(0, -1, 0) + 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) @@ -258,88 +290,189 @@ func BackdateBlocks(t *testing.T, oldBlockLocators []string) { } } -func GetStatus(t *testing.T) { - for i := 0; i < len(keepServers); i++ { - resp := MakeRequest(t, keepServers[i]+"/status.json") - log.Printf("Status from keepserver %d = %s", i, resp) +func GetStatus(t *testing.T, path string) interface{} { + client := http.Client{} + req, err := http.NewRequest("GET", path, nil) + req.Header.Add("Authorization", "OAuth2 "+keep.GetDataManagerToken(nil)) + req.Header.Add("Content-Type", "application/octet-stream") + resp, err := client.Do(req) + defer resp.Body.Close() + + if err != nil { + t.Fatalf("Error during %s %s", path, err) + } + + var s interface{} + json.NewDecoder(resp.Body).Decode(&s) + + return s +} + +func WaitUntilQueuesFinishWork(t *testing.T) { + // Wait until PullQueue and TrashQueue finish their work + for { + var done [2]bool + for i := 0; i < 2; i++ { + s := GetStatus(t, keepServers[i]+"/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"] == float64(0) && + pullQueueStatus.(map[string]interface{})["InProgress"] == float64(0) && + trashQueueStatus.(map[string]interface{})["Queued"] == float64(0) && + trashQueueStatus.(map[string]interface{})["InProgress"] == float64(0) { + done[i] = true + } + } + if done[0] && done[1] { + break + } else { + time.Sleep(1 * time.Second) + } } } +/* +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 before ttl + // 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 oldBlockLocators []string oldBlockData := "this block will have older mtime" - for i := 0; i < 2; i++ { - oldBlockLocators = append(oldBlockLocators, PutBlock(t, oldBlockData+string(i))) + for i := 0; i < 5; i++ { + oldBlockLocators = append(oldBlockLocators, PutBlock(t, fmt.Sprintf("%s%d", oldBlockData, i))) } - for i := 0; i < 2; i++ { - GetBlock(t, oldBlockLocators[i], oldBlockData+string(i)) + for i := 0; i < 5; i++ { + GetBlock(t, oldBlockLocators[i], fmt.Sprintf("%s%d", oldBlockData, i)) } - // Put some more blocks whose mtime won't be changed + // 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 that would be deleted - to_delete_collection_uuid := CreateCollection(t, "some data for collection creation") - to_delete_collection_locator := GetCollection(t, to_delete_collection_uuid) + // Create a collection that would be deleted later on + toBeDeletedCollectionUuid := CreateCollection(t, "some data for collection creation") + toBeDeletedCollectionLocator := GetCollectionManifest(t, toBeDeletedCollectionUuid) // Create another collection that has the same data as the one of the old blocks - old_block_collection_uuid := CreateCollection(t, "this block will have older mtime0") - old_block_collection_locator := GetCollection(t, old_block_collection_uuid) - exists := ValueInArray(strings.Split(old_block_collection_locator, "+")[0], oldBlockLocators) + oldBlockCollectionUuid := CreateCollection(t, "this block will have older mtime0") + oldBlockCollectionLocator := GetCollectionManifest(t, oldBlockCollectionUuid) + exists := ValueInArray(strings.Split(oldBlockCollectionLocator, "+")[0], oldBlockLocators) if exists { - t.Fatalf("Locator of the collection with the same data as old block is different %s", old_block_collection_locator) + t.Fatalf("Locator of the collection with the same data as old block is different %s", oldBlockCollectionLocator) } - GetStatus(t) + // Create another collection whose replication level will be changed + replicationCollectionUuid := CreateCollection(t, "replication level on this collection will be reduced") + replicationCollectionLocator := GetCollectionManifest(t, replicationCollectionUuid) - /* - // 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, to_delete_collection_locator) + // 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 := GetCollectionManifest(t, oneOfTwoWithSameDataUuid) + secondOfTwoWithSameDataUuid := CreateCollection(t, dataForTwoCollections) + secondOfTwoWithSameDataLocator := GetCollectionManifest(t, secondOfTwoWithSameDataUuid) + if oneOfTwoWithSameDataLocator != secondOfTwoWithSameDataLocator { + t.Fatalf("Locators for both these collections expected to be same: %s %s", oneOfTwoWithSameDataLocator, secondOfTwoWithSameDataLocator) + } - VerifyBlocks(t, nil, expected) + // Verify blocks before doing any backdating / deleting. + var expected []string + expected = append(expected, oldBlockLocators...) + expected = append(expected, newBlockLocators...) + expected = append(expected, toBeDeletedCollectionLocator) + expected = append(expected, replicationCollectionLocator) + expected = append(expected, oneOfTwoWithSameDataLocator) + expected = append(expected, secondOfTwoWithSameDataLocator) - // Run datamanager in singlerun mode - DataManagerSingleRun(t) - */ + VerifyBlocks(t, nil, expected) - // Change mtime on old blocks and delete the collection - DeleteCollection(t, to_delete_collection_uuid) + // Run datamanager in singlerun mode + DataManagerSingleRun(t) + WaitUntilQueuesFinishWork(t) + + log.Print("Backdating blocks and deleting collection now") + + // Backdate the to-be old blocks and delete the collections BackdateBlocks(t, oldBlockLocators) + DeleteCollection(t, toBeDeletedCollectionUuid) + DeleteCollection(t, secondOfTwoWithSameDataUuid) + + // Run data manager again + time.Sleep(1 * time.Second) + DataManagerSingleRun(t) + WaitUntilQueuesFinishWork(t) + + // Get block indexes and verify that all backdated blocks except the first one are not included. + // The first block was also used in a collection that is not deleted and hence should remain. + var notExpected []string + notExpected = append(notExpected, oldBlockLocators[1:]...) + + expected = expected[:0] + expected = append(expected, oldBlockLocators[0]) + expected = append(expected, newBlockLocators...) + expected = append(expected, toBeDeletedCollectionLocator) + expected = append(expected, replicationCollectionLocator) + expected = append(expected, oneOfTwoWithSameDataLocator) + expected = append(expected, secondOfTwoWithSameDataLocator) + + VerifyBlocks(t, notExpected, expected) + + // Reduce replication on replicationCollectionUuid collection and verify that the overreplicated blocks are untouched. + + // Default replication level is 2; first verify that the replicationCollectionLocator appears in both volumes + for i := 0; i < len(keepServers); i++ { + indexes := GetBlockIndexesForServer(t, i) + if !ValueInArray(replicationCollectionLocator, indexes) { + t.Fatalf("Not found block in index %s", replicationCollectionLocator) + } + } + + // Now reduce replication level on this collection and verify that it still appears in both volumes + 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"]) + } + // Run data manager again time.Sleep(1 * time.Second) DataManagerSingleRun(t) + WaitUntilQueuesFinishWork(t) - // Give some time for pull worker and trash worker to finish - time.Sleep(5 * time.Second) + for i := 0; i < len(keepServers); i++ { + indexes := GetBlockIndexesForServer(t, i) + if !ValueInArray(replicationCollectionLocator, indexes) { + t.Fatalf("Not found block in index %s", replicationCollectionLocator) + } + } - // 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, to_delete_collection_locator) + // Done testing reduce replication on collection - GetStatus(t) - VerifyBlocks(t, not_expected, newBlockLocators) + // Verify blocks one more time + VerifyBlocks(t, notExpected, expected) } -// 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) { + log.Print("TestDatamanagerSingleRunRepeatedly start") + defer TearDownDataManagerTest(t) SetupDataManagerTest(t) @@ -351,3 +484,30 @@ func x_TestInvokeDatamanagerSingleRunRepeatedly(t *testing.T) { time.Sleep(1 * time.Second) } } + +func _TestGetStatusRepeatedly(t *testing.T) { + log.Print("TestGetStatusRepeatedly start") + + 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(1 * time.Second) + } + } +}