Fix tag_all pseudo policy on default-namespaced wokloads

This fixes a regression introduced by fluxcd#1442 in which applying the `tag_all`
pseudo policy failed for workload manifests in which the namespace is omitted.

The effective namespace of the workload wasn't set after parsing, causing a
mismatch in the resource identifier (the parsed resource identifier was
cluster-scoped due to the lack of explicit namespace in the manifest).

cluster/kubernetes/manifests.go

@@ -60,7 +60,7 @@ func getCRDScopes(manifests map[string]kresource.KubeManifest) ResourceScopes {
 	return result
 }
 
-func postProcess(manifests map[string]kresource.KubeManifest, nser namespacer) (map[string]resource.Resource, error) {
+func setEffectiveNamespaces(manifests map[string]kresource.KubeManifest, nser namespacer) (map[string]resource.Resource, error) {
 	knownScopes := getCRDScopes(manifests)
 	result := map[string]resource.Resource{}
 	for _, km := range manifests {
@@ -76,15 +76,15 @@ func postProcess(manifests map[string]kresource.KubeManifest, nser namespacer) (
 	return result, nil
 }
 
-func (c *Manifests) LoadManifests(base string, paths []string) (map[string]resource.Resource, error) {
+func (m *Manifests) LoadManifests(base string, paths []string) (map[string]resource.Resource, error) {
 	manifests, err := kresource.Load(base, paths)
 	if err != nil {
 		return nil, err
 	}
-	return postProcess(manifests, c.Namespacer)
+	return setEffectiveNamespaces(manifests, m.Namespacer)
 }
 
-func (c *Manifests) UpdateImage(def []byte, id flux.ResourceID, container string, image image.Ref) ([]byte, error) {
+func (m *Manifests) UpdateImage(def []byte, id flux.ResourceID, container string, image image.Ref) ([]byte, error) {
 	return updateWorkload(def, id, container, image)
 }
 

cluster/kubernetes/policies.go

@@ -4,7 +4,6 @@ import (
 	"fmt"
 
 	"github.com/pkg/errors"
-	"gopkg.in/yaml.v2"
 
 	"github.com/weaveworks/flux"
 	kresource "github.com/weaveworks/flux/cluster/kubernetes/resource"
@@ -21,7 +20,7 @@ func (m *Manifests) UpdatePolicies(def []byte, id flux.ResourceID, update policy
 	// what all the containers are.
 	if tagAll, ok := update.Add.Get(policy.TagAll); ok {
 		add = add.Without(policy.TagAll)
-		containers, err := extractContainers(def, id)
+		containers, err := m.extractWorkloadContainers(def, id)
 		if err != nil {
 			return nil, err
 		}
@@ -49,25 +48,17 @@ func (m *Manifests) UpdatePolicies(def []byte, id flux.ResourceID, update policy
 	return (KubeYAML{}).Annotate(def, ns, kind, name, args...)
 }
 
-type manifest struct {
-	Metadata struct {
-		Annotations map[string]string `yaml:"annotations"`
-	} `yaml:"metadata"`
-}
-
-func extractAnnotations(def []byte) (map[string]string, error) {
-	var m manifest
-	if err := yaml.Unmarshal(def, &m); err != nil {
-		return nil, errors.Wrap(err, "decoding manifest for annotations")
-	}
-	if m.Metadata.Annotations == nil {
-		return map[string]string{}, nil
+func (m *Manifests) extractWorkloadContainers(def []byte, id flux.ResourceID) ([]resource.Container, error) {
+	kresources, err := kresource.ParseMultidoc(def, "stdin")
+	if err != nil {
+		return nil, err
 	}
-	return m.Metadata.Annotations, nil
-}
-
-func extractContainers(def []byte, id flux.ResourceID) ([]resource.Container, error) {
-	resources, err := kresource.ParseMultidoc(def, "stdin")
+	// Note: setEffectiveNamespaces() won't work for CRD instances whose CRD is yet to be created
+	// (due to the CRD not being present in kresources).
+	// We could get out of our way to fix this (or give a better error) but:
+	// 1. With the exception of HelmReleases CRD instances are not workloads anyways.
+	// 2. The problem is eventually fixed by the first successful sync.
+	resources, err := setEffectiveNamespaces(kresources, m.Namespacer)
 	if err != nil {
 		return nil, err
 	}

cluster/kubernetes/policies_test.go

@@ -8,9 +8,16 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/weaveworks/flux"
+	kresource "github.com/weaveworks/flux/cluster/kubernetes/resource"
 	"github.com/weaveworks/flux/policy"
 )
 
+type constNamespacer string
+
+func (ns constNamespacer) EffectiveNamespace(manifest kresource.KubeManifest, _ ResourceScopes) (string, error) {
+	return string(ns), nil
+}
+
 func TestUpdatePolicies(t *testing.T) {
 	for _, c := range []struct {
 		name    string
@@ -166,13 +173,21 @@ func TestUpdatePolicies(t *testing.T) {
 			},
 			wantErr: true,
 		},
+		{
+			name: "set tag to all containers",
+			in:   nil,
+			out:  []string{"flux.weave.works/tag.nginx", "semver:*"},
+			update: policy.Update{
+				Add: policy.Set{policy.TagAll: "semver:*"},
+			},
+		},
 	} {
 		t.Run(c.name, func(t *testing.T) {
 			caseIn := templToString(t, annotationsTemplate, c.in)
 			caseOut := templToString(t, annotationsTemplate, c.out)
 			resourceID := flux.MustParseResourceID("default:deployment/nginx")
-			out, err := (&Manifests{}).UpdatePolicies([]byte(caseIn), resourceID, c.update)
-			assert.Equal(t, c.wantErr, err != nil)
+			out, err := (&Manifests{constNamespacer("default")}).UpdatePolicies([]byte(caseIn), resourceID, c.update)
+			assert.Equal(t, c.wantErr, err != nil, "unexpected error value: %s", err)
 			if !c.wantErr {
 				assert.Equal(t, string(out), caseOut)
 			}

cluster/kubernetes/sync_test.go

@@ -319,7 +319,7 @@ metadata:
 		}
 
 		// Needed to get from KubeManifest to resource.Resource
-		resources, err := postProcess(resources0, namespacer)
+		resources, err := setEffectiveNamespaces(resources0, namespacer)
 		if err != nil {
 			t.Fatal(err)
 		}