services/keepstore/trash_worker_test.go

   1 package main
   2
   3 import (
   4         "container/list"
   5         "testing"
   6         "time"
   7 )
   8
   9 type TrashWorkerTestData struct {
  10         Locator1    string
  11         Block1      []byte
  12         BlockMtime1 int64
  13
  14         Locator2    string
  15         Block2      []byte
  16         BlockMtime2 int64
  17
  18         CreateData       bool
  19         CreateInVolume1  bool
  20         UseDelayToCreate bool
  21
  22         UseTrashLifeTime bool
  23
  24         DeleteLocator string
  25
  26         ExpectLocator1 bool
  27         ExpectLocator2 bool
  28 }
  29
  30 /* Delete block that does not exist in any of the keep volumes.
  31    Expect no errors.
  32 */
  33 func TestTrashWorkerIntegration_GetNonExistingLocator(t *testing.T) {
  34         testData := TrashWorkerTestData{
  35                 Locator1: "5d41402abc4b2a76b9719d911017c592",
  36                 Block1:   []byte("hello"),
  37
  38                 Locator2: "5d41402abc4b2a76b9719d911017c592",
  39                 Block2:   []byte("hello"),
  40
  41                 CreateData: false,
  42
  43                 DeleteLocator: "5d41402abc4b2a76b9719d911017c592",
  44
  45                 ExpectLocator1: false,
  46                 ExpectLocator2: false,
  47         }
  48         performTrashWorkerTest(testData, t)
  49 }
  50
  51 /* Delete a block that exists on volume 1 of the keep servers.
  52    Expect the second locator in volume 2 to be unaffected.
  53 */
  54 func TestTrashWorkerIntegration_LocatorInVolume1(t *testing.T) {
  55         testData := TrashWorkerTestData{
  56                 Locator1: TEST_HASH,
  57                 Block1:   TEST_BLOCK,
  58
  59                 Locator2: TEST_HASH_2,
  60                 Block2:   TEST_BLOCK_2,
  61
  62                 CreateData: true,
  63
  64                 DeleteLocator: TEST_HASH, // first locator
  65
  66                 ExpectLocator1: false,
  67                 ExpectLocator2: true,
  68         }
  69         performTrashWorkerTest(testData, t)
  70 }
  71
  72 /* Delete a block that exists on volume 2 of the keep servers.
  73    Expect the first locator in volume 1 to be unaffected.
  74 */
  75 func TestTrashWorkerIntegration_LocatorInVolume2(t *testing.T) {
  76         testData := TrashWorkerTestData{
  77                 Locator1: TEST_HASH,
  78                 Block1:   TEST_BLOCK,
  79
  80                 Locator2: TEST_HASH_2,
  81                 Block2:   TEST_BLOCK_2,
  82
  83                 CreateData: true,
  84
  85                 DeleteLocator: TEST_HASH_2, // locator 2
  86
  87                 ExpectLocator1: true,
  88                 ExpectLocator2: false,
  89         }
  90         performTrashWorkerTest(testData, t)
  91 }
  92
  93 /* Delete a block with matching mtime for locator in both volumes.
  94    Expect locator to be deleted from both volumes.
  95 */
  96 func TestTrashWorkerIntegration_LocatorInBothVolumes(t *testing.T) {
  97         testData := TrashWorkerTestData{
  98                 Locator1: TEST_HASH,
  99                 Block1:   TEST_BLOCK,
 100
 101                 Locator2: TEST_HASH,
 102                 Block2:   TEST_BLOCK,
 103
 104                 CreateData: true,
 105
 106                 DeleteLocator: TEST_HASH,
 107
 108                 ExpectLocator1: false,
 109                 ExpectLocator2: false,
 110         }
 111         performTrashWorkerTest(testData, t)
 112 }
 113
 114 /* Same locator with different Mtimes exists in both volumes.
 115    Delete the second and expect the first to be still around.
 116 */
 117 func TestTrashWorkerIntegration_MtimeMatchesForLocator1ButNotForLocator2(t *testing.T) {
 118         testData := TrashWorkerTestData{
 119                 Locator1: TEST_HASH,
 120                 Block1:   TEST_BLOCK,
 121
 122                 Locator2: TEST_HASH,
 123                 Block2:   TEST_BLOCK,
 124
 125                 CreateData:       true,
 126                 UseDelayToCreate: true,
 127
 128                 DeleteLocator: TEST_HASH,
 129
 130                 ExpectLocator1: true,
 131                 ExpectLocator2: false,
 132         }
 133         performTrashWorkerTest(testData, t)
 134 }
 135
 136 /* Two different locators in volume 1.
 137    Delete one of them.
 138    Expect the other unaffected.
 139 */
 140 func TestTrashWorkerIntegration_TwoDifferentLocatorsInVolume1(t *testing.T) {
 141         testData := TrashWorkerTestData{
 142                 Locator1: TEST_HASH,
 143                 Block1:   TEST_BLOCK,
 144
 145                 Locator2: TEST_HASH_2,
 146                 Block2:   TEST_BLOCK_2,
 147
 148                 CreateData:      true,
 149                 CreateInVolume1: true,
 150
 151                 DeleteLocator: TEST_HASH, // locator 1
 152
 153                 ExpectLocator1: false,
 154                 ExpectLocator2: true,
 155         }
 156         performTrashWorkerTest(testData, t)
 157 }
 158
 159 /* Allow default Trash Life time to be used. Thus, the newly created block
 160    will not be deleted becuase its Mtime is within the trash life time.
 161 */
 162 func TestTrashWorkerIntegration_SameLocatorInTwoVolumesWithDefaultTrashLifeTime(t *testing.T) {
 163         testData := TrashWorkerTestData{
 164                 Locator1: TEST_HASH,
 165                 Block1:   TEST_BLOCK,
 166
 167                 Locator2: TEST_HASH_2,
 168                 Block2:   TEST_BLOCK_2,
 169
 170                 CreateData:      true,
 171                 CreateInVolume1: true,
 172
 173                 UseTrashLifeTime: true,
 174
 175                 DeleteLocator: TEST_HASH, // locator 1
 176
 177                 // Since trash life time is in effect, block won't be deleted.
 178                 ExpectLocator1: true,
 179                 ExpectLocator2: true,
 180         }
 181         performTrashWorkerTest(testData, t)
 182 }
 183
 184 /* Perform the test */
 185 func performTrashWorkerTest(testData TrashWorkerTestData, t *testing.T) {
 186         actual_permission_ttl := permission_ttl
 187
 188         // Create Keep Volumes
 189         KeepVM = MakeTestVolumeManager(2)
 190         defer KeepVM.Close()
 191
 192         // Put test content
 193         vols := KeepVM.AllWritable()
 194         if testData.CreateData {
 195                 vols[0].Put(testData.Locator1, testData.Block1)
 196                 vols[0].Put(testData.Locator1+".meta", []byte("metadata"))
 197
 198                 // One of the tests deletes a locator with different Mtimes in two different volumes
 199                 if testData.UseDelayToCreate {
 200                         time.Sleep(1 * time.Second)
 201                 }
 202
 203                 if testData.CreateInVolume1 {
 204                         vols[0].Put(testData.Locator2, testData.Block2)
 205                         vols[0].Put(testData.Locator2+".meta", []byte("metadata"))
 206                 } else {
 207                         vols[1].Put(testData.Locator2, testData.Block2)
 208                         vols[1].Put(testData.Locator2+".meta", []byte("metadata"))
 209                 }
 210         }
 211
 212         // Create TrashRequest for the test
 213         trashRequest := TrashRequest{
 214                 Locator:    testData.DeleteLocator,
 215                 BlockMtime: time.Now().Unix(),
 216         }
 217
 218         // delay by permission_ttl to allow deletes to work
 219         time.Sleep(1 * time.Second)
 220
 221         // Run trash worker and put the trashRequest on trashq
 222         trashList := list.New()
 223         trashList.PushBack(trashRequest)
 224         trashq = NewWorkQueue()
 225
 226         // Trash worker would not delete block if its Mtime is within trash life time.
 227         // Hence, we will have to bypass it to allow the deletion to succeed.
 228         if !testData.UseTrashLifeTime {
 229                 permission_ttl = time.Duration(1) * time.Second
 230         }
 231         go RunTrashWorker(trashq)
 232
 233         trashq.ReplaceQueue(trashList)
 234         time.Sleep(10 * time.Millisecond) // give a moment to finish processing the list
 235
 236         // Verify Locator1 to be un/deleted as expected
 237         data, _ := GetBlock(testData.Locator1, false)
 238         if testData.ExpectLocator1 {
 239                 if len(data) == 0 {
 240                         t.Errorf("Expected Locator1 to be still present: %s", testData.Locator1)
 241                 }
 242         } else {
 243                 if len(data) > 0 {
 244                         t.Errorf("Expected Locator1 to be deleted: %s", testData.Locator1)
 245                 }
 246         }
 247
 248         // Verify Locator2 to be un/deleted as expected
 249         if testData.Locator1 != testData.Locator2 {
 250                 data, _ = GetBlock(testData.Locator2, false)
 251                 if testData.ExpectLocator2 {
 252                         if len(data) == 0 {
 253                                 t.Errorf("Expected Locator2 to be still present: %s", testData.Locator2)
 254                         }
 255                 } else {
 256                         if len(data) > 0 {
 257                                 t.Errorf("Expected Locator2 to be deleted: %s", testData.Locator2)
 258                         }
 259                 }
 260         }
 261
 262         // One test used the same locator in two different volumes but with different Mtime values
 263         // Hence let's verify that only one volume has it and the other is deleted
 264         if (testData.ExpectLocator1) &&
 265                 (testData.Locator1 == testData.Locator2) {
 266                 locatorFoundIn := 0
 267                 for _, volume := range KeepVM.AllReadable() {
 268                         if _, err := volume.Get(testData.Locator1); err == nil {
 269                                 locatorFoundIn = locatorFoundIn + 1
 270                         }
 271                 }
 272                 if locatorFoundIn != 1 {
 273                         t.Errorf("Expected locator to be found in only one volume after deleting. But found: %s", locatorFoundIn)
 274                 }
 275         }
 276
 277         // Done
 278         permission_ttl = actual_permission_ttl
 279         trashq.Close()
 280 }