+func (s *HandlerSuite) TestDiscoveryDocCache(c *check.C) {
+ countRailsReqs := func() int {
+ n := 0
+ for _, req := range s.railsSpy.RequestDumps {
+ if bytes.Contains(req, []byte("/discovery/v1/apis/arvados/v1/rest")) {
+ n++
+ }
+ }
+ return n
+ }
+ getDD := func() int {
+ req := httptest.NewRequest(http.MethodGet, "/discovery/v1/apis/arvados/v1/rest", nil)
+ resp := httptest.NewRecorder()
+ s.handler.ServeHTTP(resp, req)
+ if resp.Code == http.StatusOK {
+ var dd arvados.DiscoveryDocument
+ err := json.Unmarshal(resp.Body.Bytes(), &dd)
+ c.Check(err, check.IsNil)
+ c.Check(dd.Schemas["Collection"].UUIDPrefix, check.Equals, "4zz18")
+ }
+ return resp.Code
+ }
+ getDDConcurrently := func(n int, expectCode int, checkArgs ...interface{}) *sync.WaitGroup {
+ var wg sync.WaitGroup
+ for i := 0; i < n; i++ {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ c.Check(getDD(), check.Equals, append([]interface{}{expectCode}, checkArgs...)...)
+ }()
+ }
+ return &wg
+ }
+ clearCache := func() {
+ for _, ent := range s.handler.cache {
+ ent.refreshLock.Lock()
+ ent.mtx.Lock()
+ ent.body, ent.header, ent.refreshAfter = nil, nil, time.Time{}
+ ent.mtx.Unlock()
+ ent.refreshLock.Unlock()
+ }
+ }
+ waitPendingUpdates := func() {
+ for _, ent := range s.handler.cache {
+ ent.refreshLock.Lock()
+ defer ent.refreshLock.Unlock()
+ ent.mtx.Lock()
+ defer ent.mtx.Unlock()
+ }
+ }
+ refreshNow := func() {
+ waitPendingUpdates()
+ for _, ent := range s.handler.cache {
+ ent.refreshAfter = time.Now()
+ }
+ }
+ expireNow := func() {
+ waitPendingUpdates()
+ for _, ent := range s.handler.cache {
+ ent.expireAfter = time.Now()
+ }
+ }
+
+ // Easy path: first req fetches, subsequent reqs use cache.
+ c.Check(countRailsReqs(), check.Equals, 0)
+ c.Check(getDD(), check.Equals, http.StatusOK)
+ c.Check(countRailsReqs(), check.Equals, 1)
+ c.Check(getDD(), check.Equals, http.StatusOK)
+ c.Check(countRailsReqs(), check.Equals, 1)
+ c.Check(getDD(), check.Equals, http.StatusOK)
+ c.Check(countRailsReqs(), check.Equals, 1)
+
+ // To guarantee we have concurrent requests, we set up
+ // railsSpy to hold up the Handler's outgoing requests until
+ // we send to (or close) holdReqs.
+ holdReqs := make(chan struct{})
+ s.railsSpy.Director = func(*http.Request) {
+ <-holdReqs
+ }
+
+ // Race at startup: first req fetches, other concurrent reqs
+ // wait for the initial fetch to complete, then all return.
+ clearCache()
+ reqsBefore := countRailsReqs()
+ wg := getDDConcurrently(5, http.StatusOK, check.Commentf("race at startup"))
+ close(holdReqs)
+ wg.Wait()
+ c.Check(countRailsReqs(), check.Equals, reqsBefore+1)
+
+ // Race after expiry: concurrent reqs return the cached data
+ // but initiate a new fetch in the background.
+ refreshNow()
+ holdReqs = make(chan struct{})
+ wg = getDDConcurrently(5, http.StatusOK, check.Commentf("race after expiry"))
+ reqsBefore = countRailsReqs()
+ close(holdReqs)
+ wg.Wait()
+ for deadline := time.Now().Add(time.Second); time.Now().Before(deadline) && countRailsReqs() < reqsBefore+1; {
+ time.Sleep(time.Second / 100)
+ }
+ c.Check(countRailsReqs(), check.Equals, reqsBefore+1)
+
+ // Configure railsSpy to return an error or bad content
+ // depending on flags.
+ var wantError, wantBadContent bool
+ s.railsSpy.Director = func(req *http.Request) {
+ if wantError {
+ req.Method = "MAKE-COFFEE"
+ } else if wantBadContent {
+ req.URL.Path = "/_health/ping"
+ req.Header.Set("Authorization", "Bearer "+arvadostest.ManagementToken)
+ }
+ }
+
+ // Error at startup (empty cache) => caller gets error, and we
+ // make an upstream attempt for each incoming request because
+ // we have nothing better to return
+ clearCache()
+ wantError, wantBadContent = true, false
+ reqsBefore = countRailsReqs()
+ holdReqs = make(chan struct{})
+ wg = getDDConcurrently(5, http.StatusBadGateway, check.Commentf("error at startup"))
+ close(holdReqs)
+ wg.Wait()
+ c.Check(countRailsReqs(), check.Equals, reqsBefore+5)
+
+ // Response status is OK but body is not a discovery document
+ wantError, wantBadContent = false, true
+ reqsBefore = countRailsReqs()
+ c.Check(getDD(), check.Equals, http.StatusBadGateway)
+ c.Check(countRailsReqs(), check.Equals, reqsBefore+1)
+
+ // Error condition clears => caller gets OK, cache is warmed
+ // up
+ wantError, wantBadContent = false, false
+ reqsBefore = countRailsReqs()
+ getDDConcurrently(5, http.StatusOK, check.Commentf("success after errors at startup")).Wait()
+ c.Check(countRailsReqs(), check.Equals, reqsBefore+1)
+
+ // Error with warm cache => caller gets OK (with no attempt to
+ // re-fetch)
+ wantError, wantBadContent = true, false
+ reqsBefore = countRailsReqs()
+ getDDConcurrently(5, http.StatusOK, check.Commentf("error with warm cache")).Wait()
+ c.Check(countRailsReqs(), check.Equals, reqsBefore)
+
+ // Error with stale cache => caller gets OK with stale data
+ // while the re-fetch is attempted in the background
+ refreshNow()
+ wantError, wantBadContent = true, false
+ reqsBefore = countRailsReqs()
+ holdReqs = make(chan struct{})
+ getDDConcurrently(5, http.StatusOK, check.Commentf("error with stale cache")).Wait()
+ close(holdReqs)
+ // Only one attempt to re-fetch (holdReqs ensured the first
+ // update took long enough for the last incoming request to
+ // arrive)
+ c.Check(countRailsReqs(), check.Equals, reqsBefore+1)
+
+ refreshNow()
+ wantError, wantBadContent = false, false
+ reqsBefore = countRailsReqs()
+ holdReqs = make(chan struct{})
+ getDDConcurrently(5, http.StatusOK, check.Commentf("refresh cache after error condition clears")).Wait()
+ close(holdReqs)
+ waitPendingUpdates()
+ c.Check(countRailsReqs(), check.Equals, reqsBefore+1)
+
+ // Make sure expireAfter is getting set
+ waitPendingUpdates()
+ exp := s.handler.cache["/discovery/v1/apis/arvados/v1/rest"].expireAfter.Sub(time.Now())
+ c.Check(exp > cacheTTL, check.Equals, true)
+ c.Check(exp < cacheExpire, check.Equals, true)
+
+ // After the cache *expires* it behaves as if uninitialized:
+ // each incoming request does a new upstream request until one
+ // succeeds.
+ //
+ // First check failure after expiry:
+ expireNow()
+ wantError, wantBadContent = true, false
+ reqsBefore = countRailsReqs()
+ holdReqs = make(chan struct{})
+ wg = getDDConcurrently(5, http.StatusBadGateway, check.Commentf("error after expiry"))
+ close(holdReqs)
+ wg.Wait()
+ c.Check(countRailsReqs(), check.Equals, reqsBefore+5)
+
+ // Success after expiry:
+ wantError, wantBadContent = false, false
+ reqsBefore = countRailsReqs()
+ holdReqs = make(chan struct{})
+ wg = getDDConcurrently(5, http.StatusOK, check.Commentf("success after expiry"))
+ close(holdReqs)
+ wg.Wait()
+ c.Check(countRailsReqs(), check.Equals, reqsBefore+1)
+}
+
+func (s *HandlerSuite) TestVocabularyExport(c *check.C) {
+ voc := `{
+ "strict_tags": false,
+ "tags": {
+ "IDTAGIMPORTANCE": {
+ "strict": false,
+ "labels": [{"label": "Importance"}],
+ "values": {
+ "HIGH": {
+ "labels": [{"label": "High"}]
+ },
+ "LOW": {
+ "labels": [{"label": "Low"}]
+ }
+ }
+ }
+ }
+ }`
+ f, err := os.CreateTemp("", "test-vocabulary-*.json")
+ c.Assert(err, check.IsNil)
+ defer os.Remove(f.Name())
+ _, err = f.WriteString(voc)
+ c.Assert(err, check.IsNil)
+ f.Close()
+ s.cluster.API.VocabularyPath = f.Name()
+ for _, method := range []string{"GET", "OPTIONS"} {
+ c.Log(c.TestName()+" ", method)
+ req := httptest.NewRequest(method, "/arvados/v1/vocabulary", nil)
+ resp := httptest.NewRecorder()
+ s.handler.ServeHTTP(resp, req)
+ c.Log(resp.Body.String())
+ if !c.Check(resp.Code, check.Equals, http.StatusOK) {
+ continue
+ }
+ c.Check(resp.Header().Get("Access-Control-Allow-Origin"), check.Equals, `*`)
+ c.Check(resp.Header().Get("Access-Control-Allow-Methods"), check.Matches, `.*\bGET\b.*`)
+ c.Check(resp.Header().Get("Access-Control-Allow-Headers"), check.Matches, `.+`)
+ if method == "OPTIONS" {
+ c.Check(resp.Body.String(), check.HasLen, 0)
+ continue
+ }
+ var expectedVoc, receivedVoc *arvados.Vocabulary
+ err := json.Unmarshal([]byte(voc), &expectedVoc)
+ c.Check(err, check.IsNil)
+ err = json.Unmarshal(resp.Body.Bytes(), &receivedVoc)
+ c.Check(err, check.IsNil)
+ c.Check(receivedVoc, check.DeepEquals, expectedVoc)
+ }
+}
+
+func (s *HandlerSuite) TestVocabularyFailedCheckStatus(c *check.C) {
+ voc := `{
+ "strict_tags": false,
+ "tags": {
+ "IDTAGIMPORTANCE": {
+ "strict": true,
+ "labels": [{"label": "Importance"}],
+ "values": {
+ "HIGH": {
+ "labels": [{"label": "High"}]
+ },
+ "LOW": {
+ "labels": [{"label": "Low"}]
+ }
+ }
+ }
+ }
+ }`
+ f, err := os.CreateTemp("", "test-vocabulary-*.json")
+ c.Assert(err, check.IsNil)
+ defer os.Remove(f.Name())
+ _, err = f.WriteString(voc)
+ c.Assert(err, check.IsNil)
+ f.Close()
+ s.cluster.API.VocabularyPath = f.Name()
+
+ req := httptest.NewRequest("POST", "/arvados/v1/collections",
+ strings.NewReader(`{
+ "collection": {
+ "properties": {
+ "IDTAGIMPORTANCE": "Critical"
+ }
+ }
+ }`))
+ req.Header.Set("Authorization", "Bearer "+arvadostest.ActiveToken)
+ req.Header.Set("Content-type", "application/json")
+
+ resp := httptest.NewRecorder()
+ s.handler.ServeHTTP(resp, req)
+ c.Log(resp.Body.String())
+ c.Assert(resp.Code, check.Equals, http.StatusBadRequest)
+ var jresp httpserver.ErrorResponse
+ err = json.Unmarshal(resp.Body.Bytes(), &jresp)
+ c.Check(err, check.IsNil)
+ c.Assert(len(jresp.Errors), check.Equals, 1)
+ c.Check(jresp.Errors[0], check.Matches, `.*tag value.*is not valid for key.*`)
+}
+