Refactor

charts/vald/templates/index/operator/configmap.yaml

@@ -45,4 +45,6 @@ data:
       agent_name: {{ $agent.name }}
       agent_namespace: {{ $agent.namespace }}
       concurrency: 1
+      read_replica_enabled: {{ $agent.readreplica.enabled }}
+      read_replica_label_key: {{ $agent.readreplica.label_key }}
 {{- end }}

cmd/index/operator/sample.yaml

@@ -72,6 +72,8 @@ operator:
   agent_name: "vald-agent"
   agent_namespace: "default"
   concurrency: 1
+  read_replica_enabled: true
+  read_replica_label_key: "vald-readreplica-id"
 observability:
   enabled: false
   otlp:

internal/config/index_operator.go

@@ -23,6 +23,12 @@ type IndexOperator struct {
 
 	// Concurrency represents indexing concurrency.
 	Concurrency int `json:"concurrency" yaml:"concurrency"`
+
+	// ReadReplicaEnabled represents whether read replica is enabled or not.
+	ReadReplicaEnabled bool `json:"read_replica_enabled" yaml:"read_replica_enabled"`
+
+	// ReadReplicaLabelKey represents the label key for read replica.
+	ReadReplicaLabelKey string `json:"read_replica_label_key" yaml:"read_replica_label_key"`
 }
 
 func (ic *IndexOperator) Bind() *IndexOperator {

pkg/index/operator/service/operator.go

@@ -21,6 +21,7 @@ import (
 
 	"github.com/vdaas/vald/internal/errors"
 	"github.com/vdaas/vald/internal/k8s"
+	"github.com/vdaas/vald/internal/k8s/client"
 	"github.com/vdaas/vald/internal/k8s/job"
 	"github.com/vdaas/vald/internal/k8s/pod"
 	"github.com/vdaas/vald/internal/k8s/vald"
@@ -44,9 +45,12 @@ type Operator interface {
 }
 
 type operator struct {
-	ctrl      k8s.Controller
-	eg        errgroup.Group
-	namespace string
+	ctrl                k8s.Controller
+	eg                  errgroup.Group
+	namespace           string
+	client              client.Client
+	readReplicaEnabled  bool
+	readReplicaLabelKey string
 }
 
 // New returns Indexer object if no error occurs.
@@ -95,6 +99,13 @@ func New(agentName string, opts ...Option) (o Operator, err error) {
 	if err != nil {
 		return nil, err
 	}
+
+	client, err := client.New()
+	if err != nil {
+		return nil, err
+	}
+	operator.client = client
+
 	return operator, nil
 }
 
@@ -131,29 +142,30 @@ func (o *operator) Start(ctx context.Context) (<-chan error, error) {
 
 // TODO: implement agent pod reconcile logic to detect conditions to start indexing and saving.
 func (o *operator) podOnReconcile(ctx context.Context, podList map[string][]pod.Pod) {
-	client := o.ctrl.GetManager().GetClient()
 	for k, v := range podList {
 		for _, pod := range v {
 			log.Debug("key", k, "name:", pod.Name, "annotations:", pod.Annotations)
 
 			// rotate read replica if needed
-			if err := o.rotateIfNeeded(ctx, client, pod); err != nil {
-				log.Error(err)
+			if o.readReplicaEnabled {
+				if err := o.rotateIfNeeded(ctx, pod); err != nil {
+					log.Error(err)
+				}
 			}
 		}
 	}
 }
 
 // TODO: implement job reconcile logic to detect save job completion and to start rotation.
 func (o *operator) jobOnReconcile(ctx context.Context, jobList map[string][]job.Job) {
-	for k, v := range jobList {
-		for _, job := range v {
-			log.Debug("key", k, "name:", job.Name, "status:", job.Status)
-		}
-	}
+	// for k, v := range jobList {
+	// 	for _, job := range v {
+	// 		log.Debug("key", k, "name:", job.Name, "status:", job.Status)
+	// 	}
+	// }
 }
 
-func (o *operator) rotateIfNeeded(ctx context.Context, client crclient.Client, pod pod.Pod) error {
+func (o *operator) rotateIfNeeded(ctx context.Context, pod pod.Pod) error {
 	t, ok := pod.Annotations[vald.LastTimeSaveIndexTimestampAnnotationsKey]
 	if !ok {
 		log.Info("the agent pod has not saved index yet. skipping...")
@@ -169,11 +181,10 @@ func (o *operator) rotateIfNeeded(ctx context.Context, client crclient.Client, p
 		log.Info("no index label found. the agent is not StatefulSet? skipping...")
 		return nil
 	}
-	// FIXME: get the key from config
-	label := crclient.MatchingLabels(map[string]string{"vald-readreplica-id": podIdx})
 
 	var depList appsv1.DeploymentList
-	if err := client.List(ctx, &depList, label); err != nil {
+	label := crclient.MatchingLabels(map[string]string{o.readReplicaLabelKey: podIdx})
+	if err := o.client.List(ctx, &depList, label); err != nil {
 		return err
 	}
 	if len(depList.Items) == 0 {
@@ -194,7 +205,7 @@ func (o *operator) rotateIfNeeded(ctx context.Context, client crclient.Client, p
 			return nil
 		}
 	}
-	log.Info("rotation required. creating rotator job...")
+	log.Infof("rotation required for agent id: %s. creating rotator job...", podIdx)
 	// TODO: check if the rotator job already exists or queued
 	return nil
 }

pkg/index/operator/service/options.go

@@ -30,3 +30,17 @@ func WithErrGroup(eg errgroup.Group) Option {
 		return nil
 	}
 }
+
+func WithReadReplicaEnabled(enabled bool) Option {
+	return func(o *operator) error {
+		o.readReplicaEnabled = enabled
+		return nil
+	}
+}
+
+func WithReadReplicaLabelKey(key string) Option {
+	return func(o *operator) error {
+		o.readReplicaLabelKey = key
+		return nil
+	}
+}

pkg/index/operator/usecase/operator.go

@@ -41,7 +41,11 @@ type run struct {
 // New returns Runner instance.
 func New(cfg *config.Data) (_ runner.Runner, err error) {
 	eg := errgroup.Get()
-	operator, err := service.New(cfg.Operator.AgentName)
+	operator, err := service.New(
+		cfg.Operator.AgentName,
+		service.WithReadReplicaEnabled(cfg.Operator.ReadReplicaEnabled),
+		service.WithReadReplicaLabelKey(cfg.Operator.ReadReplicaLabelKey),
+	)
 	if err != nil {
 		return nil, err
 	}