Revert "Add verification for index correction e2e and add clusterrole…

… cronjobs for operator to deploy index correction (#2205)"

This reverts commit b4028e7.

.github/helm/values/values-lb.yaml

@@ -74,4 +74,3 @@ manager:
       # suspend because you do not want corrector to start automatically in CI
       # instead run it manually
       suspend: true
-      schedule: "1 2 3 4 5"

.github/valdrelease/valdrelease.yaml

@@ -76,7 +76,3 @@ spec:
         auto_index_duration_limit: 2m
         auto_index_check_duration: 30s
         auto_index_length: 1000
-      corrector:
-        enabled: true
-        suspend: true
-        schedule: "1 2 3 4 5"

charts/vald-helm-operator/templates/clusterrole.yaml

@@ -175,17 +175,4 @@ rules:
       - get
       - patch
       - update
-  - apiGroups:
-      - batch
-    resources:
-      - cronjobs
-    verbs:
-      - create
-      - delete
-      - get
-      - list
-      - patch
-      - update
-      - watch
-
 {{- end }}

charts/vald/values.yaml

@@ -2682,7 +2682,7 @@ manager:
       enabled: false
       # @schema {"name": "manager.index.corrector.schedule", "type": "string"}
       # manager.index.corrector.schedule -- CronJob schedule setting for index correction
-      schedule: "6 3 * * *"
+      schedule: "5 * * * *"
       # @schema {"name": "manager.index.corrector.suspend", "type": "boolean"}
       # manager.index.corrector.suspend -- CronJob suspend setting for index correction
       suspend: false

k8s/operator/helm/clusterrole.yaml

@@ -175,15 +175,3 @@ rules:
       - get
       - patch
       - update
-  - apiGroups:
-      - batch
-    resources:
-      - cronjobs
-    verbs:
-      - create
-      - delete
-      - get
-      - list
-      - patch
-      - update
-      - watch

tests/e2e/crud/crud_test.go

@@ -24,7 +24,6 @@ import (
 	"flag"
 	"fmt"
 	"os"
-	"os/exec"
 	"testing"
 	"time"
 
@@ -43,14 +42,13 @@ var (
 	port int
 	ds   *hdf5.Dataset
 
-	insertNum           int
-	correctionInsertNum int
-	searchNum           int
-	searchByIDNum       int
-	getObjectNum        int
-	updateNum           int
-	upsertNum           int
-	removeNum           int
+	insertNum     int
+	searchNum     int
+	searchByIDNum int
+	getObjectNum  int
+	updateNum     int
+	upsertNum     int
+	removeNum     int
 
 	insertFrom     int
 	searchFrom     int
@@ -75,7 +73,6 @@ func init() {
 	flag.IntVar(&port, "port", 8081, "gRPC port")
 
 	flag.IntVar(&insertNum, "insert-num", 10000, "number of id-vector pairs used for insert")
-	flag.IntVar(&correctionInsertNum, "correction-insert-num", 3000, "number of id-vector pairs used for insert")
 	flag.IntVar(&searchNum, "search-num", 10000, "number of id-vector pairs used for search")
 	flag.IntVar(&searchByIDNum, "search-by-id-num", 100, "number of id-vector pairs used for search-by-id")
 	flag.IntVar(&getObjectNum, "get-object-num", 100, "number of id-vector pairs used for get-object")
@@ -761,9 +758,7 @@ func TestE2EIndexJobCorrection(t *testing.T) {
 		t.Fatalf("an error occurred: %s", err)
 	}
 
-	// prepare train data
-	train := ds.Train[insertFrom : insertFrom+correctionInsertNum]
-
+	train := ds.Train[insertFrom : insertFrom+insertNum]
 	err = op.Insert(t, ctx, operation.Dataset{
 		Train: train,
 	})
@@ -773,49 +768,12 @@ func TestE2EIndexJobCorrection(t *testing.T) {
 
 	sleep(t, waitAfterInsertDuration)
 
-	t.Log("Test case 1: just execute index correction and check if replica number is correct after correction")
 	exe := operation.NewCronJobExecutor("vald-index-correction")
 	err = exe.CreateAndWait(t, ctx, "correction-test")
 	if err != nil {
 		t.Fatalf("an error occurred: %s", err)
 	}
 
-	// check if replica number is correct
-	err = op.StreamListObject(t, ctx, operation.Dataset{
-		Train: train,
-	})
-	if err != nil {
-		t.Fatalf("an error occurred: %s", err)
-	}
-
-	t.Log("Test case 2: execute index correction after one agent removed")
-	t.Log("removing vald-agent-ngt-0...")
-	cmd := exec.CommandContext(ctx, "sh", "-c", "kubectl delete pod vald-agent-ngt-0 && kubectl wait --for=condition=Ready pod/vald-agent-ngt-0")
-	out, err := cmd.Output()
-	if err != nil {
-		if exitErr, ok := err.(*exec.ExitError); ok {
-			t.Fatalf("%s, %s, %v", string(out), string(exitErr.Stderr), err)
-		} else {
-			t.Fatalf("unexpected error on creating job: %v", err)
-		}
-	}
-	t.Log(string(out))
-
-	// correct the deleted index
-	err = exe.CreateAndWait(t, ctx, "correction-test")
-	if err != nil {
-		t.Fatalf("an error occurred: %s", err)
-	}
-
-	// check if replica number is correct
-	err = op.StreamListObject(t, ctx, operation.Dataset{
-		Train: train,
-	})
-	if err != nil {
-		t.Fatalf("an error occurred: %s", err)
-	}
-
-	t.Log("Tear down. Removing all vectors...")
 	err = op.Remove(t, ctx, operation.Dataset{
 		Train: train,
 	})

tests/e2e/operation/job.go

@@ -56,8 +56,8 @@ func deleteJob(t *testing.T, jobName string) error {
 func waitJob(t *testing.T, ctx context.Context, jobName string) error {
 	t.Helper()
 	t.Log("waiting for the correction job to complete or fail")
-	waitCompleteCmd := fmt.Sprintf("kubectl wait --timeout=10m job/%s --for=condition=complete", jobName)
-	waitFailedCmd := fmt.Sprintf("kubectl wait --timeout=10m job/%s --for=condition=failed", jobName)
+	waitCompleteCmd := fmt.Sprintf("kubectl wait --timeout=-1s job/%s --for=condition=complete", jobName)
+	waitFailedCmd := fmt.Sprintf("kubectl wait --timeout=-1s job/%s --for=condition=failed", jobName)
 
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()

tests/e2e/operation/stream.go

@@ -1224,7 +1224,5 @@ exit_loop:
 			return fmt.Errorf("the number of vectors returned is different at index id %v: got %v, want %v", k, v, replica)
 		}
 	}
-
-	t.Log("StreamListObject operation finished successfully and all vectors are returned with correct replica number")
 	return nil
 }