Merge pull request #2184 from git-lfs/streaming-log-stats

Streaming log stats

commands/run.go

@@ -66,7 +66,7 @@ func Run() {
 	}
 
 	root.Execute()
-	logHTTPStats(getAPIClient())
+	getAPIClient().Close()
 }
 
 func gitlfsCommand(cmd *cobra.Command, args []string) {
@@ -79,10 +79,12 @@ func gitlfsCommand(cmd *cobra.Command, args []string) {
 // will resolve the localstorage directories.
 func resolveLocalStorage(cmd *cobra.Command, args []string) {
 	localstorage.ResolveDirs()
+	setupHTTPLogger(getAPIClient())
 }
 
 func setupLocalStorage(cmd *cobra.Command, args []string) {
 	config.ResolveGitBasicDirs()
+	setupHTTPLogger(getAPIClient())
 }
 
 func helpCommand(cmd *cobra.Command, args []string) {
@@ -106,27 +108,22 @@ func printHelp(commandName string) {
 	}
 }
 
-func logHTTPStats(c *lfsapi.Client) {
-	if !c.LoggingStats {
+func setupHTTPLogger(c *lfsapi.Client) {
+	if c == nil || len(os.Getenv("GIT_LOG_STATS")) < 1 {
 		return
 	}
 
-	file, err := statsLogFile()
-	if err != nil {
-		fmt.Fprintf(os.Stderr, "Error logging http stats: %s\n", err)
-		return
-	}
-
-	defer file.Close()
-	c.LogStats(file)
-}
-
-func statsLogFile() (*os.File, error) {
 	logBase := filepath.Join(config.LocalLogDir, "http")
 	if err := os.MkdirAll(logBase, 0755); err != nil {
-		return nil, err
+		fmt.Fprintf(os.Stderr, "Error logging http stats: %s\n", err)
+		return
 	}
 
 	logFile := fmt.Sprintf("http-%d.log", time.Now().Unix())
-	return os.Create(filepath.Join(logBase, logFile))
+	file, err := os.Create(filepath.Join(logBase, logFile))
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Error logging http stats: %s\n", err)
+	} else {
+		c.LogHTTPStats(file)
+	}
 }

docs/man/git-lfs-config.5.ronn

@@ -31,15 +31,21 @@ be scoped inside the configuration for a remote.
 
 * `lfs.dialtimeout`
 
-  Sets the maximum time, in seconds, that the HTTP client will wait initiate a
-  connection. This does not include the time to send a request and wait for a
+  Sets the maximum time, in seconds, that the HTTP client will wait to initiate
+  a connection. This does not include the time to send a request and wait for a
   response. Default: 30 seconds
 
 * `lfs.tlstimeout`
 
   Sets the maximum time, in seconds, that the HTTP client will wait for a TLS
   handshake. Default: 30 seconds.
 
+* `lfs.activitytimeout` / `lfs.https://<host>.activitytimeout`
+
+  Sets the maximum time, in seconds, that the HTTP client will wait for the
+  next tcp read or write. If < 1, no activity timeout is used at all.
+  Default: 10 seconds
+
 * `lfs.keepalive`
 
   Sets the maximum time, in seconds, for the HTTP client to maintain keepalive

lfsapi/client.go

@@ -1,11 +1,14 @@
 package lfsapi
 
 import (
+	"context"
 	"crypto/tls"
+	"fmt"
 	"net"
 	"net/http"
 	"net/url"
 	"os"
+	"strconv"
 	"strings"
 	"time"
 
@@ -77,6 +80,11 @@ func (c *Client) Do(req *http.Request) (*http.Response, error) {
 	return res, c.handleResponse(res)
 }
 
+// Close closes any resources that this client opened.
+func (c *Client) Close() error {
+	return c.httpLogger.Close()
+}
+
 func (c *Client) extraHeadersFor(req *http.Request) http.Header {
 	copy := make(http.Header, len(req.Header))
 	for k, vs := range req.Header {
@@ -109,19 +117,18 @@ func (c *Client) extraHeaders(u *url.URL) map[string][]string {
 }
 
 func (c *Client) doWithRedirects(cli *http.Client, req *http.Request, via []*http.Request) (*http.Response, error) {
-	c.traceRequest(req)
-	if err := c.prepareRequestBody(req); err != nil {
+	tracedReq, err := c.traceRequest(req)
+	if err != nil {
 		return nil, err
 	}
 
-	start := time.Now()
 	res, err := cli.Do(req)
 	if err != nil {
+		c.traceResponse(req, tracedReq, nil)
 		return res, err
 	}
 
-	c.traceResponse(res)
-	c.startResponseStats(res, start)
+	c.traceResponse(req, tracedReq, res)
 
 	if res.StatusCode != 307 {
 		return res, err
@@ -188,15 +195,43 @@ func (c *Client) httpClient(host string) *http.Client {
 	}
 
 	tr := &http.Transport{
-		Proxy: proxyFromClient(c),
-		Dial: (&net.Dialer{
-			Timeout:   time.Duration(dialtime) * time.Second,
-			KeepAlive: time.Duration(keepalivetime) * time.Second,
-		}).Dial,
+		Proxy:               proxyFromClient(c),
 		TLSHandshakeTimeout: time.Duration(tlstime) * time.Second,
 		MaxIdleConnsPerHost: concurrentTransfers,
 	}
 
+	activityTimeout := 10
+	if v, ok := c.uc.Get("lfs", fmt.Sprintf("https://%v", host), "activitytimeout"); ok {
+		if i, err := strconv.Atoi(v); err == nil {
+			activityTimeout = i
+		} else {
+			activityTimeout = 0
+		}
+	}
+
+	dialer := &net.Dialer{
+		Timeout:   time.Duration(dialtime) * time.Second,
+		KeepAlive: time.Duration(keepalivetime) * time.Second,
+		DualStack: true,
+	}
+
+	if activityTimeout > 0 {
+		activityDuration := time.Duration(activityTimeout) * time.Second
+		tr.DialContext = func(ctx context.Context, network, addr string) (net.Conn, error) {
+			c, err := dialer.DialContext(ctx, network, addr)
+			if c == nil {
+				return c, err
+			}
+			if tc, ok := c.(*net.TCPConn); ok {
+				tc.SetKeepAlive(true)
+				tc.SetKeepAlivePeriod(dialer.KeepAlive)
+			}
+			return &deadlineConn{Timeout: activityDuration, Conn: c}, err
+		}
+	} else {
+		tr.DialContext = dialer.DialContext
+	}
+
 	tr.TLSClientConfig = &tls.Config{}
 
 	if isClientCertEnabledForHost(c, host) {
@@ -256,6 +291,26 @@ func newRequestForRetry(req *http.Request, location string) (*http.Request, erro
 	return newReq, nil
 }
 
+type deadlineConn struct {
+	Timeout time.Duration
+	net.Conn
+}
+
+func (c *deadlineConn) Read(b []byte) (int, error) {
+	if err := c.Conn.SetDeadline(time.Now().Add(c.Timeout)); err != nil {
+		return 0, err
+	}
+	return c.Conn.Read(b)
+}
+
+func (c *deadlineConn) Write(b []byte) (int, error) {
+	if err := c.Conn.SetDeadline(time.Now().Add(c.Timeout)); err != nil {
+		return 0, err
+	}
+
+	return c.Conn.Write(b)
+}
+
 func init() {
 	UserAgent = config.VersionDesc
 }

lfsapi/lfsapi.go

@@ -37,7 +37,6 @@ type Client struct {
 
 	Verbose          bool
 	DebuggingVerbose bool
-	LoggingStats     bool
 	VerboseOut       io.Writer
 
 	hostClients map[string]*http.Client
@@ -46,10 +45,9 @@ type Client struct {
 	ntlmSessions map[string]ntlm.ClientSession
 	ntlmMu       sync.Mutex
 
-	transferBuckets  map[string][]*http.Response
-	transferBucketMu sync.Mutex
-	transfers        map[*http.Response]*httpTransfer
-	transferMu       sync.Mutex
+	httpLogger *syncLogger
+
+	LoggingStats bool // DEPRECATED
 
 	// only used for per-host ssl certs
 	gitEnv Env
@@ -95,7 +93,6 @@ func NewClient(osEnv Env, gitEnv Env) (*Client, error) {
 		SkipSSLVerify:       !gitEnv.Bool("http.sslverify", true) || osEnv.Bool("GIT_SSL_NO_VERIFY", false),
 		Verbose:             osEnv.Bool("GIT_CURL_VERBOSE", false),
 		DebuggingVerbose:    osEnv.Bool("LFS_DEBUG_HTTP", false),
-		LoggingStats:        osEnv.Bool("GIT_LOG_STATS", false),
 		HTTPSProxy:          httpsProxy,
 		HTTPProxy:           httpProxy,
 		NoProxy:             noProxy,