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 var kcSrc, kcDst *keepclient.KeepClient
  33 var srcKeepServicesJSON, dstKeepServicesJSON, blobSigningKey string
  34
  35 func (s *ServerRequiredSuite) SetUpTest(c *C) {
  36         arvadostest.ResetEnv()
  37
  38         // reset all variables between tests
  39         blobSigningKey = ""
  40         srcKeepServicesJSON = ""
  41         dstKeepServicesJSON = ""
  42         kcSrc = &keepclient.KeepClient{}
  43         kcDst = &keepclient.KeepClient{}
  44 }
  45
  46 func (s *ServerRequiredSuite) TearDownSuite(c *C) {
  47         arvadostest.StopKeepServers(3)
  48         arvadostest.StopAPI()
  49 }
  50
  51 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 }"
  52
  53 // Testing keep-rsync needs two sets of keep services: src and dst.
  54 // The test setup hence creates 3 servers instead of the default 2,
  55 // and uses the first 2 as src and the 3rd as dst keep servers.
  56 func setupRsync(c *C, enforcePermissions, updateDstReplications bool, replications int) {
  57         // srcConfig
  58         var srcConfig apiConfig
  59         srcConfig.APIHost = os.Getenv("ARVADOS_API_HOST")
  60         srcConfig.APIToken = os.Getenv("ARVADOS_API_TOKEN")
  61         srcConfig.APIHostInsecure = matchTrue.MatchString(os.Getenv("ARVADOS_API_HOST_INSECURE"))
  62
  63         // dstConfig
  64         var dstConfig apiConfig
  65         dstConfig.APIHost = os.Getenv("ARVADOS_API_HOST")
  66         dstConfig.APIToken = os.Getenv("ARVADOS_API_TOKEN")
  67         dstConfig.APIHostInsecure = matchTrue.MatchString(os.Getenv("ARVADOS_API_HOST_INSECURE"))
  68
  69         if enforcePermissions {
  70                 blobSigningKey = "zfhgfenhffzltr9dixws36j1yhksjoll2grmku38mi7yxd66h5j4q9w4jzanezacp8s6q0ro3hxakfye02152hncy6zml2ed0uc"
  71         }
  72
  73         // Start API and Keep servers
  74         arvadostest.StartAPI()
  75         arvadostest.StartKeepWithParams(3, enforcePermissions)
  76
  77         // setup keepclients
  78         var err error
  79         kcSrc, err = setupKeepClient(srcConfig, srcKeepServicesJSON, false, 0)
  80         c.Check(err, IsNil)
  81
  82         kcDst, err = setupKeepClient(dstConfig, dstKeepServicesJSON, true, replications)
  83         c.Check(err, IsNil)
  84
  85         for uuid := range kcSrc.LocalRoots() {
  86                 if strings.HasSuffix(uuid, "02") {
  87                         delete(kcSrc.LocalRoots(), uuid)
  88                 }
  89         }
  90         for uuid := range kcSrc.GatewayRoots() {
  91                 if strings.HasSuffix(uuid, "02") {
  92                         delete(kcSrc.GatewayRoots(), uuid)
  93                 }
  94         }
  95         for uuid := range kcSrc.WritableLocalRoots() {
  96                 if strings.HasSuffix(uuid, "02") {
  97                         delete(kcSrc.WritableLocalRoots(), uuid)
  98                 }
  99         }
 100
 101         for uuid := range kcDst.LocalRoots() {
 102                 if strings.HasSuffix(uuid, "00") || strings.HasSuffix(uuid, "01") {
 103                         delete(kcDst.LocalRoots(), uuid)
 104                 }
 105         }
 106         for uuid := range kcDst.GatewayRoots() {
 107                 if strings.HasSuffix(uuid, "00") || strings.HasSuffix(uuid, "01") {
 108                         delete(kcDst.GatewayRoots(), uuid)
 109                 }
 110         }
 111         for uuid := range kcDst.WritableLocalRoots() {
 112                 if strings.HasSuffix(uuid, "00") || strings.HasSuffix(uuid, "01") {
 113                         delete(kcDst.WritableLocalRoots(), uuid)
 114                 }
 115         }
 116
 117         if updateDstReplications {
 118                 kcDst.Want_replicas = replications
 119         }
 120 }
 121
 122 // Test keep-rsync initialization, with src and dst keep servers.
 123 // Do a Put and Get in src, both of which should succeed.
 124 // Do a Put and Get in dst, both of which should succeed.
 125 // Do a Get in dst for the src hash, which should raise block not found error.
 126 // Do a Get in src for the dst hash, which should raise block not found error.
 127 func (s *ServerRequiredSuite) TestRsyncPutInOne_GetFromOtherShouldFail(c *C) {
 128         setupRsync(c, false, true, 1)
 129
 130         // Put a block in src using kcSrc and Get it
 131         srcData := []byte("test-data1")
 132         locatorInSrc := fmt.Sprintf("%x", md5.Sum(srcData))
 133
 134         hash, rep, err := kcSrc.PutB(srcData)
 135         c.Check(hash, Matches, fmt.Sprintf(`^%s\+10(\+.+)?$`, locatorInSrc))
 136         c.Check(rep, Equals, 2)
 137         c.Check(err, Equals, nil)
 138
 139         reader, blocklen, _, err := kcSrc.Get(locatorInSrc)
 140         c.Check(err, IsNil)
 141         c.Check(blocklen, Equals, int64(10))
 142         all, err := ioutil.ReadAll(reader)
 143         c.Check(all, DeepEquals, srcData)
 144
 145         // Put a different block in src using kcSrc and Get it
 146         dstData := []byte("test-data2")
 147         locatorInDst := fmt.Sprintf("%x", md5.Sum(dstData))
 148
 149         hash, rep, err = kcDst.PutB(dstData)
 150         c.Check(hash, Matches, fmt.Sprintf(`^%s\+10(\+.+)?$`, locatorInDst))
 151         c.Check(rep, Equals, 1)
 152         c.Check(err, Equals, nil)
 153
 154         reader, blocklen, _, err = kcDst.Get(locatorInDst)
 155         c.Check(err, IsNil)
 156         c.Check(blocklen, Equals, int64(10))
 157         all, err = ioutil.ReadAll(reader)
 158         c.Check(all, DeepEquals, dstData)
 159
 160         // Get srcLocator using kcDst should fail with Not Found error
 161         _, _, _, err = kcDst.Get(locatorInSrc)
 162         c.Assert(err.Error(), Equals, "Block not found")
 163
 164         // Get dstLocator using kcSrc should fail with Not Found error
 165         _, _, _, err = kcSrc.Get(locatorInDst)
 166         c.Assert(err.Error(), Equals, "Block not found")
 167 }
 168
 169 // Test keep-rsync initialization, with srcKeepServicesJSON
 170 func (s *ServerRequiredSuite) TestRsyncInitializeWithKeepServicesJSON(c *C) {
 171         srcKeepServicesJSON = testKeepServicesJSON
 172
 173         setupRsync(c, false, true, 1)
 174
 175         localRoots := kcSrc.LocalRoots()
 176         c.Check(localRoots, NotNil)
 177
 178         foundIt := false
 179         for k := range localRoots {
 180                 if k == "zzzzz-bi6l4-123456789012340" {
 181                         foundIt = true
 182                 }
 183         }
 184         c.Check(foundIt, Equals, true)
 185
 186         foundIt = false
 187         for k := range localRoots {
 188                 if k == "zzzzz-bi6l4-123456789012341" {
 189                         foundIt = true
 190                 }
 191         }
 192         c.Check(foundIt, Equals, true)
 193 }
 194
 195 // Test keep-rsync initialization, with src and dst keep servers with blobSigningKey.
 196 // Do a Put and Get in src, both of which should succeed.
 197 // Do a Put and Get in dst, both of which should succeed.
 198 // Do a Get in dst for the src hash, which should raise block not found error.
 199 // Do a Get in src for the dst hash, which should raise block not found error.
 200 func (s *ServerRequiredSuite) TestRsyncWithBlobSigning_PutInOne_GetFromOtherShouldFail(c *C) {
 201         setupRsync(c, true, true, 1)
 202
 203         // Put a block in src using kcSrc and Get it
 204         srcData := []byte("test-data1")
 205         locatorInSrc := fmt.Sprintf("%x", md5.Sum(srcData))
 206
 207         hash, rep, err := kcSrc.PutB(srcData)
 208         c.Check(hash, Matches, fmt.Sprintf(`^%s\+10(\+.+)?$`, locatorInSrc))
 209         c.Check(rep, Equals, 2)
 210         c.Check(err, Equals, nil)
 211
 212         tomorrow := time.Now().AddDate(0, 0, 1)
 213         signedLocator := keepclient.SignLocator(locatorInSrc, kcSrc.Arvados.ApiToken, tomorrow, []byte(blobSigningKey))
 214
 215         reader, blocklen, _, err := kcSrc.Get(signedLocator)
 216         c.Check(err, IsNil)
 217         c.Check(blocklen, Equals, int64(10))
 218         all, err := ioutil.ReadAll(reader)
 219         c.Check(all, DeepEquals, srcData)
 220
 221         // Put a different block in src using kcSrc and Get it
 222         dstData := []byte("test-data2")
 223         locatorInDst := fmt.Sprintf("%x", md5.Sum(dstData))
 224
 225         hash, rep, err = kcDst.PutB(dstData)
 226         c.Check(hash, Matches, fmt.Sprintf(`^%s\+10(\+.+)?$`, locatorInDst))
 227         c.Check(rep, Equals, 1)
 228         c.Check(err, Equals, nil)
 229
 230         signedLocator = keepclient.SignLocator(locatorInDst, kcDst.Arvados.ApiToken, tomorrow, []byte(blobSigningKey))
 231
 232         reader, blocklen, _, err = kcDst.Get(signedLocator)
 233         c.Check(err, IsNil)
 234         c.Check(blocklen, Equals, int64(10))
 235         all, err = ioutil.ReadAll(reader)
 236         c.Check(all, DeepEquals, dstData)
 237
 238         // Get srcLocator using kcDst should fail with Not Found error
 239         signedLocator = keepclient.SignLocator(locatorInSrc, kcDst.Arvados.ApiToken, tomorrow, []byte(blobSigningKey))
 240         _, _, _, err = kcDst.Get(locatorInSrc)
 241         c.Assert(err.Error(), Equals, "Block not found")
 242
 243         // Get dstLocator using kcSrc should fail with Not Found error
 244         signedLocator = keepclient.SignLocator(locatorInDst, kcSrc.Arvados.ApiToken, tomorrow, []byte(blobSigningKey))
 245         _, _, _, err = kcSrc.Get(locatorInDst)
 246         c.Assert(err.Error(), Equals, "Block not found")
 247 }
 248
 249 // Test keep-rsync initialization with default replications count
 250 func (s *ServerRequiredSuite) TestInitializeRsyncDefaultReplicationsCount(c *C) {
 251         setupRsync(c, false, false, 0)
 252
 253         // Must have got default replications value as 2 from dst discovery document
 254         c.Assert(kcDst.Want_replicas, Equals, 2)
 255 }
 256
 257 // Test keep-rsync initialization with replications count argument
 258 func (s *ServerRequiredSuite) TestInitializeRsyncReplicationsCount(c *C) {
 259         setupRsync(c, false, false, 3)
 260
 261         // Since replications value is provided, default is not used
 262         c.Assert(kcDst.Want_replicas, Equals, 3)
 263 }
 264
 265 // Put some blocks in Src and some more in Dst
 266 // And copy missing blocks from Src to Dst
 267 func (s *ServerRequiredSuite) TestKeepRsync(c *C) {
 268         testKeepRsync(c, false, "")
 269 }
 270
 271 // Put some blocks in Src and some more in Dst with blob signing enabled.
 272 // And copy missing blocks from Src to Dst
 273 func (s *ServerRequiredSuite) TestKeepRsync_WithBlobSigning(c *C) {
 274         testKeepRsync(c, true, "")
 275 }
 276
 277 // Put some blocks in Src and some more in Dst
 278 // Use prefix while doing rsync
 279 // And copy missing blocks from Src to Dst
 280 func (s *ServerRequiredSuite) TestKeepRsync_WithPrefix(c *C) {
 281         data := []byte("test-data-4")
 282         hash := fmt.Sprintf("%x", md5.Sum(data))
 283
 284         testKeepRsync(c, false, hash[0:3])
 285 }
 286
 287 // Put some blocks in Src and some more in Dst
 288 // Use prefix not in src while doing rsync
 289 // And copy missing blocks from Src to Dst
 290 func (s *ServerRequiredSuite) TestKeepRsync_WithNoSuchPrefixInSrc(c *C) {
 291         testKeepRsync(c, false, "999")
 292 }
 293
 294 // Put 5 blocks in src. Put 2 of those blocks in dst
 295 // Hence there are 3 additional blocks in src
 296 // Also, put 2 extra blocks in dst; they are hence only in dst
 297 // Run rsync and verify that those 7 blocks are now available in dst
 298 func testKeepRsync(c *C, enforcePermissions bool, prefix string) {
 299         setupRsync(c, enforcePermissions, true, 1)
 300
 301         // setupTestData
 302         setupTestData(c, prefix)
 303
 304         err := performKeepRsync(kcSrc, kcDst, blobSigningKey, prefix)
 305         c.Check(err, IsNil)
 306
 307         // Now GetIndex from dst and verify that all 5 from src and the 2 extra blocks are found
 308         dstIndex, err := getUniqueLocators(kcDst, "")
 309         c.Check(err, IsNil)
 310
 311         if prefix == "" {
 312                 for _, locator := range srcLocators {
 313                         _, ok := dstIndex[locator]
 314                         c.Assert(ok, Equals, true)
 315                 }
 316         } else {
 317                 for _, locator := range srcLocatorsMatchingPrefix {
 318                         _, ok := dstIndex[locator]
 319                         c.Assert(ok, Equals, true)
 320                 }
 321         }
 322
 323         for _, locator := range extraDstLocators {
 324                 _, ok := dstIndex[locator]
 325                 c.Assert(ok, Equals, true)
 326         }
 327
 328         if prefix == "" {
 329                 // all blocks from src and the two extra blocks
 330                 c.Assert(len(dstIndex), Equals, len(srcLocators)+len(extraDstLocators))
 331         } else {
 332                 // one matching prefix, 2 that were initially copied into dst along with src, and the extra blocks
 333                 c.Assert(len(dstIndex), Equals, len(srcLocatorsMatchingPrefix)+len(extraDstLocators)+2)
 334         }
 335 }
 336
 337 // Setup test data in src and dst.
 338 var srcLocators, srcLocatorsMatchingPrefix, dstLocators, extraDstLocators []string
 339
 340 func setupTestData(c *C, indexPrefix string) {
 341         srcLocators = []string{}
 342         srcLocatorsMatchingPrefix = []string{}
 343         dstLocators = []string{}
 344         extraDstLocators = []string{}
 345
 346         // Put a few blocks in src using kcSrc
 347         for i := 0; i < 5; i++ {
 348                 hash, _, err := kcSrc.PutB([]byte(fmt.Sprintf("test-data-%d", i)))
 349                 c.Check(err, IsNil)
 350
 351                 srcLocators = append(srcLocators, strings.Split(hash, "+A")[0])
 352                 if strings.HasPrefix(hash, indexPrefix) {
 353                         srcLocatorsMatchingPrefix = append(srcLocatorsMatchingPrefix, strings.Split(hash, "+A")[0])
 354                 }
 355         }
 356
 357         // Put first two of those src blocks in dst using kcDst
 358         for i := 0; i < 2; i++ {
 359                 hash, _, err := kcDst.PutB([]byte(fmt.Sprintf("test-data-%d", i)))
 360                 c.Check(err, IsNil)
 361                 dstLocators = append(dstLocators, strings.Split(hash, "+A")[0])
 362         }
 363
 364         // Put two more blocks in dst; they are not in src at all
 365         for i := 0; i < 2; i++ {
 366                 hash, _, err := kcDst.PutB([]byte(fmt.Sprintf("other-data-%d", i)))
 367                 c.Check(err, IsNil)
 368                 dstLocators = append(dstLocators, strings.Split(hash, "+A")[0])
 369                 extraDstLocators = append(extraDstLocators, strings.Split(hash, "+A")[0])
 370         }
 371 }
 372
 373 // Setup rsync using srcKeepServicesJSON with fake keepservers.
 374 // Expect error during performKeepRsync due to unreachable src keepservers.
 375 func (s *ServerRequiredSuite) TestErrorDuringRsync_FakeSrcKeepservers(c *C) {
 376         srcKeepServicesJSON = testKeepServicesJSON
 377
 378         setupRsync(c, false, false, 1)
 379
 380         err := performKeepRsync(kcSrc, kcDst, "", "")
 381         c.Check(strings.HasSuffix(err.Error(), "no such host"), Equals, true)
 382 }
 383
 384 // Setup rsync using dstKeepServicesJSON with fake keepservers.
 385 // Expect error during performKeepRsync due to unreachable dst keepservers.
 386 func (s *ServerRequiredSuite) TestErrorDuringRsync_FakeDstKeepservers(c *C) {
 387         dstKeepServicesJSON = testKeepServicesJSON
 388
 389         setupRsync(c, false, false, 1)
 390
 391         err := performKeepRsync(kcSrc, kcDst, "", "")
 392         c.Check(strings.HasSuffix(err.Error(), "no such host"), Equals, true)
 393 }
 394
 395 // Test rsync with signature error during Get from src.
 396 func (s *ServerRequiredSuite) TestErrorDuringRsync_ErrorGettingBlockFromSrc(c *C) {
 397         setupRsync(c, true, true, 1)
 398
 399         // put some blocks in src and dst
 400         setupTestData(c, "")
 401
 402         // Change blob signing key to a fake key, so that Get from src fails
 403         blobSigningKey = "thisisfakeblobsigningkey"
 404
 405         err := performKeepRsync(kcSrc, kcDst, blobSigningKey, "")
 406         c.Check(strings.HasSuffix(err.Error(), "Block not found"), Equals, true)
 407 }
 408
 409 // Test rsync with error during Put to src.
 410 func (s *ServerRequiredSuite) TestErrorDuringRsync_ErrorPuttingBlockInDst(c *C) {
 411         setupRsync(c, false, true, 1)
 412
 413         // put some blocks in src and dst
 414         setupTestData(c, "")
 415
 416         // Increase Want_replicas on dst to result in insufficient replicas error during Put
 417         kcDst.Want_replicas = 2
 418
 419         err := performKeepRsync(kcSrc, kcDst, blobSigningKey, "")
 420         c.Check(strings.HasSuffix(err.Error(), "Could not write sufficient replicas"), Equals, true)
 421 }
 422
 423 // Test loadConfig func
 424 func (s *ServerRequiredSuite) TestLoadConfig(c *C) {
 425         // Setup a src config file
 426         srcFile := setupConfigFile(c, "src-config")
 427         defer os.Remove(srcFile.Name())
 428         srcConfigFile := srcFile.Name()
 429
 430         // Setup a dst config file
 431         dstFile := setupConfigFile(c, "dst-config")
 432         defer os.Remove(dstFile.Name())
 433         dstConfigFile := dstFile.Name()
 434
 435         // load configuration from those files
 436         srcConfig, srcBlobSigningKey, err := loadConfig(srcConfigFile)
 437         c.Check(err, IsNil)
 438
 439         c.Assert(srcConfig.APIHost, Equals, "testhost")
 440         c.Assert(srcConfig.APIToken, Equals, "testtoken")
 441         c.Assert(srcConfig.APIHostInsecure, Equals, true)
 442         c.Assert(srcConfig.ExternalClient, Equals, false)
 443
 444         dstConfig, _, err := loadConfig(dstConfigFile)
 445         c.Check(err, IsNil)
 446
 447         c.Assert(dstConfig.APIHost, Equals, "testhost")
 448         c.Assert(dstConfig.APIToken, Equals, "testtoken")
 449         c.Assert(dstConfig.APIHostInsecure, Equals, true)
 450         c.Assert(dstConfig.ExternalClient, Equals, false)
 451
 452         c.Assert(srcBlobSigningKey, Equals, "abcdefg")
 453 }
 454
 455 // Test loadConfig func without setting up the config files
 456 func (s *ServerRequiredSuite) TestLoadConfig_MissingSrcConfig(c *C) {
 457         _, _, err := loadConfig("")
 458         c.Assert(err.Error(), Equals, "config file not specified")
 459 }
 460
 461 // Test loadConfig func - error reading config
 462 func (s *ServerRequiredSuite) TestLoadConfig_ErrorLoadingSrcConfig(c *C) {
 463         _, _, err := loadConfig("no-such-config-file")
 464         c.Assert(strings.HasSuffix(err.Error(), "no such file or directory"), Equals, true)
 465 }
 466
 467 func setupConfigFile(c *C, name string) *os.File {
 468         // Setup a config file
 469         file, err := ioutil.TempFile(os.TempDir(), name)
 470         c.Check(err, IsNil)
 471
 472         fileContent := "ARVADOS_API_HOST=testhost\n"
 473         fileContent += "ARVADOS_API_TOKEN=testtoken\n"
 474         fileContent += "ARVADOS_API_HOST_INSECURE=true\n"
 475         fileContent += "ARVADOS_BLOB_SIGNING_KEY=abcdefg"
 476
 477         _, err = file.Write([]byte(fileContent))
 478         c.Check(err, IsNil)
 479
 480         return file
 481 }