Merge pull request #2313 from ripienaar/2312

(#2312) Support a query on JSON data in kv watcher

Author: ripienaar

aagent/watchers/kvwatcher/kv.go

@@ -17,6 +17,7 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/nats-io/nats.go"
+	"github.com/tidwall/gjson"
 
 	"github.com/choria-io/ccm/hiera"
 	"github.com/choria-io/go-choria/aagent/model"
@@ -59,6 +60,7 @@ type properties struct {
 	RepublishTrigger          string `mapstructure:"republish_trigger"`
 	HieraConfig               bool   `mapstructure:"hiera_config"`
 	StoreKey                  string `mapstructure:"store_key"`
+	Query                     string `mapstructure:"query"`
 }
 
 type Watcher struct {
@@ -337,6 +339,15 @@ func (w *Watcher) parseValue(val []byte) (any, error) {
 			}
 		}
 
+		if w.properties.Query != "" {
+			j, err := json.Marshal(parsedValue)
+			if err != nil {
+				return nil, fmt.Errorf("could not marshal data for query: %w", err)
+			}
+
+			res := gjson.GetBytes(j, w.properties.Query)
+			parsedValue = res.Value()
+		}
 	} else if bytes.HasPrefix(v, []byte("[")) && bytes.HasSuffix(v, []byte("]")) {
 		parsedValue = []any{}
 		err := json.Unmarshal(v, &parsedValue)

aagent/watchers/kvwatcher/kv_test.go

@@ -83,6 +83,14 @@ var _ = Describe("AAgent/Watchers/KvWatcher", func() {
 			Expect(err).ToNot(HaveOccurred())
 		})
 
+		It("Should support data queries", func() {
+			w.properties.Query = "spec"
+			kve.EXPECT().Value().Return([]byte("{\"spec\": \"foo\"}")).MinTimes(1)
+			machine.EXPECT().DataPut("machines", "foo").Return(nil).Times(1)
+			_, err := w.poll()
+			Expect(err).ToNot(HaveOccurred())
+		})
+
 		It("Should handle a leading and trailing unicode whitespace", func() {
 			kve.EXPECT().Value().Return([]byte("\n   \t{\"spec\": \"foo\"}\t  \n")).MinTimes(1)
 			machine.EXPECT().DataPut("machines", map[string]any{"spec": "foo"}).Return(nil).Times(1)