tools/keep-rsync/keep-rsync_test.go

   1 package main
   2
   3 import (
   4         "crypto/md5"
   5         "fmt"
   6         "io/ioutil"
   7         "os"
   8         "strings"
   9         "testing"
  10         "time"
  11
  12         "git.curoverse.com/arvados.git/sdk/go/arvadostest"
  13         "git.curoverse.com/arvados.git/sdk/go/keepclient"
  14
  15         . "gopkg.in/check.v1"
  16 )
  17
  18 // Gocheck boilerplate
  19 func Test(t *testing.T) {
  20         TestingT(t)
  21 }
  22
  23 // Gocheck boilerplate
  24 var _ = Suite(&ServerRequiredSuite{})
  25
  26 // Tests that require the Keep server running
  27 type ServerRequiredSuite struct{}
  28
  29 func (s *ServerRequiredSuite) SetUpSuite(c *C) {
  30 }
  31
  32 func (s *ServerRequiredSuite) SetUpTest(c *C) {
  33         arvadostest.ResetEnv()
  34         srcKeepServicesJSON = ""
  35         dstKeepServicesJSON = ""
  36         replications = 0
  37         prefix = ""
  38         blobSigningKey = ""
  39 }
  40
  41 func (s *ServerRequiredSuite) TearDownSuite(c *C) {
  42         arvadostest.StopKeep()
  43         arvadostest.StopAPI()
  44 }
  45
  46 var testKeepServicesJSON = "{ \"kind\":\"arvados#keepServiceList\", \"etag\":\"\", \"self_link\":\"\", \"offset\":null, \"limit\":null, \"items\":[ { \"href\":\"/keep_services/zzzzz-bi6l4-123456789012340\", \"kind\":\"arvados#keepService\", \"etag\":\"641234567890enhj7hzx432e5\", \"uuid\":\"zzzzz-bi6l4-123456789012340\", \"owner_uuid\":\"zzzzz-tpzed-123456789012345\", \"service_host\":\"keep0.zzzzz.arvadosapi.com\", \"service_port\":25107, \"service_ssl_flag\":false, \"service_type\":\"disk\", \"read_only\":false }, { \"href\":\"/keep_services/zzzzz-bi6l4-123456789012341\", \"kind\":\"arvados#keepService\", \"etag\":\"641234567890enhj7hzx432e5\", \"uuid\":\"zzzzz-bi6l4-123456789012341\", \"owner_uuid\":\"zzzzz-tpzed-123456789012345\", \"service_host\":\"keep0.zzzzz.arvadosapi.com\", \"service_port\":25108, \"service_ssl_flag\":false, \"service_type\":\"disk\", \"read_only\":false } ], \"items_available\":2 }"
  47
  48 // Testing keep-rsync needs two sets of keep services: src and dst.
  49 // The test setup hence tweaks keep-rsync initialization to achieve this.
  50 // First invoke initializeKeepRsync and then invoke StartKeepWithParams
  51 // to create the keep servers to be used as destination.
  52 func setupRsync(c *C, enforcePermissions bool, overwrite bool) {
  53         // srcConfig
  54         srcConfig.APIHost = os.Getenv("ARVADOS_API_HOST")
  55         srcConfig.APIToken = os.Getenv("ARVADOS_API_TOKEN")
  56         srcConfig.APIHostInsecure = matchTrue.MatchString(os.Getenv("ARVADOS_API_HOST_INSECURE"))
  57
  58         // dstConfig
  59         dstConfig.APIHost = os.Getenv("ARVADOS_API_HOST")
  60         dstConfig.APIToken = os.Getenv("ARVADOS_API_TOKEN")
  61         dstConfig.APIHostInsecure = matchTrue.MatchString(os.Getenv("ARVADOS_API_HOST_INSECURE"))
  62
  63         if enforcePermissions {
  64                 blobSigningKey = "zfhgfenhffzltr9dixws36j1yhksjoll2grmku38mi7yxd66h5j4q9w4jzanezacp8s6q0ro3hxakfye02152hncy6zml2ed0uc"
  65         }
  66
  67         // Start API and Keep servers
  68         arvadostest.StartAPI()
  69         arvadostest.StartKeepWithParams(false, enforcePermissions)
  70
  71         // initialize keep-rsync
  72         err := initializeKeepRsync()
  73         c.Check(err, IsNil)
  74
  75         // Create an additional keep server to be used as destination and reload kcDst
  76         // Set replications to 1 since those many keep servers were created for dst.
  77         if overwrite {
  78                 arvadostest.StartKeepWithParams(true, enforcePermissions)
  79                 replications = 1
  80
  81                 kcDst, err = keepclient.MakeKeepClient(&arvDst)
  82                 c.Check(err, IsNil)
  83                 kcDst.Want_replicas = 1
  84         }
  85 }
  86
  87 // Test readConfigFromFile method
  88 func (s *ServerRequiredSuite) TestReadConfigFromFile(c *C) {
  89         // Setup a test config file
  90         file, err := ioutil.TempFile(os.TempDir(), "config")
  91         c.Check(err, IsNil)
  92         defer os.Remove(file.Name())
  93
  94         fileContent := "ARVADOS_API_HOST=testhost\n"
  95         fileContent += "ARVADOS_API_TOKEN=testtoken\n"
  96         fileContent += "ARVADOS_API_HOST_INSECURE=true\n"
  97         fileContent += "ARVADOS_BLOB_SIGNING_KEY=abcdefg"
  98
  99         _, err = file.Write([]byte(fileContent))
 100
 101         // Invoke readConfigFromFile method with this test filename
 102         config, err := readConfigFromFile(file.Name())
 103         c.Check(err, IsNil)
 104         c.Assert(config.APIHost, Equals, "testhost")
 105         c.Assert(config.APIToken, Equals, "testtoken")
 106         c.Assert(config.APIHostInsecure, Equals, true)
 107         c.Assert(config.ExternalClient, Equals, false)
 108         c.Assert(blobSigningKey, Equals, "abcdefg")
 109 }
 110
 111 // Test keep-rsync initialization, with src and dst keep servers.
 112 // Do a Put and Get in src, both of which should succeed.
 113 // Do a Put and Get in dst, both of which should succeed.
 114 // Do a Get in dst for the src hash, which should raise block not found error.
 115 // Do a Get in src for the dst hash, which should raise block not found error.
 116 func (s *ServerRequiredSuite) TestRsyncPutInOne_GetFromOtherShouldFail(c *C) {
 117         setupRsync(c, false, true)
 118
 119         // Put a block in src using kcSrc and Get it
 120         srcData := []byte("test-data1")
 121         locatorInSrc := fmt.Sprintf("%x", md5.Sum(srcData))
 122
 123         hash, rep, err := kcSrc.PutB(srcData)
 124         c.Check(hash, Matches, fmt.Sprintf(`^%s\+10(\+.+)?$`, locatorInSrc))
 125         c.Check(rep, Equals, 2)
 126         c.Check(err, Equals, nil)
 127
 128         reader, blocklen, _, err := kcSrc.Get(locatorInSrc)
 129         c.Check(err, IsNil)
 130         c.Check(blocklen, Equals, int64(10))
 131         all, err := ioutil.ReadAll(reader)
 132         c.Check(all, DeepEquals, srcData)
 133
 134         // Put a different block in src using kcSrc and Get it
 135         dstData := []byte("test-data2")
 136         locatorInDst := fmt.Sprintf("%x", md5.Sum(dstData))
 137
 138         hash, rep, err = kcDst.PutB(dstData)
 139         c.Check(hash, Matches, fmt.Sprintf(`^%s\+10(\+.+)?$`, locatorInDst))
 140         c.Check(rep, Equals, 1)
 141         c.Check(err, Equals, nil)
 142
 143         reader, blocklen, _, err = kcDst.Get(locatorInDst)
 144         c.Check(err, IsNil)
 145         c.Check(blocklen, Equals, int64(10))
 146         all, err = ioutil.ReadAll(reader)
 147         c.Check(all, DeepEquals, dstData)
 148
 149         // Get srcLocator using kcDst should fail with Not Found error
 150         _, _, _, err = kcDst.Get(locatorInSrc)
 151         c.Assert(err.Error(), Equals, "Block not found")
 152
 153         // Get dstLocator using kcSrc should fail with Not Found error
 154         _, _, _, err = kcSrc.Get(locatorInDst)
 155         c.Assert(err.Error(), Equals, "Block not found")
 156 }
 157
 158 // Test keep-rsync initialization, with srcKeepServicesJSON
 159 func (s *ServerRequiredSuite) TestRsyncInitializeWithKeepServicesJSON(c *C) {
 160         srcKeepServicesJSON = testKeepServicesJSON
 161
 162         setupRsync(c, false, true)
 163
 164         localRoots := kcSrc.LocalRoots()
 165         c.Check(localRoots != nil, Equals, true)
 166
 167         foundIt := false
 168         for k := range localRoots {
 169                 if k == "zzzzz-bi6l4-123456789012340" {
 170                         foundIt = true
 171                 }
 172         }
 173         c.Check(foundIt, Equals, true)
 174
 175         foundIt = false
 176         for k := range localRoots {
 177                 if k == "zzzzz-bi6l4-123456789012341" {
 178                         foundIt = true
 179                 }
 180         }
 181         c.Check(foundIt, Equals, true)
 182 }
 183
 184 // Test keep-rsync initialization, with src and dst keep servers with blobSigningKey.
 185 // Do a Put and Get in src, both of which should succeed.
 186 // Do a Put and Get in dst, both of which should succeed.
 187 // Do a Get in dst for the src hash, which should raise block not found error.
 188 // Do a Get in src for the dst hash, which should raise block not found error.
 189 func (s *ServerRequiredSuite) TestRsyncWithBlobSigning_PutInOne_GetFromOtherShouldFail(c *C) {
 190         setupRsync(c, true, true)
 191
 192         // Put a block in src using kcSrc and Get it
 193         srcData := []byte("test-data1")
 194         locatorInSrc := fmt.Sprintf("%x", md5.Sum(srcData))
 195
 196         hash, rep, err := kcSrc.PutB(srcData)
 197         c.Check(hash, Matches, fmt.Sprintf(`^%s\+10(\+.+)?$`, locatorInSrc))
 198         c.Check(rep, Equals, 2)
 199         c.Check(err, Equals, nil)
 200
 201         tomorrow := time.Now().AddDate(0, 0, 1)
 202         signedLocator := keepclient.SignLocator(locatorInSrc, arvSrc.ApiToken, tomorrow, []byte(blobSigningKey))
 203
 204         reader, blocklen, _, err := kcSrc.Get(signedLocator)
 205         c.Check(err, IsNil)
 206         c.Check(blocklen, Equals, int64(10))
 207         all, err := ioutil.ReadAll(reader)
 208         c.Check(all, DeepEquals, srcData)
 209
 210         // Put a different block in src using kcSrc and Get it
 211         dstData := []byte("test-data2")
 212         locatorInDst := fmt.Sprintf("%x", md5.Sum(dstData))
 213
 214         hash, rep, err = kcDst.PutB(dstData)
 215         c.Check(hash, Matches, fmt.Sprintf(`^%s\+10(\+.+)?$`, locatorInDst))
 216         c.Check(rep, Equals, 1)
 217         c.Check(err, Equals, nil)
 218
 219         signedLocator = keepclient.SignLocator(locatorInDst, arvDst.ApiToken, tomorrow, []byte(blobSigningKey))
 220
 221         reader, blocklen, _, err = kcDst.Get(signedLocator)
 222         c.Check(err, IsNil)
 223         c.Check(blocklen, Equals, int64(10))
 224         all, err = ioutil.ReadAll(reader)
 225         c.Check(all, DeepEquals, dstData)
 226
 227         // Get srcLocator using kcDst should fail with Not Found error
 228         signedLocator = keepclient.SignLocator(locatorInSrc, arvDst.ApiToken, tomorrow, []byte(blobSigningKey))
 229         _, _, _, err = kcDst.Get(locatorInSrc)
 230         c.Assert(err.Error(), Equals, "Block not found")
 231
 232         // Get dstLocator using kcSrc should fail with Not Found error
 233         signedLocator = keepclient.SignLocator(locatorInDst, arvSrc.ApiToken, tomorrow, []byte(blobSigningKey))
 234         _, _, _, err = kcSrc.Get(locatorInDst)
 235         c.Assert(err.Error(), Equals, "Block not found")
 236 }
 237
 238 // Test keep-rsync initialization with default replications count
 239 func (s *ServerRequiredSuite) TestInitializeRsyncDefaultReplicationsCount(c *C) {
 240         setupRsync(c, false, false)
 241
 242         // Must have got default replications value as 2 from dst discovery document
 243         c.Assert(replications, Equals, 2)
 244 }
 245
 246 // Test keep-rsync initialization with replications count argument
 247 func (s *ServerRequiredSuite) TestInitializeRsyncReplicationsCount(c *C) {
 248         // First set replications to 3 to mimic passing input argument
 249         replications = 3
 250
 251         setupRsync(c, false, false)
 252
 253         // Since replications value is provided, default is not used
 254         c.Assert(replications, Equals, 3)
 255 }
 256
 257 // Put some blocks in Src and some more in Dst
 258 // And copy missing blocks from Src to Dst
 259 func (s *ServerRequiredSuite) TestKeepRsync(c *C) {
 260         testKeepRsync(c, false, "")
 261 }
 262
 263 // Put some blocks in Src and some more in Dst with blob signing enabled.
 264 // And copy missing blocks from Src to Dst
 265 func (s *ServerRequiredSuite) TestKeepRsync_WithBlobSigning(c *C) {
 266         testKeepRsync(c, true, "")
 267 }
 268
 269 // Put some blocks in Src and some more in Dst
 270 // Use prefix while doing rsync
 271 // And copy missing blocks from Src to Dst
 272 func (s *ServerRequiredSuite) TestKeepRsync_WithPrefix(c *C) {
 273         data := []byte("test-data-4")
 274         hash := fmt.Sprintf("%x", md5.Sum(data))
 275
 276         testKeepRsync(c, false, hash[0:3])
 277 }
 278
 279 // Put some blocks in Src and some more in Dst
 280 // Use prefix not in src while doing rsync
 281 // And copy missing blocks from Src to Dst
 282 func (s *ServerRequiredSuite) TestKeepRsync_WithNoSuchPrefixInSrc(c *C) {
 283         testKeepRsync(c, false, "999")
 284 }
 285
 286 // Put 5 blocks in src. Put 2 of those blocks in dst
 287 // Hence there are 3 additional blocks in src
 288 // Also, put 2 extra blocks in dst; they are hence only in dst
 289 // Run rsync and verify that those 7 blocks are now available in dst
 290 func testKeepRsync(c *C, enforcePermissions bool, indexPrefix string) {
 291         setupRsync(c, enforcePermissions, true)
 292
 293         prefix = indexPrefix
 294
 295         tomorrow := time.Now().AddDate(0, 0, 1)
 296
 297         // Put a few blocks in src using kcSrc
 298         var srcLocators []string
 299         var srcLocatorsMatchingPrefix []string
 300         for i := 0; i < 5; i++ {
 301                 data := []byte(fmt.Sprintf("test-data-%d", i))
 302                 hash := fmt.Sprintf("%x", md5.Sum(data))
 303
 304                 hash2, rep, err := kcSrc.PutB(data)
 305                 c.Check(hash2, Matches, fmt.Sprintf(`^%s\+11(\+.+)?$`, hash))
 306                 c.Check(rep, Equals, 2)
 307                 c.Check(err, IsNil)
 308
 309                 getLocator := hash
 310                 if enforcePermissions {
 311                         getLocator = keepclient.SignLocator(getLocator, arvSrc.ApiToken, tomorrow, []byte(blobSigningKey))
 312                 }
 313
 314                 reader, blocklen, _, err := kcSrc.Get(getLocator)
 315                 c.Check(err, IsNil)
 316                 c.Check(blocklen, Equals, int64(11))
 317                 all, err := ioutil.ReadAll(reader)
 318                 c.Check(all, DeepEquals, data)
 319
 320                 srcLocators = append(srcLocators, fmt.Sprintf("%s+%d", hash, blocklen))
 321                 if strings.HasPrefix(hash, indexPrefix) {
 322                         srcLocatorsMatchingPrefix = append(srcLocatorsMatchingPrefix, fmt.Sprintf("%s+%d", hash, blocklen))
 323                 }
 324         }
 325
 326         // Put first two of those src blocks in dst using kcDst
 327         var dstLocators []string
 328         for i := 0; i < 2; i++ {
 329                 data := []byte(fmt.Sprintf("test-data-%d", i))
 330                 hash := fmt.Sprintf("%x", md5.Sum(data))
 331
 332                 hash2, rep, err := kcDst.PutB(data)
 333                 c.Check(hash2, Matches, fmt.Sprintf(`^%s\+11(\+.+)?$`, hash))
 334                 c.Check(rep, Equals, 1)
 335                 c.Check(err, IsNil)
 336
 337                 getLocator := hash
 338                 if enforcePermissions {
 339                         getLocator = keepclient.SignLocator(getLocator, arvDst.ApiToken, tomorrow, []byte(blobSigningKey))
 340                 }
 341
 342                 reader, blocklen, _, err := kcDst.Get(getLocator)
 343                 c.Check(err, IsNil)
 344                 c.Check(blocklen, Equals, int64(11))
 345                 all, err := ioutil.ReadAll(reader)
 346                 c.Check(all, DeepEquals, data)
 347
 348                 dstLocators = append(dstLocators, fmt.Sprintf("%s+%d", hash, blocklen))
 349         }
 350
 351         // Put two more blocks in dst; they are not in src at all
 352         var extraDstLocators []string
 353         for i := 0; i < 2; i++ {
 354                 data := []byte(fmt.Sprintf("other-data-%d", i))
 355                 hash := fmt.Sprintf("%x", md5.Sum(data))
 356
 357                 hash2, rep, err := kcDst.PutB(data)
 358                 c.Check(hash2, Matches, fmt.Sprintf(`^%s\+12(\+.+)?$`, hash))
 359                 c.Check(rep, Equals, 1)
 360                 c.Check(err, IsNil)
 361
 362                 getLocator := hash
 363                 if enforcePermissions {
 364                         getLocator = keepclient.SignLocator(getLocator, arvDst.ApiToken, tomorrow, []byte(blobSigningKey))
 365                 }
 366
 367                 reader, blocklen, _, err := kcDst.Get(getLocator)
 368                 c.Check(err, IsNil)
 369                 c.Check(blocklen, Equals, int64(12))
 370                 all, err := ioutil.ReadAll(reader)
 371                 c.Check(all, DeepEquals, data)
 372
 373                 extraDstLocators = append(extraDstLocators, fmt.Sprintf("%s+%d", hash, blocklen))
 374         }
 375
 376         err := performKeepRsync()
 377         c.Check(err, IsNil)
 378
 379         // Now GetIndex from dst and verify that all 5 from src and the 2 extra blocks are found
 380         dstIndex, err := getUniqueLocators(kcDst, "")
 381         c.Check(err, IsNil)
 382
 383         if prefix == "" {
 384                 for _, locator := range srcLocators {
 385                         _, ok := dstIndex[locator]
 386                         c.Assert(ok, Equals, true)
 387                 }
 388         } else {
 389                 for _, locator := range srcLocatorsMatchingPrefix {
 390                         _, ok := dstIndex[locator]
 391                         c.Assert(ok, Equals, true)
 392                 }
 393         }
 394
 395         for _, locator := range extraDstLocators {
 396                 _, ok := dstIndex[locator]
 397                 c.Assert(ok, Equals, true)
 398         }
 399
 400         if prefix == "" {
 401                 // all blocks from src and the two extra blocks
 402                 c.Assert(len(dstIndex), Equals, len(srcLocators)+len(extraDstLocators))
 403         } else {
 404                 // one matching prefix, 2 that were initially copied into dst along with src, and the extra blocks
 405                 c.Assert(len(dstIndex), Equals, len(srcLocatorsMatchingPrefix)+len(extraDstLocators)+2)
 406         }
 407 }
 408
 409 // Setup rsync using srcKeepServicesJSON with fake keepservers.
 410 // Expect error during performKeepRsync due to unreachable src keepservers.
 411 func (s *ServerRequiredSuite) TestErrorDuringRsync_FakeSrcKeepservers(c *C) {
 412         srcKeepServicesJSON = testKeepServicesJSON
 413
 414         setupRsync(c, false, true)
 415
 416         err := performKeepRsync()
 417         c.Check(err, NotNil)
 418 }
 419
 420 // Setup rsync using dstKeepServicesJSON with fake keepservers.
 421 // Expect error during performKeepRsync due to unreachable dst keepservers.
 422 func (s *ServerRequiredSuite) TestErrorDuringRsync_FakeDstKeepservers(c *C) {
 423         dstKeepServicesJSON = testKeepServicesJSON
 424
 425         setupRsync(c, false, false)
 426
 427         err := performKeepRsync()
 428         c.Check(err, NotNil)
 429 }