feature: Add draft implementation for @gather

Author: rg0now

pkg/cache/delta.go

@@ -9,15 +9,15 @@ import (
 	"github.com/hsnlab/dcontroller/pkg/object"
 )
 
-type DeltaType = toolscache.DeltaType
+type DeltaType string
 
 const (
-	Added    = toolscache.Added
-	Deleted  = toolscache.Deleted
-	Updated  = toolscache.Updated
-	Replaced = toolscache.Replaced
-	Sync     = toolscache.Sync
-	Upserted = "Upserted" // for events that are either an update/replace or an add
+	Added    = DeltaType(toolscache.Added)
+	Deleted  = DeltaType(toolscache.Deleted)
+	Updated  = DeltaType(toolscache.Updated)
+	Replaced = DeltaType(toolscache.Replaced)
+	Sync     = DeltaType(toolscache.Sync)
+	Upserted = DeltaType("Upserted") // for events that are either an update/replace or an add
 )
 
 // NilDelta is a placeholder for a delta that means no change.

pkg/expression/expression.go

@@ -216,23 +216,21 @@ func (e *Expression) Evaluate(ctx EvalCtx) (any, error) {
 				return nil, NewExpressionError(e, err)
 			}
 
-			var ret any
 			// evaluate expressons
 			for _, arg := range args {
 				res, err := arg.Evaluate(ctx)
 				if err != nil {
 					return nil, errors.New("failed to evaluate expression")
 				}
-				ret, err = object.MergeAny(ret, res)
+				ctx.Object, err = object.MergeAny(ctx.Object, res)
 				if err != nil {
 					return nil, err
 				}
-				ctx.Object = ret
 			}
 
-			ctx.Log.V(8).Info("eval ready", "expression", e.String(), "result", ret)
+			ctx.Log.V(8).Info("eval ready", "expression", e.String(), "result", ctx.Object)
 
-			return ret, nil
+			return ctx.Object, nil
 
 		case "@filter":
 			args, err := AsExpOrExpList(e.Arg)

pkg/expression/expression_test.go

@@ -644,8 +644,8 @@ var _ = Describe("Expressions", func() {
 			}
 		})
 
-		It("should merge two JSONPath setters", func() {
-			jsonData := `{"@merge":[{"spec":{"a":"aaa"}},{"$.spec.b.d":12}]}`
+		It("should merge two setters with a JSONPath setter", func() {
+			jsonData := `{"@merge":[{"spec":1},{"spec":{"a":"aaa"}},{"$.spec.b.d":12}]}`
 			var exp Expression
 			err := json.Unmarshal([]byte(jsonData), &exp)
 			Expect(err).NotTo(HaveOccurred())

pkg/expression/jsonpath.go

@@ -29,7 +29,7 @@ func GetJSONPath(ctx EvalCtx, key string) (any, error) {
 		subject = ctx.Subject
 	}
 
-	ret, err := GetJSONPathExp(key, subject)
+	ret, err := GetJSONPathRaw(key, subject)
 	if err != nil {
 		return nil, err
 	}
@@ -83,7 +83,7 @@ func SetJSONPath(ctx EvalCtx, key string, value, data any) error {
 	}
 
 	// then call the low-level set util
-	if err := SetJSONPathExp(key, value, data); err != nil {
+	if err := SetJSONPathRaw(key, value, data); err != nil {
 		return fmt.Errorf("JSONPath expression error: cannot set "+
 			"key %q to value %q: %w", key, value, err)
 	}
@@ -93,9 +93,9 @@ func SetJSONPath(ctx EvalCtx, key string, value, data any) error {
 
 // low-level utils
 
-// GetJSONPathExp evaluates a JSONPath expression on the specified object and returns the result or
+// GetJSONPathRaw evaluates a JSONPath expression on the specified object and returns the result or
 // an error.
-func GetJSONPathExp(query string, object any) (any, error) {
+func GetJSONPathRaw(query string, object any) (any, error) {
 	je, err := jp.ParseString(query)
 	if err != nil {
 		return nil, err
@@ -112,10 +112,10 @@ func GetJSONPathExp(query string, object any) (any, error) {
 	return values[0], nil
 }
 
-// SetJSONPathExp sets a key (possibly represented with a JSONPath expression) to a value (can also
+// SetJSONPathRaw sets a key (possibly represented with a JSONPath expression) to a value (can also
 // be a JSONPath expression, which will be evaluated using the object argument) in the given data
 // structure.
-func SetJSONPathExp(key string, value, target any) error {
+func SetJSONPathRaw(key string, value, target any) error {
 	je, err := jp.ParseString(key)
 	if err != nil {
 		return err
@@ -124,14 +124,16 @@ func SetJSONPathExp(key string, value, target any) error {
 	return je.Set(target, value)
 }
 
-// lit := []Expression{}
-// for _, arg := range args {
-// 	lit = append(lit, Expression{Op: "@any", Literal: arg, Raw: util.Stringify(arg)})
-// }
-// argList, err := asExpList(exp.Arg)
-// Expect(err).NotTo(HaveOccurred())
-// argList = append(argList, Expression{
-// 	Op:      "@list",
-// 	Literal: lit,
-// })
-// exp.Arg.Literal = argList
+// SetJSONPathRawExp is the same as SetJSONPathRaw but takes the key as a string expression.
+func SetJSONPathRawExp(keyExp *Expression, value, data any) error {
+	key, err := keyExp.GetLiteralString()
+	if err != nil {
+		return err
+	}
+
+	if err := SetJSONPathRaw(key, value, data); err != nil {
+		return err
+	}
+
+	return nil
+}

pkg/expression/jsonpath_test.go

@@ -87,19 +87,19 @@ var _ = Describe("JSONPath", func() {
 			obj, err := object.NewViewObjectFromNativeObject("Service", input)
 			Expect(err).NotTo(HaveOccurred())
 
-			res, err := GetJSONPathExp(`$.metadata.name`, obj.UnstructuredContent())
+			res, err := GetJSONPathRaw(`$.metadata.name`, obj.UnstructuredContent())
 			Expect(err).NotTo(HaveOccurred())
 			s, err := AsString(res)
 			Expect(err).NotTo(HaveOccurred())
 			Expect(s).To(Equal("testservice-ok"))
 
-			res, err = GetJSONPathExp(`$["metadata"]["namespace"]`, obj.UnstructuredContent())
+			res, err = GetJSONPathRaw(`$["metadata"]["namespace"]`, obj.UnstructuredContent())
 			Expect(err).NotTo(HaveOccurred())
 			s, err = AsString(res)
 			Expect(err).NotTo(HaveOccurred())
 			Expect(s).To(Equal("testnamespace"))
 
-			res, err = GetJSONPathExp(`$.metadata`, obj.UnstructuredContent())
+			res, err = GetJSONPathRaw(`$.metadata`, obj.UnstructuredContent())
 			Expect(err).NotTo(HaveOccurred())
 			d, ok := res.(Unstructured)
 			Expect(ok).To(BeTrue())
@@ -108,13 +108,13 @@ var _ = Describe("JSONPath", func() {
 			Expect(d).To(HaveKey("name"))
 			Expect(d["name"]).To(Equal("testservice-ok"))
 
-			res, err = GetJSONPathExp(`$.spec.ports[1].port`, obj.UnstructuredContent())
+			res, err = GetJSONPathRaw(`$.spec.ports[1].port`, obj.UnstructuredContent())
 			Expect(err).NotTo(HaveOccurred())
 			i, err := AsInt(res)
 			Expect(err).NotTo(HaveOccurred())
 			Expect(i).To(Equal(int64(2)))
 
-			res, err = GetJSONPathExp(`$.spec.ports[?(@.name == 'udp-ok')].protocol`,
+			res, err = GetJSONPathRaw(`$.spec.ports[?(@.name == 'udp-ok')].protocol`,
 				obj.UnstructuredContent())
 			Expect(err).NotTo(HaveOccurred())
 			s, err = AsString(res)
@@ -138,14 +138,14 @@ var _ = Describe("JSONPath", func() {
 			})
 
 			// must use the alternative form
-			res, err := GetJSONPathExp(`$["metadata"]["annotations"]["kubernetes.io/service-name"]`,
+			res, err := GetJSONPathRaw(`$["metadata"]["annotations"]["kubernetes.io/service-name"]`,
 				obj.UnstructuredContent())
 			Expect(err).NotTo(HaveOccurred())
 			s, err := AsString(res)
 			Expect(err).NotTo(HaveOccurred())
 			Expect(s).To(Equal("example"))
 
-			res, err = GetJSONPathExp(`$["metadata"]["annotations"]["kubernetes.io[service-name]"]`,
+			res, err = GetJSONPathRaw(`$["metadata"]["annotations"]["kubernetes.io[service-name]"]`,
 				obj.UnstructuredContent())
 			Expect(err).NotTo(HaveOccurred())
 			s, err = AsString(res)

pkg/pipeline/aggregation.go

@@ -16,6 +16,7 @@ const aggregateOp = "@aggregate"
 // of objects in a view.
 type Aggregation struct {
 	*opv1a1.Aggregation
+	Stages []*Stage
 	engine Engine
 }
 
@@ -24,16 +25,24 @@ func NewAggregation(engine Engine, config *opv1a1.Aggregation) *Aggregation {
 	if config == nil {
 		return nil
 	}
-	return &Aggregation{
+
+	a := &Aggregation{
 		Aggregation: config,
+		Stages:      make([]*Stage, len(config.Expressions)),
 		engine:      engine,
 	}
+
+	for i, e := range config.Expressions {
+		a.Stages[i] = NewStage(engine, &e)
+	}
+
+	return a
 }
 
 func (a *Aggregation) String() string {
 	ss := []string{}
-	for _, e := range a.Expressions {
-		ss = append(ss, e.String())
+	for _, s := range a.Stages {
+		ss = append(ss, s.String())
 	}
 	return fmt.Sprintf("%s:[%s]", aggregateOp, strings.Join(ss, ","))
 }