Add Filter query configuration to Continuous Benchmark Job

charts/vald-benchmark-operator/crds/valdbenchmarkjob.yaml

@@ -328,11 +328,11 @@ spec:
                 object_config:
                   type: object
                   properties:
-                    filter_config:
-                      type: object
-                      properties:
-                        host:
-                          type: string
+                    filter_configs:
+                      type: array
+                      items:
+                        type: object
+                        x-kubernetes-preserve-unknown-fields: true
                 remove_config:
                   type: object
                   properties:

charts/vald-benchmark-operator/schemas/job-values.yaml

@@ -140,15 +140,14 @@ remove_config:
 # @schema {"name": "object_config", "type": "object"}
 # object_config -- object config
 object_config:
-  # @schema {"name": "object_config.filter_config", "type": "object"}
-  # object_config.filter_config -- filter target config
-  filter_config:
-    # @schema {"name": "object_config.filter_config.host", "type": "string"}
-    # object_config.filter_config.host -- filter target host
-    host: 0.0.0.0
-    # @schema {"name": "object_config.filter_config.host", "type": "integer"}
-    # object_config.filter_config.port -- filter target host
-    port: 8081
+  # @schema {"name": "object_config.filter_configs", "type": "array", "items": {"type": "object"}}
+  # object_config.filter_configs -- filter configs
+  filter_configs:
+    - target:
+        host: 0.0.0.0
+        port: 8081
+      query:
+        query: ""
 # @schema {"name": "client_config", "type": "object"}
 # client_config -- gRPC client config for request to the Vald cluster
 client_config:

internal/config/benchmark.go

@@ -156,11 +156,15 @@
 
 // ObjectConfig defines the desired state of object config
 type ObjectConfig struct {
-	FilterConfig FilterConfig `json:"filter_config,omitempty" yaml:"filter_config"`
+	FilterConfigs []*FilterConfig `json:"filter_configs,omitempty" yaml:"filter_configs"`
 }
 
 func (cfg *ObjectConfig) Bind() *ObjectConfig {
-	cfg.FilterConfig = *cfg.FilterConfig.Bind()
+	for i := 0; i < len(cfg.FilterConfigs); i++ {
+		if cfg.FilterConfigs[i] != nil {
+			cfg.FilterConfigs[i] = cfg.FilterConfigs[i].Bind()
+		}
+	}
 	return cfg
 }
 
@@ -175,14 +179,33 @@
 	return cfg
 }
 
+// FilterQuery defines the query passed to filter target.
+type FilterQuery struct {
+	Query string `json:"query,omitempty" yaml:"query"`
+}
+
+func (cfg *FilterQuery) Bind() *FilterQuery {
+	cfg.Query = GetActualValue(cfg.Query)
+	return cfg
+}
+
 // FilterConfig defines the desired state of filter config
 type FilterConfig struct {
-	Targets []*FilterTarget `json:"target,omitempty" yaml:"target"`
+	Target *FilterTarget `json:"target,omitempty" yaml:"target"`
+	Query  *FilterQuery  `json:"query,omitempty"  yaml:"query"`
 }
 
 func (cfg *FilterConfig) Bind() *FilterConfig {
-	for i := 0; i < len(cfg.Targets); i++ {
-		cfg.Targets[i] = cfg.Targets[i].Bind()
+	if cfg.Target != nil {
+		cfg.Target = cfg.Target.Bind()
+	} else {
+		cfg.Target = (&FilterTarget{}).Bind()
+	}
+
+	if cfg.Query != nil {
+		cfg.Query = cfg.Query.Bind()
+	} else {
+		cfg.Query = (&FilterQuery{}).Bind()
 	}
 	return cfg
 }

pkg/tools/benchmark/job/service/object.go

@@ -80,12 +80,29 @@ func (j *job) getObject(ctx context.Context, ech chan error) error {
 	eg.SetLimit(j.concurrencyLimit)
 	for i := j.dataset.Range.Start; i <= j.dataset.Range.End; i++ {
 		log.Infof("[benchmark job] Start get object: iter = %d", i)
-		ft := []*payload.Filter_Target{}
+		fcfgs := []*payload.Filter_Config{}
 		if j.objectConfig != nil {
-			for i, target := range j.objectConfig.FilterConfig.Targets {
-				ft[i] = &payload.Filter_Target{
-					Host: target.Host,
-					Port: uint32(target.Port),
+			for _, cfg := range j.objectConfig.FilterConfigs {
+				if cfg != nil {
+					var (
+						target *payload.Filter_Target
+						query  *payload.Filter_Query
+					)
+					if cfg.Target != nil {
+						target = &payload.Filter_Target{
+							Host: cfg.Target.Host,
+							Port: uint32(cfg.Target.Port),
+						}
+					}
+					if cfg.Query != nil {
+						query = &payload.Filter_Query{
+							Query: cfg.Query.Query,
+						}
+					}
+					fcfgs = append(fcfgs, &payload.Filter_Config{
+						Target: target,
+						Query:  query,
+					})
 				}
 			}
 		}
@@ -108,9 +125,7 @@ func (j *job) getObject(ctx context.Context, ech chan error) error {
 				Id: &payload.Object_ID{
 					Id: strconv.Itoa(idx),
 				},
-				Filters: &payload.Filter_Config{
-					Targets: ft,
-				},
+				Filters: fcfgs,
 			})
 			if err != nil {
 				select {