Use kubeyaml for YAML updates

Makefile

@@ -37,7 +37,7 @@ realclean: clean
 	rm -rf ./cache
 
 test:
-	go test ${TEST_FLAGS} $(shell go list ./... | grep -v "^github.com/weaveworks/flux/vendor" | sort -u)
+	PATH=${PATH}:${PWD}/bin go test ${TEST_FLAGS} $(shell go list ./... | grep -v "^github.com/weaveworks/flux/vendor" | sort -u)
 
 build/.%.done: docker/Dockerfile.%
 	mkdir -p ./build/docker/$*

bin/kubeyaml

@@ -0,0 +1,2 @@
+#!/bin/sh
+docker run --rm -i squaremo/kubeyaml:latest "$@"

cluster/kubernetes/kubeyaml.go

@@ -0,0 +1,42 @@
+package kubernetes
+
+import (
+	"bytes"
+	"errors"
+	"os/exec"
+	"strings"
+)
+
+// KubeYAML is a placeholder value for calling the helper executable
+// `kubeyaml`.
+type KubeYAML struct {
+}
+
+// Image calls the kubeyaml subcommand `image` with the arguments given.
+func (k KubeYAML) Image(in []byte, ns, kind, name, container, image string) ([]byte, error) {
+	args := []string{"image", "--namespace", ns, "--kind", kind, "--name", name}
+	args = append(args, "--container", container, "--image", image)
+	return execKubeyaml(in, args)
+}
+
+// Annotate calls the kubeyaml subcommand `annotate` with the arguments as given.
+func (k KubeYAML) Annotate(in []byte, ns, kind, name string, policies ...string) ([]byte, error) {
+	args := []string{"annotate", "--namespace", ns, "--kind", kind, "--name", name}
+	args = append(args, policies...)
+	return execKubeyaml(in, args)
+}
+
+func execKubeyaml(in []byte, args []string) ([]byte, error) {
+	cmd := exec.Command("kubeyaml", args...)
+	out := &bytes.Buffer{}
+	errOut := &bytes.Buffer{}
+	cmd.Stdin = bytes.NewBuffer(in)
+	cmd.Stdout = out
+	cmd.Stderr = errOut
+
+	err := cmd.Run()
+	if err != nil {
+		return nil, errors.New(strings.TrimSpace(errOut.String()))
+	}
+	return out.Bytes(), nil
+}

cluster/kubernetes/policies.go

@@ -1,8 +1,7 @@
 package kubernetes
 
 import (
-	"regexp"
-	"strings"
+	"fmt"
 
 	"github.com/pkg/errors"
 	yaml "gopkg.in/yaml.v2"
@@ -14,6 +13,7 @@ import (
 )
 
 func (m *Manifests) UpdatePolicies(def []byte, id flux.ResourceID, update policy.Update) ([]byte, error) {
+	ns, kind, name := id.Components()
 	add, del := update.Add, update.Remove
 
 	// We may be sent the pseudo-policy `policy.TagAll`, which means
@@ -35,78 +35,15 @@ func (m *Manifests) UpdatePolicies(def []byte, id flux.ResourceID, update policy
 		}
 	}
 
-	return updateAnnotations(def, id, func(old map[string]string) {
-		for k, v := range add {
-			old[kresource.PolicyPrefix+string(k)] = v
-		}
-		for k := range del {
-			delete(old, kresource.PolicyPrefix+string(k))
-		}
-	})
-}
-
-func updateAnnotations(def []byte, id flux.ResourceID, f func(map[string]string)) ([]byte, error) {
-	annotations, err := extractAnnotations(def)
-	if err != nil {
-		return nil, err
-	}
-	f(annotations)
-
-	// Write the new annotations back into the manifest
-	// Generate a fragment of the new annotations.
-	var fragment string
-	if len(annotations) > 0 {
-		fragmentB, err := yaml.Marshal(map[string]map[string]string{
-			"annotations": annotations,
-		})
-		if err != nil {
-			return nil, err
-		}
-
-		fragment = string(fragmentB)
-
-		// Remove the last newline, so it fits in better
-		fragment = strings.TrimSuffix(fragment, "\n")
-
-		// indent the fragment 2 spaces
-		// TODO: delete all regular expressions which are used to modify YAML.
-		// See #1019. Modifying this is not recommended.
-		fragment = regexp.MustCompile(`(.+)`).ReplaceAllString(fragment, "  $1")
-
-		// Add a newline if it's not blank
-		if len(fragment) > 0 {
-			fragment = "\n" + fragment
-		}
-	}
-
-	// Find where to insert the fragment.
-	// TODO: delete all regular expressions which are used to modify YAML.
-	// See #1019. Modifying this is not recommended.
-	replaced := false
-	annotationsRE := regexp.MustCompile(`(?m:\n  annotations:\s*(?:#.*)*(?:\n    .*|\n)*$)`)
-	newDef := annotationsRE.ReplaceAllStringFunc(string(def), func(found string) string {
-		if !replaced {
-			replaced = true
-			return fragment
-		}
-		return found
-	})
-	if !replaced {
-		metadataRE := multilineRE(`(metadata:\s*(?:#.*)*)`)
-		newDef = metadataRE.ReplaceAllStringFunc(string(def), func(found string) string {
-			if !replaced {
-				replaced = true
-				f := found + fragment
-				return f
-			}
-			return found
-		})
+	args := []string{}
+	for pol, val := range add {
+		args = append(args, fmt.Sprintf("%s%s=%s", kresource.PolicyPrefix, pol, val))
 	}
-	if !replaced {
-		return nil, errors.New("Could not update resource annotations")
+	for pol, _ := range del {
+		args = append(args, fmt.Sprintf("%s%s=", kresource.PolicyPrefix, pol))
 	}
 
-	return []byte(newDef), err
+	return (KubeYAML{}).Annotate(def, ns, kind, name, args...)
 }
 
 type manifest struct {

cluster/kubernetes/policies_test.go

@@ -2,8 +2,6 @@ package kubernetes
 
 import (
 	"bytes"
-	"fmt"
-	"strings"
 	"testing"
 	"text/template"
 
@@ -14,45 +12,45 @@ import (
 func TestUpdatePolicies(t *testing.T) {
 	for _, c := range []struct {
 		name    string
-		in, out map[string]string
+		in, out []string
 		update  policy.Update
 	}{
 		{
 			name: "adding annotation with others existing",
-			in:   map[string]string{"prometheus.io.scrape": "false"},
-			out:  map[string]string{"flux.weave.works/automated": "true", "prometheus.io.scrape": "false"},
+			in:   []string{"prometheus.io.scrape", "'false'"},
+			out:  []string{"prometheus.io.scrape", "'false'", "flux.weave.works/automated", "'true'"},
 			update: policy.Update{
 				Add: policy.Set{policy.Automated: "true"},
 			},
 		},
 		{
 			name: "adding annotation when already has annotation",
-			in:   map[string]string{"flux.weave.works/automated": "true"},
-			out:  map[string]string{"flux.weave.works/automated": "true"},
+			in:   []string{"flux.weave.works/automated", "'true'"},
+			out:  []string{"flux.weave.works/automated", "'true'"},
 			update: policy.Update{
 				Add: policy.Set{policy.Automated: "true"},
 			},
 		},
 		{
 			name: "adding annotation when already has annotation and others",
-			in:   map[string]string{"flux.weave.works/automated": "true", "prometheus.io.scrape": "false"},
-			out:  map[string]string{"flux.weave.works/automated": "true", "prometheus.io.scrape": "false"},
+			in:   []string{"flux.weave.works/automated", "'true'", "prometheus.io.scrape", "'false'"},
+			out:  []string{"flux.weave.works/automated", "'true'", "prometheus.io.scrape", "'false'"},
 			update: policy.Update{
 				Add: policy.Set{policy.Automated: "true"},
 			},
 		},
 		{
 			name: "adding first annotation",
 			in:   nil,
-			out:  map[string]string{"flux.weave.works/automated": "true"},
+			out:  []string{"flux.weave.works/automated", "'true'"},
 			update: policy.Update{
 				Add: policy.Set{policy.Automated: "true"},
 			},
 		},
 		{
 			name: "add and remove different annotations at the same time",
-			in:   map[string]string{"flux.weave.works/automated": "true", "prometheus.io.scrape": "false"},
-			out:  map[string]string{"flux.weave.works/locked": "true", "prometheus.io.scrape": "false"},
+			in:   []string{"flux.weave.works/automated", "'true'", "prometheus.io.scrape", "'false'"},
+			out:  []string{"prometheus.io.scrape", "'false'", "flux.weave.works/locked", "'true'"},
 			update: policy.Update{
 				Add:    policy.Set{policy.Locked: "true"},
 				Remove: policy.Set{policy.Automated: "true"},
@@ -69,15 +67,15 @@ func TestUpdatePolicies(t *testing.T) {
 		},
 		{
 			name: "remove annotation with others existing",
-			in:   map[string]string{"flux.weave.works/automated": "true", "prometheus.io.scrape": "false"},
-			out:  map[string]string{"prometheus.io.scrape": "false"},
+			in:   []string{"flux.weave.works/automated", "true", "prometheus.io.scrape", "false"},
+			out:  []string{"prometheus.io.scrape", "false"},
 			update: policy.Update{
 				Remove: policy.Set{policy.Automated: "true"},
 			},
 		},
 		{
 			name: "remove last annotation",
-			in:   map[string]string{"flux.weave.works/automated": "true"},
+			in:   []string{"flux.weave.works/automated", "true"},
 			out:  nil,
 			update: policy.Update{
 				Remove: policy.Set{policy.Automated: "true"},
@@ -93,23 +91,23 @@ func TestUpdatePolicies(t *testing.T) {
 		},
 		{
 			name: "remove annotation with only others",
-			in:   map[string]string{"prometheus.io.scrape": "false"},
-			out:  map[string]string{"prometheus.io.scrape": "false"},
+			in:   []string{"prometheus.io.scrape", "false"},
+			out:  []string{"prometheus.io.scrape", "false"},
 			update: policy.Update{
 				Remove: policy.Set{policy.Automated: "true"},
 			},
 		},
 		{
 			name: "multiline",
-			in:   map[string]string{"flux.weave.works/locked_msg": "|-\n      first\n      second"},
+			in:   []string{"flux.weave.works/locked_msg", "|-\n      first\n      second"},
 			out:  nil,
 			update: policy.Update{
 				Remove: policy.Set{policy.LockedMsg: "foo"},
 			},
 		},
 		{
 			name: "multiline with empty line",
-			in:   map[string]string{"flux.weave.works/locked_msg": "|-\n      first\n\n      third"},
+			in:   []string{"flux.weave.works/locked_msg", "|-\n      first\n\n      third"},
 			out:  nil,
 			update: policy.Update{
 				Remove: policy.Set{policy.LockedMsg: "foo"},
@@ -118,8 +116,8 @@ func TestUpdatePolicies(t *testing.T) {
 	} {
 		caseIn := templToString(t, annotationsTemplate, c.in)
 		caseOut := templToString(t, annotationsTemplate, c.out)
-		id := flux.MustParseResourceID("default:deplot/nginx")
-		out, err := (&Manifests{}).UpdatePolicies([]byte(caseIn), id, c.update)
+		resourceID := flux.MustParseResourceID("default:deployment/nginx")
+		out, err := (&Manifests{}).UpdatePolicies([]byte(caseIn), resourceID, c.update)
 		if err != nil {
 			t.Errorf("[%s] %v", c.name, err)
 		} else if string(out) != caseOut {
@@ -132,9 +130,9 @@ var annotationsTemplate = template.Must(template.New("").Parse(`---
 apiVersion: extensions/v1beta1
 kind: Deployment
 metadata: # comment really close to the war zone
-  {{with .}}annotations:{{range $k, $v := .}}
-    {{$k}}: {{printf "%s" $v}}{{end}}
-  {{end}}name: nginx
+  name: nginx{{with .}}
+  annotations:{{range .}}
+    {{index . 0}}: {{printf "%s" (index . 1)}}{{end}}{{end}}
 spec:
   replicas: 1
   template:
@@ -149,15 +147,13 @@ spec:
         - containerPort: 80
 `))
 
-func templToString(t *testing.T, templ *template.Template, annotations map[string]string) string {
-	for k, v := range annotations {
-		// Don't wrap multilines
-		if !strings.HasPrefix(v, "|") {
-			annotations[k] = fmt.Sprintf("%q", v)
-		}
+func templToString(t *testing.T, templ *template.Template, data []string) string {
+	var pairs [][]string
+	for i := 0; i < len(data); i += 2 {
+		pairs = append(pairs, []string{data[i], data[i+1]})
 	}
 	out := &bytes.Buffer{}
-	err := templ.Execute(out, annotations)
+	err := templ.Execute(out, pairs)
 	if err != nil {
 		t.Fatal(err)
 	}

cluster/kubernetes/resource/list.go

@@ -0,0 +1,10 @@
+package resource
+
+import (
+	"github.com/weaveworks/flux/resource"
+)
+
+type List struct {
+	baseObject
+	Items []resource.Resource
+}

cluster/kubernetes/resource/load.go

@@ -146,7 +146,15 @@ func ParseMultidoc(multidoc []byte, source string) (map[string]resource.Resource
 		if obj == nil {
 			continue
 		}
-		objs[obj.ResourceID().String()] = obj
+		// Lists must be treated specially, since it's the
+		// contained resources we are after.
+		if list, ok := obj.(*List); ok {
+			for _, item := range list.Items {
+				objs[item.ResourceID().String()] = item
+			}
+		} else {
+			objs[obj.ResourceID().String()] = obj
+		}
 	}
 
 	if err := chunks.Err(); err != nil {

cluster/kubernetes/resource/load_test.go

@@ -6,6 +6,7 @@ import (
 	"reflect"
 	"testing"
 
+	"github.com/weaveworks/flux"
 	"github.com/weaveworks/flux/cluster/kubernetes/testfiles"
 	"github.com/weaveworks/flux/resource"
 )
@@ -118,6 +119,39 @@ data:
 	}
 }
 
+func TestUnmarshalList(t *testing.T) {
+	doc := `---
+kind: List
+metadata:
+  name: list
+items:
+- kind: Deployment
+  metadata:
+    name: foo
+- kind: Service
+  metadata:
+    name: bar
+`
+	res, err := unmarshalObject("", []byte(doc))
+	if err != nil {
+		t.Fatal(err)
+	}
+	list, ok := res.(*List)
+	if !ok {
+		t.Fatal("did not parse as a list")
+	}
+	if len(list.Items) != 2 {
+		t.Fatalf("expected two items, got %+v", list.Items)
+	}
+	for i, id := range []flux.ResourceID{
+		flux.MustParseResourceID("default:deployment/foo"),
+		flux.MustParseResourceID("default:service/bar")} {
+		if list.Items[i].ResourceID() != id {
+			t.Errorf("At %d, expected %q, got %q", i, id, list.Items[i].ResourceID())
+		}
+	}
+}
+
 func debyte(r resource.Resource) resource.Resource {
 	if res, ok := r.(interface {
 		debyte()