Introduce Durability Policy in VTop

deploy/crds/planetscale.com_vitessclusters.yaml

@@ -958,6 +958,8 @@ spec:
                       type: object
                     databaseName:
                       type: string
+                    durabilityPolicy:
+                      type: string
                     name:
                       maxLength: 63
                       minLength: 1

deploy/crds/planetscale.com_vitesskeyspaces.yaml

@@ -153,6 +153,8 @@ spec:
                 type: array
               databaseName:
                 type: string
+              durabilityPolicy:
+                type: string
               extraVitessFlags:
                 additionalProperties:
                   type: string

docs/api/index.html

@@ -5777,6 +5777,18 @@ <h3 id="planetscale.com/v2.VitessKeyspaceTemplate">VitessKeyspaceTemplate
 </tr>
 <tr>
 <td>
+<code>durabilityPolicy</code></br>
+<em>
+string
+</em>
+</td>
+<td>
+<p>DurabilityPolicy is the name of the durability policy to use for the keyspace.
+If unspecified, vtop will not set the durability policy.</p>
+</td>
+</tr>
+<tr>
+<td>
 <code>vitessOrchestrator</code></br>
 <em>
 <a href="#planetscale.com/v2.VitessOrchestratorSpec">

pkg/apis/planetscale/v2/vitesskeyspace_types.go

@@ -113,6 +113,10 @@ type VitessKeyspaceTemplate struct {
 	// Default: Add a "vt_" prefix to the keyspace name.
 	DatabaseName string `json:"databaseName,omitempty"`
 
+	// DurabilityPolicy is the name of the durability policy to use for the keyspace.
+	// If unspecified, vtop will not set the durability policy.
+	DurabilityPolicy string `json:"durabilityPolicy,omitempty"`
+
 	// VitessOrchestrator deploys a set of Vitess Orchestrator (vtorc) servers for the Keyspace.
 	// It is highly recommended that you set disable_active_reparents=true
 	// and enable_semi_sync=false for the vtablets if enabling vtorc.

pkg/controller/vitesskeyspace/reconcile_keyspace_information.go

@@ -0,0 +1,81 @@
+/*
+Copyright 2022 PlanetScale Inc.
+
+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
+
+    http://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 vitesskeyspace
+
+import (
+	"context"
+	corev1 "k8s.io/api/core/v1"
+
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+
+	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
+	vtctldatapb "vitess.io/vitess/go/vt/proto/vtctldata"
+	"vitess.io/vitess/go/vt/topo"
+
+	"planetscale.dev/vitess-operator/pkg/operator/results"
+)
+
+func (r *reconcileHandler) reconcileKeyspaceInformation(ctx context.Context) (reconcile.Result, error) {
+	resultBuilder := &results.Builder{}
+
+	// Initialize the topo server before using it.
+	// This call is idempotent, so it is safe to call each time
+	// before using the topo server.
+	err := r.tsInit(ctx)
+	if err != nil {
+		return resultBuilder.RequeueAfter(topoRequeueDelay)
+	}
+
+	topoServer := r.ts.Server
+	keyspaceName := r.vtk.Spec.Name
+	durabilityPolicy := r.vtk.Spec.DurabilityPolicy
+	keyspaceInfo, err := topoServer.GetKeyspace(ctx, keyspaceName)
+	if err != nil {
+		// The keyspace information record does not exist in the topo server.
+		// We should create the record
+		if !topo.IsErrType(err, topo.NoNode) {
+			// Create a normal keyspace with the requested durability policy
+			_, err := r.wr.VtctldServer().CreateKeyspace(ctx, &vtctldatapb.CreateKeyspaceRequest{
+				Name:             keyspaceName,
+				Type:             topodatapb.KeyspaceType_NORMAL,
+				DurabilityPolicy: durabilityPolicy,
+			})
+			if err != nil {
+				resultBuilder.Error(err)
+			}
+			return resultBuilder.Result()
+		}
+		// GetKeyspace returned a error other than NoNode.
+		// Maybe the topo server is temporarily unreachable
+		// We should retry after some time.
+		r.recorder.Eventf(r.vtk, corev1.EventTypeWarning, "GetKeyspace", "failed to get keyspace %v: %v", keyspaceName, err)
+		return resultBuilder.RequeueAfter(topoRequeueDelay)
+	}
+
+	// DurabilityPolicy doesn't match the one requested by the user
+	// We change the durability policy using the SetKeyspaceDurabilityPolicy rpc
+	if durabilityPolicy != "" && keyspaceInfo.DurabilityPolicy != durabilityPolicy {
+		_, err := r.wr.VtctldServer().SetKeyspaceDurabilityPolicy(ctx, &vtctldatapb.SetKeyspaceDurabilityPolicyRequest{
+			Keyspace:         keyspaceName,
+			DurabilityPolicy: durabilityPolicy,
+		})
+		if err != nil {
+			resultBuilder.Error(err)
+		}
+	}
+	return resultBuilder.Result()
+}

pkg/controller/vitesskeyspace/vitesskeyspace_controller.go

@@ -168,6 +168,10 @@ func (r *ReconcileVitessKeyspace) Reconcile(cctx context.Context, request reconc
 		}
 	}()
 
+	// Create/update keyspace record in the topo server
+	keyspaceInfoRes, err := handler.reconcileKeyspaceInformation(ctx)
+	resultBuilder.Merge(keyspaceInfoRes, err)
+
 	// Create/update desired VitessShards.
 	if err := handler.reconcileShards(ctx); err != nil {
 		resultBuilder.Error(err)

test/endtoend/operator/101_initial_cluster_backup.yaml

@@ -52,6 +52,7 @@ spec:
 
   keyspaces:
   - name: commerce
+    durabilityPolicy: semi_sync
     turndownPolicy: Immediate
     partitionings:
     - equal:

test/endtoend/operator/101_initial_cluster_vtadmin.yaml

@@ -68,6 +68,7 @@ spec:
 
   keyspaces:
   - name: commerce
+    durabilityPolicy: semi_sync
     turndownPolicy: Immediate
     partitionings:
     - equal:

test/endtoend/operator/201_customer_tablets.yaml

@@ -41,6 +41,7 @@ spec:
 
   keyspaces:
   - name: commerce
+    durabilityPolicy: semi_sync
     turndownPolicy: Immediate
     partitionings:
     - equal:
@@ -77,6 +78,7 @@ spec:
                 requests:
                   storage: 10Gi
   - name: customer
+    durabilityPolicy: semi_sync
     turndownPolicy: Immediate
     partitionings:
     - equal:

test/endtoend/operator/302_new_shards.yaml

@@ -41,6 +41,7 @@ spec:
 
   keyspaces:
   - name: commerce
+    durabilityPolicy: semi_sync
     turndownPolicy: Immediate
     partitionings:
     - equal:
@@ -77,6 +78,7 @@ spec:
                 requests:
                   storage: 10Gi
   - name: customer
+    durabilityPolicy: semi_sync
     turndownPolicy: Immediate
     partitionings:
     - equal:

test/endtoend/operator/306_down_shard_0.yaml

@@ -41,6 +41,7 @@ spec:
 
   keyspaces:
   - name: commerce
+    durabilityPolicy: semi_sync
     turndownPolicy: Immediate
     partitionings:
     - equal:
@@ -73,6 +74,7 @@ spec:
                 requests:
                   storage: 10Gi
   - name: customer
+    durabilityPolicy: semi_sync
     turndownPolicy: Immediate
     partitionings:
     - equal:

test/endtoend/operator/cluster_upgrade.yaml

@@ -45,6 +45,7 @@ spec:
 
   keyspaces:
   - name: commerce
+    durabilityPolicy: semi_sync
     turndownPolicy: Immediate
     partitionings:
     - equal:

test/endtoend/operator/operator-latest.yaml

@@ -1,3 +1,5 @@
+
+---
 apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
@@ -2089,6 +2091,8 @@ spec:
                         type: object
                       databaseName:
                         type: string
+                      durabilityPolicy:
+                        type: string
                       name:
                         maxLength: 63
                         minLength: 1
@@ -3574,6 +3578,8 @@ spec:
                   type: array
                 databaseName:
                   type: string
+                durabilityPolicy:
+                  type: string
                 extraVitessFlags:
                   additionalProperties:
                     type: string

test/endtoend/upgrade_test.sh

@@ -170,6 +170,8 @@ function upgradeToLatest() {
   echo "Apply operator-latest.yaml "
   kubectl apply -f operator-latest.yaml
 
+  sleep 2
+
   echo "Upgrade all the other binaries"
   kubectl apply -f cluster_upgrade.yaml
 
@@ -231,9 +233,13 @@ setupKubectlAccessForCI
 get_started "operator.yaml" "101_initial_cluster.yaml"
 verifyVtGateVersion "14.0.0"
 checkSemiSyncSetup
+# Initially no durability policy is specified
+verifyDurabilityPolicy "commerce" ""
 upgradeToLatest
 verifyVtGateVersion "15.0.0"
 checkSemiSyncSetup
+# After upgrading, we set the durability policy to semi_sync
+verifyDurabilityPolicy "commerce" "semi_sync"
 move_tables
 resharding
 

test/endtoend/utils.sh

@@ -123,6 +123,20 @@ function verifyVtGateVersion() {
   fi
 }
 
+
+# verifyDurabilityPolicy verifies the durability policy
+# in the given keyspace
+function verifyDurabilityPolicy() {
+  keyspace=$1
+  durabilityPolicy=$2
+  data=$(vtctlclient GetKeyspace "$keyspace")
+  echo "$data" | grep "\"durability_policy\": \"$durabilityPolicy\"" > /dev/null 2>&1
+  if [ $? -ne 0 ]; then
+    echo -e "The durability policy in $keyspace is incorrect, got:\n$data"
+    exit 1
+  fi
+}
+
 function waitForKeyspaceToBeServing() {
   ks=$1
   shard=$2