services/keepproxy/keepproxy_test.go

   1 package main
   2
   3 import (
   4         "crypto/md5"
   5         "fmt"
   6         "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
   7         "git.curoverse.com/arvados.git/sdk/go/arvadostest"
   8         "git.curoverse.com/arvados.git/sdk/go/keepclient"
   9         "io/ioutil"
  10         "log"
  11         "net/http"
  12         "os"
  13         "strings"
  14         "testing"
  15         "time"
  16
  17         . "gopkg.in/check.v1"
  18 )
  19
  20 // Gocheck boilerplate
  21 func Test(t *testing.T) {
  22         TestingT(t)
  23 }
  24
  25 // Gocheck boilerplate
  26 var _ = Suite(&ServerRequiredSuite{})
  27
  28 // Tests that require the Keep server running
  29 type ServerRequiredSuite struct{}
  30
  31 // Gocheck boilerplate
  32 var _ = Suite(&NoKeepServerSuite{})
  33
  34 // Test with no keepserver to simulate errors
  35 type NoKeepServerSuite struct{}
  36
  37 var TestProxyUUID = "zzzzz-bi6l4-lrixqc4fxofbmzz"
  38
  39 // Wait (up to 1 second) for keepproxy to listen on a port. This
  40 // avoids a race condition where we hit a "connection refused" error
  41 // because we start testing the proxy too soon.
  42 func waitForListener() {
  43         const (
  44                 ms = 5
  45         )
  46         for i := 0; listener == nil && i < 10000; i += ms {
  47                 time.Sleep(ms * time.Millisecond)
  48         }
  49         if listener == nil {
  50                 log.Fatalf("Timed out waiting for listener to start")
  51         }
  52 }
  53
  54 func closeListener() {
  55         if listener != nil {
  56                 listener.Close()
  57         }
  58 }
  59
  60 func (s *ServerRequiredSuite) SetUpSuite(c *C) {
  61         arvadostest.StartAPI()
  62         arvadostest.StartKeep(2, false)
  63 }
  64
  65 func (s *ServerRequiredSuite) SetUpTest(c *C) {
  66         arvadostest.ResetEnv()
  67 }
  68
  69 func (s *ServerRequiredSuite) TearDownSuite(c *C) {
  70         arvadostest.StopKeep(2)
  71         arvadostest.StopAPI()
  72 }
  73
  74 func (s *NoKeepServerSuite) SetUpSuite(c *C) {
  75         arvadostest.StartAPI()
  76         // We need API to have some keep services listed, but the
  77         // services themselves should be unresponsive.
  78         arvadostest.StartKeep(2, false)
  79         arvadostest.StopKeep(2)
  80 }
  81
  82 func (s *NoKeepServerSuite) SetUpTest(c *C) {
  83         arvadostest.ResetEnv()
  84 }
  85
  86 func (s *NoKeepServerSuite) TearDownSuite(c *C) {
  87         arvadostest.StopAPI()
  88 }
  89
  90 func runProxy(c *C, args []string, bogusClientToken bool) *keepclient.KeepClient {
  91         args = append([]string{"keepproxy"}, args...)
  92         os.Args = append(args, "-listen=:0")
  93         listener = nil
  94         go main()
  95         waitForListener()
  96
  97         arv, err := arvadosclient.MakeArvadosClient()
  98         c.Assert(err, Equals, nil)
  99         if bogusClientToken {
 100                 arv.ApiToken = "bogus-token"
 101         }
 102         kc := keepclient.New(&arv)
 103         sr := map[string]string{
 104                 TestProxyUUID: "http://" + listener.Addr().String(),
 105         }
 106         kc.SetServiceRoots(sr, sr, sr)
 107         kc.Arvados.External = true
 108
 109         return kc
 110 }
 111
 112 func (s *ServerRequiredSuite) TestDesiredReplicas(c *C) {
 113         kc := runProxy(c, nil, false)
 114         defer closeListener()
 115
 116         content := []byte("TestDesiredReplicas")
 117         hash := fmt.Sprintf("%x", md5.Sum(content))
 118
 119         for _, kc.Want_replicas = range []int{0, 1, 2} {
 120                 locator, rep, err := kc.PutB(content)
 121                 c.Check(err, Equals, nil)
 122                 c.Check(rep, Equals, kc.Want_replicas)
 123                 if rep > 0 {
 124                         c.Check(locator, Matches, fmt.Sprintf(`^%s\+%d(\+.+)?$`, hash, len(content)))
 125                 }
 126         }
 127 }
 128
 129 func (s *ServerRequiredSuite) TestPutAskGet(c *C) {
 130         kc := runProxy(c, nil, false)
 131         defer closeListener()
 132
 133         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 134         var hash2 string
 135
 136         {
 137                 _, _, err := kc.Ask(hash)
 138                 c.Check(err, Equals, keepclient.BlockNotFound)
 139                 log.Print("Finished Ask (expected BlockNotFound)")
 140         }
 141
 142         {
 143                 reader, _, _, err := kc.Get(hash)
 144                 c.Check(reader, Equals, nil)
 145                 c.Check(err, Equals, keepclient.BlockNotFound)
 146                 log.Print("Finished Get (expected BlockNotFound)")
 147         }
 148
 149         // Note in bug #5309 among other errors keepproxy would set
 150         // Content-Length incorrectly on the 404 BlockNotFound response, this
 151         // would result in a protocol violation that would prevent reuse of the
 152         // connection, which would manifest by the next attempt to use the
 153         // connection (in this case the PutB below) failing.  So to test for
 154         // that bug it's necessary to trigger an error response (such as
 155         // BlockNotFound) and then do something else with the same httpClient
 156         // connection.
 157
 158         {
 159                 var rep int
 160                 var err error
 161                 hash2, rep, err = kc.PutB([]byte("foo"))
 162                 c.Check(hash2, Matches, fmt.Sprintf(`^%s\+3(\+.+)?$`, hash))
 163                 c.Check(rep, Equals, 2)
 164                 c.Check(err, Equals, nil)
 165                 log.Print("Finished PutB (expected success)")
 166         }
 167
 168         {
 169                 blocklen, _, err := kc.Ask(hash2)
 170                 c.Assert(err, Equals, nil)
 171                 c.Check(blocklen, Equals, int64(3))
 172                 log.Print("Finished Ask (expected success)")
 173         }
 174
 175         {
 176                 reader, blocklen, _, err := kc.Get(hash2)
 177                 c.Assert(err, Equals, nil)
 178                 all, err := ioutil.ReadAll(reader)
 179                 c.Check(all, DeepEquals, []byte("foo"))
 180                 c.Check(blocklen, Equals, int64(3))
 181                 log.Print("Finished Get (expected success)")
 182         }
 183
 184         {
 185                 var rep int
 186                 var err error
 187                 hash2, rep, err = kc.PutB([]byte(""))
 188                 c.Check(hash2, Matches, `^d41d8cd98f00b204e9800998ecf8427e\+0(\+.+)?$`)
 189                 c.Check(rep, Equals, 2)
 190                 c.Check(err, Equals, nil)
 191                 log.Print("Finished PutB zero block")
 192         }
 193
 194         {
 195                 reader, blocklen, _, err := kc.Get("d41d8cd98f00b204e9800998ecf8427e")
 196                 c.Assert(err, Equals, nil)
 197                 all, err := ioutil.ReadAll(reader)
 198                 c.Check(all, DeepEquals, []byte(""))
 199                 c.Check(blocklen, Equals, int64(0))
 200                 log.Print("Finished Get zero block")
 201         }
 202 }
 203
 204 func (s *ServerRequiredSuite) TestPutAskGetForbidden(c *C) {
 205         kc := runProxy(c, nil, true)
 206         defer closeListener()
 207
 208         hash := fmt.Sprintf("%x", md5.Sum([]byte("bar")))
 209
 210         {
 211                 _, _, err := kc.Ask(hash)
 212                 errNotFound, _ := err.(keepclient.ErrNotFound)
 213                 c.Check(errNotFound, NotNil)
 214                 c.Assert(strings.Contains(err.Error(), "HTTP 403"), Equals, true)
 215                 log.Print("Ask 1")
 216         }
 217
 218         {
 219                 hash2, rep, err := kc.PutB([]byte("bar"))
 220                 c.Check(hash2, Equals, "")
 221                 c.Check(rep, Equals, 0)
 222                 c.Check(err, Equals, keepclient.InsufficientReplicasError)
 223                 log.Print("PutB")
 224         }
 225
 226         {
 227                 blocklen, _, err := kc.Ask(hash)
 228                 errNotFound, _ := err.(keepclient.ErrNotFound)
 229                 c.Check(errNotFound, NotNil)
 230                 c.Assert(strings.Contains(err.Error(), "HTTP 403"), Equals, true)
 231                 c.Check(blocklen, Equals, int64(0))
 232                 log.Print("Ask 2")
 233         }
 234
 235         {
 236                 _, blocklen, _, err := kc.Get(hash)
 237                 errNotFound, _ := err.(keepclient.ErrNotFound)
 238                 c.Check(errNotFound, NotNil)
 239                 c.Assert(strings.Contains(err.Error(), "HTTP 403"), Equals, true)
 240                 c.Check(blocklen, Equals, int64(0))
 241                 log.Print("Get")
 242         }
 243 }
 244
 245 func (s *ServerRequiredSuite) TestGetDisabled(c *C) {
 246         kc := runProxy(c, []string{"-no-get"}, false)
 247         defer closeListener()
 248
 249         hash := fmt.Sprintf("%x", md5.Sum([]byte("baz")))
 250
 251         {
 252                 _, _, err := kc.Ask(hash)
 253                 errNotFound, _ := err.(keepclient.ErrNotFound)
 254                 c.Check(errNotFound, NotNil)
 255                 c.Assert(strings.Contains(err.Error(), "HTTP 400"), Equals, true)
 256                 log.Print("Ask 1")
 257         }
 258
 259         {
 260                 hash2, rep, err := kc.PutB([]byte("baz"))
 261                 c.Check(hash2, Matches, fmt.Sprintf(`^%s\+3(\+.+)?$`, hash))
 262                 c.Check(rep, Equals, 2)
 263                 c.Check(err, Equals, nil)
 264                 log.Print("PutB")
 265         }
 266
 267         {
 268                 blocklen, _, err := kc.Ask(hash)
 269                 errNotFound, _ := err.(keepclient.ErrNotFound)
 270                 c.Check(errNotFound, NotNil)
 271                 c.Assert(strings.Contains(err.Error(), "HTTP 400"), Equals, true)
 272                 c.Check(blocklen, Equals, int64(0))
 273                 log.Print("Ask 2")
 274         }
 275
 276         {
 277                 _, blocklen, _, err := kc.Get(hash)
 278                 errNotFound, _ := err.(keepclient.ErrNotFound)
 279                 c.Check(errNotFound, NotNil)
 280                 c.Assert(strings.Contains(err.Error(), "HTTP 400"), Equals, true)
 281                 c.Check(blocklen, Equals, int64(0))
 282                 log.Print("Get")
 283         }
 284 }
 285
 286 func (s *ServerRequiredSuite) TestPutDisabled(c *C) {
 287         kc := runProxy(c, []string{"-no-put"}, false)
 288         defer closeListener()
 289
 290         hash2, rep, err := kc.PutB([]byte("quux"))
 291         c.Check(hash2, Equals, "")
 292         c.Check(rep, Equals, 0)
 293         c.Check(err, Equals, keepclient.InsufficientReplicasError)
 294 }
 295
 296 func (s *ServerRequiredSuite) TestCorsHeaders(c *C) {
 297         runProxy(c, nil, false)
 298         defer closeListener()
 299
 300         {
 301                 client := http.Client{}
 302                 req, err := http.NewRequest("OPTIONS",
 303                         fmt.Sprintf("http://%s/%x+3", listener.Addr().String(), md5.Sum([]byte("foo"))),
 304                         nil)
 305                 req.Header.Add("Access-Control-Request-Method", "PUT")
 306                 req.Header.Add("Access-Control-Request-Headers", "Authorization, X-Keep-Desired-Replicas")
 307                 resp, err := client.Do(req)
 308                 c.Check(err, Equals, nil)
 309                 c.Check(resp.StatusCode, Equals, 200)
 310                 body, err := ioutil.ReadAll(resp.Body)
 311                 c.Check(string(body), Equals, "")
 312                 c.Check(resp.Header.Get("Access-Control-Allow-Methods"), Equals, "GET, HEAD, POST, PUT, OPTIONS")
 313                 c.Check(resp.Header.Get("Access-Control-Allow-Origin"), Equals, "*")
 314         }
 315
 316         {
 317                 resp, err := http.Get(
 318                         fmt.Sprintf("http://%s/%x+3", listener.Addr().String(), md5.Sum([]byte("foo"))))
 319                 c.Check(err, Equals, nil)
 320                 c.Check(resp.Header.Get("Access-Control-Allow-Headers"), Equals, "Authorization, Content-Length, Content-Type, X-Keep-Desired-Replicas")
 321                 c.Check(resp.Header.Get("Access-Control-Allow-Origin"), Equals, "*")
 322         }
 323 }
 324
 325 func (s *ServerRequiredSuite) TestPostWithoutHash(c *C) {
 326         runProxy(c, nil, false)
 327         defer closeListener()
 328
 329         {
 330                 client := http.Client{}
 331                 req, err := http.NewRequest("POST",
 332                         "http://"+listener.Addr().String()+"/",
 333                         strings.NewReader("qux"))
 334                 req.Header.Add("Authorization", "OAuth2 4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h")
 335                 req.Header.Add("Content-Type", "application/octet-stream")
 336                 resp, err := client.Do(req)
 337                 c.Check(err, Equals, nil)
 338                 body, err := ioutil.ReadAll(resp.Body)
 339                 c.Check(err, Equals, nil)
 340                 c.Check(string(body), Matches,
 341                         fmt.Sprintf(`^%x\+3(\+.+)?$`, md5.Sum([]byte("qux"))))
 342         }
 343 }
 344
 345 func (s *ServerRequiredSuite) TestStripHint(c *C) {
 346         c.Check(removeHint.ReplaceAllString("http://keep.zzzzz.arvadosapi.com:25107/2228819a18d3727630fa30c81853d23f+67108864+A37b6ab198qqqq28d903b975266b23ee711e1852c@55635f73+K@zzzzz", "$1"),
 347                 Equals,
 348                 "http://keep.zzzzz.arvadosapi.com:25107/2228819a18d3727630fa30c81853d23f+67108864+A37b6ab198qqqq28d903b975266b23ee711e1852c@55635f73")
 349         c.Check(removeHint.ReplaceAllString("http://keep.zzzzz.arvadosapi.com:25107/2228819a18d3727630fa30c81853d23f+67108864+K@zzzzz+A37b6ab198qqqq28d903b975266b23ee711e1852c@55635f73", "$1"),
 350                 Equals,
 351                 "http://keep.zzzzz.arvadosapi.com:25107/2228819a18d3727630fa30c81853d23f+67108864+A37b6ab198qqqq28d903b975266b23ee711e1852c@55635f73")
 352         c.Check(removeHint.ReplaceAllString("http://keep.zzzzz.arvadosapi.com:25107/2228819a18d3727630fa30c81853d23f+67108864+A37b6ab198qqqq28d903b975266b23ee711e1852c@55635f73+K@zzzzz-zzzzz-zzzzzzzzzzzzzzz", "$1"),
 353                 Equals,
 354                 "http://keep.zzzzz.arvadosapi.com:25107/2228819a18d3727630fa30c81853d23f+67108864+A37b6ab198qqqq28d903b975266b23ee711e1852c@55635f73+K@zzzzz-zzzzz-zzzzzzzzzzzzzzz")
 355         c.Check(removeHint.ReplaceAllString("http://keep.zzzzz.arvadosapi.com:25107/2228819a18d3727630fa30c81853d23f+67108864+K@zzzzz-zzzzz-zzzzzzzzzzzzzzz+A37b6ab198qqqq28d903b975266b23ee711e1852c@55635f73", "$1"),
 356                 Equals,
 357                 "http://keep.zzzzz.arvadosapi.com:25107/2228819a18d3727630fa30c81853d23f+67108864+K@zzzzz-zzzzz-zzzzzzzzzzzzzzz+A37b6ab198qqqq28d903b975266b23ee711e1852c@55635f73")
 358
 359 }
 360
 361 // Test GetIndex
 362 //   Put one block, with 2 replicas
 363 //   With no prefix (expect the block locator, twice)
 364 //   With an existing prefix (expect the block locator, twice)
 365 //   With a valid but non-existing prefix (expect "\n")
 366 //   With an invalid prefix (expect error)
 367 func (s *ServerRequiredSuite) TestGetIndex(c *C) {
 368         kc := runProxy(c, nil, false)
 369         defer closeListener()
 370
 371         // Put "index-data" blocks
 372         data := []byte("index-data")
 373         hash := fmt.Sprintf("%x", md5.Sum(data))
 374
 375         hash2, rep, err := kc.PutB(data)
 376         c.Check(hash2, Matches, fmt.Sprintf(`^%s\+10(\+.+)?$`, hash))
 377         c.Check(rep, Equals, 2)
 378         c.Check(err, Equals, nil)
 379
 380         reader, blocklen, _, err := kc.Get(hash)
 381         c.Assert(err, Equals, nil)
 382         c.Check(blocklen, Equals, int64(10))
 383         all, err := ioutil.ReadAll(reader)
 384         c.Check(all, DeepEquals, data)
 385
 386         // Put some more blocks
 387         _, rep, err = kc.PutB([]byte("some-more-index-data"))
 388         c.Check(err, Equals, nil)
 389
 390         kc.Arvados.ApiToken = arvadostest.DataManagerToken
 391
 392         // Invoke GetIndex
 393         for _, spec := range []struct {
 394                 prefix         string
 395                 expectTestHash bool
 396                 expectOther    bool
 397         }{
 398                 {"", true, true},         // with no prefix
 399                 {hash[:3], true, false},  // with matching prefix
 400                 {"abcdef", false, false}, // with no such prefix
 401         } {
 402                 indexReader, err := kc.GetIndex(TestProxyUUID, spec.prefix)
 403                 c.Assert(err, Equals, nil)
 404                 indexResp, err := ioutil.ReadAll(indexReader)
 405                 c.Assert(err, Equals, nil)
 406                 locators := strings.Split(string(indexResp), "\n")
 407                 gotTestHash := 0
 408                 gotOther := 0
 409                 for _, locator := range locators {
 410                         if locator == "" {
 411                                 continue
 412                         }
 413                         c.Check(locator[:len(spec.prefix)], Equals, spec.prefix)
 414                         if locator[:32] == hash {
 415                                 gotTestHash++
 416                         } else {
 417                                 gotOther++
 418                         }
 419                 }
 420                 c.Check(gotTestHash == 2, Equals, spec.expectTestHash)
 421                 c.Check(gotOther > 0, Equals, spec.expectOther)
 422         }
 423
 424         // GetIndex with invalid prefix
 425         _, err = kc.GetIndex(TestProxyUUID, "xyz")
 426         c.Assert((err != nil), Equals, true)
 427 }
 428
 429 func (s *ServerRequiredSuite) TestPutAskGetInvalidToken(c *C) {
 430         kc := runProxy(c, nil, false)
 431         defer closeListener()
 432
 433         // Put a test block
 434         hash, rep, err := kc.PutB([]byte("foo"))
 435         c.Check(err, Equals, nil)
 436         c.Check(rep, Equals, 2)
 437
 438         for _, token := range []string{
 439                 "nosuchtoken",
 440                 "2ym314ysp27sk7h943q6vtc378srb06se3pq6ghurylyf3pdmx", // expired
 441         } {
 442                 // Change token to given bad token
 443                 kc.Arvados.ApiToken = token
 444
 445                 // Ask should result in error
 446                 _, _, err = kc.Ask(hash)
 447                 c.Check(err, NotNil)
 448                 errNotFound, _ := err.(keepclient.ErrNotFound)
 449                 c.Check(errNotFound.Temporary(), Equals, false)
 450                 c.Assert(strings.Contains(err.Error(), "HTTP 403"), Equals, true)
 451
 452                 // Get should result in error
 453                 _, _, _, err = kc.Get(hash)
 454                 c.Check(err, NotNil)
 455                 errNotFound, _ = err.(keepclient.ErrNotFound)
 456                 c.Check(errNotFound.Temporary(), Equals, false)
 457                 c.Assert(strings.Contains(err.Error(), "HTTP 403 \"Missing or invalid Authorization header\""), Equals, true)
 458         }
 459 }
 460
 461 func (s *ServerRequiredSuite) TestAskGetKeepProxyConnectionError(c *C) {
 462         arv, err := arvadosclient.MakeArvadosClient()
 463         c.Assert(err, Equals, nil)
 464
 465         // keepclient with no such keep server
 466         kc := keepclient.New(&arv)
 467         locals := map[string]string{
 468                 TestProxyUUID: "http://localhost:12345",
 469         }
 470         kc.SetServiceRoots(locals, nil, nil)
 471
 472         // Ask should result in temporary connection refused error
 473         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 474         _, _, err = kc.Ask(hash)
 475         c.Check(err, NotNil)
 476         errNotFound, _ := err.(*keepclient.ErrNotFound)
 477         c.Check(errNotFound.Temporary(), Equals, true)
 478         c.Assert(strings.Contains(err.Error(), "connection refused"), Equals, true)
 479
 480         // Get should result in temporary connection refused error
 481         _, _, _, err = kc.Get(hash)
 482         c.Check(err, NotNil)
 483         errNotFound, _ = err.(*keepclient.ErrNotFound)
 484         c.Check(errNotFound.Temporary(), Equals, true)
 485         c.Assert(strings.Contains(err.Error(), "connection refused"), Equals, true)
 486 }
 487
 488 func (s *NoKeepServerSuite) TestAskGetNoKeepServerError(c *C) {
 489         kc := runProxy(c, nil, false)
 490         defer closeListener()
 491
 492         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 493         for _, f := range []func() error{
 494                 func() error {
 495                         _, _, err := kc.Ask(hash)
 496                         return err
 497                 },
 498                 func() error {
 499                         _, _, _, err := kc.Get(hash)
 500                         return err
 501                 },
 502         } {
 503                 err := f()
 504                 c.Assert(err, NotNil)
 505                 errNotFound, _ := err.(*keepclient.ErrNotFound)
 506                 c.Check(errNotFound.Temporary(), Equals, true)
 507                 c.Check(err, ErrorMatches, `.*HTTP 502.*`)
 508         }
 509 }