Configurable image deadline

- Use WithDeadline rather than WithTimeout
- Make deadline configurable

Fixes #713

cmd/fluxd/main.go

@@ -83,6 +83,8 @@ func main() {
 		registryPollInterval = fs.Duration("registry-poll-interval", 5*time.Minute, "period at which to poll registry for new images")
 		registryRPS          = fs.Int("registry-rps", 200, "maximum registry requests per second per host")
 		registryBurst        = fs.Int("registry-burst", defaultRemoteConnections, "maximum registry request burst per host (default matched to number of http worker goroutines)")
+		imageTimeout         = fs.Duration("registry-image-timeout", 10*time.Second, "maximum time allowed for all requests related to a single image")
+		registryTimeout      = fs.Duration("registry-timeout", 10*time.Second, "maximum time allowed for individual remote registry requests")
 		// k8s-secret backed ssh keyring configuration
 		k8sSecretName            = fs.String("k8s-secret-name", "flux-git-deploy", "Name of the k8s secret used to store the private SSH key")
 		k8sSecretVolumeMountPath = fs.String("k8s-secret-volume-mount-path", "/etc/fluxd/ssh", "Mount location of the k8s secret storing the private SSH key")
@@ -236,7 +238,7 @@ func main() {
 		remoteFactory := registry.NewRemoteClientFactory(registryLogger, registryMiddleware.RateLimiterConfig{
 			RPS:   *registryRPS,
 			Burst: *registryBurst,
-		})
+		}, *imageTimeout, *registryTimeout)
 
 		// Warmer
 		warmerLogger := log.NewContext(logger).With("component", "warmer")

registry/client_factory.go

@@ -28,16 +28,20 @@ type ClientFactory interface {
 
 // ---
 // A new ClientFactory for a Remote.
-func NewRemoteClientFactory(l log.Logger, rlc middleware.RateLimiterConfig) ClientFactory {
+func NewRemoteClientFactory(l log.Logger, rlc middleware.RateLimiterConfig, imgTimeout, rmtTimeout time.Duration) ClientFactory {
 	return &remoteClientFactory{
-		Logger: l,
-		rlConf: rlc,
+		Logger:        l,
+		rlConf:        rlc,
+		ImageTimeout:  imgTimeout,
+		RemoteTimeout: rmtTimeout,
 	}
 }
 
 type remoteClientFactory struct {
-	Logger log.Logger
-	rlConf middleware.RateLimiterConfig
+	Logger        log.Logger
+	rlConf        middleware.RateLimiterConfig
+	ImageTimeout  time.Duration // Max amount of time per image (all requests)
+	RemoteTimeout time.Duration // Max amount of time per HTTP request
 }
 
 func (f *remoteClientFactory) ClientFor(host string, creds Credentials) (Client, error) {
@@ -56,7 +60,8 @@ func (f *remoteClientFactory) ClientFor(host string, creds Credentials) (Client,
 	// A context we'll use to cancel requests on error
 	ctx, cancel := context.WithCancel(context.Background())
 	// Add a timeout to the request
-	ctx, cancel = context.WithTimeout(ctx, requestTimeout)
+	ctx, cancel = context.WithDeadline(ctx, time.Now().Add(f.ImageTimeout))
+	println(time.Now().String(), time.Now().Add(f.ImageTimeout).String())
 
 	// Use the wrapper to fix headers for quay.io, and remember bearer tokens
 	var transport http.RoundTripper
@@ -76,7 +81,7 @@ func (f *remoteClientFactory) ClientFor(host string, creds Credentials) (Client,
 			Client: &http.Client{
 				Transport: transport,
 				Jar:       jar,
-				Timeout:   requestTimeout,
+				Timeout:   f.RemoteTimeout,
 			},
 			Logf: dockerregistry.Quiet,
 		},

registry/integration_test.go

@@ -39,6 +39,8 @@ func TestWarming_WarmerWriteCacheRead(t *testing.T) {
 	remote := NewRemoteClientFactory(
 		logger.With("component", "client"),
 		middleware.RateLimiterConfig{200, 10},
+		10*time.Second,
+		10*time.Second,
 	)
 
 	cache := NewCacheClientFactory(

registry/registry_test.go

@@ -90,7 +90,7 @@ func TestRemoteFactory_RawClient(t *testing.T) {
 	fact := NewRemoteClientFactory(log.NewNopLogger(), middleware.RateLimiterConfig{
 		RPS:   200,
 		Burst: 1,
-	})
+	}, 10*time.Second, 10*time.Second)
 
 	// Refresh tags first
 	var tags []string
@@ -127,7 +127,7 @@ func TestRemoteFactory_RawClient(t *testing.T) {
 }
 
 func TestRemoteFactory_InvalidHost(t *testing.T) {
-	fact := NewRemoteClientFactory(log.NewNopLogger(), middleware.RateLimiterConfig{})
+	fact := NewRemoteClientFactory(log.NewNopLogger(), middleware.RateLimiterConfig{}, time.Second, time.Second)
 	invalidId, err := flux.ParseImageID("invalid.host/library/alpine:latest")
 	if err != nil {
 		t.Fatal(err)

registry/warming.go

@@ -4,7 +4,6 @@ package registry
 import (
 	"context"
 	"encoding/json"
-	"net"
 	"strings"
 	"sync"
 	"time"
@@ -141,10 +140,10 @@ func (w *Warmer) warm(id flux.ImageID, creds Credentials) {
 			// Get the image from the remote
 			img, err := client.Manifest(imageID)
 			if err != nil {
-				if err, ok := errors.Cause(err).(net.Error); ok && err.Timeout() {
-					// This was due to a context timeout, don't bother logging
-					return
-				}
+				//if err, ok := errors.Cause(err).(net.Error); ok && err.Timeout() {
+				//	// This was due to a context timeout, don't bother logging
+				//	return
+				//}
 				w.Logger.Log("err", errors.Wrap(err, "requesting manifests"))
 				return
 			}