Instrumentation: Proxy status code correction and various improvements (#47473)

For a proxied request, e.g. Grafana's datasource or plugin proxy:
If the request is cancelled, e.g. from the browser, the HTTP status code is 
now 499 Client closed request instead of 502 Bad gateway.
If the request times out, e.g. takes longer time than allowed, the HTTP status 
code is now 504 Gateway timeout instead of 502 Bad gateway.
This also means that request metrics and logs will get their status codes 
adjusted according to above.

Fixes #46337
Fixes #46338

pkg/api/app_routes.go

@@ -27,10 +27,10 @@ func (hs *HTTPServer) initAppPluginRoutes(r *web.Mux) {
 			Renegotiation:      tls.RenegotiateFreelyAsClient,
 		},
 		Proxy: http.ProxyFromEnvironment,
-		Dial: (&net.Dialer{
+		DialContext: (&net.Dialer{
 			Timeout:   30 * time.Second,
 			KeepAlive: 30 * time.Second,
-		}).Dial,
+		}).DialContext,
 		TLSHandshakeTimeout: 10 * time.Second,
 	}
 
@@ -65,6 +65,7 @@ func AppPluginRoute(route *plugins.Route, appID string, hs *HTTPServer) web.Hand
 
 		proxy := pluginproxy.NewApiPluginProxy(c, path, route, appID, hs.Cfg, hs.PluginSettings, hs.SecretsService)
 		proxy.Transport = pluginProxyTransport
+
 		proxy.ServeHTTP(c.Resp, c.Req)
 	}
 }

pkg/api/grafana_com_proxy.go

@@ -7,22 +7,24 @@ import (
 	"net/url"
 	"time"
 
+	"github.com/grafana/grafana/pkg/infra/log"
 	"github.com/grafana/grafana/pkg/models"
 	"github.com/grafana/grafana/pkg/setting"
 	"github.com/grafana/grafana/pkg/util"
+	"github.com/grafana/grafana/pkg/util/proxyutil"
 	"github.com/grafana/grafana/pkg/web"
 )
 
 var grafanaComProxyTransport = &http.Transport{
 	Proxy: http.ProxyFromEnvironment,
-	Dial: (&net.Dialer{
+	DialContext: (&net.Dialer{
 		Timeout:   30 * time.Second,
 		KeepAlive: 30 * time.Second,
-	}).Dial,
+	}).DialContext,
 	TLSHandshakeTimeout: 10 * time.Second,
 }
 
-func ReverseProxyGnetReq(proxyPath string, version string) *httputil.ReverseProxy {
+func ReverseProxyGnetReq(logger log.Logger, proxyPath string, version string) *httputil.ReverseProxy {
 	url, _ := url.Parse(setting.GrafanaComUrl)
 
 	director := func(req *http.Request) {
@@ -41,13 +43,12 @@ func ReverseProxyGnetReq(proxyPath string, version string) *httputil.ReverseProx
 		req.Header.Add("grafana-version", version)
 	}
 
-	return &httputil.ReverseProxy{Director: director}
+	return proxyutil.NewReverseProxy(logger, director)
 }
 
 func (hs *HTTPServer) ProxyGnetRequest(c *models.ReqContext) {
 	proxyPath := web.Params(c.Req)["*"]
-	proxy := ReverseProxyGnetReq(proxyPath, hs.Cfg.BuildVersion)
+	proxy := ReverseProxyGnetReq(c.Logger, proxyPath, hs.Cfg.BuildVersion)
 	proxy.Transport = grafanaComProxyTransport
 	proxy.ServeHTTP(c.Resp, c.Req)
-	c.Resp.Header().Del("Set-Cookie")
 }

pkg/api/pluginproxy/ds_proxy.go

@@ -5,9 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"io/ioutil"
-	"log"
 	"net/http"
-	"net/http/httputil"
 	"net/url"
 	"strconv"
 	"strings"
@@ -48,35 +46,10 @@ type DataSourceProxy struct {
 	secretsService     secrets.Service
 }
 
-type handleResponseTransport struct {
-	transport http.RoundTripper
-}
-
-func (t *handleResponseTransport) RoundTrip(req *http.Request) (*http.Response, error) {
-	res, err := t.transport.RoundTrip(req)
-	if err != nil {
-		return nil, err
-	}
-	res.Header.Del("Set-Cookie")
-	proxyutil.SetProxyResponseHeaders(res.Header)
-	return res, nil
-}
-
 type httpClient interface {
 	Do(req *http.Request) (*http.Response, error)
 }
 
-type logWrapper struct {
-	logger glog.Logger
-}
-
-// Write writes log messages as bytes from proxy
-func (lw *logWrapper) Write(p []byte) (n int, err error) {
-	withoutNewline := strings.TrimSuffix(string(p), "\n")
-	lw.logger.Error("Data proxy error", "error", withoutNewline)
-	return len(p), nil
-}
-
 // NewDataSourceProxy creates a new Datasource proxy
 func NewDataSourceProxy(ds *models.DataSource, pluginRoutes []*plugins.Route, ctx *models.ReqContext,
 	proxyPath string, cfg *setting.Cfg, clientProvider httpclient.Provider,
@@ -115,8 +88,14 @@ func (proxy *DataSourceProxy) HandleRequest() {
 		return
 	}
 
-	proxyErrorLogger := logger.New("userId", proxy.ctx.UserId, "orgId", proxy.ctx.OrgId, "uname", proxy.ctx.Login,
-		"path", proxy.ctx.Req.URL.Path, "remote_addr", proxy.ctx.RemoteAddr(), "referer", proxy.ctx.Req.Referer())
+	proxyErrorLogger := logger.New(
+		"userId", proxy.ctx.UserId,
+		"orgId", proxy.ctx.OrgId,
+		"uname", proxy.ctx.Login,
+		"path", proxy.ctx.Req.URL.Path,
+		"remote_addr", proxy.ctx.RemoteAddr(),
+		"referer", proxy.ctx.Req.Referer(),
+	)
 
 	transport, err := proxy.dataSourcesService.GetHTTPTransport(proxy.ds, proxy.clientProvider)
 	if err != nil {
@@ -124,36 +103,36 @@ func (proxy *DataSourceProxy) HandleRequest() {
 		return
 	}
 
-	reverseProxy := &httputil.ReverseProxy{
-		Director:      proxy.director,
-		FlushInterval: time.Millisecond * 200,
-		ErrorLog:      log.New(&logWrapper{logger: proxyErrorLogger}, "", 0),
-		Transport: &handleResponseTransport{
-			transport: transport,
-		},
-		ModifyResponse: func(resp *http.Response) error {
-			if resp.StatusCode == 401 {
-				// The data source rejected the request as unauthorized, convert to 400 (bad request)
-				body, err := ioutil.ReadAll(resp.Body)
-				if err != nil {
-					return fmt.Errorf("failed to read data source response body: %w", err)
-				}
-				_ = resp.Body.Close()
-
-				proxyErrorLogger.Info("Authentication to data source failed", "body", string(body), "statusCode",
-					resp.StatusCode)
-				msg := "Authentication to data source failed"
-				*resp = http.Response{
-					StatusCode:    400,
-					Status:        "Bad Request",
-					Body:          ioutil.NopCloser(strings.NewReader(msg)),
-					ContentLength: int64(len(msg)),
-				}
+	modifyResponse := func(resp *http.Response) error {
+		if resp.StatusCode == 401 {
+			// The data source rejected the request as unauthorized, convert to 400 (bad request)
+			body, err := ioutil.ReadAll(resp.Body)
+			if err != nil {
+				return fmt.Errorf("failed to read data source response body: %w", err)
 			}
-			return nil
-		},
+			_ = resp.Body.Close()
+
+			proxyErrorLogger.Info("Authentication to data source failed", "body", string(body), "statusCode",
+				resp.StatusCode)
+			msg := "Authentication to data source failed"
+			*resp = http.Response{
+				StatusCode:    400,
+				Status:        "Bad Request",
+				Body:          ioutil.NopCloser(strings.NewReader(msg)),
+				ContentLength: int64(len(msg)),
+				Header:        http.Header{},
+			}
+		}
+		return nil
 	}
 
+	reverseProxy := proxyutil.NewReverseProxy(
+		proxyErrorLogger,
+		proxy.director,
+		proxyutil.WithTransport(transport),
+		proxyutil.WithModifyResponse(modifyResponse),
+	)
+
 	proxy.logRequest()
 	ctx, span := proxy.tracer.Start(proxy.ctx.Req.Context(), "datasource reverse proxy")
 	defer span.End()
@@ -236,14 +215,8 @@ func (proxy *DataSourceProxy) director(req *http.Request) {
 	}
 
 	proxyutil.ClearCookieHeader(req, keepCookieNames)
-	proxyutil.PrepareProxyRequest(req)
-
 	req.Header.Set("User-Agent", fmt.Sprintf("Grafana/%s", setting.BuildVersion))
 
-	// Clear Origin and Referer to avoid CORS issues
-	req.Header.Del("Origin")
-	req.Header.Del("Referer")
-
 	jsonData := make(map[string]interface{})
 	if proxy.ds.JsonData != nil {
 		jsonData, err = proxy.ds.JsonData.Map()

pkg/api/pluginproxy/ds_proxy_test.go

@@ -456,8 +456,6 @@ func TestDataSourceProxy_routeRule(t *testing.T) {
 
 		assert.Equal(t, "http://host/root/path/to/folder/", req.URL.String())
 
-		assert.Empty(t, req.Header.Get("Origin"))
-		assert.Empty(t, req.Header.Get("Referer"))
 		assert.Equal(t, "stillthere", req.Header.Get("X-Canary"))
 	})
 

pkg/api/pluginproxy/pluginproxy.go

@@ -1,7 +1,9 @@
 package pluginproxy
 
 import (
+	"bytes"
 	"encoding/json"
+	"io/ioutil"
 	"net/http"
 	"net/http/httputil"
 	"net/url"
@@ -22,7 +24,18 @@ type templateData struct {
 
 // NewApiPluginProxy create a plugin proxy
 func NewApiPluginProxy(ctx *models.ReqContext, proxyPath string, route *plugins.Route,
-	appID string, cfg *setting.Cfg, pluginSettingsService pluginsettings.Service, secretsService secrets.Service) *httputil.ReverseProxy {
+	appID string, cfg *setting.Cfg, pluginSettingsService pluginsettings.Service,
+	secretsService secrets.Service) *httputil.ReverseProxy {
+	appProxyLogger := logger.New(
+		"userId", ctx.UserId,
+		"orgId", ctx.OrgId,
+		"uname", ctx.Login,
+		"app", appID,
+		"path", ctx.Req.URL.Path,
+		"remote_addr", ctx.RemoteAddr(),
+		"referer", ctx.Req.Referer(),
+	)
+
 	director := func(req *http.Request) {
 		query := pluginsettings.GetByPluginIDArgs{OrgID: ctx.OrgId, PluginID: appID}
 		ps, err := pluginSettingsService.GetPluginSettingByPluginID(ctx.Req.Context(), &query)
@@ -61,8 +74,6 @@ func NewApiPluginProxy(ctx *models.ReqContext, proxyPath string, route *plugins.
 		req.Header.Del("Cookie")
 		req.Header.Del("Set-Cookie")
 
-		proxyutil.PrepareProxyRequest(req)
-
 		// Create a HTTP header with the context in it.
 		ctxJSON, err := json.Marshal(ctx.SignedInUser)
 		if err != nil {
@@ -80,15 +91,35 @@ func NewApiPluginProxy(ctx *models.ReqContext, proxyPath string, route *plugins.
 		}
 
 		if err := setBodyContent(req, route, data); err != nil {
-			logger.Error("Failed to set plugin route body content", "error", err)
+			appProxyLogger.Error("Failed to set plugin route body content", "error", err)
 		}
 	}
 
-	return &httputil.ReverseProxy{Director: director, ModifyResponse: modifyResponse}
+	logAppPluginProxyRequest(appID, cfg, ctx)
+
+	return proxyutil.NewReverseProxy(appProxyLogger, director)
 }
 
-func modifyResponse(resp *http.Response) error {
-	proxyutil.SetProxyResponseHeaders(resp.Header)
+func logAppPluginProxyRequest(appID string, cfg *setting.Cfg, c *models.ReqContext) {
+	if !cfg.DataProxyLogging {
+		return
+	}
 
-	return nil
+	var body string
+	if c.Req.Body != nil {
+		buffer, err := ioutil.ReadAll(c.Req.Body)
+		if err == nil {
+			c.Req.Body = ioutil.NopCloser(bytes.NewBuffer(buffer))
+			body = string(buffer)
+		}
+	}
+
+	logger.Info("Proxying incoming request",
+		"userid", c.UserId,
+		"orgid", c.OrgId,
+		"username", c.Login,
+		"app", appID,
+		"uri", c.Req.RequestURI,
+		"method", c.Req.Method,
+		"body", body)
 }

pkg/util/proxyutil/reverse_proxy.go

@@ -0,0 +1,146 @@
+package proxyutil
+
+import (
+	"context"
+	"errors"
+	"log"
+	"net/http"
+	"net/http/httputil"
+	"strings"
+	"time"
+
+	glog "github.com/grafana/grafana/pkg/infra/log"
+)
+
+// StatusClientClosedRequest A non-standard status code introduced by nginx
+// for the case when a client closes the connection while nginx is processing
+// the request.
+// https://httpstatus.in/499/
+const StatusClientClosedRequest = 499
+
+// ReverseProxyOption reverse proxy option to configure a httputil.ReverseProxy.
+type ReverseProxyOption func(*httputil.ReverseProxy)
+
+// NewReverseProxy creates a new httputil.ReverseProxy with sane default configuration.
+func NewReverseProxy(logger glog.Logger, director func(*http.Request), opts ...ReverseProxyOption) *httputil.ReverseProxy {
+	if logger == nil {
+		panic("logger cannot be nil")
+	}
+
+	if director == nil {
+		panic("director cannot be nil")
+	}
+
+	p := &httputil.ReverseProxy{
+		FlushInterval: time.Millisecond * 200,
+		ErrorHandler:  errorHandler(logger),
+		ErrorLog:      log.New(&logWrapper{logger: logger}, "", 0),
+		Director:      director,
+	}
+
+	for _, opt := range opts {
+		opt(p)
+	}
+
+	origDirector := p.Director
+	p.Director = wrapDirector(origDirector)
+
+	if p.ModifyResponse == nil {
+		// nolint:bodyclose
+		p.ModifyResponse = modifyResponse(logger)
+	} else {
+		modResponse := p.ModifyResponse
+		p.ModifyResponse = func(resp *http.Response) error {
+			if err := modResponse(resp); err != nil {
+				return err
+			}
+
+			// nolint:bodyclose
+			return modifyResponse(logger)(resp)
+		}
+	}
+
+	return p
+}
+
+// wrapDirector wraps a director and adds additional functionality.
+func wrapDirector(d func(*http.Request)) func(req *http.Request) {
+	return func(req *http.Request) {
+		d(req)
+		PrepareProxyRequest(req)
+
+		// Clear Origin and Referer to avoid CORS issues
+		req.Header.Del("Origin")
+		req.Header.Del("Referer")
+	}
+}
+
+// modifyResponse enforces certain constraints on http.Response.
+func modifyResponse(logger glog.Logger) func(resp *http.Response) error {
+	return func(resp *http.Response) error {
+		resp.Header.Del("Set-Cookie")
+		SetProxyResponseHeaders(resp.Header)
+		return nil
+	}
+}
+
+type timeoutError interface {
+	error
+	Timeout() bool
+}
+
+// errorHandler handles any errors happening while proxying a request and enforces
+// certain HTTP status based on the kind of error.
+// If client cancel/close the request we return 499 StatusClientClosedRequest.
+// If timeout happens while communicating with upstream server we return http.StatusGatewayTimeout.
+// If any other error we return http.StatusBadGateway.
+func errorHandler(logger glog.Logger) func(http.ResponseWriter, *http.Request, error) {
+	return func(w http.ResponseWriter, r *http.Request, err error) {
+		if errors.Is(err, context.Canceled) {
+			logger.Debug("Proxy request cancelled by client")
+			w.WriteHeader(StatusClientClosedRequest)
+			return
+		}
+
+		// nolint:errorlint
+		if timeoutErr, ok := err.(timeoutError); ok && timeoutErr.Timeout() {
+			logger.Error("Proxy request timed out", "err", err)
+			w.WriteHeader(http.StatusGatewayTimeout)
+			return
+		}
+
+		logger.Error("Proxy request failed", "err", err)
+		w.WriteHeader(http.StatusBadGateway)
+	}
+}
+
+type logWrapper struct {
+	logger glog.Logger
+}
+
+// Write writes log messages as bytes from proxy.
+func (lw *logWrapper) Write(p []byte) (n int, err error) {
+	withoutNewline := strings.TrimSuffix(string(p), "\n")
+	lw.logger.Error("Proxy request error", "error", withoutNewline)
+	return len(p), nil
+}
+
+func WithTransport(transport http.RoundTripper) ReverseProxyOption {
+	if transport == nil {
+		panic("transport cannot be nil")
+	}
+
+	return ReverseProxyOption(func(rp *httputil.ReverseProxy) {
+		rp.Transport = transport
+	})
+}
+
+func WithModifyResponse(fn func(*http.Response) error) ReverseProxyOption {
+	if fn == nil {
+		panic("fn cannot be nil")
+	}
+
+	return ReverseProxyOption(func(rp *httputil.ReverseProxy) {
+		rp.ModifyResponse = fn
+	})
+}

pkg/util/proxyutil/reverse_proxy_test.go

@@ -0,0 +1,157 @@
+package proxyutil
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/grafana/grafana/pkg/infra/log"
+	"github.com/stretchr/testify/require"
+)
+
+func TestReverseProxy(t *testing.T) {
+	t.Run("When proxying a request should enforce request and response constraints", func(t *testing.T) {
+		var actualReq *http.Request
+		upstream := newUpstreamServer(t, http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+			actualReq = req
+			http.SetCookie(w, &http.Cookie{Name: "test"})
+			w.WriteHeader(http.StatusOK)
+		}))
+		t.Cleanup(upstream.Close)
+		rec := httptest.NewRecorder()
+		req := httptest.NewRequest(http.MethodGet, upstream.URL, nil)
+		req.Header.Set("X-Forwarded-Host", "forwarded.host.com")
+		req.Header.Set("X-Forwarded-Port", "8080")
+		req.Header.Set("X-Forwarded-Proto", "https")
+		req.Header.Set("Origin", "test.com")
+		req.Header.Set("Referer", "https://test.com/api")
+		req.RemoteAddr = "10.0.0.1"
+
+		rp := NewReverseProxy(log.New("test"), func(req *http.Request) {
+			req.Header.Set("X-KEY", "value")
+		})
+		require.NotNil(t, rp)
+		require.NotNil(t, rp.ModifyResponse)
+		rp.ServeHTTP(rec, req)
+
+		require.NotNil(t, actualReq)
+		require.Empty(t, actualReq.Header.Get("X-Forwarded-Host"))
+		require.Empty(t, actualReq.Header.Get("X-Forwarded-Port"))
+		require.Empty(t, actualReq.Header.Get("X-Forwarded-Proto"))
+		require.Equal(t, "10.0.0.1", actualReq.Header.Get("X-Forwarded-For"))
+		require.Empty(t, actualReq.Header.Get("Origin"))
+		require.Empty(t, actualReq.Header.Get("Referer"))
+		require.Equal(t, "value", actualReq.Header.Get("X-KEY"))
+		resp := rec.Result()
+		require.Empty(t, resp.Cookies())
+		require.Equal(t, "sandbox", resp.Header.Get("Content-Security-Policy"))
+		require.NoError(t, resp.Body.Close())
+	})
+
+	t.Run("When proxying a request using WithModifyResponse should call it before default ModifyResponse func", func(t *testing.T) {
+		var actualReq *http.Request
+		upstream := newUpstreamServer(t, http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+			actualReq = req
+			http.SetCookie(w, &http.Cookie{Name: "test"})
+			w.WriteHeader(http.StatusOK)
+		}))
+		t.Cleanup(upstream.Close)
+		rec := httptest.NewRecorder()
+		req := httptest.NewRequest(http.MethodGet, upstream.URL, nil)
+		rp := NewReverseProxy(
+			log.New("test"),
+			func(req *http.Request) {
+				req.Header.Set("X-KEY", "value")
+			},
+			WithModifyResponse(func(r *http.Response) error {
+				r.Header.Set("X-KEY2", "value2")
+				return nil
+			}),
+		)
+		require.NotNil(t, rp)
+		require.NotNil(t, rp.ModifyResponse)
+		rp.ServeHTTP(rec, req)
+
+		require.NotNil(t, actualReq)
+		require.Equal(t, "value", actualReq.Header.Get("X-KEY"))
+		resp := rec.Result()
+		require.Empty(t, resp.Cookies())
+		require.Equal(t, "sandbox", resp.Header.Get("Content-Security-Policy"))
+		require.Equal(t, "value2", resp.Header.Get("X-KEY2"))
+		require.NoError(t, resp.Body.Close())
+	})
+
+	t.Run("Error handling should convert status codes depending on what kind of error it is", func(t *testing.T) {
+		timedOutTransport := http.DefaultTransport.(*http.Transport)
+		timedOutTransport.ResponseHeaderTimeout = time.Nanosecond
+
+		testCases := []struct {
+			desc               string
+			transport          http.RoundTripper
+			expectedStatusCode int
+		}{
+			{
+				desc:               "Cancelled request should return 499 Client closed request",
+				transport:          &cancelledRoundTripper{},
+				expectedStatusCode: StatusClientClosedRequest,
+			},
+			{
+				desc:               "Timed out request should return 504 Gateway timeout",
+				transport:          timedOutTransport,
+				expectedStatusCode: http.StatusGatewayTimeout,
+			},
+			{
+				desc:               "Failed request should return 502 Bad gateway",
+				transport:          &failingRoundTripper{},
+				expectedStatusCode: http.StatusBadGateway,
+			},
+		}
+
+		for _, tc := range testCases {
+			t.Run(tc.desc, func(t *testing.T) {
+				upstream := newUpstreamServer(t, http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+					w.WriteHeader(http.StatusOK)
+				}))
+				t.Cleanup(upstream.Close)
+				rec := httptest.NewRecorder()
+				req := httptest.NewRequest(http.MethodGet, upstream.URL, nil)
+
+				rp := NewReverseProxy(
+					log.New("test"),
+					func(req *http.Request) {},
+					WithTransport(tc.transport),
+				)
+				require.NotNil(t, rp)
+				require.NotNil(t, rp.Transport)
+				require.Same(t, tc.transport, rp.Transport)
+				rp.ServeHTTP(rec, req)
+
+				resp := rec.Result()
+				require.Equal(t, tc.expectedStatusCode, resp.StatusCode)
+				require.NoError(t, resp.Body.Close())
+			})
+		}
+	})
+}
+
+func newUpstreamServer(t *testing.T, handler http.Handler) *httptest.Server {
+	t.Helper()
+
+	upstream := httptest.NewServer(handler)
+	return upstream
+}
+
+type cancelledRoundTripper struct{}
+
+func (cancelledRoundTripper) RoundTrip(*http.Request) (*http.Response, error) {
+	return nil, context.Canceled
+}
+
+type failingRoundTripper struct{}
+
+func (failingRoundTripper) RoundTrip(*http.Request) (*http.Response, error) {
+	return nil, errors.New("some error")
+}