X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/44321d4b5e7677df2ad70d88db6358b2778b58e5..86e1730f97383b3ae1685445323aa253b99ee821:/services/datamanager/datamanager_test.go diff --git a/services/datamanager/datamanager_test.go b/services/datamanager/datamanager_test.go index 39e545dcb8..0ff6b77734 100644 --- a/services/datamanager/datamanager_test.go +++ b/services/datamanager/datamanager_test.go @@ -6,12 +6,13 @@ import ( "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" + "git.curoverse.com/arvados.git/services/datamanager/collection" + "git.curoverse.com/arvados.git/services/datamanager/summary" "io/ioutil" - "log" "net/http" "os" "os/exec" + "path" "regexp" "strings" "testing" @@ -28,35 +29,37 @@ func SetupDataManagerTest(t *testing.T) { // 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.Fatalf("Error setting up arvados client: %s", err) + t.Fatalf("Error making arvados client: %s", err) } + arv.ApiToken = arvadostest.DataManagerToken // keep client keepClient = &keepclient.KeepClient{ Arvados: &arv, Want_replicas: 2, - Using_proxy: true, Client: &http.Client{}, } // discover keep services - if err := keepClient.DiscoverKeepServers(); err != nil { + if err = keepClient.DiscoverKeepServers(); err != nil { 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() + summary.WriteDataTo = "" + collection.HeapProfileFilename = "" } func putBlock(t *testing.T, data string) string { @@ -113,7 +116,7 @@ func createCollection(t *testing.T, data string) string { // Get collection locator var locatorMatcher = regexp.MustCompile(`^([0-9a-f]{32})\+(\d*)(.*)$`) -func getCollectionLocator(t *testing.T, uuid string) string { +func getFirstLocatorFromCollection(t *testing.T, uuid string) string { manifest := getCollection(t, uuid)["manifest_text"].(string) locator := strings.Split(manifest, " ")[1] @@ -125,7 +128,18 @@ func getCollectionLocator(t *testing.T, uuid string) string { return match[1] + "+" + match[2] } +func switchToken(t string) func() { + orig := arv.ApiToken + restore := func() { + arv.ApiToken = orig + } + arv.ApiToken = t + return restore +} + func getCollection(t *testing.T, uuid string) Dict { + defer switchToken(arvadostest.AdminToken)() + getback := make(Dict) err := arv.Get("collections", uuid, nil, &getback) if err != nil { @@ -139,6 +153,8 @@ func getCollection(t *testing.T, uuid string) Dict { } func updateCollection(t *testing.T, uuid string, paramName string, paramValue string) { + defer switchToken(arvadostest.AdminToken)() + err := arv.Update("collections", uuid, arvadosclient.Dict{ "collection": arvadosclient.Dict{ paramName: paramValue, @@ -153,6 +169,8 @@ func updateCollection(t *testing.T, uuid string, paramName string, paramValue st type Dict map[string]interface{} func deleteCollection(t *testing.T, uuid string) { + defer switchToken(arvadostest.AdminToken)() + getback := make(Dict) err := arv.Delete("collections", uuid, nil, &getback) if err != nil { @@ -164,7 +182,7 @@ func deleteCollection(t *testing.T, uuid string) { } func dataManagerSingleRun(t *testing.T) { - err := singlerun() + err := singlerun(arv) if err != nil { t.Fatalf("Error during singlerun %s", err) } @@ -176,7 +194,7 @@ func getBlockIndexesForServer(t *testing.T, i int) []string { path := keepServers[i] + "/index" client := http.Client{} req, err := http.NewRequest("GET", path, nil) - req.Header.Add("Authorization", "OAuth2 "+keep.GetDataManagerToken(nil)) + req.Header.Add("Authorization", "OAuth2 "+arvadostest.DataManagerToken) req.Header.Add("Content-Type", "application/octet-stream") resp, err := client.Do(req) defer resp.Body.Close() @@ -207,33 +225,26 @@ func getBlockIndexes(t *testing.T) [][]string { return indexes } -func verifyBlocks(t *testing.T, notExpected []string, expected []string) { +func verifyBlocks(t *testing.T, notExpected []string, expected []string, minReplication int) { blocks := getBlockIndexes(t) + for _, block := range notExpected { - for i := 0; i < len(blocks); i++ { - exists := valueInArray(block, blocks[i]) - if exists { - t.Fatalf("Found unexpected block in index %s", block) + for _, idx := range blocks { + if valueInArray(block, idx) { + t.Fatalf("Found unexpected block %s", block) } } } - // var blockExists [][]string - blockExists := make(map[string][]string) for _, block := range expected { - var blockArray []string - for i := 0; i < len(blocks); i++ { - exists := valueInArray(block, blocks[i]) - if exists { - blockArray = append(blockArray, block) + nFound := 0 + for _, idx := range blocks { + if valueInArray(block, idx) { + nFound++ } } - blockExists[block] = blockArray - } - - for _, block := range expected { - if blockExists[block] == nil || len(blockExists[block]) != 2 { - t.Fatalf("Expected to find two replicas for block %s; found %d", block, len(blockExists[block])) + if nFound < minReplication { + t.Fatalf("Found %d replicas of block %s, expected >= %d", nFound, block, minReplication) } } } @@ -247,13 +258,10 @@ func valueInArray(value string, list []string) bool { 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 -*/ +// 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 @@ -305,14 +313,13 @@ func backdateBlocks(t *testing.T, oldUnusedBlockLocators []string) { 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("Authorization", "OAuth2 "+arvadostest.DataManagerToken) 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) } + defer resp.Body.Close() var s interface{} json.NewDecoder(resp.Body).Decode(&s) @@ -320,39 +327,26 @@ func getStatus(t *testing.T, path string) interface{} { return s } +// Wait until PullQueue and TrashQueue are empty on all keepServers. 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 { + 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. -*/ +// 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) @@ -385,30 +379,34 @@ func TestPutAndGetBlocks(t *testing.T) { } // Create a collection that would be deleted later on - toBeDeletedCollectionUuid := createCollection(t, "some data for collection creation") - toBeDeletedCollectionLocator := getCollectionLocator(t, toBeDeletedCollectionUuid) + toBeDeletedCollectionUUID := createCollection(t, "some data for collection creation") + toBeDeletedCollectionLocator := getFirstLocatorFromCollection(t, toBeDeletedCollectionUUID) // Create another collection that has the same data as the one of the old blocks - oldUsedBlockCollectionUuid := createCollection(t, oldUsedBlockData) - oldUsedBlockCollectionLocator := getCollectionLocator(t, oldUsedBlockCollectionUuid) + 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) } // Create another collection whose replication level will be changed - replicationCollectionUuid := createCollection(t, "replication level on this collection will be reduced") - replicationCollectionLocator := getCollectionLocator(t, replicationCollectionUuid) + 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 := getCollectionLocator(t, oneOfTwoWithSameDataUuid) - secondOfTwoWithSameDataUuid := createCollection(t, dataForTwoCollections) - secondOfTwoWithSameDataLocator := getCollectionLocator(t, secondOfTwoWithSameDataUuid) + 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) } + // create collection with empty manifest text + emptyBlockLocator := putBlock(t, "") + emptyCollection := createCollection(t, "") + // Verify blocks before doing any backdating / deleting. var expected []string expected = append(expected, oldUnusedBlockLocators...) @@ -417,19 +415,22 @@ func TestPutAndGetBlocks(t *testing.T) { expected = append(expected, replicationCollectionLocator) expected = append(expected, oneOfTwoWithSameDataLocator) expected = append(expected, secondOfTwoWithSameDataLocator) + expected = append(expected, emptyBlockLocator) - verifyBlocks(t, nil, expected) + verifyBlocks(t, nil, expected, 2) // Run datamanager in singlerun mode dataManagerSingleRun(t) waitUntilQueuesFinishWork(t) - verifyBlocks(t, nil, expected) + verifyBlocks(t, nil, expected, 2) // Backdate the to-be old blocks and delete the collections backdateBlocks(t, oldUnusedBlockLocators) - deleteCollection(t, toBeDeletedCollectionUuid) - deleteCollection(t, secondOfTwoWithSameDataUuid) + deleteCollection(t, toBeDeletedCollectionUUID) + deleteCollection(t, secondOfTwoWithSameDataUUID) + backdateBlocks(t, []string{emptyBlockLocator}) + deleteCollection(t, emptyCollection) // Run data manager again dataManagerSingleRun(t) @@ -440,64 +441,51 @@ func TestPutAndGetBlocks(t *testing.T) { expected = append(expected, oldUsedBlockLocator) expected = append(expected, newBlockLocators...) expected = append(expected, toBeDeletedCollectionLocator) - expected = append(expected, replicationCollectionLocator) expected = append(expected, oneOfTwoWithSameDataLocator) expected = append(expected, secondOfTwoWithSameDataLocator) + expected = append(expected, emptyBlockLocator) // even when unreferenced, this remains - verifyBlocks(t, oldUnusedBlockLocators, expected) - - // Reduce replication on replicationCollectionUuid collection and verify that the overreplicated blocks are untouched. + verifyBlocks(t, oldUnusedBlockLocators, expected, 2) - // 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) - } - } + // 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.) - // 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) + // 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"]) } + // Verify data is currently overreplicated. + verifyBlocks(t, nil, []string{replicationCollectionLocator}, 2) + // Run data manager again - time.Sleep(100 * time.Millisecond) dataManagerSingleRun(t) waitUntilQueuesFinishWork(t) - for i := 0; i < len(keepServers); i++ { - indexes := getBlockIndexesForServer(t, i) - if !valueInArray(replicationCollectionLocator, indexes) { - t.Fatalf("Not found block in index %s", replicationCollectionLocator) - } - } - // Done testing reduce replication on collection + // Verify data is not underreplicated. + verifyBlocks(t, nil, []string{replicationCollectionLocator}, 1) - // Verify blocks one more time - verifyBlocks(t, oldUnusedBlockLocators, expected) + // Verify *other* collections' data is not underreplicated. + verifyBlocks(t, oldUnusedBlockLocators, expected, 2) } -func _TestDatamanagerSingleRunRepeatedly(t *testing.T) { - log.Print("TestDatamanagerSingleRunRepeatedly start") - +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(100 * time.Millisecond) } } -func _TestGetStatusRepeatedly(t *testing.T) { - log.Print("TestGetStatusRepeatedly start") - +func TestGetStatusRepeatedly(t *testing.T) { defer TearDownDataManagerTest(t) SetupDataManagerTest(t) @@ -521,3 +509,224 @@ func _TestGetStatusRepeatedly(t *testing.T) { } } } + +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 = arvadostest.ActiveToken + + err := singlerun(arv) + if err == nil { + t.Fatalf("Expected error during singlerun as non-admin user") + } +} + +func TestPutAndGetBlocks_NoErrorDuringSingleRun(t *testing.T) { + testOldBlocksNotDeletedOnDataManagerError(t, "", "", false, false) +} + +func TestPutAndGetBlocks_ErrorDuringGetCollectionsBadWriteTo(t *testing.T) { + badpath, err := arvadostest.CreateBadPath() + if err != nil { + t.Fatalf(err.Error()) + } + defer func() { + err = arvadostest.DestroyBadPath(badpath) + if err != nil { + t.Fatalf(err.Error()) + } + }() + testOldBlocksNotDeletedOnDataManagerError(t, path.Join(badpath, "writetofile"), "", true, true) +} + +func TestPutAndGetBlocks_ErrorDuringGetCollectionsBadHeapProfileFilename(t *testing.T) { + badpath, err := arvadostest.CreateBadPath() + if err != nil { + t.Fatalf(err.Error()) + } + defer func() { + err = arvadostest.DestroyBadPath(badpath) + if err != nil { + t.Fatalf(err.Error()) + } + }() + testOldBlocksNotDeletedOnDataManagerError(t, "", path.Join(badpath, "heapprofilefile"), true, true) +} + +// Create some blocks and backdate some of them. +// Run datamanager while producing an error condition. +// Verify that the blocks are hence not deleted. +func testOldBlocksNotDeletedOnDataManagerError(t *testing.T, writeDataTo string, heapProfileFile string, expectError bool, expectOldBlocks bool) { + defer TearDownDataManagerTest(t) + SetupDataManagerTest(t) + + // Put some blocks and backdate them. + 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))) + } + backdateBlocks(t, oldUnusedBlockLocators) + + // Run data manager + summary.WriteDataTo = writeDataTo + collection.HeapProfileFilename = heapProfileFile + + err := singlerun(arv) + if !expectError { + if err != nil { + t.Fatalf("Got an error during datamanager singlerun: %v", err) + } + } else { + if err == nil { + t.Fatalf("Expected error during datamanager singlerun") + } + } + waitUntilQueuesFinishWork(t) + + // Get block indexes and verify that all backdated blocks are not/deleted as expected + if expectOldBlocks { + verifyBlocks(t, nil, oldUnusedBlockLocators, 2) + } else { + verifyBlocks(t, oldUnusedBlockLocators, nil, 2) + } +} + +// Create a collection with multiple streams and blocks +func createMultiStreamBlockCollection(t *testing.T, data string, numStreams, numBlocks int) (string, []string) { + defer switchToken(arvadostest.AdminToken)() + + manifest := "" + locators := make(map[string]bool) + for s := 0; s < numStreams; s++ { + manifest += fmt.Sprintf("./stream%d ", s) + for b := 0; b < numBlocks; b++ { + locator, _, err := keepClient.PutB([]byte(fmt.Sprintf("%s in stream %d and block %d", data, s, b))) + if err != nil { + t.Fatalf("Error creating block %d in stream %d: %v", b, s, err) + } + locators[strings.Split(locator, "+A")[0]] = true + manifest += locator + " " + } + manifest += "0:1:dummyfile.txt\n" + } + + collection := make(Dict) + err := arv.Create("collections", + arvadosclient.Dict{"collection": arvadosclient.Dict{"manifest_text": manifest}}, + &collection) + + if err != nil { + t.Fatalf("Error creating collection %v", err) + } + + var locs []string + for k := range locators { + locs = append(locs, k) + } + + return collection["uuid"].(string), locs +} + +// Create collection with multiple streams and blocks; backdate the blocks and but do not delete the collection. +// Also, create stray block and backdate it. +// After datamanager run: expect blocks from the collection, but not the stray block. +func TestManifestWithMultipleStreamsAndBlocks(t *testing.T) { + testManifestWithMultipleStreamsAndBlocks(t, 100, 10, "", false) +} + +// Same test as TestManifestWithMultipleStreamsAndBlocks with an additional +// keepstore of a service type other than "disk". Only the "disk" type services +// will be indexed by datamanager and hence should work the same way. +func TestManifestWithMultipleStreamsAndBlocks_WithOneUnsupportedKeepServer(t *testing.T) { + testManifestWithMultipleStreamsAndBlocks(t, 2, 2, "testblobstore", false) +} + +// Test datamanager with dry-run. Expect no block to be deleted. +func TestManifestWithMultipleStreamsAndBlocks_DryRun(t *testing.T) { + testManifestWithMultipleStreamsAndBlocks(t, 2, 2, "", true) +} + +func testManifestWithMultipleStreamsAndBlocks(t *testing.T, numStreams, numBlocks int, createExtraKeepServerWithType string, isDryRun bool) { + defer TearDownDataManagerTest(t) + SetupDataManagerTest(t) + + // create collection whose blocks will be backdated + collectionWithOldBlocks, oldBlocks := createMultiStreamBlockCollection(t, "old block", numStreams, numBlocks) + if collectionWithOldBlocks == "" { + t.Fatalf("Failed to create collection with %d blocks", numStreams*numBlocks) + } + if len(oldBlocks) != numStreams*numBlocks { + t.Fatalf("Not all blocks are created: expected %v, found %v", 1000, len(oldBlocks)) + } + + // create a stray block that will be backdated + strayOldBlock := putBlock(t, "this stray block is old") + + expected := []string{strayOldBlock} + expected = append(expected, oldBlocks...) + verifyBlocks(t, nil, expected, 2) + + // Backdate old blocks; but the collection still references these blocks + backdateBlocks(t, oldBlocks) + + // also backdate the stray old block + backdateBlocks(t, []string{strayOldBlock}) + + // If requested, create an extra keepserver with the given type + // This should be ignored during indexing and hence not change the datamanager outcome + var extraKeepServerUUID string + if createExtraKeepServerWithType != "" { + extraKeepServerUUID = addExtraKeepServer(t, createExtraKeepServerWithType) + defer deleteExtraKeepServer(extraKeepServerUUID) + } + + // run datamanager + dryRun = isDryRun + dataManagerSingleRun(t) + + if dryRun { + // verify that all blocks, including strayOldBlock, are still to be found + verifyBlocks(t, nil, expected, 2) + } else { + // verify that strayOldBlock is not to be found, but the collections blocks are still there + verifyBlocks(t, []string{strayOldBlock}, oldBlocks, 2) + } +} + +// Add one more keepstore with the given service type +func addExtraKeepServer(t *testing.T, serviceType string) string { + defer switchToken(arvadostest.AdminToken)() + + extraKeepService := make(arvadosclient.Dict) + err := arv.Create("keep_services", + arvadosclient.Dict{"keep_service": arvadosclient.Dict{ + "service_host": "localhost", + "service_port": "21321", + "service_ssl_flag": false, + "service_type": serviceType}}, + &extraKeepService) + if err != nil { + t.Fatal(err) + } + + return extraKeepService["uuid"].(string) +} + +func deleteExtraKeepServer(uuid string) { + defer switchToken(arvadostest.AdminToken)() + arv.Delete("keep_services", uuid, nil, nil) +}