Merge branch 'mk/async-geo-proxy-api-call' into 'master'

Geo: Implement async Geo Proxy API calls

See merge request gitlab-org/gitlab!66491

workhorse/internal/api/api.go

@@ -3,7 +3,6 @@ package api
 import (
 	"bytes"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -40,8 +39,6 @@ type API struct {
 	Version string
 }
 
-var ErrNotGeoSecondary = errors.New("this is not a Geo secondary site")
-
 var (
 	requestsCounter = promauto.NewCounterVec(
 		prometheus.CounterOpts{
@@ -399,7 +396,6 @@ func validResponseContentType(resp *http.Response) bool {
 	return helper.IsContentType(ResponseContentType, resp.Header.Get("Content-Type"))
 }
 
-// TODO: Cache the result of the API requests https://gitlab.com/gitlab-org/gitlab/-/issues/329671
 func (api *API) GetGeoProxyURL() (*url.URL, error) {
 	geoProxyApiUrl := *api.URL
 	geoProxyApiUrl.Path, geoProxyApiUrl.RawPath = joinURLPath(api.URL, geoProxyEndpointPath)
@@ -424,10 +420,6 @@ func (api *API) GetGeoProxyURL() (*url.URL, error) {
 		return nil, fmt.Errorf("GetGeoProxyURL: decode response: %v", err)
 	}
 
-	if response.GeoProxyURL == "" {
-		return nil, ErrNotGeoSecondary
-	}
-
 	geoProxyURL, err := url.Parse(response.GeoProxyURL)
 	if err != nil {
 		return nil, fmt.Errorf("GetGeoProxyURL: Could not parse Geo proxy URL: %v, err: %v", response.GeoProxyURL, err)

workhorse/internal/api/api_test.go

@@ -22,16 +22,14 @@ func TestGetGeoProxyURLWhenGeoSecondary(t *testing.T) {
 	geoProxyURL, err := getGeoProxyURLGivenResponse(t, `{"geo_proxy_url":"http://primary"}`)
 
 	require.NoError(t, err)
-	require.NotNil(t, geoProxyURL)
 	require.Equal(t, "http://primary", geoProxyURL.String())
 }
 
 func TestGetGeoProxyURLWhenGeoPrimaryOrNonGeo(t *testing.T) {
 	geoProxyURL, err := getGeoProxyURLGivenResponse(t, "{}")
 
-	require.Error(t, err)
-	require.Equal(t, ErrNotGeoSecondary, err)
-	require.Nil(t, geoProxyURL)
+	require.NoError(t, err)
+	require.Equal(t, "", geoProxyURL.String())
 }
 
 func getGeoProxyURLGivenResponse(t *testing.T, givenInternalApiResponse string) (*url.URL, error) {

workhorse/internal/upstream/upstream.go

@@ -10,6 +10,7 @@ import (
 	"fmt"
 	"os"
 	"sync"
+	"time"
 
 	"net/http"
 	"net/url"
@@ -35,6 +36,7 @@ var (
 	requestHeaderBlacklist = []string{
 		upload.RewrittenFieldsHeader,
 	}
+	geoProxyApiPollingInterval = 10 * time.Second
 )
 
 type upstream struct {
@@ -48,6 +50,7 @@ type upstream struct {
 	geoLocalRoutes        []routeEntry
 	geoProxyCableRoute    routeEntry
 	geoProxyRoute         routeEntry
+	geoProxyTestChannel   chan struct{}
 	accessLogger          *logrus.Logger
 	enableGeoProxyFeature bool
 	mu                    sync.RWMutex
@@ -61,6 +64,9 @@ func newUpstream(cfg config.Config, accessLogger *logrus.Logger, routesCallback
 	up := upstream{
 		Config:       cfg,
 		accessLogger: accessLogger,
+		// Kind of a feature flag. See https://gitlab.com/groups/gitlab-org/-/epics/5914#note_564974130
+		enableGeoProxyFeature: os.Getenv("GEO_SECONDARY_PROXY") == "1",
+		geoProxyBackend:       &url.URL{},
 	}
 	if up.Backend == nil {
 		up.Backend = DefaultBackend
@@ -79,10 +85,13 @@ func newUpstream(cfg config.Config, accessLogger *logrus.Logger, routesCallback
 		up.Version,
 		up.RoundTripper,
 	)
-	// Kind of a feature flag. See https://gitlab.com/groups/gitlab-org/-/epics/5914#note_564974130
-	up.enableGeoProxyFeature = os.Getenv("GEO_SECONDARY_PROXY") == "1"
+
 	routesCallback(&up)
 
+	if up.enableGeoProxyFeature {
+		go up.pollGeoProxyAPI()
+	}
+
 	var correlationOpts []correlation.InboundHandlerOption
 	if cfg.PropagateCorrelationID {
 		correlationOpts = append(correlationOpts, correlation.WithPropagation())
@@ -168,19 +177,14 @@ func (u *upstream) findRoute(cleanedPath string, r *http.Request) *routeEntry {
 }
 
 func (u *upstream) findGeoProxyRoute(cleanedPath string, r *http.Request) *routeEntry {
-	geoProxyURL, err := u.APIClient.GetGeoProxyURL()
+	u.mu.RLock()
+	defer u.mu.RUnlock()
 
-	if err == nil {
-		u.setGeoProxyRoutes(geoProxyURL)
-		return u.matchGeoProxyRoute(cleanedPath, r)
-	} else if err != apipkg.ErrNotGeoSecondary {
-		log.WithRequest(r).WithError(err).Error("Geo Proxy: Unable to determine Geo Proxy URL. Falling back to normal routing")
+	if u.geoProxyBackend.String() == "" {
+		log.WithRequest(r).Debug("Geo Proxy: Not a Geo proxy")
+		return nil
 	}
 
-	return nil
-}
-
-func (u *upstream) matchGeoProxyRoute(cleanedPath string, r *http.Request) *routeEntry {
 	// Some routes are safe to serve from this GitLab instance
 	for _, ro := range u.geoLocalRoutes {
 		if ro.isMatch(cleanedPath, r) {
@@ -191,8 +195,6 @@ func (u *upstream) matchGeoProxyRoute(cleanedPath string, r *http.Request) *rout
 
 	log.WithRequest(r).WithFields(log.Fields{"geoProxyBackend": u.geoProxyBackend}).Debug("Geo Proxy: Forward this request")
 
-	u.mu.RLock()
-	defer u.mu.RUnlock()
 	if cleanedPath == "/-/cable" {
 		return &u.geoProxyCableRoute
 	}
@@ -200,15 +202,40 @@ func (u *upstream) matchGeoProxyRoute(cleanedPath string, r *http.Request) *rout
 	return &u.geoProxyRoute
 }
 
-func (u *upstream) setGeoProxyRoutes(geoProxyURL *url.URL) {
+func (u *upstream) pollGeoProxyAPI() {
+	for {
+		u.callGeoProxyAPI()
+
+		// Notify tests when callGeoProxyAPI() finishes
+		if u.geoProxyTestChannel != nil {
+			u.geoProxyTestChannel <- struct{}{}
+		}
+
+		time.Sleep(geoProxyApiPollingInterval)
+	}
+}
+
+// Calls /api/v4/geo/proxy and sets up routes
+func (u *upstream) callGeoProxyAPI() {
+	geoProxyURL, err := u.APIClient.GetGeoProxyURL()
+	if err != nil {
+		log.WithError(err).WithFields(log.Fields{"geoProxyBackend": u.geoProxyBackend}).Error("Geo Proxy: Unable to determine Geo Proxy URL. Fallback on cached value.")
+		return
+	}
+
+	if u.geoProxyBackend.String() != geoProxyURL.String() {
+		log.WithFields(log.Fields{"oldGeoProxyURL": u.geoProxyBackend, "newGeoProxyURL": geoProxyURL}).Info("Geo Proxy: URL changed")
+		u.updateGeoProxyFields(geoProxyURL)
+	}
+}
+
+func (u *upstream) updateGeoProxyFields(geoProxyURL *url.URL) {
 	u.mu.Lock()
 	defer u.mu.Unlock()
-	if u.geoProxyBackend == nil || u.geoProxyBackend.String() != geoProxyURL.String() {
-		log.WithFields(log.Fields{"geoProxyURL": geoProxyURL}).Debug("Geo Proxy: Update GeoProxyRoute")
-		u.geoProxyBackend = geoProxyURL
-		geoProxyRoundTripper := roundtripper.NewBackendRoundTripper(u.geoProxyBackend, "", u.ProxyHeadersTimeout, u.DevelopmentMode)
-		geoProxyUpstream := proxypkg.NewProxy(u.geoProxyBackend, u.Version, geoProxyRoundTripper)
-		u.geoProxyCableRoute = u.wsRoute(`^/-/cable\z`, geoProxyUpstream)
-		u.geoProxyRoute = u.route("", "", geoProxyUpstream)
-	}
+
+	u.geoProxyBackend = geoProxyURL
+	geoProxyRoundTripper := roundtripper.NewBackendRoundTripper(u.geoProxyBackend, "", u.ProxyHeadersTimeout, u.DevelopmentMode)
+	geoProxyUpstream := proxypkg.NewProxy(u.geoProxyBackend, u.Version, geoProxyRoundTripper)
+	u.geoProxyCableRoute = u.wsRoute(`^/-/cable\z`, geoProxyUpstream)
+	u.geoProxyRoute = u.route("", "", geoProxyUpstream)
 }

workhorse/internal/upstream/upstream_test.go

@@ -141,7 +141,7 @@ func TestGeoProxyFeatureEnabledOnNonGeoSecondarySite(t *testing.T) {
 	runTestCases(t, ws, testCases)
 }
 
-func TestGeoProxyWithAPIError(t *testing.T) {
+func TestGeoProxyFeatureEnabledButWithAPIError(t *testing.T) {
 	geoProxyEndpointResponseBody := "Invalid response"
 	railsServer, deferredClose := startRailsServer("Local Rails server", geoProxyEndpointResponseBody)
 	defer deferredClose()
@@ -214,10 +214,15 @@ func startRailsServer(railsServerName string, geoProxyEndpointResponseBody strin
 }
 
 func startWorkhorseServer(railsServerURL string, enableGeoProxyFeature bool) (*httptest.Server, func()) {
+	geoProxyTestChannel := make(chan struct{})
+
 	myConfigureRoutes := func(u *upstream) {
 		// Enable environment variable "feature flag"
 		u.enableGeoProxyFeature = enableGeoProxyFeature
 
+		// An empty message will be sent to this channel after every callGeoProxyAPI()
+		u.geoProxyTestChannel = geoProxyTestChannel
+
 		// call original
 		configureRoutes(u)
 	}
@@ -226,5 +231,13 @@ func startWorkhorseServer(railsServerURL string, enableGeoProxyFeature bool) (*h
 	ws := httptest.NewServer(upstreamHandler)
 	testhelper.ConfigureSecret()
 
+	if enableGeoProxyFeature {
+		// Wait for an empty message from callGeoProxyAPI(). This should be done on
+		// all tests where enableGeoProxyFeature is true, including the ones where
+		// we expect geoProxyURL to be nil or error, to ensure the tests do not pass
+		// by coincidence.
+		<-geoProxyTestChannel
+	}
+
 	return ws, ws.Close
 }