fix: Perform fields filtering server side (#4595)

Signed-off-by: Simon Behar <simbeh7@gmail.com>

pkg/apiclient/workflow/forwarder_overwrite.go

@@ -9,6 +9,4 @@ func init() {
 	forward_WorkflowService_WatchEvents_0 = http.StreamForwarder
 	forward_WorkflowService_PodLogs_0 = http.StreamForwarder
 	forward_WorkflowService_WorkflowLogs_0 = http.StreamForwarder
-	forward_WorkflowService_ListWorkflows_0 = http.UnaryForwarder
-	forward_WorkflowService_GetWorkflow_0 = http.UnaryForwarder
 }

server/workflow/workflow_server.go

@@ -19,6 +19,7 @@ import (
 	"github.com/argoproj/argo/pkg/client/clientset/versioned"
 	"github.com/argoproj/argo/server/auth"
 	argoutil "github.com/argoproj/argo/util"
+	"github.com/argoproj/argo/util/fields"
 	"github.com/argoproj/argo/util/instanceid"
 	"github.com/argoproj/argo/util/logs"
 	"github.com/argoproj/argo/workflow/common"
@@ -105,7 +106,23 @@ func (s *workflowServer) GetWorkflow(ctx context.Context, req *workflowpkg.Workf
 	if err != nil {
 		return nil, err
 	}
-	return wf, err
+	if req.Fields != "" {
+		wfBytes, err := json.Marshal(wf)
+		if err != nil {
+			return nil, err
+		}
+		resClean, err := fields.CleanFields(req.Fields, wfBytes)
+		if err != nil {
+			return nil, fmt.Errorf("unable to CleanFields in request: %w", err)
+		}
+		var newWf wfv1.Workflow
+		err = json.Unmarshal(resClean, &newWf)
+		if err != nil {
+			return nil, err
+		}
+		return &newWf, nil
+	}
+	return wf, nil
 }
 
 func (s *workflowServer) ListWorkflows(ctx context.Context, req *workflowpkg.WorkflowListRequest) (*wfv1.WorkflowList, error) {
@@ -139,7 +156,24 @@ func (s *workflowServer) ListWorkflows(ctx context.Context, req *workflowpkg.Wor
 	// we make no promises about the overall list sorting, we just sort each page
 	sort.Sort(wfList.Items)
 
-	return &wfv1.WorkflowList{ListMeta: metav1.ListMeta{Continue: wfList.Continue, ResourceVersion: wfList.ResourceVersion}, Items: wfList.Items}, nil
+	res := &wfv1.WorkflowList{ListMeta: metav1.ListMeta{Continue: wfList.Continue, ResourceVersion: wfList.ResourceVersion}, Items: wfList.Items}
+	if req.Fields != "" {
+		resBytes, err := json.Marshal(res)
+		if err != nil {
+			return nil, err
+		}
+		resClean, err := fields.CleanFields(req.Fields, resBytes)
+		if err != nil {
+			return nil, fmt.Errorf("unable to CleanFields in request: %w", err)
+		}
+		var newRes wfv1.WorkflowList
+		err = json.Unmarshal(resClean, &newRes)
+		if err != nil {
+			return nil, err
+		}
+		return &newRes, nil
+	}
+	return res, nil
 }
 
 func (s *workflowServer) WatchWorkflows(req *workflowpkg.WatchWorkflowsRequest, ws workflowpkg.WorkflowService_WatchWorkflowsServer) error {

util/fields/fields.go

@@ -0,0 +1,74 @@
+package fields
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+)
+
+func CleanFields(fieldsQuery string, dataBytes []byte) ([]byte, error) {
+	exclude := false
+	fields := make(map[string]interface{})
+	if fieldsQuery != "" {
+		if strings.HasPrefix(fieldsQuery, "-") {
+			fieldsQuery = fieldsQuery[1:]
+			exclude = true
+		}
+		for _, field := range strings.Split(fieldsQuery, ",") {
+			fields[field] = true
+		}
+	}
+
+	data := make(map[string]interface{})
+	err := json.Unmarshal(dataBytes, &data)
+	if err != nil {
+		return nil, err
+	}
+	err = processItem([]string{}, data, exclude, fields)
+	if err != nil {
+		return nil, err
+	}
+	clean, err := json.Marshal(data)
+	if err != nil {
+		return nil, err
+	}
+	return clean, nil
+}
+
+func processItem(path []string, item interface{}, exclude bool, fields map[string]interface{}) error {
+	if mapItem, ok := item.(map[string]interface{}); ok {
+		for k, v := range mapItem {
+
+			fieldPath := strings.Join(append(path, k), ".")
+			_, pathIn := fields[fieldPath]
+			parentPathIn := pathIn
+			if !parentPathIn {
+				for k := range fields {
+					if strings.HasPrefix(k, fieldPath) {
+						parentPathIn = true
+						break
+					}
+				}
+			}
+
+			if exclude && !pathIn || !exclude && parentPathIn {
+				if !pathIn {
+					if err := processItem(append(path, k), v, exclude, fields); err != nil {
+						return err
+					}
+				}
+			} else {
+				delete(mapItem, k)
+			}
+		}
+		return nil
+	} else if arrayItem, ok := item.([]interface{}); ok {
+		for i := range arrayItem {
+			if err := processItem(path, arrayItem[i], exclude, fields); err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+	return fmt.Errorf("cannot process item for fields, unknown format")
+}

util/fields/fields_test.go

@@ -0,0 +1,97 @@
+package fields
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"sigs.k8s.io/yaml"
+
+	"github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
+)
+
+var sampleWorkflow = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  name: hello-world-qgpxz
+spec:
+  arguments: {}
+  entrypoint: whalesay
+  templates:
+  - arguments: {}
+    container:
+      args:
+      - hello world
+      command:
+      - cowsay
+      image: docker/whalesay:latest
+      name: ""
+      resources: {}
+    inputs: {}
+    metadata: {}
+    name: whalesay
+    outputs: {}
+status:
+  artifactRepositoryRef:
+    configMap: artifact-repositories
+    key: default-v1
+    namespace: argo
+  conditions:
+  - status: "True"
+    type: Completed
+  finishedAt: "2020-12-01T17:30:51Z"
+  nodes:
+    hello-world-qgpxz:
+      displayName: hello-world-qgpxz
+  phase: Succeeded
+  progress: 1/1
+  resourcesDuration:
+    cpu: 3
+    memory: 1
+  startedAt: "2020-12-01T17:30:46Z"
+`
+
+func TestCleanFields(t *testing.T) {
+	var wf v1alpha1.Workflow
+	err := yaml.Unmarshal([]byte(sampleWorkflow), &wf)
+	assert.NoError(t, err)
+
+	jsonWf, err := json.Marshal(wf)
+	assert.NoError(t, err)
+
+	cleanJsonWf, err := CleanFields("status.phase,metadata.name,spec.entrypoint", jsonWf)
+	assert.NoError(t, err)
+
+	var cleanWf v1alpha1.Workflow
+	err = json.Unmarshal(cleanJsonWf, &cleanWf)
+	assert.NoError(t, err)
+
+	assert.Equal(t, "Succeeded", string(cleanWf.Status.Phase))
+	assert.Equal(t, "whalesay", cleanWf.Spec.Entrypoint)
+	assert.Equal(t, "hello-world-qgpxz", cleanWf.Name)
+
+	assert.Nil(t, cleanWf.Status.Nodes)
+}
+
+func TestCleanFieldsExclude(t *testing.T) {
+	var wf v1alpha1.Workflow
+	err := yaml.Unmarshal([]byte(sampleWorkflow), &wf)
+	assert.NoError(t, err)
+
+	jsonWf, err := json.Marshal(wf)
+	assert.NoError(t, err)
+
+	cleanJsonWf, err := CleanFields("-status.phase,metadata.name,spec.entrypoint", jsonWf)
+	assert.NoError(t, err)
+
+	var cleanWf v1alpha1.Workflow
+	err = json.Unmarshal(cleanJsonWf, &cleanWf)
+	assert.NoError(t, err)
+
+	assert.Nil(t, string(cleanWf.Status.Phase))
+	assert.Nil(t, cleanWf.Spec.Entrypoint)
+	assert.Nil(t, cleanWf.Name)
+
+	assert.NotNil(t, cleanWf.Status.Nodes)
+}

util/json/json.go

@@ -34,48 +34,3 @@ func (j *JSONMarshaler) NewEncoder(w io.Writer) gwruntime.Encoder {
 func (j *JSONMarshaler) Unmarshal(data []byte, v interface{}) error {
 	return json.Unmarshal(data, v)
 }
-
-// https://github.com/ksonnet/ksonnet/blob/master/pkg/kubecfg/diff.go
-func removeFields(config, live interface{}) interface{} {
-	switch c := config.(type) {
-	case map[string]interface{}:
-		return RemoveMapFields(c, live.(map[string]interface{}))
-	case []interface{}:
-		return removeListFields(c, live.([]interface{}))
-	default:
-		return live
-	}
-}
-
-// RemoveMapFields remove all non-existent fields in the live that don't exist in the config
-func RemoveMapFields(config, live map[string]interface{}) map[string]interface{} {
-	result := map[string]interface{}{}
-	for k, v1 := range config {
-		v2, ok := live[k]
-		if !ok {
-			continue
-		}
-		if v2 != nil {
-			v2 = removeFields(v1, v2)
-		}
-		result[k] = v2
-	}
-	return result
-}
-
-func removeListFields(config, live []interface{}) []interface{} {
-	// If live is longer than config, then the extra elements at the end of the
-	// list will be returned as-is so they appear in the diff.
-	result := make([]interface{}, 0, len(live))
-	for i, v2 := range live {
-		if len(config) > i {
-			if v2 != nil {
-				v2 = removeFields(config[i], v2)
-			}
-			result = append(result, v2)
-		} else {
-			result = append(result, v2)
-		}
-	}
-	return result
-}