Add agent endpoint for resolving tags

agent/agentserver/client.go

@@ -16,6 +16,7 @@ package agentserver
 import (
 	"fmt"
 	"io"
+	"io/ioutil"
 	"net/url"
 
 	"github.com/uber/kraken/core"
@@ -32,6 +33,23 @@ func NewClient(addr string) *Client {
 	return &Client{addr}
 }
 
+func (c *Client) GetTag(tag string) (core.Digest, error) {
+	resp, err := httputil.Get(fmt.Sprintf("http://%s/tags/%s", c.addr, url.PathEscape(tag)))
+	if err != nil {
+		return core.Digest{}, err
+	}
+	defer resp.Body.Close()
+	b, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return core.Digest{}, fmt.Errorf("read body: %s", err)
+	}
+	d, err := core.ParseSHA256Digest(string(b))
+	if err != nil {
+		return core.Digest{}, fmt.Errorf("parse digest: %s", err)
+	}
+	return d, nil
+}
+
 // Download returns the blob for namespace / d. Callers should close the
 // returned ReadCloser when done reading the blob.
 func (c *Client) Download(namespace string, d core.Digest) (io.ReadCloser, error) {

agent/agentserver/server.go

@@ -24,6 +24,7 @@ import (
 	"github.com/pressly/chi"
 	"github.com/uber-go/tally"
 
+	"github.com/uber/kraken/build-index/tagclient"
 	"github.com/uber/kraken/core"
 	"github.com/uber/kraken/lib/middleware"
 	"github.com/uber/kraken/lib/store"
@@ -41,19 +42,21 @@ type Server struct {
 	stats  tally.Scope
 	cads   *store.CADownloadStore
 	sched  scheduler.ReloadableScheduler
+	tags   tagclient.Client
 }
 
 // New creates a new Server.
 func New(
 	config Config,
 	stats tally.Scope,
 	cads *store.CADownloadStore,
-	sched scheduler.ReloadableScheduler) *Server {
+	sched scheduler.ReloadableScheduler,
+	tags tagclient.Client) *Server {
 
 	stats = stats.Tagged(map[string]string{
 		"module": "agentserver",
 	})
-	return &Server{config, stats, cads, sched}
+	return &Server{config, stats, cads, sched, tags}
 }
 
 // Handler returns the HTTP handler.
@@ -65,6 +68,8 @@ func (s *Server) Handler() http.Handler {
 
 	r.Get("/health", handler.Wrap(s.healthHandler))
 
+	r.Get("/tags/{tag}", handler.Wrap(s.getTagHandler))
+
 	r.Get("/namespace/{namespace}/blobs/{digest}", handler.Wrap(s.downloadBlobHandler))
 
 	r.Delete("/blobs/{digest}", handler.Wrap(s.deleteBlobHandler))
@@ -80,6 +85,23 @@ func (s *Server) Handler() http.Handler {
 	return r
 }
 
+// getTagHandler proxies get tag requests to the build-index.
+func (s *Server) getTagHandler(w http.ResponseWriter, r *http.Request) error {
+	tag, err := httputil.ParseParam(r, "tag")
+	if err != nil {
+		return err
+	}
+	d, err := s.tags.Get(tag)
+	if err != nil {
+		if err == tagclient.ErrTagNotFound {
+			return handler.ErrorStatus(http.StatusNotFound)
+		}
+		return handler.Errorf("get tag: %s", err)
+	}
+	io.WriteString(w, d.String())
+	return nil
+}
+
 // downloadBlobHandler downloads a blob through p2p.
 func (s *Server) downloadBlobHandler(w http.ResponseWriter, r *http.Request) error {
 	namespace, err := httputil.ParseParam(r, "namespace")

agent/agentserver/server_test.go

@@ -22,6 +22,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/uber/kraken/build-index/tagclient"
 	"github.com/uber/kraken/core"
 	"github.com/uber/kraken/lib/store"
 	"github.com/uber/kraken/lib/torrent/scheduler"
@@ -31,6 +32,41 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func TestGetTag(t *testing.T) {
+	require := require.New(t)
+
+	mocks, cleanup := newServerMocks(t)
+	defer cleanup()
+
+	tag := core.TagFixture()
+	d := core.DigestFixture()
+
+	mocks.tags.EXPECT().Get(tag).Return(d, nil)
+
+	c := NewClient(mocks.startServer())
+
+	result, err := c.GetTag(tag)
+	require.NoError(err)
+	require.Equal(d, result)
+}
+
+func TestGetTagNotFound(t *testing.T) {
+	require := require.New(t)
+
+	mocks, cleanup := newServerMocks(t)
+	defer cleanup()
+
+	tag := core.TagFixture()
+
+	mocks.tags.EXPECT().Get(tag).Return(core.Digest{}, tagclient.ErrTagNotFound)
+
+	c := NewClient(mocks.startServer())
+
+	_, err := c.GetTag(tag)
+	require.Error(err)
+	require.True(httputil.IsNotFound(err))
+}
+
 func TestDownload(t *testing.T) {
 	require := require.New(t)
 

agent/agentserver/testutils_test.go

@@ -21,12 +21,14 @@ import (
 
 	"github.com/uber/kraken/lib/store"
 	"github.com/uber/kraken/mocks/lib/torrent/scheduler"
+	"github.com/uber/kraken/mocks/build-index/tagclient"
 	"github.com/uber/kraken/utils/testutil"
 )
 
 type serverMocks struct {
 	cads    *store.CADownloadStore
 	sched   *mockscheduler.MockReloadableScheduler
+	tags    *mocktagclient.MockClient
 	cleanup *testutil.Cleanup
 }
 
@@ -41,11 +43,13 @@ func newServerMocks(t *testing.T) (*serverMocks, func()) {
 
 	sched := mockscheduler.NewMockReloadableScheduler(ctrl)
 
-	return &serverMocks{cads, sched, &cleanup}, cleanup.Run
+	tags := mocktagclient.NewMockClient(ctrl)
+
+	return &serverMocks{cads, sched, tags, &cleanup}, cleanup.Run
 }
 
 func (m *serverMocks) startServer() string {
-	s := New(Config{}, tally.NoopScope, m.cads, m.sched)
+	s := New(Config{}, tally.NoopScope, m.cads, m.sched, m.tags)
 	addr, stop := testutil.StartServer(s.Handler())
 	m.cleanup.Add(stop)
 	return addr

agent/cmd/cmd.go

@@ -157,7 +157,7 @@ func Run(flags *Flags) {
 		log.Fatalf("Failed to init registry: %s", err)
 	}
 
-	agentServer := agentserver.New(config.AgentServer, stats, cads, sched)
+	agentServer := agentserver.New(config.AgentServer, stats, cads, sched, tagClient)
 	addr := fmt.Sprintf(":%d", flags.AgentServerPort)
 	log.Infof("Starting agent server on %s", addr)
 	go func() {