Fix e2e for read replica and add e2e for index operator

.github/actions/detect-docker-image-tags/action.yaml

@@ -23,7 +23,7 @@ inputs:
   images:
     description: "Image names"
     required: false
-    default: "vdaas/vald-agent-ngt vdaas/vald-agent-faiss vdaas/vald-agent-sidecar vdaas/vald-discoverer-k8s vdaas/vald-lb-gateway vdaas/vald-filter-gateway vdaas/vald-mirror-gateway vdaas/vald-manager-index"
+    default: "vdaas/vald-agent-ngt vdaas/vald-agent-faiss vdaas/vald-agent-sidecar vdaas/vald-discoverer-k8s vdaas/vald-lb-gateway vdaas/vald-filter-gateway vdaas/vald-mirror-gateway vdaas/vald-manager-index vdaas/vald-index-operator vdaas/vald-readreplica-rotate"
 outputs:
   HELM_EXTRA_OPTIONS:
     description: "Helm extra options that specifies image tags"
@@ -49,8 +49,10 @@ runs:
           ["vdaas/vald-manager-index"]="manager.index.image.tag"
           ["vdaas/vald-index-creation"]="manager.index.creator.image.tag"
           ["vdaas/vald-index-save"]="manager.index.saver.image.tag"
+          ["vdaas/vald-readreplica-rotate"]="manager.index.readreplica.rotator.image.tag"
           ["vdaas/vald-helm-operator"]="image.tag"
           ["vdaas/vald-ci-container"]=""
+          ["vdaas/vald-index-operator"]="manager.index.operator.image.tag"
         )
 
         for image in ${IMAGES}; do

.github/actions/setup-e2e/action.yaml

@@ -43,7 +43,7 @@ inputs:
   target_images:
     description: "Image names"
     required: false
-    default: "vdaas/vald-agent-ngt vdaas/vald-agent-faiss vdaas/vald-agent-sidecar vdaas/vald-discoverer-k8s vdaas/vald-lb-gateway vdaas/vald-filter-gateway vdaas/vald-mirror-gateway vdaas/vald-manager-index"
+    default: "vdaas/vald-agent-ngt vdaas/vald-agent-faiss vdaas/vald-agent-sidecar vdaas/vald-discoverer-k8s vdaas/vald-lb-gateway vdaas/vald-filter-gateway vdaas/vald-mirror-gateway vdaas/vald-manager-index vdaas/vald-index-operator vdaas/vald-readreplica-rotate"
 outputs:
   HELM_EXTRA_OPTIONS:
     description: "Helm extra options that specifies E2E target image tags"

.github/actions/wait-for-docker-image/action.yaml

@@ -19,7 +19,7 @@ inputs:
   images:
     description: "image names"
     required: false
-    default: "vdaas/vald-agent-ngt vdaas/vald-agent-faiss vdaas/vald-agent-sidecar vdaas/vald-discoverer-k8s vdaas/vald-lb-gateway vdaas/vald-filter-gateway vdaas/vald-mirror-gateway vdaas/vald-manager-index"
+    default: "vdaas/vald-agent-ngt vdaas/vald-agent-faiss vdaas/vald-agent-sidecar vdaas/vald-discoverer-k8s vdaas/vald-lb-gateway vdaas/vald-filter-gateway vdaas/vald-mirror-gateway vdaas/vald-manager-index vdaas/vald-index-operator vdaas/vald-readreplica-rotate"
 outputs: {}
 runs:
   using: "composite"

.github/helm/values/values-readreplica.yaml

@@ -48,6 +48,7 @@ agent:
     dimension: 784
     index_path: /var/ngt/index
     enable_in_memory_mode: false
+    enable_export_index_info_to_k8s: true
   persistentVolume:
     enabled: true
     accessMode: ReadWriteOnce
@@ -80,3 +81,6 @@ manager:
     readreplica:
       rotator:
         enabled: true
+    operator:
+      enabled: true
+      rotation_job_concurrency: 10

.github/workflows/e2e.yml

@@ -279,9 +279,6 @@ jobs:
   e2e-stream-crud-with-readreplica:
     name: "E2E test (Stream CRUD) with read replica"
     needs: [dump-contexts-to-log]
-    # FIXME: This job is disabled because it is not working properly for a moment.
-    #        Needs to fix TestE2EReadReplica not to use CronJob since there is no CronJob for read replica anymore.
-    if: false
     runs-on: ubuntu-latest
     timeout-minutes: 60
     steps:

tests/e2e/crud/crud_test.go

@@ -25,7 +25,6 @@ import (
 	"fmt"
 	"os"
 	"os/exec"
-	"strconv"
 	"testing"
 	"time"
 
@@ -834,6 +833,7 @@ func TestE2EIndexJobCorrection(t *testing.T) {
 	}
 }
 
+// TestE2EReadReplica tests that search requests succeed with read replica resources.
 func TestE2EReadReplica(t *testing.T) {
 	t.Cleanup(teardown)
 
@@ -861,29 +861,15 @@ func TestE2EReadReplica(t *testing.T) {
 
 	sleep(t, waitAfterInsertDuration)
 
-	t.Log("starting to restart all the agent pods to make it backup index to pvc...")
-	if err := kubectl.RolloutResource(ctx, t, "statefulsets/vald-agent"); err != nil {
-		t.Fatalf("failed to restart all the agent pods: %s", err)
-	}
-
-	t.Log("starting to create read replica rotators...")
-	pods, err := kubeClient.GetPods(ctx, namespace, "app=vald-agent")
-	if err != nil {
-		t.Fatalf("GetPods failed: %s", err)
-	}
-	cronJobs, err := kubeClient.ListCronJob(ctx, namespace, "app=vald-readreplica-rotate")
-	if err != nil {
-		t.Fatalf("ListCronJob failed: %s", err)
-	}
-	cronJob := cronJobs[0]
-	for id := 0; id < len(pods); id++ {
-		// the annotation key comes from `manager.index.readreplica.rotator.target_read_replica_id_annotations_key`
-		cronJob.Spec.JobTemplate.Spec.Template.GetObjectMeta().SetAnnotations(map[string]string{"vald.vdaas.org/target-read-replica-id": strconv.Itoa(id)})
-		kubeClient.CreateJobFromCronJob(ctx, "vald-readreplica-rotate-"+strconv.Itoa(id), namespace, &cronJob)
-	}
-
+	t.Log("index operator should be creating read replica rotator jobs")
 	t.Log("waiting for read replica rotator jobs to complete...")
-	if err := kubectl.WaitResources(ctx, t, "job", "app=vald-readreplica-rotate", "complete", "120s"); err != nil {
+	if err := kubectl.WaitResources(ctx, t, "job", "app=vald-readreplica-rotate", "complete", "60s"); err != nil {
+		t.Log("wait failed. printing yaml of vald-readreplica-rotate")
+		kubectl.KubectlCmd(ctx, t, "get", "pod", "-l", "app=vald-readreplica-rotate", "-oyaml")
+		t.Log("wait failed. printing log of vald-index-operator")
+		kubectl.DebugLog(ctx, t, "app=vald-index-operator")
+		t.Log("wait failed. printing log of vald-readreplica-rotate")
+		kubectl.DebugLog(ctx, t, "app=vald-readreplica-rotate")
 		t.Fatalf("failed to wait for read replica rotator jobs to complete: %s", err)
 	}
 

tests/e2e/kubernetes/kubectl/kubectl.go

@@ -48,6 +48,18 @@ func WaitResources(ctx context.Context, t *testing.T, resource, labelSelector, c
 	return runCmd(t, cmd)
 }
 
+func DebugLog(ctx context.Context, t *testing.T, label string) error {
+	t.Helper()
+	cmd := exec.CommandContext(ctx, "kubectl", "logs", "-l", label, "--tail=-1")
+	return runCmd(t, cmd)
+}
+
+func KubectlCmd(ctx context.Context, t *testing.T, subcmds ...string) error {
+	t.Helper()
+	cmd := exec.CommandContext(ctx, "kubectl", subcmds...)
+	return runCmd(t, cmd)
+}
+
 func runCmd(t *testing.T, cmd *exec.Cmd) error {
 	t.Helper()
 	out, err := cmd.Output()