Repository: weaveworks/flux
Pull request #2203 (2019-06-27 14:02:50)

Release 1.13.1
Cherry picked PRs and changelog for 1.13.1.


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index ac78490f..551f5bdc 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,6 +1,33 @@
 This is the changelog for the Flux daemon; the changelog for the Helm
 operator is in [./CHANGELOG-helmop.md](./CHANGELOG-helmop.md).
 
+## 1.13.1 (2019-06-27)
+
+This is a patch release.
+
+### Fixes
+
+- Use a context with a timeout for every request that comes through
+  the upstream connection, so they may be abandoned if taking too long [weaveworks/flux#2171][]
+- Initialise the high-water mark once, so it doesn't get continually
+  reset and cause notification noise [weaveworks/flux#2177][]
+- Force tag updates when making local clones, to account for changes
+  in git 2.20 [weaveworks/flux#2184][]
+
+### Thanks
+
+Cheers to the following people for their contributions: @2opremio,
+@J-Lou, @aarnaud, @adrian, @airmap-madison, @alanjcastonguay,
+@arsiesys, @atbe-crowe, @azazel75, @bia, @carlosjgp, @chriscorn-takt,
+@cristian-radu, @davidkarlsen, @derrickburns, @dholbach, @dlespiau,
+@errordeveloper, @ewoutp, @hiddeco, @humayunjamal, @isen-ng,
+@judewin-alef, @kevinm444, @muhlba91, @roaddemon, @runningman84,
+@squaremo, @starkers, @stefanprodan, @sukrit007, @willholley.
+
+[weaveworks/flux#2171]: https://github.com/weaveworks/flux/pull/2171
+[weaveworks/flux#2177]: https://github.com/weaveworks/flux/pull/2177
+[weaveworks/flux#2184]: https://github.com/weaveworks/flux/pull/2184
+
 ## 1.13.0 (2019-06-13)
 
 This feature release contains an experimental feature for [generating
diff --git a/chart/flux/values.yaml b/chart/flux/values.yaml
index 53f046d5..344134e8 100644
--- a/chart/flux/values.yaml
+++ b/chart/flux/values.yaml
@@ -9,7 +9,7 @@ logFormat: fmt
 
 image:
   repository: docker.io/weaveworks/flux
-  tag: 1.13.0
+  tag: 1.13.1
   pullPolicy: IfNotPresent
   pullSecret:
 
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/daemon/loop.go b/daemon/loop.go
index 2eee8c26..c9106ad9 100644
--- a/daemon/loop.go
+++ b/daemon/loop.go
@@ -46,14 +46,14 @@ func (d *Daemon) Loop(stop chan struct{}, wg *sync.WaitGroup, logger log.Logger)
 	// every timer tick as well as every mirror refresh.
 	syncHead := ""
 
+	// In-memory sync tag state
+	lastKnownSyncTag := &lastKnownSyncTag{logger: logger, syncTag: d.GitConfig.SyncTag}
+
 	// Ask for a sync, and to poll images, straight away
 	d.AskForSync()
 	d.AskForImagePoll()
 
 	for {
-		var (
-			lastKnownSyncTag = &lastKnownSyncTag{logger: logger, syncTag: d.GitConfig.SyncTag}
-		)
 		select {
 		case <-stop:
 			logger.Log("stopping", "true")
diff --git a/deploy/flux-deployment.yaml b/deploy/flux-deployment.yaml
index 7c1ac984..86d23557 100644
--- a/deploy/flux-deployment.yaml
+++ b/deploy/flux-deployment.yaml
@@ -63,7 +63,7 @@ spec:
         # There are no ":latest" images for flux. Find the most recent
         # release or image version at https://hub.docker.com/r/weaveworks/flux/tags
         # and replace the tag here.
-        image: docker.io/weaveworks/flux:1.13.0
+        image: docker.io/weaveworks/flux:1.13.1
         imagePullPolicy: IfNotPresent
         resources:
           requests:
diff --git a/git/working.go b/git/working.go
index 326d1bca..b501cb56 100644
--- a/git/working.go
+++ b/git/working.go
@@ -82,6 +82,18 @@ func (r *Repo) Clone(ctx context.Context, conf Config) (*Checkout, error) {
 	}
 
 	r.mu.RLock()
+	// Here is where we mimic `git fetch --tags --force`, but
+	// _without_ overwriting head refs. This is only required for a
+	// `Checkout` and _not_ for `Repo` as (bare) mirrors will happily
+	// accept any ref changes to tags.
+	//
+	// NB: do this before any other fetch actions, as otherwise we may
+	// get an 'existing tag clobber' error back.
+	if err := fetch(ctx, repoDir, r.dir, `'+refs/tags/*:refs/tags/*'`); err != nil {
+		os.RemoveAll(repoDir)
+		r.mu.RUnlock()
+		return nil, err
+	}
 	if err := fetch(ctx, repoDir, r.dir, realNotesRef+":"+realNotesRef); err != nil {
 		os.RemoveAll(repoDir)
 		r.mu.RUnlock()
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)
 	}