Repository: weaveworks/flux
Pull request #2171 (2019-06-19 20:35:19)

Implement configurable timeout for RPC operations
This assures that in a scenario where for example the Kubernetes API
is temporary rate limited, requests that are taking too long get
abandoned, rather than piling up.

As we have observed excessive amount of goroutines getting spawned
in a short amount of time, sometimes resulting in a shutdown of the
Flux daemon due to the spike in CPU (and RAM) that comes with it.

squaremo: Adding a context to the RPCs is :100:

I think this will need to go further though. The problem being addressed (in large part) is that Kubernetes API requests pile up, once they start needing to wait for rate limiting. To be able to abandon those when a request times out, the context must be propagated to those calls (i.e., to the cluster.Cluster methods, then to the Kubernetes client methods).

Sadly, the client-go API doesn't support contexts. What I suggest is, for the meantime, is that a context is passed through, and checked before each Kubernetes API call (or anything that occurs in a loop, at least).

Patch:
diff --git a/cluster/cluster.go b/cluster/cluster.go
index e24883eb..5fedba27 100644
--- a/cluster/cluster.go
+++ b/cluster/cluster.go
@@ -1,6 +1,7 @@
 package cluster
 
 import (
+	"context"
 	"errors"
 
 	"github.com/weaveworks/flux"
@@ -25,11 +26,11 @@ const (
 // are distinct interfaces.
 type Cluster interface {
 	// Get all of the services (optionally, from a specific namespace), excluding those
-	AllWorkloads(maybeNamespace string) ([]Workload, error)
-	SomeWorkloads([]flux.ResourceID) ([]Workload, error)
+	AllWorkloads(ctx context.Context, maybeNamespace string) ([]Workload, error)
+	SomeWorkloads(ctx context.Context, ids []flux.ResourceID) ([]Workload, error)
 	IsAllowedResource(flux.ResourceID) bool
 	Ping() error
-	Export() ([]byte, error)
+	Export(ctx context.Context) ([]byte, error)
 	Sync(SyncSet) error
 	PublicSSHKey(regenerate bool) (ssh.PublicKey, error)
 }
diff --git a/cluster/kubernetes/images.go b/cluster/kubernetes/images.go
index dccf3068..e4fdcc91 100644
--- a/cluster/kubernetes/images.go
+++ b/cluster/kubernetes/images.go
@@ -1,6 +1,7 @@
 package kubernetes
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/go-kit/kit/log"
@@ -122,8 +123,9 @@ func mergeCredentials(log func(...interface{}) error,
 // ImagesToFetch is a k8s specific method to get a list of images to update along with their credentials
 func (c *Cluster) ImagesToFetch() registry.ImageCreds {
 	allImageCreds := make(registry.ImageCreds)
+	ctx := context.Background()
 
-	namespaces, err := c.getAllowedAndExistingNamespaces()
+	namespaces, err := c.getAllowedAndExistingNamespaces(ctx)
 	if err != nil {
 		c.logger.Log("err", errors.Wrap(err, "getting namespaces"))
 		return allImageCreds
@@ -132,7 +134,7 @@ func (c *Cluster) ImagesToFetch() registry.ImageCreds {
 	for _, ns := range namespaces {
 		seenCreds := make(map[string]registry.Credentials)
 		for kind, resourceKind := range resourceKinds {
-			workloads, err := resourceKind.getWorkloads(c, ns.Name)
+			workloads, err := resourceKind.getWorkloads(ctx, c, ns.Name)
 			if err != nil {
 				if apierrors.IsNotFound(err) || apierrors.IsForbidden(err) {
 					// Skip unsupported or forbidden resource kinds
diff --git a/cluster/kubernetes/kubernetes.go b/cluster/kubernetes/kubernetes.go
index 5de47558..188f6b0d 100644
--- a/cluster/kubernetes/kubernetes.go
+++ b/cluster/kubernetes/kubernetes.go
@@ -2,6 +2,7 @@ package kubernetes
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"fmt"
 	"sync"
@@ -127,7 +128,7 @@ func NewCluster(client ExtendedClient, applier Applier, sshKeyRing ssh.KeyRing,
 // SomeWorkloads returns the workloads named, missing out any that don't
 // exist in the cluster or aren't in an allowed namespace.
 // They do not necessarily have to be returned in the order requested.
-func (c *Cluster) SomeWorkloads(ids []flux.ResourceID) (res []cluster.Workload, err error) {
+func (c *Cluster) SomeWorkloads(ctx context.Context, ids []flux.ResourceID) (res []cluster.Workload, err error) {
 	var workloads []cluster.Workload
 	for _, id := range ids {
 		if !c.IsAllowedResource(id) {
@@ -141,7 +142,7 @@ func (c *Cluster) SomeWorkloads(ids []flux.ResourceID) (res []cluster.Workload,
 			continue
 		}
 
-		workload, err := resourceKind.getWorkload(c, ns, name)
+		workload, err := resourceKind.getWorkload(ctx, c, ns, name)
 		if err != nil {
 			if apierrors.IsForbidden(err) || apierrors.IsNotFound(err) {
 				continue
@@ -161,8 +162,8 @@ func (c *Cluster) SomeWorkloads(ids []flux.ResourceID) (res []cluster.Workload,
 
 // AllWorkloads returns all workloads in allowed namespaces matching the criteria; that is, in
 // the namespace (or any namespace if that argument is empty)
-func (c *Cluster) AllWorkloads(namespace string) (res []cluster.Workload, err error) {
-	namespaces, err := c.getAllowedAndExistingNamespaces()
+func (c *Cluster) AllWorkloads(ctx context.Context, namespace string) (res []cluster.Workload, err error) {
+	namespaces, err := c.getAllowedAndExistingNamespaces(ctx)
 	if err != nil {
 		return nil, errors.Wrap(err, "getting namespaces")
 	}
@@ -174,7 +175,7 @@ func (c *Cluster) AllWorkloads(namespace string) (res []cluster.Workload, err er
 		}
 
 		for kind, resourceKind := range resourceKinds {
-			workloads, err := resourceKind.getWorkloads(c, ns.Name)
+			workloads, err := resourceKind.getWorkloads(ctx, c, ns.Name)
 			if err != nil {
 				switch {
 				case apierrors.IsNotFound(err):
@@ -219,10 +220,10 @@ func (c *Cluster) Ping() error {
 }
 
 // Export exports cluster resources
-func (c *Cluster) Export() ([]byte, error) {
+func (c *Cluster) Export(ctx context.Context) ([]byte, error) {
 	var config bytes.Buffer
 
-	namespaces, err := c.getAllowedAndExistingNamespaces()
+	namespaces, err := c.getAllowedAndExistingNamespaces(ctx)
 	if err != nil {
 		return nil, errors.Wrap(err, "getting namespaces")
 	}
@@ -240,7 +241,7 @@ func (c *Cluster) Export() ([]byte, error) {
 		}
 
 		for _, resourceKind := range resourceKinds {
-			workloads, err := resourceKind.getWorkloads(c, ns.Name)
+			workloads, err := resourceKind.getWorkloads(ctx, c, ns.Name)
 			if err != nil {
 				switch {
 				case apierrors.IsNotFound(err):
@@ -281,10 +282,13 @@ func (c *Cluster) PublicSSHKey(regenerate bool) (ssh.PublicKey, error) {
 // the Flux instance is expected to have access to and can look for resources inside of.
 // It returns a list of all namespaces unless an explicit list of allowed namespaces
 // has been set on the Cluster instance.
-func (c *Cluster) getAllowedAndExistingNamespaces() ([]apiv1.Namespace, error) {
+func (c *Cluster) getAllowedAndExistingNamespaces(ctx context.Context) ([]apiv1.Namespace, error) {
 	if len(c.allowedNamespaces) > 0 {
 		nsList := []apiv1.Namespace{}
 		for _, name := range c.allowedNamespaces {
+			if err := ctx.Err(); err != nil {
+				return nil, err
+			}
 			ns, err := c.client.CoreV1().Namespaces().Get(name, meta_v1.GetOptions{})
 			switch {
 			case err == nil:
@@ -303,6 +307,9 @@ func (c *Cluster) getAllowedAndExistingNamespaces() ([]apiv1.Namespace, error) {
 		return nsList, nil
 	}
 
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
 	namespaces, err := c.client.CoreV1().Namespaces().List(meta_v1.ListOptions{})
 	if err != nil {
 		return nil, err
diff --git a/cluster/kubernetes/resourcekinds.go b/cluster/kubernetes/resourcekinds.go
index dab1bd20..a7416c64 100644
--- a/cluster/kubernetes/resourcekinds.go
+++ b/cluster/kubernetes/resourcekinds.go
@@ -1,6 +1,7 @@
 package kubernetes
 
 import (
+	"context"
 	"strings"
 
 	apiapps "k8s.io/api/apps/v1"
@@ -30,8 +31,8 @@ const AntecedentAnnotation = "flux.weave.works/antecedent"
 // Kind registry
 
 type resourceKind interface {
-	getWorkload(c *Cluster, namespace, name string) (workload, error)
-	getWorkloads(c *Cluster, namespace string) ([]workload, error)
+	getWorkload(ctx context.Context, c *Cluster, namespace, name string) (workload, error)
+	getWorkloads(ctx context.Context, c *Cluster, namespace string) ([]workload, error)
 }
 
 var (
@@ -114,7 +115,10 @@ func (w workload) toClusterWorkload(resourceID flux.ResourceID) cluster.Workload
 
 type deploymentKind struct{}
 
-func (dk *deploymentKind) getWorkload(c *Cluster, namespace, name string) (workload, error) {
+func (dk *deploymentKind) getWorkload(ctx context.Context, c *Cluster, namespace, name string) (workload, error) {
+	if err := ctx.Err(); err != nil {
+		return workload{}, err
+	}
 	deployment, err := c.client.AppsV1().Deployments(namespace).Get(name, meta_v1.GetOptions{})
 	if err != nil {
 		return workload{}, err
@@ -123,7 +127,10 @@ func (dk *deploymentKind) getWorkload(c *Cluster, namespace, name string) (workl
 	return makeDeploymentWorkload(deployment), nil
 }
 
-func (dk *deploymentKind) getWorkloads(c *Cluster, namespace string) ([]workload, error) {
+func (dk *deploymentKind) getWorkloads(ctx context.Context, c *Cluster, namespace string) ([]workload, error) {
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
 	deployments, err := c.client.AppsV1().Deployments(namespace).List(meta_v1.ListOptions{})
 	if err != nil {
 		return nil, err
@@ -191,7 +198,10 @@ func makeDeploymentWorkload(deployment *apiapps.Deployment) workload {
 
 type daemonSetKind struct{}
 
-func (dk *daemonSetKind) getWorkload(c *Cluster, namespace, name string) (workload, error) {
+func (dk *daemonSetKind) getWorkload(ctx context.Context, c *Cluster, namespace, name string) (workload, error) {
+	if err := ctx.Err(); err != nil {
+		return workload{}, err
+	}
 	daemonSet, err := c.client.AppsV1().DaemonSets(namespace).Get(name, meta_v1.GetOptions{})
 	if err != nil {
 		return workload{}, err
@@ -200,7 +210,10 @@ func (dk *daemonSetKind) getWorkload(c *Cluster, namespace, name string) (worklo
 	return makeDaemonSetWorkload(daemonSet), nil
 }
 
-func (dk *daemonSetKind) getWorkloads(c *Cluster, namespace string) ([]workload, error) {
+func (dk *daemonSetKind) getWorkloads(ctx context.Context, c *Cluster, namespace string) ([]workload, error) {
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
 	daemonSets, err := c.client.AppsV1().DaemonSets(namespace).List(meta_v1.ListOptions{})
 	if err != nil {
 		return nil, err
@@ -252,7 +265,10 @@ func makeDaemonSetWorkload(daemonSet *apiapps.DaemonSet) workload {
 
 type statefulSetKind struct{}
 
-func (dk *statefulSetKind) getWorkload(c *Cluster, namespace, name string) (workload, error) {
+func (dk *statefulSetKind) getWorkload(ctx context.Context, c *Cluster, namespace, name string) (workload, error) {
+	if err := ctx.Err(); err != nil {
+		return workload{}, err
+	}
 	statefulSet, err := c.client.AppsV1().StatefulSets(namespace).Get(name, meta_v1.GetOptions{})
 	if err != nil {
 		return workload{}, err
@@ -261,7 +277,10 @@ func (dk *statefulSetKind) getWorkload(c *Cluster, namespace, name string) (work
 	return makeStatefulSetWorkload(statefulSet), nil
 }
 
-func (dk *statefulSetKind) getWorkloads(c *Cluster, namespace string) ([]workload, error) {
+func (dk *statefulSetKind) getWorkloads(ctx context.Context, c *Cluster, namespace string) ([]workload, error) {
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
 	statefulSets, err := c.client.AppsV1().StatefulSets(namespace).List(meta_v1.ListOptions{})
 	if err != nil {
 		return nil, err
@@ -345,7 +364,10 @@ func makeStatefulSetWorkload(statefulSet *apiapps.StatefulSet) workload {
 
 type cronJobKind struct{}
 
-func (dk *cronJobKind) getWorkload(c *Cluster, namespace, name string) (workload, error) {
+func (dk *cronJobKind) getWorkload(ctx context.Context, c *Cluster, namespace, name string) (workload, error) {
+	if err := ctx.Err(); err != nil {
+		return workload{}, err
+	}
 	cronJob, err := c.client.BatchV1beta1().CronJobs(namespace).Get(name, meta_v1.GetOptions{})
 	if err != nil {
 		return workload{}, err
@@ -354,7 +376,10 @@ func (dk *cronJobKind) getWorkload(c *Cluster, namespace, name string) (workload
 	return makeCronJobWorkload(cronJob), nil
 }
 
-func (dk *cronJobKind) getWorkloads(c *Cluster, namespace string) ([]workload, error) {
+func (dk *cronJobKind) getWorkloads(ctx context.Context, c *Cluster, namespace string) ([]workload, error) {
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
 	cronJobs, err := c.client.BatchV1beta1().CronJobs(namespace).List(meta_v1.ListOptions{})
 	if err != nil {
 		return nil, err
@@ -382,7 +407,10 @@ func makeCronJobWorkload(cronJob *apibatch.CronJob) workload {
 
 type fluxHelmReleaseKind struct{}
 
-func (fhr *fluxHelmReleaseKind) getWorkload(c *Cluster, namespace, name string) (workload, error) {
+func (fhr *fluxHelmReleaseKind) getWorkload(ctx context.Context, c *Cluster, namespace, name string) (workload, error) {
+	if err := ctx.Err(); err != nil {
+		return workload{}, err
+	}
 	fluxHelmRelease, err := c.client.HelmV1alpha2().FluxHelmReleases(namespace).Get(name, meta_v1.GetOptions{})
 	if err != nil {
 		return workload{}, err
@@ -390,7 +418,10 @@ func (fhr *fluxHelmReleaseKind) getWorkload(c *Cluster, namespace, name string)
 	return makeFluxHelmReleaseWorkload(fluxHelmRelease), nil
 }
 
-func (fhr *fluxHelmReleaseKind) getWorkloads(c *Cluster, namespace string) ([]workload, error) {
+func (fhr *fluxHelmReleaseKind) getWorkloads(ctx context.Context, c *Cluster, namespace string) ([]workload, error) {
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
 	fluxHelmReleases, err := c.client.HelmV1alpha2().FluxHelmReleases(namespace).List(meta_v1.ListOptions{})
 	if err != nil {
 		return nil, err
@@ -444,7 +475,10 @@ func createK8sFHRContainers(values map[string]interface{}) []apiv1.Container {
 
 type helmReleaseKind struct{}
 
-func (hr *helmReleaseKind) getWorkload(c *Cluster, namespace, name string) (workload, error) {
+func (hr *helmReleaseKind) getWorkload(ctx context.Context, c *Cluster, namespace, name string) (workload, error) {
+	if err := ctx.Err(); err != nil {
+		return workload{}, err
+	}
 	helmRelease, err := c.client.FluxV1beta1().HelmReleases(namespace).Get(name, meta_v1.GetOptions{})
 	if err != nil {
 		return workload{}, err
@@ -452,7 +486,10 @@ func (hr *helmReleaseKind) getWorkload(c *Cluster, namespace, name string) (work
 	return makeHelmReleaseWorkload(helmRelease), nil
 }
 
-func (hr *helmReleaseKind) getWorkloads(c *Cluster, namespace string) ([]workload, error) {
+func (hr *helmReleaseKind) getWorkloads(ctx context.Context, c *Cluster, namespace string) ([]workload, error) {
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
 	helmReleases, err := c.client.FluxV1beta1().HelmReleases(namespace).List(meta_v1.ListOptions{})
 	if err != nil {
 		return nil, err
diff --git a/cluster/kubernetes/sync.go b/cluster/kubernetes/sync.go
index acc858be..502de85f 100644
--- a/cluster/kubernetes/sync.go
+++ b/cluster/kubernetes/sync.go
@@ -2,6 +2,7 @@ package kubernetes
 
 import (
 	"bytes"
+	"context"
 	"crypto/sha1"
 	"crypto/sha256"
 	"encoding/base64"
@@ -292,7 +293,7 @@ func (c *Cluster) listAllowedResources(
 	}
 
 	// List resources only from the allowed namespaces
-	namespaces, err := c.getAllowedAndExistingNamespaces()
+	namespaces, err := c.getAllowedAndExistingNamespaces(context.Background())
 	if err != nil {
 		return nil, err
 	}
diff --git a/cluster/mock/mock.go b/cluster/mock/mock.go
index ae6c47be..a16633fd 100644
--- a/cluster/mock/mock.go
+++ b/cluster/mock/mock.go
@@ -2,6 +2,7 @@ package mock
 
 import (
 	"bytes"
+	"context"
 
 	"github.com/weaveworks/flux"
 	"github.com/weaveworks/flux/cluster"
@@ -14,11 +15,11 @@ import (
 
 // Doubles as a cluster.Cluster and cluster.Manifests implementation
 type Mock struct {
-	AllWorkloadsFunc              func(maybeNamespace string) ([]cluster.Workload, error)
-	SomeWorkloadsFunc             func([]flux.ResourceID) ([]cluster.Workload, error)
+	AllWorkloadsFunc              func(ctx context.Context, maybeNamespace string) ([]cluster.Workload, error)
+	SomeWorkloadsFunc             func(ctx context.Context, ids []flux.ResourceID) ([]cluster.Workload, error)
 	IsAllowedResourceFunc         func(flux.ResourceID) bool
 	PingFunc                      func() error
-	ExportFunc                    func() ([]byte, error)
+	ExportFunc                    func(ctx context.Context) ([]byte, error)
 	SyncFunc                      func(cluster.SyncSet) error
 	PublicSSHKeyFunc              func(regenerate bool) (ssh.PublicKey, error)
 	SetWorkloadContainerImageFunc func(def []byte, id flux.ResourceID, container string, newImageID image.Ref) ([]byte, error)
@@ -33,12 +34,12 @@ type Mock struct {
 var _ cluster.Cluster = &Mock{}
 var _ manifests.Manifests = &Mock{}
 
-func (m *Mock) AllWorkloads(maybeNamespace string) ([]cluster.Workload, error) {
-	return m.AllWorkloadsFunc(maybeNamespace)
+func (m *Mock) AllWorkloads(ctx context.Context, maybeNamespace string) ([]cluster.Workload, error) {
+	return m.AllWorkloadsFunc(ctx, maybeNamespace)
 }
 
-func (m *Mock) SomeWorkloads(s []flux.ResourceID) ([]cluster.Workload, error) {
-	return m.SomeWorkloadsFunc(s)
+func (m *Mock) SomeWorkloads(ctx context.Context, ids []flux.ResourceID) ([]cluster.Workload, error) {
+	return m.SomeWorkloadsFunc(ctx, ids)
 }
 
 func (m *Mock) IsAllowedResource(id flux.ResourceID) bool {
@@ -49,8 +50,8 @@ func (m *Mock) Ping() error {
 	return m.PingFunc()
 }
 
-func (m *Mock) Export() ([]byte, error) {
-	return m.ExportFunc()
+func (m *Mock) Export(ctx context.Context) ([]byte, error) {
+	return m.ExportFunc(ctx)
 }
 
 func (m *Mock) Sync(c cluster.SyncSet) error {
diff --git a/cmd/fluxd/main.go b/cmd/fluxd/main.go
index 1e4dd210..75979707 100644
--- a/cmd/fluxd/main.go
+++ b/cmd/fluxd/main.go
@@ -169,8 +169,10 @@ func main() {
 		// manifest generation
 		manifestGeneration = fs.Bool("manifest-generation", false, "experimental; search for .flux.yaml files to generate manifests")
 
+		// upstream connection settings
 		upstreamURL = fs.String("connect", "", "connect to an upstream service e.g., Weave Cloud, at this base address")
 		token       = fs.String("token", "", "authentication token for upstream service")
+		rpcTimeout  = fs.Duration("rpc-timeout", 10*time.Second, "maximum time an operation requested by the upstream may take")
 
 		dockerConfig = fs.String("docker-config", "", "path to a docker config to use for image registry credentials")
 
@@ -601,6 +603,7 @@ func main() {
 				transport.NewUpstreamRouter(),
 				*upstreamURL,
 				remote.NewErrorLoggingUpstreamServer(daemon, upstreamLogger),
+				*rpcTimeout,
 				upstreamLogger,
 			)
 			if err != nil {
diff --git a/daemon/daemon.go b/daemon/daemon.go
index f7be8b00..eee5991b 100644
--- a/daemon/daemon.go
+++ b/daemon/daemon.go
@@ -71,7 +71,7 @@ func (d *Daemon) Ping(ctx context.Context) error {
 }
 
 func (d *Daemon) Export(ctx context.Context) ([]byte, error) {
-	return d.Cluster.Export()
+	return d.Cluster.Export(ctx)
 }
 
 func (d *Daemon) getManifestStore(checkout *git.Checkout) (manifests.Store, error) {
@@ -122,9 +122,9 @@ func (d *Daemon) ListServicesWithOptions(ctx context.Context, opts v11.ListServi
 	var clusterWorkloads []cluster.Workload
 	var err error
 	if len(opts.Services) > 0 {
-		clusterWorkloads, err = d.Cluster.SomeWorkloads(opts.Services)
+		clusterWorkloads, err = d.Cluster.SomeWorkloads(ctx, opts.Services)
 	} else {
-		clusterWorkloads, err = d.Cluster.AllWorkloads(opts.Namespace)
+		clusterWorkloads, err = d.Cluster.AllWorkloads(ctx, opts.Namespace)
 	}
 	if err != nil {
 		return nil, errors.Wrap(err, "getting workloads from cluster")
@@ -199,12 +199,12 @@ func (d *Daemon) ListImagesWithOptions(ctx context.Context, opts v10.ListImagesO
 		if err != nil {
 			return nil, errors.Wrap(err, "treating workload spec as ID")
 		}
-		workloads, err = d.Cluster.SomeWorkloads([]flux.ResourceID{id})
+		workloads, err = d.Cluster.SomeWorkloads(ctx, []flux.ResourceID{id})
 		if err != nil {
 			return nil, errors.Wrap(err, "getting some workloads")
 		}
 	} else {
-		workloads, err = d.Cluster.AllWorkloads(opts.Namespace)
+		workloads, err = d.Cluster.AllWorkloads(ctx, opts.Namespace)
 		if err != nil {
 			return nil, errors.Wrap(err, "getting all workloads")
 		}
diff --git a/daemon/images.go b/daemon/images.go
index 6fca631d..6404b825 100644
--- a/daemon/images.go
+++ b/daemon/images.go
@@ -29,7 +29,7 @@ func (d *Daemon) pollForNewImages(logger log.Logger) {
 		return
 	}
 	// Find images to check
-	workloads, err := d.Cluster.SomeWorkloads(candidateWorkloads.IDs())
+	workloads, err := d.Cluster.SomeWorkloads(ctx, candidateWorkloads.IDs())
 	if err != nil {
 		logger.Log("error", errors.Wrap(err, "checking workloads for new images"))
 		return
diff --git a/http/daemon/upstream.go b/http/daemon/upstream.go
index 53458f63..cdea3e43 100644
--- a/http/daemon/upstream.go
+++ b/http/daemon/upstream.go
@@ -30,6 +30,7 @@ type Upstream struct {
 	endpoint  string
 	apiClient *fluxclient.Client
 	server    api.UpstreamServer
+	timeout   time.Duration
 	logger    log.Logger
 	quit      chan struct{}
 
@@ -46,7 +47,7 @@ var (
 	}, []string{"target"})
 )
 
-func NewUpstream(client *http.Client, ua string, t fluxclient.Token, router *mux.Router, endpoint string, s api.UpstreamServer, logger log.Logger) (*Upstream, error) {
+func NewUpstream(client *http.Client, ua string, t fluxclient.Token, router *mux.Router, endpoint string, s api.UpstreamServer, timeout time.Duration, logger log.Logger) (*Upstream, error) {
 	httpEndpoint, wsEndpoint, err := inferEndpoints(endpoint)
 	if err != nil {
 		return nil, errors.Wrap(err, "inferring WS/HTTP endpoints")
@@ -65,6 +66,7 @@ func NewUpstream(client *http.Client, ua string, t fluxclient.Token, router *mux
 		endpoint:  wsEndpoint,
 		apiClient: fluxclient.New(client, router, httpEndpoint, t),
 		server:    s,
+		timeout:   timeout,
 		logger:    logger,
 		quit:      make(chan struct{}),
 	}
@@ -161,7 +163,7 @@ func (a *Upstream) connect() error {
 
 	// Hook up the rpc server. We are a websocket _client_, but an RPC
 	// _server_.
-	rpcserver, err := rpc.NewServer(a.server)
+	rpcserver, err := rpc.NewServer(a.server, a.timeout)
 	if err != nil {
 		return errors.Wrap(err, "initializing rpc server")
 	}
diff --git a/release/context.go b/release/context.go
index c307e951..ac279ae4 100644
--- a/release/context.go
+++ b/release/context.go
@@ -85,7 +85,7 @@ func (rc *ReleaseContext) SelectWorkloads(ctx context.Context, results update.Re
 	}
 
 	// Ask the cluster about those that we're still interested in
-	definedAndRunning, err := rc.cluster.SomeWorkloads(toAskClusterAbout)
+	definedAndRunning, err := rc.cluster.SomeWorkloads(ctx, toAskClusterAbout)
 	if err != nil {
 		return nil, err
 	}
diff --git a/remote/rpc/server.go b/remote/rpc/server.go
index 4b9f9e5b..256c8c7b 100644
--- a/remote/rpc/server.go
+++ b/remote/rpc/server.go
@@ -5,6 +5,7 @@ import (
 	"io"
 	"net/rpc"
 	"net/rpc/jsonrpc"
+	"time"
 
 	"github.com/weaveworks/flux/api/v10"
 
@@ -20,17 +21,17 @@ import (
 
 // Server takes an api.Server and makes it available over RPC.
 type Server struct {
-	server *rpc.Server
+	server  *rpc.Server
 }
 
 // NewServer instantiates a new RPC server, handling requests on the
 // conn by invoking methods on the underlying (assumed local) server.
-func NewServer(s api.UpstreamServer) (*Server, error) {
+func NewServer(s api.UpstreamServer, t time.Duration) (*Server, error) {
 	server := rpc.NewServer()
-	if err := server.Register(&RPCServer{s}); err != nil {
+	if err := server.Register(&RPCServer{s, t}); err != nil {
 		return nil, err
 	}
-	return &Server{server: server}, nil
+	return &Server{server}, nil
 }
 
 func (c *Server) ServeConn(conn io.ReadWriteCloser) {
@@ -38,15 +39,20 @@ func (c *Server) ServeConn(conn io.ReadWriteCloser) {
 }
 
 type RPCServer struct {
-	s api.UpstreamServer
+	s       api.UpstreamServer
+	timeout time.Duration
 }
 
 func (p *RPCServer) Ping(_ struct{}, _ *struct{}) error {
-	return p.s.Ping(context.Background())
+	ctx, cancel := context.WithTimeout(context.Background(), p.timeout)
+	defer cancel()
+	return p.s.Ping(ctx)
 }
 
 func (p *RPCServer) Version(_ struct{}, resp *string) error {
-	v, err := p.s.Version(context.Background())
+	ctx, cancel := context.WithTimeout(context.Background(), p.timeout)
+	defer cancel()
+	v, err := p.s.Version(ctx)
 	*resp = v
 	return err
 }
@@ -57,7 +63,9 @@ type ExportResponse struct {
 }
 
 func (p *RPCServer) Export(_ struct{}, resp *ExportResponse) error {
-	v, err := p.s.Export(context.Background())
+	ctx, cancel := context.WithTimeout(context.Background(), p.timeout)
+	defer cancel()
+	v, err := p.s.Export(ctx)
 	resp.Result = v
 	if err != nil {
 		if err, ok := errors.Cause(err).(*fluxerr.Error); ok {
@@ -74,7 +82,9 @@ type ListServicesResponse struct {
 }
 
 func (p *RPCServer) ListServices(namespace string, resp *ListServicesResponse) error {
-	v, err := p.s.ListServices(context.Background(), namespace)
+	ctx, cancel := context.WithTimeout(context.Background(), p.timeout)
+	defer cancel()
+	v, err := p.s.ListServices(ctx, namespace)
 	resp.Result = v
 	if err != nil {
 		if err, ok := errors.Cause(err).(*fluxerr.Error); ok {
@@ -91,7 +101,9 @@ type ListImagesResponse struct {
 }
 
 func (p *RPCServer) ListImages(spec update.ResourceSpec, resp *ListImagesResponse) error {
-	v, err := p.s.ListImages(context.Background(), spec)
+	ctx, cancel := context.WithTimeout(context.Background(), p.timeout)
+	defer cancel()
+	v, err := p.s.ListImages(ctx, spec)
 	resp.Result = v
 	if err != nil {
 		if err, ok := errors.Cause(err).(*fluxerr.Error); ok {
@@ -103,7 +115,9 @@ func (p *RPCServer) ListImages(spec update.ResourceSpec, resp *ListImagesRespons
 }
 
 func (p *RPCServer) ListImagesWithOptions(opts v10.ListImagesOptions, resp *ListImagesResponse) error {
-	v, err := p.s.ListImagesWithOptions(context.Background(), opts)
+	ctx, cancel := context.WithTimeout(context.Background(), p.timeout)
+	defer cancel()
+	v, err := p.s.ListImagesWithOptions(ctx, opts)
 	resp.Result = v
 	if err != nil {
 		if err, ok := errors.Cause(err).(*fluxerr.Error); ok {
@@ -120,7 +134,9 @@ type UpdateManifestsResponse struct {
 }
 
 func (p *RPCServer) UpdateManifests(spec update.Spec, resp *UpdateManifestsResponse) error {
-	v, err := p.s.UpdateManifests(context.Background(), spec)
+	ctx, cancel := context.WithTimeout(context.Background(), p.timeout)
+	defer cancel()
+	v, err := p.s.UpdateManifests(ctx, spec)
 	resp.Result = v
 	if err != nil {
 		if err, ok := errors.Cause(err).(*fluxerr.Error); ok {
@@ -136,7 +152,9 @@ type NotifyChangeResponse struct {
 }
 
 func (p *RPCServer) NotifyChange(c v9.Change, resp *NotifyChangeResponse) error {
-	err := p.s.NotifyChange(context.Background(), c)
+	ctx, cancel := context.WithTimeout(context.Background(), p.timeout)
+	defer cancel()
+	err := p.s.NotifyChange(ctx, c)
 	if err != nil {
 		if err, ok := errors.Cause(err).(*fluxerr.Error); ok {
 			resp.ApplicationError = err
@@ -152,7 +170,9 @@ type JobStatusResponse struct {
 }
 
 func (p *RPCServer) JobStatus(jobID job.ID, resp *JobStatusResponse) error {
-	v, err := p.s.JobStatus(context.Background(), jobID)
+	ctx, cancel := context.WithTimeout(context.Background(), p.timeout)
+	defer cancel()
+	v, err := p.s.JobStatus(ctx, jobID)
 	resp.Result = v
 	if err != nil {
 		if err, ok := errors.Cause(err).(*fluxerr.Error); ok {
@@ -169,7 +189,9 @@ type SyncStatusResponse struct {
 }
 
 func (p *RPCServer) SyncStatus(ref string, resp *SyncStatusResponse) error {
-	v, err := p.s.SyncStatus(context.Background(), ref)
+	ctx, cancel := context.WithTimeout(context.Background(), p.timeout)
+	defer cancel()
+	v, err := p.s.SyncStatus(ctx, ref)
 	resp.Result = v
 	if err != nil {
 		if err, ok := errors.Cause(err).(*fluxerr.Error); ok {
@@ -186,7 +208,9 @@ type GitRepoConfigResponse struct {
 }
 
 func (p *RPCServer) GitRepoConfig(regenerate bool, resp *GitRepoConfigResponse) error {
-	v, err := p.s.GitRepoConfig(context.Background(), regenerate)
+	ctx, cancel := context.WithTimeout(context.Background(), p.timeout)
+	defer cancel()
+	v, err := p.s.GitRepoConfig(ctx, regenerate)
 	resp.Result = v
 	if err != nil {
 		if err, ok := errors.Cause(err).(*fluxerr.Error); ok {


Test patch:
diff --git a/cluster/kubernetes/kubernetes_test.go b/cluster/kubernetes/kubernetes_test.go
index 4e16c6c2..6ebfee91 100644
--- a/cluster/kubernetes/kubernetes_test.go
+++ b/cluster/kubernetes/kubernetes_test.go
@@ -1,6 +1,7 @@
 package kubernetes
 
 import (
+	"context"
 	"reflect"
 	"testing"
 
@@ -28,7 +29,7 @@ func testGetAllowedNamespaces(t *testing.T, namespace []string, expected []strin
 	client := ExtendedClient{coreClient: clientset}
 	c := NewCluster(client, nil, nil, log.NewNopLogger(), namespace, []string{})
 
-	namespaces, err := c.getAllowedAndExistingNamespaces()
+	namespaces, err := c.getAllowedAndExistingNamespaces(context.Background())
 	if err != nil {
 		t.Errorf("The error should be nil, not: %s", err)
 	}
diff --git a/daemon/daemon_test.go b/daemon/daemon_test.go
index 6600c5df..bf27c521 100644
--- a/daemon/daemon_test.go
+++ b/daemon/daemon_test.go
@@ -587,7 +587,7 @@ func TestDaemon_Automated(t *testing.T) {
 			},
 		},
 	}
-	k8s.SomeWorkloadsFunc = func([]flux.ResourceID) ([]cluster.Workload, error) {
+	k8s.SomeWorkloadsFunc = func(ctx context.Context, ids []flux.ResourceID) ([]cluster.Workload, error) {
 		return []cluster.Workload{workload}, nil
 	}
 	start()
@@ -613,7 +613,7 @@ func TestDaemon_Automated_semver(t *testing.T) {
 			},
 		},
 	}
-	k8s.SomeWorkloadsFunc = func([]flux.ResourceID) ([]cluster.Workload, error) {
+	k8s.SomeWorkloadsFunc = func(ctx context.Context, ids []flux.ResourceID) ([]cluster.Workload, error) {
 		return []cluster.Workload{workload}, nil
 	}
 	start()
@@ -675,7 +675,7 @@ func mockDaemon(t *testing.T) (*Daemon, func(), func(), *mock.Mock, *mockEventWr
 	var k8s *mock.Mock
 	{
 		k8s = &mock.Mock{}
-		k8s.AllWorkloadsFunc = func(maybeNamespace string) ([]cluster.Workload, error) {
+		k8s.AllWorkloadsFunc = func(ctx context.Context, maybeNamespace string) ([]cluster.Workload, error) {
 			if maybeNamespace == ns {
 				return []cluster.Workload{
 					singleService,
@@ -686,9 +686,9 @@ func mockDaemon(t *testing.T) (*Daemon, func(), func(), *mock.Mock, *mockEventWr
 			return []cluster.Workload{}, nil
 		}
 		k8s.IsAllowedResourceFunc = func(flux.ResourceID) bool { return true }
-		k8s.ExportFunc = func() ([]byte, error) { return testBytes, nil }
+		k8s.ExportFunc = func(ctx context.Context) ([]byte, error) { return testBytes, nil }
 		k8s.PingFunc = func() error { return nil }
-		k8s.SomeWorkloadsFunc = func([]flux.ResourceID) ([]cluster.Workload, error) {
+		k8s.SomeWorkloadsFunc = func(ctx context.Context, ids []flux.ResourceID) ([]cluster.Workload, error) {
 			return []cluster.Workload{
 				singleService,
 			}, nil
diff --git a/daemon/sync_test.go b/daemon/sync_test.go
index 7e65bc3b..2c0d1906 100644
--- a/daemon/sync_test.go
+++ b/daemon/sync_test.go
@@ -44,7 +44,7 @@ func daemon(t *testing.T) (*Daemon, func()) {
 	repo, repoCleanup := gittest.Repo(t)
 
 	k8s = &mock.Mock{}
-	k8s.ExportFunc = func() ([]byte, error) { return nil, nil }
+	k8s.ExportFunc = func(ctx context.Context) ([]byte, error) { return nil, nil }
 
 	events = &mockEventWriter{}
 
diff --git a/release/releaser_test.go b/release/releaser_test.go
index 8b050562..f4d73308 100644
--- a/release/releaser_test.go
+++ b/release/releaser_test.go
@@ -145,10 +145,10 @@ var (
 
 func mockCluster(running ...cluster.Workload) *mock.Mock {
 	return &mock.Mock{
-		AllWorkloadsFunc: func(string) ([]cluster.Workload, error) {
+		AllWorkloadsFunc: func(ctx context.Context, maybeNamespace string) ([]cluster.Workload, error) {
 			return running, nil
 		},
-		SomeWorkloadsFunc: func(ids []flux.ResourceID) ([]cluster.Workload, error) {
+		SomeWorkloadsFunc: func(ctx context.Context, ids []flux.ResourceID) ([]cluster.Workload, error) {
 			var res []cluster.Workload
 			for _, id := range ids {
 				for _, svc := range running {
diff --git a/remote/rpc/rpc_test.go b/remote/rpc/rpc_test.go
index 93f1f8e0..88b3ef48 100644
--- a/remote/rpc/rpc_test.go
+++ b/remote/rpc/rpc_test.go
@@ -6,6 +6,7 @@ import (
 	"io"
 	"reflect"
 	"testing"
+	"time"
 
 	"github.com/weaveworks/flux/api"
 	"github.com/weaveworks/flux/remote"
@@ -26,7 +27,7 @@ func TestRPC(t *testing.T) {
 	wrap := func(mock api.UpstreamServer) api.UpstreamServer {
 		clientConn, serverConn := pipes()
 
-		server, err := NewServer(mock)
+		server, err := NewServer(mock, 10*time.Second)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -61,7 +62,7 @@ func TestBadRPC(t *testing.T) {
 	ctx := context.Background()
 	mock := &remote.MockServer{}
 	clientConn, serverConn := faultyPipes()
-	server, err := NewServer(mock)
+	server, err := NewServer(mock, 10*time.Second)
 	if err != nil {
 		t.Fatal(err)
 	}