Add tracker health check client

build-index/tagclient/client.go

@@ -4,7 +4,7 @@
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-//     http://www.apache.org/licenses/LICENSE-2.0
+//	http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -38,6 +38,7 @@ var (
 
 // Client wraps tagserver endpoints.
 type Client interface {
+	CheckReadiness() error
 	Put(tag string, d core.Digest) error
 	PutAndReplicate(tag string, d core.Digest) error
 	Get(tag string) (core.Digest, error)
@@ -70,6 +71,14 @@ func NewSingleClient(addr string, config *tls.Config) Client {
 	return &singleClient{addr, config}
 }
 
+func (c *singleClient) CheckReadiness() error {
+	_, err := httputil.Get(
+		fmt.Sprintf("http://%s/readiness", c.addr),
+		httputil.SendTimeout(5*time.Second),
+		httputil.SendTLS(c.tls))
+	return err
+}
+
 func (c *singleClient) Put(tag string, d core.Digest) error {
 	_, err := httputil.Put(
 		fmt.Sprintf("http://%s/tags/%s/digest/%s", c.addr, url.PathEscape(tag), d.String()),
@@ -311,6 +320,33 @@ func (cc *clusterClient) do(request func(c Client) error) error {
 	return err
 }
 
+// doOnce tries the request on only one randomly chosen client without any retries if it fails.
+func (cc *clusterClient) doOnce(request func(c Client) error) error {
+	addrs := cc.hosts.Resolve().Sample(1)
+	if len(addrs) == 0 {
+		return errors.New("cluster client: no hosts could be resolved")
+	}
+	// read the only sampled addr
+	var addr string
+	for addr = range addrs {
+	}
+	err := request(NewSingleClient(addr, cc.tls))
+	if httputil.IsNetworkError(err) {
+		cc.hosts.Failed(addr)
+	}
+	return err
+}
+
+func (cc *clusterClient) CheckReadiness() error {
+	return cc.doOnce(func(c Client) error {
+		err := c.CheckReadiness()
+		if err != nil {
+			return fmt.Errorf("build index not ready: %v", err)
+		}
+		return nil
+	})
+}
+
 func (cc *clusterClient) Put(tag string, d core.Digest) error {
 	return cc.do(func(c Client) error { return c.Put(tag, d) })
 }

build-index/tagserver/server.go

@@ -4,7 +4,7 @@
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-//     http://www.apache.org/licenses/LICENSE-2.0
+//	http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -108,6 +108,7 @@ func (s *Server) Handler() http.Handler {
 	r.Use(middleware.LatencyTimer(s.stats))
 
 	r.Get("/health", handler.Wrap(s.healthHandler))
+	r.Get("/readiness", handler.Wrap(s.readinessCheckHandler))
 
 	r.Put("/tags/{tag}/digest/{digest}", handler.Wrap(s.putTagHandler))
 	r.Head("/tags/{tag}", handler.Wrap(s.hasTagHandler))
@@ -145,6 +146,15 @@ func (s *Server) healthHandler(w http.ResponseWriter, r *http.Request) error {
 	return nil
 }
 
+func (s *Server) readinessCheckHandler(w http.ResponseWriter, r *http.Request) error {
+	err := s.backends.CheckReadiness()
+	if err != nil {
+		return handler.Errorf("not ready to serve traffic: %s", err).Status(http.StatusServiceUnavailable)
+	}
+	fmt.Fprintln(w, "OK")
+	return nil
+}
+
 func (s *Server) putTagHandler(w http.ResponseWriter, r *http.Request) error {
 	tag, err := httputil.ParseParam(r, "tag")
 	if err != nil {

build-index/tagserver/server_test.go

@@ -4,7 +4,7 @@
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-//     http://www.apache.org/licenses/LICENSE-2.0
+//	http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -14,10 +14,12 @@
 package tagserver
 
 import (
+	"errors"
 	"fmt"
 	"io/ioutil"
 	"net/http"
 	"net/url"
+	"regexp"
 	"strconv"
 	"testing"
 	"time"
@@ -149,6 +151,53 @@ func TestHealth(t *testing.T) {
 	require.Equal("OK\n", string(b))
 }
 
+func TestCheckReadiness(t *testing.T) {
+	for _, tc := range []struct {
+		name                  string
+		mockStatErr           error
+		expectedErrMsgPattern string
+	}{
+		{
+			name:                  "success",
+			mockStatErr:           nil,
+			expectedErrMsgPattern: "",
+		},
+		{
+			name:                  "failure, 503 (since Stat fails)",
+			mockStatErr:           errors.New("test error"),
+			expectedErrMsgPattern: fmt.Sprintf(`build index not ready: GET http://127\.0\.0\.1:\d+/readiness 503: not ready to serve traffic: backend for namespace 'foo-bar/\*' not ready: test error`),
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			require := require.New(t)
+
+			mocks, cleanup := newServerMocks(t)
+			defer cleanup()
+
+			addr, stop := testutil.StartServer(mocks.handler())
+			defer stop()
+
+			client := newClusterClient(addr)
+			backendClient := mockbackend.NewMockClient(mocks.ctrl)
+			require.NoError(mocks.backends.Register("foo-bar/*", backendClient, true))
+
+			mockStat := &core.BlobInfo{}
+			if tc.mockStatErr != nil {
+				mockStat = nil
+			}
+			backendClient.EXPECT().Stat(backend.ReadinessCheckNamespace, backend.ReadinessCheckName).Return(mockStat, tc.mockStatErr)
+
+			err := client.CheckReadiness()
+			if tc.expectedErrMsgPattern == "" {
+				require.Nil(err)
+			} else {
+				r, _ := regexp.Compile(tc.expectedErrMsgPattern)
+				require.True(r.MatchString(err.Error()))
+			}
+		})
+	}
+}
+
 func TestPut(t *testing.T) {
 	require := require.New(t)
 

core/digest.go

@@ -23,11 +23,6 @@ import (
 	"strings"
 )
 
-const (
-	// DigestEmptyTar is the sha256 digest of an empty tar file.
-	DigestEmptyTar = "sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
-)
-
 // DigestList is a list of digests.
 type DigestList []Digest
 

lib/backend/constants.go

@@ -30,4 +30,5 @@ const (
 var (
 	ReadinessCheckNamespace string = core.NamespaceFixture()
 	ReadinessCheckName      string = core.DigestFixture().Hex()
+	ReadinessCheckDigest, _        = core.NewSHA256DigestFromHex(ReadinessCheckName)
 )

lib/backend/manager.go

@@ -143,7 +143,7 @@ func (m *Manager) GetClient(namespace string) (Client, error) {
 	return nil, ErrNamespaceNotFound
 }
 
-// IsReady returns whether the backends are ready (reachable).
+// CheckReadiness returns whether the backends are ready (available).
 // A backend must be explicitly configured as required for readiness to be checked.
 func (m *Manager) CheckReadiness() error {
 	for _, b := range m.backends {

lib/backend/manager_test.go

@@ -19,7 +19,6 @@ import (
 
 	"github.com/uber-go/tally"
 	"github.com/uber/kraken/core"
-	"github.com/uber/kraken/lib/backend"
 	. "github.com/uber/kraken/lib/backend"
 	"github.com/uber/kraken/lib/backend/backenderrors"
 	"github.com/uber/kraken/lib/backend/namepath"
@@ -222,8 +221,8 @@ func TestManagerCheckReadiness(t *testing.T) {
 				mockStat2 = nil
 			}
 
-			c1.EXPECT().Stat(backend.ReadinessCheckNamespace, backend.ReadinessCheckName).Return(mockStat1, tc.mockStat1Err).AnyTimes()
-			c2.EXPECT().Stat(backend.ReadinessCheckNamespace, backend.ReadinessCheckName).Return(mockStat2, tc.mockStat2Err).AnyTimes()
+			c1.EXPECT().Stat(ReadinessCheckNamespace, ReadinessCheckName).Return(mockStat1, tc.mockStat1Err).AnyTimes()
+			c2.EXPECT().Stat(ReadinessCheckNamespace, ReadinessCheckName).Return(mockStat2, tc.mockStat2Err).AnyTimes()
 
 			require.NoError(m.Register(n1, c1, tc.mustReady1))
 			require.NoError(m.Register(n2, c2, tc.mustReady2))