Merge branch 'feature/gitaly-feature-flag' into 'master'

Proxy GET /info/refs to Gitaly

See merge request !105

internal/api/api.go

@@ -88,6 +88,10 @@ type Response struct {
 	Entry string `json:"entry"`
 	// Used to communicate terminal session details
 	Terminal *TerminalSettings
+	// Path to Gitaly Socket
+	GitalySocketPath string
+	// Path to Gitaly HTTP resource
+	GitalyResourcePath string
 }
 
 // singleJoiningSlash is taken from reverseproxy.go:NewSingleHostReverseProxy

internal/config/config.go

+package config
+
+import (
+	"net/url"
+	"time"
+)
+
+type Config struct {
+	Backend             *url.URL
+	Version             string
+	DocumentRoot        string
+	DevelopmentMode     bool
+	Socket              string
+	ProxyHeadersTimeout time.Duration
+	APILimit            uint
+	APIQueueLimit       uint
+	APIQueueTimeout     time.Duration
+}

internal/git/git-http.go

@@ -17,11 +17,19 @@ import (
 	"strings"
 
 	"gitlab.com/gitlab-org/gitlab-workhorse/internal/api"
+	"gitlab.com/gitlab-org/gitlab-workhorse/internal/config"
+	"gitlab.com/gitlab-org/gitlab-workhorse/internal/gitaly"
 	"gitlab.com/gitlab-org/gitlab-workhorse/internal/helper"
 )
 
-func GetInfoRefs(a *api.API) http.Handler {
-	return repoPreAuthorizeHandler(a, handleGetInfoRefs)
+func GetInfoRefsHandler(a *api.API, cfg *config.Config) http.Handler {
+	return repoPreAuthorizeHandler(a, func(rw http.ResponseWriter, r *http.Request, apiResponse *api.Response) {
+		if apiResponse.GitalySocketPath == "" {
+			handleGetInfoRefs(rw, r, apiResponse)
+		} else {
+			handleGetInfoRefsWithGitaly(rw, r, apiResponse, gitaly.NewClient(apiResponse.GitalySocketPath, cfg))
+		}
+	})
 }
 
 func PostRPC(a *api.API) http.Handler {
@@ -54,6 +62,17 @@ func repoPreAuthorizeHandler(myAPI *api.API, handleFunc api.HandleFunc) http.Han
 	}, "")
 }
 
+func handleGetInfoRefsWithGitaly(rw http.ResponseWriter, r *http.Request, a *api.Response, gitalyClient *gitaly.Client) {
+	req := *r // Make a copy of r
+	req.Header = helper.HeaderClone(r.Header)
+	req.Header.Add("Gitaly-Repo-Path", a.RepoPath)
+	req.Header.Add("Gitaly-GL-Id", a.GL_ID)
+	req.URL.Path = path.Join(a.GitalyResourcePath, subCommand(getService(r)))
+	req.URL.RawQuery = ""
+
+	gitalyClient.Proxy.ServeHTTP(rw, &req)
+}
+
 func handleGetInfoRefs(rw http.ResponseWriter, r *http.Request, a *api.Response) {
 	w := NewGitHttpResponseWriter(rw)
 	// Log 0 bytes in because we ignore the request body (and there usually is none anyway).

internal/gitaly/gitaly.go

+package gitaly
+
+import (
+	"sync"
+
+	"gitlab.com/gitlab-org/gitlab-workhorse/internal/badgateway"
+	"gitlab.com/gitlab-org/gitlab-workhorse/internal/config"
+	"gitlab.com/gitlab-org/gitlab-workhorse/internal/proxy"
+)
+
+type Client struct {
+	Proxy *proxy.Proxy
+}
+
+type clientCache struct {
+	sync.RWMutex
+	clients map[string]*Client
+}
+
+var cache = clientCache{
+	clients: make(map[string]*Client),
+}
+
+func NewClient(socketPath string, cfg *config.Config) *Client {
+	if client := getClient(socketPath); client != nil {
+		return client
+	}
+
+	cache.Lock()
+	defer cache.Unlock()
+
+	if client := cache.clients[socketPath]; client != nil {
+		return client
+	}
+
+	client := &Client{}
+	roundTripper := badgateway.NewRoundTripper(nil, socketPath, cfg.ProxyHeadersTimeout, cfg.DevelopmentMode)
+	client.Proxy = proxy.NewProxy(nil, cfg.Version, roundTripper)
+	client.Proxy.AllowResponseBuffering = false
+
+	cache.clients[socketPath] = client
+
+	return client
+}
+
+func getClient(socketPath string) *Client {
+	cache.RLock()
+	defer cache.RUnlock()
+
+	return cache.clients[socketPath]
+}

internal/proxy/proxy.go

@@ -11,13 +11,23 @@ import (
 	"gitlab.com/gitlab-org/gitlab-workhorse/internal/helper"
 )
 
+var (
+	defaultTarget = helper.URLMustParse("http://localhost")
+)
+
 type Proxy struct {
 	Version                string
 	reverseProxy           *httputil.ReverseProxy
+	AllowResponseBuffering bool
 }
 
 func NewProxy(myURL *url.URL, version string, roundTripper *badgateway.RoundTripper) *Proxy {
-	p := Proxy{Version: version}
+	p := Proxy{Version: version, AllowResponseBuffering: true}
+
+	if myURL == nil {
+		myURL = defaultTarget
+	}
+
 	u := *myURL // Make a copy of p.URL
 	u.Path = ""
 	p.reverseProxy = httputil.NewSingleHostReverseProxy(&u)
@@ -34,7 +44,9 @@ func (p *Proxy) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	req.Header.Set("Gitlab-Workhorse", p.Version)
 	req.Header.Set("Gitlab-Workhorse-Proxy-Start", fmt.Sprintf("%d", time.Now().UnixNano()))
 
+	if p.AllowResponseBuffering {
 		helper.AllowResponseBuffering(w)
+	}
 
 	p.reverseProxy.ServeHTTP(w, &req)
 }

internal/upstream/routes.go

@@ -116,7 +116,7 @@ func (u *Upstream) configureRoutes() {
 
 	u.Routes = []routeEntry{
 		// Git Clone
-		route("GET", gitProjectPattern+`info/refs\z`, git.GetInfoRefs(api)),
+		route("GET", gitProjectPattern+`info/refs\z`, git.GetInfoRefsHandler(api, &u.Config)),
 		route("POST", gitProjectPattern+`git-upload-pack\z`, contentEncodingHandler(git.PostRPC(api))),
 		route("POST", gitProjectPattern+`git-receive-pack\z`, contentEncodingHandler(git.PostRPC(api))),
 		route("PUT", gitProjectPattern+`gitlab-lfs/objects/([0-9a-f]{64})/([0-9]+)\z`, lfs.PutStore(api, proxy)),

internal/upstream/upstream.go

@@ -9,11 +9,10 @@ package upstream
 import (
 	"fmt"
 	"net/http"
-	"net/url"
 	"strings"
-	"time"
 
 	"gitlab.com/gitlab-org/gitlab-workhorse/internal/badgateway"
+	"gitlab.com/gitlab-org/gitlab-workhorse/internal/config"
 	"gitlab.com/gitlab-org/gitlab-workhorse/internal/helper"
 	"gitlab.com/gitlab-org/gitlab-workhorse/internal/upload"
 	"gitlab.com/gitlab-org/gitlab-workhorse/internal/urlprefix"
@@ -26,33 +25,21 @@ var (
 	}
 )
 
-type Config struct {
-	Backend             *url.URL
-	Version             string
-	DocumentRoot        string
-	DevelopmentMode     bool
-	Socket              string
-	ProxyHeadersTimeout time.Duration
-	APILimit            uint
-	APIQueueLimit       uint
-	APIQueueTimeout     time.Duration
-}
-
 type Upstream struct {
-	Config
+	config.Config
 	URLPrefix    urlprefix.Prefix
 	Routes       []routeEntry
 	RoundTripper *badgateway.RoundTripper
 }
 
-func NewUpstream(config Config) *Upstream {
+func NewUpstream(cfg config.Config) *Upstream {
 	up := Upstream{
-		Config: config,
+		Config: cfg,
 	}
 	if up.Backend == nil {
 		up.Backend = DefaultBackend
 	}
-	up.RoundTripper = badgateway.NewRoundTripper(up.Backend, up.Socket, up.ProxyHeadersTimeout, config.DevelopmentMode)
+	up.RoundTripper = badgateway.NewRoundTripper(up.Backend, up.Socket, up.ProxyHeadersTimeout, cfg.DevelopmentMode)
 	up.configureURLPrefix()
 	up.configureRoutes()
 	return &up

main.go

@@ -24,6 +24,7 @@ import (
 	"syscall"
 	"time"
 
+	"gitlab.com/gitlab-org/gitlab-workhorse/internal/config"
 	"gitlab.com/gitlab-org/gitlab-workhorse/internal/queueing"
 	"gitlab.com/gitlab-org/gitlab-workhorse/internal/secret"
 	"gitlab.com/gitlab-org/gitlab-workhorse/internal/upstream"
@@ -108,7 +109,7 @@ func main() {
 	}
 
 	secret.SetPath(*secretPath)
-	upConfig := upstream.Config{
+	cfg := config.Config{
 		Backend:             backendURL,
 		Socket:              *authSocket,
 		Version:             Version,
@@ -120,7 +121,7 @@ func main() {
 		APIQueueTimeout:     *apiQueueTimeout,
 	}
 
-	up := wrapRaven(upstream.NewUpstream(upConfig))
+	up := wrapRaven(upstream.NewUpstream(cfg))
 
 	log.Fatal(http.Serve(listener, up))
 }

main_test.go

@@ -8,6 +8,7 @@ import (
 	"io"
 	"io/ioutil"
 	"log"
+	"net"
 	"net/http"
 	"net/http/httptest"
 	"os"
@@ -19,6 +20,7 @@ import (
 	"time"
 
 	"gitlab.com/gitlab-org/gitlab-workhorse/internal/api"
+	"gitlab.com/gitlab-org/gitlab-workhorse/internal/config"
 	"gitlab.com/gitlab-org/gitlab-workhorse/internal/helper"
 	"gitlab.com/gitlab-org/gitlab-workhorse/internal/testhelper"
 	"gitlab.com/gitlab-org/gitlab-workhorse/internal/upstream"
@@ -556,6 +558,70 @@ func TestApiContentTypeBlock(t *testing.T) {
 	}
 }
 
+func TestGetInfoRefsProxiedToGitalySuccessfully(t *testing.T) {
+	content := "0000"
+	apiResponse := gitOkBody(t)
+	apiResponse.GitalyResourcePath = "/projects/1/git-http/info-refs"
+
+	gitalyPath := path.Join(apiResponse.GitalyResourcePath, "upload-pack")
+	gitaly := startGitalyServer(regexp.MustCompile(gitalyPath), content)
+	defer gitaly.Close()
+
+	apiResponse.GitalySocketPath = gitaly.Listener.Addr().String()
+	ts := testAuthServer(nil, 200, apiResponse)
+	defer ts.Close()
+
+	ws := startWorkhorseServer(ts.URL)
+	defer ws.Close()
+
+	resource := "/gitlab-org/gitlab-test.git/info/refs?service=git-upload-pack"
+	resp, err := http.Get(ws.URL + resource)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	responseBody, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		t.Error(err)
+	}
+
+	if !bytes.Equal(responseBody, []byte(content)) {
+		t.Errorf("GET %q: Expected %q, got %q", resource, content, responseBody)
+	}
+}
+
+func TestGetInfoRefsHandledLocallyDueToEmptyGitalySocketPath(t *testing.T) {
+	gitaly := startGitalyServer(nil, "Gitaly response: should never reach the client")
+	defer gitaly.Close()
+
+	apiResponse := gitOkBody(t)
+	apiResponse.GitalySocketPath = ""
+	ts := testAuthServer(nil, 200, apiResponse)
+	defer ts.Close()
+
+	ws := startWorkhorseServer(ts.URL)
+	defer ws.Close()
+
+	resource := "/gitlab-org/gitlab-test.git/info/refs?service=git-upload-pack"
+	resp, err := http.Get(ws.URL + resource)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	responseBody, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		t.Error(err)
+	}
+
+	if resp.StatusCode != 200 {
+		t.Errorf("GET %q: expected 200, got %d", resource, resp.StatusCode)
+	}
+
+	if bytes.Contains(responseBody, []byte("Gitaly response")) {
+		t.Errorf("GET %q: request should not have been proxied to Gitaly", resource)
+	}
+}
+
 func setupStaticFile(fpath, content string) error {
 	cwd, err := os.Getwd()
 	if err != nil {
@@ -643,18 +709,47 @@ func archiveOKServer(t *testing.T, archiveName string) *httptest.Server {
 	})
 }
 
-func startWorkhorseServer(authBackend string) *httptest.Server {
-	testhelper.ConfigureSecret()
-	config := upstream.Config{
-		Backend:      helper.URLMustParse(authBackend),
+func newUpstreamConfig(authBackend string) *config.Config {
+	return &config.Config{
 		Version:      "123",
 		DocumentRoot: testDocumentRoot,
+		Backend:      helper.URLMustParse(authBackend),
 	}
+}
+
+func startWorkhorseServer(authBackend string) *httptest.Server {
+	return startWorkhorseServerWithConfig(newUpstreamConfig(authBackend))
+}
+
+func startWorkhorseServerWithConfig(cfg *config.Config) *httptest.Server {
+	testhelper.ConfigureSecret()
+	u := upstream.NewUpstream(*cfg)
 
-	u := upstream.NewUpstream(config)
 	return httptest.NewServer(u)
 }
 
+func startGitalyServer(url *regexp.Regexp, body string) *httptest.Server {
+	ts := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+
+		if url != nil && !url.MatchString(r.URL.Path) {
+			log.Println("Gitaly", r.Method, r.URL, "DENY")
+			w.WriteHeader(404)
+			return
+		}
+
+		fmt.Fprint(w, body)
+	}))
+
+	listener, err := net.Listen("unix", path.Join(scratchDir, "gitaly.sock"))
+	if err != nil {
+		log.Fatal(err)
+	}
+	ts.Listener = listener
+
+	ts.Start()
+	return ts
+}
+
 func runOrFail(t *testing.T, cmd *exec.Cmd) {
 	out, err := cmd.CombinedOutput()
 	t.Logf("%s", out)
@@ -663,7 +758,7 @@ func runOrFail(t *testing.T, cmd *exec.Cmd) {
 	}
 }
 
-func gitOkBody(t *testing.T) interface{} {
+func gitOkBody(t *testing.T) *api.Response {
 	return &api.Response{
 		GL_ID:    "user-123",
 		RepoPath: repoPath(t),