services/keepstore/trash_worker_test.go

   1 package main
   2
   3 import (
   4         "container/list"
   5         "context"
   6         "testing"
   7         "time"
   8 )
   9
  10 type TrashWorkerTestData struct {
  11         Locator1    string
  12         Block1      []byte
  13         BlockMtime1 int64
  14
  15         Locator2    string
  16         Block2      []byte
  17         BlockMtime2 int64
  18
  19         CreateData      bool
  20         CreateInVolume1 bool
  21
  22         UseTrashLifeTime bool
  23         DifferentMtimes  bool
  24
  25         DeleteLocator string
  26
  27         ExpectLocator1 bool
  28         ExpectLocator2 bool
  29 }
  30
  31 /* Delete block that does not exist in any of the keep volumes.
  32    Expect no errors.
  33 */
  34 func TestTrashWorkerIntegration_GetNonExistingLocator(t *testing.T) {
  35         theConfig.EnableDelete = true
  36         testData := TrashWorkerTestData{
  37                 Locator1: "5d41402abc4b2a76b9719d911017c592",
  38                 Block1:   []byte("hello"),
  39
  40                 Locator2: "5d41402abc4b2a76b9719d911017c592",
  41                 Block2:   []byte("hello"),
  42
  43                 CreateData: false,
  44
  45                 DeleteLocator: "5d41402abc4b2a76b9719d911017c592",
  46
  47                 ExpectLocator1: false,
  48                 ExpectLocator2: false,
  49         }
  50         performTrashWorkerTest(testData, t)
  51 }
  52
  53 /* Delete a block that exists on volume 1 of the keep servers.
  54    Expect the second locator in volume 2 to be unaffected.
  55 */
  56 func TestTrashWorkerIntegration_LocatorInVolume1(t *testing.T) {
  57         theConfig.EnableDelete = true
  58         testData := TrashWorkerTestData{
  59                 Locator1: TestHash,
  60                 Block1:   TestBlock,
  61
  62                 Locator2: TestHash2,
  63                 Block2:   TestBlock2,
  64
  65                 CreateData: true,
  66
  67                 DeleteLocator: TestHash, // first locator
  68
  69                 ExpectLocator1: false,
  70                 ExpectLocator2: true,
  71         }
  72         performTrashWorkerTest(testData, t)
  73 }
  74
  75 /* Delete a block that exists on volume 2 of the keep servers.
  76    Expect the first locator in volume 1 to be unaffected.
  77 */
  78 func TestTrashWorkerIntegration_LocatorInVolume2(t *testing.T) {
  79         theConfig.EnableDelete = true
  80         testData := TrashWorkerTestData{
  81                 Locator1: TestHash,
  82                 Block1:   TestBlock,
  83
  84                 Locator2: TestHash2,
  85                 Block2:   TestBlock2,
  86
  87                 CreateData: true,
  88
  89                 DeleteLocator: TestHash2, // locator 2
  90
  91                 ExpectLocator1: true,
  92                 ExpectLocator2: false,
  93         }
  94         performTrashWorkerTest(testData, t)
  95 }
  96
  97 /* Delete a block with matching mtime for locator in both volumes.
  98    Expect locator to be deleted from both volumes.
  99 */
 100 func TestTrashWorkerIntegration_LocatorInBothVolumes(t *testing.T) {
 101         theConfig.EnableDelete = true
 102         testData := TrashWorkerTestData{
 103                 Locator1: TestHash,
 104                 Block1:   TestBlock,
 105
 106                 Locator2: TestHash,
 107                 Block2:   TestBlock,
 108
 109                 CreateData: true,
 110
 111                 DeleteLocator: TestHash,
 112
 113                 ExpectLocator1: false,
 114                 ExpectLocator2: false,
 115         }
 116         performTrashWorkerTest(testData, t)
 117 }
 118
 119 /* Same locator with different Mtimes exists in both volumes.
 120    Delete the second and expect the first to be still around.
 121 */
 122 func TestTrashWorkerIntegration_MtimeMatchesForLocator1ButNotForLocator2(t *testing.T) {
 123         theConfig.EnableDelete = true
 124         testData := TrashWorkerTestData{
 125                 Locator1: TestHash,
 126                 Block1:   TestBlock,
 127
 128                 Locator2: TestHash,
 129                 Block2:   TestBlock,
 130
 131                 CreateData:      true,
 132                 DifferentMtimes: true,
 133
 134                 DeleteLocator: TestHash,
 135
 136                 ExpectLocator1: true,
 137                 ExpectLocator2: false,
 138         }
 139         performTrashWorkerTest(testData, t)
 140 }
 141
 142 /* Two different locators in volume 1.
 143    Delete one of them.
 144    Expect the other unaffected.
 145 */
 146 func TestTrashWorkerIntegration_TwoDifferentLocatorsInVolume1(t *testing.T) {
 147         theConfig.EnableDelete = true
 148         testData := TrashWorkerTestData{
 149                 Locator1: TestHash,
 150                 Block1:   TestBlock,
 151
 152                 Locator2: TestHash2,
 153                 Block2:   TestBlock2,
 154
 155                 CreateData:      true,
 156                 CreateInVolume1: true,
 157
 158                 DeleteLocator: TestHash, // locator 1
 159
 160                 ExpectLocator1: false,
 161                 ExpectLocator2: true,
 162         }
 163         performTrashWorkerTest(testData, t)
 164 }
 165
 166 /* Allow default Trash Life time to be used. Thus, the newly created block
 167    will not be deleted because its Mtime is within the trash life time.
 168 */
 169 func TestTrashWorkerIntegration_SameLocatorInTwoVolumesWithDefaultTrashLifeTime(t *testing.T) {
 170         theConfig.EnableDelete = true
 171         testData := TrashWorkerTestData{
 172                 Locator1: TestHash,
 173                 Block1:   TestBlock,
 174
 175                 Locator2: TestHash2,
 176                 Block2:   TestBlock2,
 177
 178                 CreateData:      true,
 179                 CreateInVolume1: true,
 180
 181                 UseTrashLifeTime: true,
 182
 183                 DeleteLocator: TestHash, // locator 1
 184
 185                 // Since trash life time is in effect, block won't be deleted.
 186                 ExpectLocator1: true,
 187                 ExpectLocator2: true,
 188         }
 189         performTrashWorkerTest(testData, t)
 190 }
 191
 192 /* Delete a block with matching mtime for locator in both volumes, but EnableDelete is false,
 193    so block won't be deleted.
 194 */
 195 func TestTrashWorkerIntegration_DisabledDelete(t *testing.T) {
 196         theConfig.EnableDelete = false
 197         testData := TrashWorkerTestData{
 198                 Locator1: TestHash,
 199                 Block1:   TestBlock,
 200
 201                 Locator2: TestHash,
 202                 Block2:   TestBlock,
 203
 204                 CreateData: true,
 205
 206                 DeleteLocator: TestHash,
 207
 208                 ExpectLocator1: true,
 209                 ExpectLocator2: true,
 210         }
 211         performTrashWorkerTest(testData, t)
 212 }
 213
 214 /* Perform the test */
 215 func performTrashWorkerTest(testData TrashWorkerTestData, t *testing.T) {
 216         // Create Keep Volumes
 217         KeepVM = MakeTestVolumeManager(2)
 218         defer KeepVM.Close()
 219
 220         // Put test content
 221         vols := KeepVM.AllWritable()
 222         if testData.CreateData {
 223                 vols[0].Put(context.Background(), testData.Locator1, testData.Block1)
 224                 vols[0].Put(context.Background(), testData.Locator1+".meta", []byte("metadata"))
 225
 226                 if testData.CreateInVolume1 {
 227                         vols[0].Put(context.Background(), testData.Locator2, testData.Block2)
 228                         vols[0].Put(context.Background(), testData.Locator2+".meta", []byte("metadata"))
 229                 } else {
 230                         vols[1].Put(context.Background(), testData.Locator2, testData.Block2)
 231                         vols[1].Put(context.Background(), testData.Locator2+".meta", []byte("metadata"))
 232                 }
 233         }
 234
 235         oldBlockTime := time.Now().Add(-theConfig.BlobSignatureTTL.Duration() - time.Minute)
 236
 237         // Create TrashRequest for the test
 238         trashRequest := TrashRequest{
 239                 Locator:    testData.DeleteLocator,
 240                 BlockMtime: oldBlockTime.UnixNano(),
 241         }
 242
 243         // Run trash worker and put the trashRequest on trashq
 244         trashList := list.New()
 245         trashList.PushBack(trashRequest)
 246         trashq = NewWorkQueue()
 247         defer trashq.Close()
 248
 249         if !testData.UseTrashLifeTime {
 250                 // Trash worker would not delete block if its Mtime is
 251                 // within trash life time. Back-date the block to
 252                 // allow the deletion to succeed.
 253                 for _, v := range vols {
 254                         v.(*MockVolume).Timestamps[testData.DeleteLocator] = oldBlockTime
 255                         if testData.DifferentMtimes {
 256                                 oldBlockTime = oldBlockTime.Add(time.Second)
 257                         }
 258                 }
 259         }
 260         go RunTrashWorker(trashq)
 261
 262         // Install gate so all local operations block until we say go
 263         gate := make(chan struct{})
 264         for _, v := range vols {
 265                 v.(*MockVolume).Gate = gate
 266         }
 267
 268         assertStatusItem := func(k string, expect float64) {
 269                 if v := getStatusItem("TrashQueue", k); v != expect {
 270                         t.Errorf("Got %s %v, expected %v", k, v, expect)
 271                 }
 272         }
 273
 274         assertStatusItem("InProgress", 0)
 275         assertStatusItem("Queued", 0)
 276
 277         listLen := trashList.Len()
 278         trashq.ReplaceQueue(trashList)
 279
 280         // Wait for worker to take request(s)
 281         expectEqualWithin(t, time.Second, listLen, func() interface{} { return trashq.Status().InProgress })
 282
 283         // Ensure status.json also reports work is happening
 284         assertStatusItem("InProgress", float64(1))
 285         assertStatusItem("Queued", float64(listLen-1))
 286
 287         // Let worker proceed
 288         close(gate)
 289
 290         // Wait for worker to finish
 291         expectEqualWithin(t, time.Second, 0, func() interface{} { return trashq.Status().InProgress })
 292
 293         // Verify Locator1 to be un/deleted as expected
 294         buf := make([]byte, BlockSize)
 295         size, err := GetBlock(context.Background(), testData.Locator1, buf, nil)
 296         if testData.ExpectLocator1 {
 297                 if size == 0 || err != nil {
 298                         t.Errorf("Expected Locator1 to be still present: %s", testData.Locator1)
 299                 }
 300         } else {
 301                 if size > 0 || err == nil {
 302                         t.Errorf("Expected Locator1 to be deleted: %s", testData.Locator1)
 303                 }
 304         }
 305
 306         // Verify Locator2 to be un/deleted as expected
 307         if testData.Locator1 != testData.Locator2 {
 308                 size, err = GetBlock(context.Background(), testData.Locator2, buf, nil)
 309                 if testData.ExpectLocator2 {
 310                         if size == 0 || err != nil {
 311                                 t.Errorf("Expected Locator2 to be still present: %s", testData.Locator2)
 312                         }
 313                 } else {
 314                         if size > 0 || err == nil {
 315                                 t.Errorf("Expected Locator2 to be deleted: %s", testData.Locator2)
 316                         }
 317                 }
 318         }
 319
 320         // The DifferentMtimes test puts the same locator in two
 321         // different volumes, but only one copy has an Mtime matching
 322         // the trash request.
 323         if testData.DifferentMtimes {
 324                 locatorFoundIn := 0
 325                 for _, volume := range KeepVM.AllReadable() {
 326                         buf := make([]byte, BlockSize)
 327                         if _, err := volume.Get(context.Background(), testData.Locator1, buf); err == nil {
 328                                 locatorFoundIn = locatorFoundIn + 1
 329                         }
 330                 }
 331                 if locatorFoundIn != 1 {
 332                         t.Errorf("Found %d copies of %s, expected 1", locatorFoundIn, testData.Locator1)
 333                 }
 334         }
 335 }