[WIP] Feature/readreplica/separate helm

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

@@ -187,5 +187,25 @@ rules:
       - patch
       - update
       - watch
-
+  - apiGroups:
+      - snapshot.storage.k8s.io
+    resources:
+      - volumesnapshots
+    verbs:
+      - get
+      - list
+      - watch
+      - create
+      - delete
+      - patch
+  - apiGroups:
+      - ""
+    resources:
+      - persistentvolumeclaims
+    verbs:
+      - get
+      - list
+      - watch
+      - create
+      - delete
 {{- end }}

charts/vald-readreplica/.helmignore

@@ -0,0 +1,23 @@
+# Patterns to ignore when building packages.
+# This supports shell glob matching, relative path matching, and
+# negation (prefixed with !). Only one pattern per line.
+.DS_Store
+# Common VCS dirs
+.git/
+.gitignore
+.bzr/
+.bzrignore
+.hg/
+.hgignore
+.svn/
+# Common backup files
+*.swp
+*.bak
+*.tmp
+*.orig
+*~
+# Various IDEs
+.project
+.idea/
+*.tmproj
+.vscode/

charts/vald-readreplica/Chart.yaml

@@ -0,0 +1,24 @@
+apiVersion: v2
+name: vald-readreplica
+description: A Helm chart for Kubernetes
+
+# A chart can be either an 'application' or a 'library' chart.
+#
+# Application charts are a collection of templates that can be packaged into versioned archives
+# to be deployed.
+#
+# Library charts provide useful utilities or functions for the chart developer. They're included as
+# a dependency of application charts to inject those utilities and functions into the rendering
+# pipeline. Library charts do not define any templates and therefore cannot be deployed.
+type: application
+
+# This is the chart version. This version number should be incremented each time you make changes
+# to the chart and its templates, including the app version.
+# Versions are expected to follow Semantic Versioning (https://semver.org/)
+version: 0.1.0
+
+# This is the version number of the application being deployed. This version number should be
+# incremented each time you make changes to the application. Versions are not expected to
+# follow Semantic Versioning. They should reflect the version the application is using.
+# It is recommended to use it with quotes.
+appVersion: "1.16.0"

charts/vald-readreplica/templates/NOTES.txt

@@ -0,0 +1,4 @@
+Release {{ .Release.Name }} is created.
+
+Readreplica feature is intended to be used with the main vald chart.
+Please look at the document of the main chart for more details.

charts/vald-readreplica/templates/_helpers.tpl

@@ -0,0 +1 @@
+../../vald/templates/_helpers.tpl

charts/vald/templates/agent/readreplica/deployment.yaml → charts/vald-readreplica/templates/deployment.yaml

@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
+{{- $values := .Values -}}
 {{- $agent := .Values.agent -}}
 {{- $readreplica := .Values.agent.readreplica -}}
 {{- $defaults := .Values.defaults -}}
@@ -63,7 +64,7 @@ spec:
       {{- if $agent.initContainers }}
       initContainers:
         {{- if $agent.initContainers }}
-        {{- $initContainers := dict "initContainers" $agent.initContainers "Values" .Values "namespace" .Release.Namespace -}}
+        {{- $initContainers := dict "initContainers" $agent.initContainers "Values" $values "namespace" $release.Namespace -}}
         {{- include "vald.initContainers" $initContainers | trim | nindent 8 }}
         {{- end }}
       {{- end }}
@@ -135,6 +136,9 @@ spec:
         - name: {{ $readreplica.volume_name }}
           persistentVolumeClaim:
             claimName: {{ $readreplica.name }}-pvc-{{ $id }}
+        {{- if $agent.volumes }}
+        {{- toYaml $agent.volumes | nindent 8 }}
+        {{- end }}
       {{- if $agent.nodeName }}
       nodeName: {{ $agent.nodeName }}
       {{- end }}

charts/vald/templates/agent/readreplica/pvc.yaml → charts/vald-readreplica/templates/pvc.yaml

@@ -30,10 +30,11 @@ spec:
     - ReadOnlyMany
   resources:
     requests:
-      storage: 1Gi
+      storage: {{ $agent.persistentVolume.size }}
   dataSource:
     name: {{ $readreplica.name }}-snapshot-{{ $id }}
     kind: VolumeSnapshot
     apiGroup: snapshot.storage.k8s.io
+  storageClassName: {{ $agent.persistentVolume.storageClass }}
 {{- end }}
 {{- end }}

charts/vald/templates/agent/readreplica/snapshot.yaml → charts/vald-readreplica/templates/snapshot.yaml

@@ -25,7 +25,7 @@ metadata:
     {{ $readreplica.label_key }}: "{{ $id }}"
   name: {{ $readreplica.name }}-snapshot-{{ $id }}
 spec:
-  volumeSnapshotClassName: csi-hostpath-snapclass
+  volumeSnapshotClassName: {{ $readreplica.snapshot_classname }}
   source:
     # Specify the PVCs for which you want to take snapshots
     persistentVolumeClaimName: {{ $agent.name }}-pvc-{{ $agent.name }}-{{ $id }}

charts/vald/templates/agent/readreplica/svc.yaml → charts/vald-readreplica/templates/svc.yaml

@@ -50,6 +50,7 @@ spec:
     app: {{ $readreplica.name }}-{{ $id }}
     app.kubernetes.io/name: {{ $valdname }}
   type: ClusterIP
+  clusterIP: None
   {{- if $agent.externalTrafficPolicy }}
   externalTrafficPolicy: {{ $agent.externalTrafficPolicy }}
   {{- end }}

charts/vald-readreplica/values.yaml

@@ -0,0 +1 @@
+../vald/values.yaml

charts/vald/templates/gateway/lb/configmap.yaml

@@ -16,6 +16,8 @@
 {{- $gateway := .Values.gateway.lb -}}
 {{- $agent := .Values.agent -}}
 {{- $discoverer := .Values.discoverer -}}
+{{- $readreplica := .Values.agent.readreplica -}}
+{{- $release := .Release -}}
 {{- if $gateway.enabled }}
 apiVersion: v1
 kind: ConfigMap
@@ -49,6 +51,7 @@ data:
       agent_namespace: {{ $gateway.gateway_config.agent_namespace | quote }}
       node_name: {{ $gateway.gateway_config.node_name | quote }}
       index_replica: {{ $gateway.gateway_config.index_replica }}
+      read_replica_replicas: {{ $readreplica.replica }}
       discoverer:
         duration: {{ $gateway.gateway_config.discoverer.duration }}
         client:
@@ -64,4 +67,20 @@ data:
         agent_client_options:
           {{- include "vald.grpc.client.addrs" (dict "Valued" $gateway.gateway_config.discoverer.agent_client_options.addrs) | nindent 10 }}
           {{- include "vald.grpc.client" (dict "Values" $gateway.gateway_config.discoverer.agent_client_options "default" .Values.defaults.grpc.client) | nindent 10 }}
+      {{- if $readreplica.enabled }}
+      read_replica_client:
+        client:
+          {{- $discovererClient := $gateway.gateway_config.discoverer.client }}
+          {{- $readReplicaPort := $agent.server_config.servers.grpc.port }}
+          {{- $defaultReadReplicaPort := default .Values.defaults.server_config.servers.grpc.port $readReplicaPort }}
+          {{- $readReplicaAddrs := list }}
+          {{- range $i := until (int $agent.minReplicas) }}
+            {{- $addr := printf "%s-%d.%s.svc.cluster.local:%d" $readreplica.name $i $release.Namespace (int64 $defaultReadReplicaPort) }}
+            {{- $readReplicaAddrs = append $readReplicaAddrs $addr }}
+          {{- end }}
+          {{- $readReplicaAddrs := dict "Values" $discovererClient.addrs "default" $readReplicaAddrs }}
+          {{- include "vald.grpc.client.addrs" $readReplicaAddrs | nindent 10 }}
+          {{- $readReplicaGRPCclient := dict "Values" $discovererClient "default" .Values.defaults.grpc.client }}
+          {{- include "vald.grpc.client" $readReplicaGRPCclient | nindent 10 }}
+      {{- end }}
 {{- end }}

charts/vald/templates/index/job/readreplica/rotate/clusterrole.yaml

@@ -14,7 +14,8 @@
 # limitations under the License.
 #
 {{- $rotator := .Values.manager.index.readreplica.rotator -}}
-{{- if and $rotator.enabled $rotator.clusterRole.enabled }}
+{{- $readreplica := .Values.agent.readreplica -}}
+{{- if and $readreplica.enabled $rotator.clusterRole.enabled }}
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRole
 metadata:

charts/vald/templates/index/job/readreplica/rotate/clusterrolebinding.yaml

@@ -14,7 +14,8 @@
 # limitations under the License.
 #
 {{- $rotator := .Values.manager.index.readreplica.rotator -}}
-{{- if and $rotator.enabled $rotator.clusterRoleBinding.enabled }}
+{{- $readreplica := .Values.agent.readreplica -}}
+{{- if and $readreplica.enabled $rotator.clusterRoleBinding.enabled }}
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding
 metadata:

charts/vald/templates/index/job/readreplica/rotate/configmap.yaml

@@ -14,11 +14,12 @@
 # limitations under the License.
 #
 {{- $rotator    := .Values.manager.index.readreplica.rotator -}}
+{{- $readreplica := .Values.agent.readreplica -}}
 {{- $gateway    := .Values.gateway.lb -}}
 {{- $index      := .Values.manager.index -}}
 {{- $agent      := .Values.agent -}}
 {{- $discoverer := .Values.discoverer -}}
-{{- if $rotator.enabled }}
+{{- if $readreplica.enabled }}
 apiVersion: v1
 kind: ConfigMap
 metadata:

charts/vald/templates/index/job/readreplica/rotate/cronjob.yaml

@@ -14,7 +14,8 @@
 # limitations under the License.
 #
 {{- $rotator := .Values.manager.index.readreplica.rotator -}}
-{{- if $rotator.enabled }}
+{{- $readreplica := .Values.agent.readreplica -}}
+{{- if $readreplica.enabled }}
 apiVersion: batch/v1
 kind: CronJob
 metadata:

charts/vald/templates/index/job/readreplica/rotate/networkpolicy.yaml

@@ -15,7 +15,8 @@
 #
 
 {{- $rotator := .Values.manager.index.readreplica.rotator -}}
-{{- if and .Values.defaults.networkPolicy.enabled $rotator.enabled }}
+{{- $readreplica := .Values.agent.readreplica -}}
+{{- if and .Values.defaults.networkPolicy.enabled $readreplica.enabled }}
 apiVersion: networking.k8s.io/v1
 kind: NetworkPolicy
 metadata:

charts/vald/templates/index/job/readreplica/rotate/serviceaccount.yaml

@@ -14,7 +14,8 @@
 # limitations under the License.
 #
 {{- $rotator := .Values.manager.index.readreplica.rotator -}}
-{{- if and $rotator.enabled $rotator.serviceAccount.enabled }}
+{{- $readreplica := .Values.agent.readreplica -}}
+{{- if and $readreplica.enabled $rotator.serviceAccount.enabled }}
 apiVersion: v1
 kind: ServiceAccount
 metadata:

charts/vald/values.yaml

@@ -1656,7 +1656,7 @@ agent:
     snapshot_classname: ""
     # @schema {"name": "agent.readreplica.replica", "type": "integer"}
     # agent.readreplica.replica -- replica number of read replica
-    replica: 2
+    replica: 1
     # @schema {"name": "agent.readreplica.service", "type": "object"}
     # agent.readreplica.service -- service settings for read replica service resources
     service:

internal/client/v1/client/discoverer/discover.go

@@ -38,6 +38,7 @@ type Client interface {
 	Start(ctx context.Context) (<-chan error, error)
 	GetAddrs(ctx context.Context) []string
 	GetClient() grpc.Client
+	GetReadClient() grpc.Client
 }
 
 type client struct {
@@ -56,6 +57,10 @@ type client struct {
 	name         string
 	namespace    string
 	nodeName     string
+	// read replica related below
+	readClient          grpc.Client
+	readReplicaReplicas uint64
+	roundRobin          atomic.Uint64
 }
 
 func New(opts ...Option) (d Client, err error) {
@@ -74,6 +79,14 @@ func (c *client) Start(ctx context.Context) (<-chan error, error) {
 		return nil, err
 	}
 
+	var rrech <-chan error
+	if c.readClient != nil {
+		rrech, err = c.readClient.StartConnectionMonitor(ctx)
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	ech := make(chan error, 100)
 	addrs, err := c.dnsDiscovery(ctx, ech)
 	if err != nil {
@@ -134,6 +147,7 @@ func (c *client) Start(ctx context.Context) (<-chan error, error) {
 				return finalize()
 			case err = <-dech:
 			case err = <-aech:
+			case err = <-rrech:
 			case <-dt.C:
 				err = c.discover(ctx, ech)
 			}
@@ -172,6 +186,29 @@ func (c *client) GetClient() grpc.Client {
 	return c.client
 }
 
+func (c *client) GetReadClient() grpc.Client {
+	// read replica
+	// 1. primary + svc cluster IP *n でここで比率を見て呼び分ける
+	//    read replicaのreplica数がHPAなどで高速に変動する場合に、実装がシンプルそうなのでこちらが良さそう
+	// 2. primary + ipsで単純にラウンドロビン
+
+	// round robin with c.client and c.readClient everytime it's called
+	// with a ratio of primary + read replica deployment replicas
+	// TODO: is this atomic operation really worth it?
+	var new uint64
+	for {
+		cur := c.roundRobin.Load()
+		new = (cur + 1) % (c.readReplicaReplicas + 1)
+		if c.roundRobin.CompareAndSwap(cur, new) {
+			break
+		}
+	}
+	if new == 0 || c.readClient == nil {
+		return c.client
+	}
+	return c.readClient
+}
+
 func (c *client) connect(ctx context.Context, addr string) (err error) {
 	if c.autoconn && c.client != nil {
 		_, err = c.client.Connect(ctx, addr)

internal/client/v1/client/discoverer/option.go

@@ -68,6 +68,13 @@ func WithDiscovererClient(gc grpc.Client) Option {
 	}
 }
 
+func WithReadReplicaClient(gc grpc.Client) Option {
+	return func(c *client) error {
+		c.readClient = gc
+		return nil
+	}
+}
+
 func WithDiscoverDuration(dur string) Option {
 	return func(c *client) error {
 		d, err := timeutil.Parse(dur)
@@ -142,3 +149,10 @@ func WithErrGroup(eg errgroup.Group) Option {
 		return nil
 	}
 }
+
+func WithReadReplicaReplicas(num uint64) Option {
+	return func(c *client) error {
+		c.readReplicaReplicas = num
+		return nil
+	}
+}

internal/config/lb.go

@@ -37,6 +37,12 @@ type LB struct {
 	// IndexReplica represents index replication count
 	IndexReplica int `json:"index_replica" yaml:"index_replica"`
 
+	// ReadReplicaReplicas represents replica count of read replica Deployment
+	ReadReplicaReplicas uint64 `json:"read_replica_replicas" yaml:"read_replica_replicas"`
+
+	// ReadReplicaClient represents read replica client configuration
+	ReadReplicaClient ReadReplicaClient `json:"read_replica_client" yaml:"read_replica_client"`
+
 	// Discoverer represent agent discoverer service configuration
 	Discoverer *DiscovererClient `json:"discoverer" yaml:"discoverer"`
 
@@ -56,3 +62,26 @@ func (g *LB) Bind() *LB {
 	}
 	return g
 }
+
+// ReadReplicaClient
+type ReadReplicaClient struct {
+	Duration           string      `json:"duration"             yaml:"duration"`
+	Client             *GRPCClient `json:"client"               yaml:"client"`
+	AgentClientOptions *GRPCClient `json:"agent_client_options" yaml:"agent_client_options"`
+}
+
+// Bind binds the actual data from the ReadReplicaClient receiver field.
+func (d *ReadReplicaClient) Bind() *ReadReplicaClient {
+	d.Duration = GetActualValue(d.Duration)
+	if d.Client != nil {
+		d.Client.Bind()
+	} else {
+		d.Client = newGRPCClientConfig()
+	}
+	if d.AgentClientOptions != nil {
+		d.AgentClientOptions.Bind()
+	} else {
+		d.AgentClientOptions = newGRPCClientConfig()
+	}
+	return d
+}