✨ Add search result aggregation option

Signed-off-by: vankichi <[email protected]>
vdaas · Jun 14, 2023 · fd47989 · fd47989
1 parent 3a0d522
commit fd47989
Show file tree

Hide file tree

Showing 13 changed files with 55 additions and 17 deletions.
diff --git a/charts/vald-benchmark-operator/crds/valdbenchmarkjob.yaml b/charts/vald-benchmark-operator/crds/valdbenchmarkjob.yaml
@@ -325,6 +325,14 @@ spec:
                 search_config:
                   type: object
                   properties:
+                    aggregation_algorithm:
+                      type: string
+                      enum:
+                        - Unknown
+                        - ConcurrentQueue
+                        - SortSlice
+                        - SortPoolSlice
+                        - PairingHeap
                     enable_linear_search:
                       type: boolean
                     epsilon:

diff --git a/charts/vald-benchmark-operator/job-values.schema.json b/charts/vald-benchmark-operator/job-values.schema.json
@@ -409,6 +409,17 @@
       "type": "object",
       "description": "upsert config",
       "properties": {
+        "aggregation_algorithm": {
+          "type": "string",
+          "description": "search result aggregation algorithm",
+          "enum": [
+            "Unknown",
+            "ConcurrentQueue",
+            "SortSlice",
+            "SortPoolSlice",
+            "PairingHeap"
+          ]
+        },
         "enable_linear_search": {
           "type": "boolean",
           "description": "enable linear search for calculation recall"

diff --git a/charts/vald-benchmark-operator/schemas/job-values.yaml b/charts/vald-benchmark-operator/schemas/job-values.yaml
@@ -114,6 +114,9 @@ search_config:
   # @schema {"name": "search_config.enable_linear_search", "type": "boolean"}
   # search_config.enable_linear_search -- enable linear search for calculation recall
   enable_linear_search: true
+  # @schema {"name": "search_config.aggregation_algorithm", "type": "string", "enum": ["Unknown", "ConcurrentQueue", "SortSlice", "SortPoolSlice", "PairingHeap"]}
+  # search_config.aggregation_algorithm -- search result aggregation algorithm
+  aggregation_algorithm: "Unknown"
 
 # @schema {"name": "remove_config", "type": "object"}
 # remove_config -- remove config

diff --git a/charts/vald-benchmark-operator/values.yaml b/charts/vald-benchmark-operator/values.yaml
@@ -46,7 +46,7 @@ image:
 job_image:
   # @schema {"name": "job_image.repository", "type": "string"}
   # image.repository -- job image repository
-  repository: vdaas/vald-benchmark-operator
+  repository: vdaas/vald-benchmark-job
   # @schema {"name": "job_image.tag", "type": "string"}
   # image.tag -- image tag for job docker image
   tag: v1.7.5

diff --git a/go.mod b/go.mod
@@ -553,6 +553,7 @@ require (
 	golang.org/x/sync v0.2.0
 	golang.org/x/sys v0.8.0
 	golang.org/x/text v0.9.0
+	golang.org/x/time v0.3.0
 	golang.org/x/tools v0.9.1
 	golang.org/x/time v0.3.0
 	gonum.org/v1/hdf5 v0.0.0-00010101000000-000000000000
@@ -656,7 +657,6 @@ require (
 	golang.org/x/image v0.7.0 // indirect
 	golang.org/x/mod v0.10.0 // indirect
 	golang.org/x/term v0.8.0 // indirect
-	golang.org/x/time v0.3.0 // indirect
 	golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 // indirect
 	gomodules.xyz/jsonpatch/v2 v2.3.0 // indirect
 	google.golang.org/api v0.125.0 // indirect

diff --git a/internal/config/benchmark.go b/internal/config/benchmark.go
@@ -127,16 +127,18 @@ func (cfg *UpsertConfig) Bind() *UpsertConfig {
 
 // SearchConfig defines the desired state of search config
 type SearchConfig struct {
-	Epsilon            float32 `json:"epsilon,omitempty"`
-	Radius             float32 `json:"radius,omitempty"`
-	Num                int32   `json:"num,omitempty"`
-	MinNum             int32   `json:"min_num,omitempty"`
-	Timeout            string  `json:"timeout,omitempty"`
-	EnableLinearSearch bool    `json:"enable_linear_search,omitempty"`
+	Epsilon              float32 `json:"epsilon,omitempty"`
+	Radius               float32 `json:"radius,omitempty"`
+	Num                  int32   `json:"num,omitempty"`
+	MinNum               int32   `json:"min_num,omitempty"`
+	Timeout              string  `json:"timeout,omitempty"`
+	EnableLinearSearch   bool    `json:"enable_linear_search,omitempty"`
+	AggregationAlgorithm string  `json:"aggregation_algorithm,omitempty"`
 }
 
 func (cfg *SearchConfig) Bind() *SearchConfig {
 	cfg.Timeout = GetActualValue(cfg.Timeout)
+	cfg.AggregationAlgorithm = GetActualValue(cfg.AggregationAlgorithm)
 	return cfg
 }
 

diff --git a/internal/k8s/job/job.go b/internal/k8s/job/job.go
@@ -30,7 +30,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
-	"sigs.k8s.io/controller-runtime/pkg/source"
 )
 
 // JobWatcher is a type alias for k8s resource controller.
@@ -153,7 +152,7 @@ func (r *reconciler) Owns() (client.Object, []builder.OwnsOption) {
 
 // Watches returns the kind of the job and the event handler.
 // It will always return nil.
-func (r *reconciler) Watches() (*source.Kind, handler.EventHandler, []builder.WatchesOption) {
+func (r *reconciler) Watches() (client.Object, handler.EventHandler, []builder.WatchesOption) {
 	// return &source.Kind{Type: new(corev1.Pod)}, &handler.EnqueueRequestForObject{}
 	return nil, nil, nil
 }
diff --git a/internal/k8s/vald/benchmark/job/job.go b/internal/k8s/vald/benchmark/job/job.go
@@ -31,7 +31,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/scheme"
-	"sigs.k8s.io/controller-runtime/pkg/source"
 )
 
 type BenchmarkJobWatcher k8s.ResourceController
@@ -134,7 +133,7 @@ func (r *reconciler) Owns() (client.Object, []builder.OwnsOption) {
 	return nil, nil
 }
 
-func (r *reconciler) Watches() (*source.Kind, handler.EventHandler, []builder.WatchesOption) {
+func (r *reconciler) Watches() (client.Object, handler.EventHandler, []builder.WatchesOption) {
 	// return &source.Kind{Type: new(corev1.Pod)}, &handler.EnqueueRequestForObject{}
 	return nil, nil, nil
 }
diff --git a/internal/k8s/vald/benchmark/scenario/scenario.go b/internal/k8s/vald/benchmark/scenario/scenario.go
@@ -29,7 +29,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
-	"sigs.k8s.io/controller-runtime/pkg/source"
 )
 
 type BenchmarkScenarioWatcher k8s.ResourceController
@@ -122,7 +121,8 @@ func (r *reconciler) Owns() (client.Object, []builder.OwnsOption) {
 	return nil, nil
 }
 
-func (r *reconciler) Watches() (*source.Kind, handler.EventHandler, []builder.WatchesOption) {
+func (r *reconciler) Watches() (client.Object, handler.EventHandler, []builder.WatchesOption) {
 	// return &source.Kind{Type: new(corev1.Pod)}, &handler.EnqueueRequestForObject{}
-	return &source.Kind{Type: new(v1.ValdBenchmarkScenario)}, &handler.EnqueueRequestForObject{}, nil
+	// return &source.Kind{Type: new(v1.ValdBenchmarkScenario)}, &handler.EnqueueRequestForObject{}, nil
+	return nil, &handler.EnqueueRequestForObject{}, nil
 }
diff --git a/k8s/tools/benchmark/operator/configmap.yaml b/k8s/tools/benchmark/operator/configmap.yaml
@@ -139,5 +139,5 @@ data:
         enabled: false
         sampling_rate: 1
     job_image:
-      image: "local-registry:5000/vdaas/vald-benchmark-job"
+      image: "vdaas/vald-benchmark-job:v1.7.5"
       pullPolicy: Always
diff --git a/k8s/tools/benchmark/operator/crds/valdbenchmarkjob.yaml b/k8s/tools/benchmark/operator/crds/valdbenchmarkjob.yaml
@@ -325,6 +325,14 @@ spec:
                 search_config:
                   type: object
                   properties:
+                    aggregation_algorithm:
+                      type: string
+                      enum:
+                        - Unknown
+                        - ConcurrentQueue
+                        - SortSlice
+                        - SortPoolSlice
+                        - PairingHeap
                     enable_linear_search:
                       type: boolean
                     epsilon:

diff --git a/k8s/tools/benchmark/operator/deployment.yaml b/k8s/tools/benchmark/operator/deployment.yaml
@@ -43,7 +43,7 @@ spec:
       serviceAccountName: vald-benchmark-operator
       containers:
         - name: vald-benchmark-operator
-          image: local-registry:5000/vdaas/vald-benchmark-operator
+          image: "vdaas/vald-benchmark-operator:v1.7.5"
           imagePullPolicy: Always
           livenessProbe:
             failureThreshold: 2

diff --git a/pkg/tools/benchmark/job/service/search.go b/pkg/tools/benchmark/job/service/search.go
@@ -37,6 +37,14 @@ func (j *job) search(ctx context.Context, ech chan error) error {
 		Radius:  float32(j.searchConfig.Radius),
 		Epsilon: float32(j.searchConfig.Epsilon),
 		Timeout: j.timeout.Nanoseconds(),
+		AggregationAlgorithm: func() payload.Search_AggregationAlgorithm {
+			if len(j.searchConfig.AggregationAlgorithm) > 0 {
+				if v, ok := payload.Search_AggregationAlgorithm_value[j.searchConfig.AggregationAlgorithm]; ok {
+					return payload.Search_AggregationAlgorithm(v)
+				}
+			}
+			return 0
+		}(),
 	}
 	sres := make([]*payload.Search_Response, len(vecs))
 	log.Infof("[benchmark job] Start search")