✅ add benchmark operator reconcile test

Signed-off-by: vankichi <[email protected]>

internal/k8s/job/job.go

@@ -48,6 +48,9 @@ type reconciler struct {
 // Job is a type alias for the k8s job definition.
 type Job = batchv1.Job
 
+// JobStatus is a type alias for the k8s job status definition.
+type JobStatus = batchv1.JobStatus
+
 // New returns the JobWatcher that implements reconciliation loop, or any errors occurred.
 func New(opts ...Option) (JobWatcher, error) {
 	r := &reconciler{

internal/test/mock/controller_runtime.go

@@ -0,0 +1,67 @@
+// Copyright (C) 2019-2023 vdaas.org vald team <[email protected]>
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// You may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package mock
+
+import (
+	"context"
+
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+)
+
+type MockSubResourceWriter struct {
+	client.SubResourceWriter
+}
+
+func (s *MockSubResourceWriter) Update(context.Context, client.Object, ...client.SubResourceUpdateOption) error {
+	return nil
+}
+
+type MockClient struct {
+	client.Client
+}
+
+func (*MockClient) Status() client.SubResourceWriter {
+	s := MockSubResourceWriter{
+		SubResourceWriter: &MockSubResourceWriter{},
+	}
+	return s.SubResourceWriter
+}
+
+func (*MockClient) Get(context.Context, client.ObjectKey, client.Object, ...client.GetOption) error {
+	return nil
+}
+
+func (*MockClient) Create(context.Context, client.Object, ...client.CreateOption) error {
+	return nil
+}
+
+func (*MockClient) Delete(context.Context, client.Object, ...client.DeleteOption) error {
+	return nil
+}
+
+func (*MockClient) DeleteAllOf(context.Context, client.Object, ...client.DeleteAllOfOption) error {
+	return nil
+}
+
+type MockManager struct {
+	manager.Manager
+}
+
+func (m *MockManager) GetClient() client.Client {
+	c := &MockClient{
+		Client: &MockClient{},
+	}
+	return c.Client
+}

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

@@ -56,6 +56,10 @@ func (j *job) insert(ctx context.Context, ech chan error) error {
 				case ech <- err:
 				}
 			}
+			// loopCnt represents the quotient of iter divided by the len(vecs).
+			// This is to account for when iter exceeds len(vecs).
+			// It is used to calculate idx to determine which index of vecs to access.
+			// idx takes between <0, len(vecs)-1>.
 			loopCnt := math.Floor(float64(iter-1) / float64(len(vecs)))
 			idx := iter - 1 - (len(vecs) * int(loopCnt))
 			res, err := j.client.Insert(egctx, &payload.Insert_Request{
@@ -71,6 +75,9 @@ func (j *job) insert(ctx context.Context, ech chan error) error {
 					log.Errorf("[benchmark job] context error is detected: %s\t%s", err.Error(), egctx.Err())
 					return errors.Join(err, egctx.Err())
 				default:
+					// TODO: count up error for observe benchmark job
+					// We should wait for refactoring internal/o11y.
+					log.Errorf("[benchmark job] err: %s", err.Error())
 				}
 			}
 			// TODO: send metrics to the Prometeus

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

@@ -254,6 +254,7 @@ func (j *job) Start(ctx context.Context) (<-chan error, error) {
 	cech, err := j.client.Start(ctx)
 	if err != nil {
 		log.Error("[benchmark job] failed to start connection monitor")
+		close(ech)
 		return nil, err
 	}
 	j.eg.Go(func() error {
@@ -280,6 +281,7 @@ func (j *job) Start(ctx context.Context) (<-chan error, error) {
 				case ech <- err:
 				}
 			}
+			close(ech)
 			if err := p.Signal(syscall.SIGTERM); err != nil {
 				log.Error(err)
 			}

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

@@ -56,6 +56,9 @@ func (j *job) exists(ctx context.Context, ech chan error) error {
 					log.Errorf("[benchmark job] context error is detected: %s\t%s", err.Error(), egctx.Err())
 					return nil
 				default:
+					// TODO: count up error for observe benchmark job
+					// We should wait for refactoring internal/o11y.
+					log.Errorf("[benchmark job] err: %s", err.Error())
 				}
 			}
 			log.Debugf("[benchmark job] Finish exists: iter= %d \n%v\n", idx, res)
@@ -115,6 +118,9 @@ func (j *job) getObject(ctx context.Context, ech chan error) error {
 					log.Errorf("[benchmark job] context error is detected: %s\t%s", err.Error(), egctx.Err())
 					return nil
 				default:
+					// TODO: count up error for observe benchmark job
+					// We should wait for refactoring internal/o11y.
+					log.Errorf("[benchmark job] err: %s", err.Error())
 				}
 			}
 			if res != nil {

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

@@ -65,6 +65,9 @@ func (j *job) remove(ctx context.Context, ech chan error) error {
 					log.Errorf("[benchmark job] context error is detected: %s\t%s", err.Error(), egctx.Err())
 					return errors.Join(err, egctx.Err())
 				default:
+					// TODO: count up error for observe benchmark job
+					// We should wait for refactoring internal/o11y.
+					log.Errorf("[benchmark job] err: %s", err.Error())
 				}
 			}
 			log.Debugf("[benchmark job] Finish remove: iter= %d \n%v", idx, res)

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

@@ -65,6 +65,10 @@ func (j *job) search(ctx context.Context, ech chan error) error {
 				case ech <- err:
 				}
 			}
+			// loopCnt represents the quotient of iter divided by the len(vecs).
+			// This is to account for when iter exceeds len(vecs).
+			// It is used to calculate idx to determine which index of vecs to access.
+			// idx takes between <0, len(vecs)-1>.
 			loopCnt := math.Floor(float64(iter-1) / float64(len(vecs)))
 			idx := iter - 1 - (len(vecs) * int(loopCnt))
 			if len(vecs[idx]) != j.dimension {
@@ -81,6 +85,9 @@ func (j *job) search(ctx context.Context, ech chan error) error {
 					log.Errorf("[benchmark job] context error is detected: %s\t%s", err.Error(), egctx.Err())
 					return nil
 				default:
+					// TODO: count up error for observe benchmark job
+					// We should wait for refactoring internal/o11y.
+					log.Errorf("[benchmark job] err: %s", err.Error())
 				}
 			}
 			if res != nil && j.searchConfig.EnableLinearSearch {
@@ -116,6 +123,10 @@ func (j *job) search(ctx context.Context, ech chan error) error {
 					}
 				}
 				log.Debugf("[benchmark job] Start linear search: iter = %d", iter)
+				// loopCnt represents the quotient of iter divided by the len(vecs).
+				// This is to account for when iter exceeds len(vecs).
+				// It is used to calculate idx to determine which index of vecs to access.
+				// idx takes between <0, len(vecs)-1>.
 				loopCnt := math.Floor(float64(i-1) / float64(len(vecs)))
 				idx := iter - 1 - (len(vecs) * int(loopCnt))
 				if len(vecs[idx]) != j.dimension {
@@ -132,6 +143,9 @@ func (j *job) search(ctx context.Context, ech chan error) error {
 						log.Errorf("[benchmark job] context error is detected: %s\t%s", err.Error(), egctx.Err())
 						return errors.Join(err, egctx.Err())
 					default:
+						// TODO: count up error for observe benchmark job
+						// We should wait for refactoring internal/o11y.
+						log.Errorf("[benchmark job] err: %s", err.Error())
 					}
 				}
 				if res != nil {

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

@@ -57,6 +57,10 @@ func (j *job) update(ctx context.Context, ech chan error) error {
 				case ech <- err:
 				}
 			}
+			// loopCnt represents the quotient of iter divided by the len(vecs).
+			// This is to account for when iter exceeds len(vecs).
+			// It is used to calculate idx to determine which index of vecs to access.
+			// idx takes between <0, len(vecs)-1>.
 			loopCnt := math.Floor(float64(iter-1) / float64(len(vecs)))
 			idx := iter - 1 - (len(vecs) * int(loopCnt))
 			res, err := j.client.Update(egctx, &payload.Update_Request{
@@ -72,6 +76,9 @@ func (j *job) update(ctx context.Context, ech chan error) error {
 					log.Errorf("[benchmark job] context error is detected: %s\t%s", err.Error(), egctx.Err())
 					return errors.Join(err, egctx.Err())
 				default:
+					// TODO: count up error for observe benchmark job
+					// We should wait for refactoring internal/o11y.
+					log.Errorf("[benchmark job] err: %s", err.Error())
 				}
 			}
 			if res != nil {

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

@@ -57,6 +57,10 @@ func (j *job) upsert(ctx context.Context, ech chan error) error {
 				case ech <- err:
 				}
 			}
+			// loopCnt represents the quotient of iter divided by the len(vecs).
+			// This is to account for when iter exceeds len(vecs).
+			// It is used to calculate idx to determine which index of vecs to access.
+			// idx takes between <0, len(vecs)-1>.
 			loopCnt := math.Floor(float64(iter-1) / float64(len(vecs)))
 			idx := iter - 1 - (len(vecs) * int(loopCnt))
 			res, err := j.client.Upsert(egctx, &payload.Upsert_Request{
@@ -72,6 +76,9 @@ func (j *job) upsert(ctx context.Context, ech chan error) error {
 					log.Errorf("[benchmark job] context error is detected: %s\t%s", err.Error(), egctx.Err())
 					return errors.Join(err, egctx.Err())
 				default:
+					// TODO: count up error for observe benchmark job
+					// We should wait for refactoring internal/o11y.
+					log.Errorf("[benchmark job] err: %s", err.Error())
 				}
 			}
 			if res != nil {

pkg/tools/benchmark/operator/service/operator.go

@@ -57,9 +57,9 @@ type operator struct {
 	jobNamespace       string
 	jobImage           string
 	jobImagePullPolicy string
-	scenarios          atomic.Pointer[map[string]*scenario]
-	benchjobs          atomic.Pointer[map[string]*v1.ValdBenchmarkJob]
-	jobs               atomic.Pointer[map[string]string]
+	scenarios          *atomic.Pointer[map[string]*scenario]
+	benchjobs          *atomic.Pointer[map[string]*v1.ValdBenchmarkJob]
+	jobs               *atomic.Pointer[map[string]string]
 	rcd                time.Duration // reconcile check duration
 	eg                 errgroup.Group
 	ctrl               k8s.Controller
@@ -137,20 +137,32 @@ func (o *operator) initCtrl() error {
 }
 
 func (o *operator) getAtomicScenario() map[string]*scenario {
+	if o.scenarios == nil {
+		o.scenarios = &atomic.Pointer[map[string]*scenario]{}
+		return nil
+	}
 	if v := o.scenarios.Load(); v != nil {
 		return *(v)
 	}
 	return nil
 }
 
 func (o *operator) getAtomicBenchJob() map[string]*v1.ValdBenchmarkJob {
+	if o.benchjobs == nil {
+		o.benchjobs = &atomic.Pointer[map[string]*v1.ValdBenchmarkJob]{}
+		return nil
+	}
 	if v := o.benchjobs.Load(); v != nil {
 		return *(v)
 	}
 	return nil
 }
 
 func (o *operator) getAtomicJob() map[string]string {
+	if o.jobs == nil {
+		o.jobs = &atomic.Pointer[map[string]string]{}
+		return nil
+	}
 	if v := o.jobs.Load(); v != nil {
 		return *(v)
 	}
@@ -171,7 +183,7 @@ func (o *operator) jobReconcile(ctx context.Context, jobList map[string][]job.Jo
 	if cjobs == nil {
 		cjobs = map[string]string{}
 	}
-	// jobStatus is used for update benchmark job resource status
+	// benchmarkJobStatus is used for update benchmark job resource status
 	benchmarkJobStatus := make(map[string]v1.BenchmarkJobStatus)
 	// jobNames is used for check whether cjobs has delted job.
 	// If cjobs has the delted job, it will be remove the end of jobReconcile function.
@@ -215,7 +227,7 @@ func (o *operator) jobReconcile(ctx context.Context, jobList map[string][]job.Jo
 	log.Debug("[reconcile job] finish")
 }
 
-// benchmarkJobReconcile gets the vald benchmark job resource list and create Job for running benchmark job.
+// benchJobReconcile gets the vald benchmark job resource list and create Job for running benchmark job.
 func (o *operator) benchJobReconcile(ctx context.Context, benchJobList map[string]v1.ValdBenchmarkJob) {
 	log.Debugf("[reconcile benchmark job resource] job list: %#v", benchJobList)
 	if len(benchJobList) == 0 {
@@ -228,6 +240,7 @@ func (o *operator) benchJobReconcile(ctx context.Context, benchJobList map[strin
 	if cbjl == nil {
 		cbjl = make(map[string]*v1.ValdBenchmarkJob, 0)
 	}
+	// jobStatus is used for update benchmarkJob CR status if updating is needed.
 	jobStatus := make(map[string]v1.BenchmarkJobStatus)
 	for k := range benchJobList {
 		// update scenario status
@@ -475,7 +488,7 @@ func (o *operator) createJob(ctx context.Context, bjr v1.ValdBenchmarkJob) error
 		benchjob.WithTTLSecondsAfterFinished(int32(bjr.Spec.TTLSecondsAfterFinished)),
 	)
 	if err != nil {
-		return nil
+		return err
 	}
 	// create job
 	c := o.ctrl.GetManager().GetClient()