Kubernetes sandbox reliability updates.

Author: thompsonja

helm/vitess/templates/_vttablet.tpl

@@ -176,9 +176,10 @@ volumes:
 {{- $tablet := index . 4 -}}
 {{- with $tablet.vttablet -}}
 {{- $0 := $.Values.vttablet -}}
+{{- $cellClean := $cell.name | replace "_" "-" -}}
 {{- $keyspaceClean := $keyspace.name | replace "_" "-" -}}
 {{- $shardClean := include "format-shard-name" $shard.name -}}
-{{- $setName := printf "%s-%s-%s" $keyspaceClean $shardClean $tablet.type | lower -}}
+{{- $setName := printf "%s-%s-%s-%s" $cellClean $keyspaceClean $shardClean $tablet.type | lower -}}
 {{- $uid := "$(cat $VTDATAROOT/init/tablet-uid)" }}
 # vttablet StatefulSet
 apiVersion: apps/v1beta1
@@ -193,6 +194,7 @@ spec:
       labels:
         app: vitess
         component: vttablet
+        cell: {{$cellClean | quote}}
         keyspace: {{$keyspace.name | quote}}
         shard: {{$shardClean | quote}}
         type: {{$tablet.type | quote}}
@@ -222,6 +224,7 @@ spec:
 {{- $cell := index . 1 -}}
 {{- $keyspace := index . 2 -}}
 {{- $shard := index . 3 -}}
+{{- $shardClean := include "format-shard-name" $shard.name -}}
 {{- $tablet := index . 4 -}}
 {{- $uid := index . 5 -}}
 {{- with $tablet.vttablet -}}
@@ -235,7 +238,7 @@ metadata:
     app: vitess
     component: vttablet
     keyspace: {{$keyspace.name | quote}}
-    shard: {{$shard.name | quote}}
+    shard: {{$shardClean | quote}}
     type: {{$tablet.type | quote}}
   annotations:
     pod.beta.kubernetes.io/init-containers: '[

test/cluster/k8s_environment.py

@@ -184,7 +184,8 @@ def restart_mysql_task(self, tablet_name, task_name, is_alloc=False):
     time.sleep(60)
 
     # Create the pod again.
-    os.system('cat %s | kubectl create -f -' % tmpfile.name)
+    os.system('cat %s | kubectl create --namespace=%s -f -' % (
+        tmpfile.name, self.cluster_name))
     while time.time() - start_time < 120:
       logging.info('Waiting for pod %s to be running', vttablet_pod_name)
       pod = subprocess.check_output(

test/cluster/sandbox/initial_reparent.py

@@ -4,10 +4,10 @@
 import json
 import logging
 import optparse
+import sys
 import time
 from vtproto import topodata_pb2
 from vttest import sharding_utils
-import sandbox_utils
 import vtctl_sandbox
 
 
@@ -23,24 +23,29 @@ def initial_reparent(keyspace, master_cell, num_shards, namespace, timeout_s):
   """Performs the first reparent."""
   successfully_reparented = []
   master_tablets = {}
+  start_time = time.time()
+  logging.info('Finding tablets to reparent to.')
   while len(master_tablets) < num_shards:
+    if time.time() - start_time > timeout_s:
+      logging.fatal('Timed out waiting to find a replica tablet')
+      return 1
     for shard_name in sharding_utils.get_shard_names(num_shards):
-      shard_name = sandbox_utils.fix_shard_name(shard_name)
+      if shard_name in master_tablets:
+        continue
       tablets = vtctl_sandbox.execute_vtctl_command(
-          ['ListShardTablets', '%s/%s' % (
-              keyspace, sandbox_utils.fix_shard_name(shard_name))],
+          ['ListShardTablets', '%s/%s' % (keyspace, shard_name)],
           namespace=namespace)[0].split('\n')
       tablets = [x.split(' ') for x in tablets if x]
       potential_masters = [
           x[0] for x in tablets if x[3] == 'replica'
           and x[0].split('-')[0] == master_cell]
       if potential_masters:
         master_tablets[shard_name] = potential_masters[0]
+        logging.info(
+            '%s selected for shard %s', potential_masters[0], shard_name)
 
-  start_time = time.time()
   while time.time() - start_time < timeout_s:
     for shard_name in sharding_utils.get_shard_names(num_shards):
-      shard_name = sandbox_utils.fix_shard_name(shard_name)
       master_tablet_id = master_tablets[shard_name]
       if is_master(master_tablet_id, namespace):
         logging.info('Tablet %s is the master of %s/%s.',
@@ -55,8 +60,9 @@ def initial_reparent(keyspace, master_cell, num_shards, namespace, timeout_s):
            master_tablet_id], namespace=namespace, timeout_s=5)
     if len(successfully_reparented) == num_shards:
       logging.info('Done with initial reparent.')
-      return
+      return 0
   logging.fatal('Timed out waiting for initial reparent.')
+  return 1
 
 
 def main():
@@ -73,9 +79,9 @@ def main():
   logging.getLogger().setLevel(logging.INFO)
 
   options, _ = parser.parse_args()
-  initial_reparent(options.keyspace, options.master_cell,
-                   options.shard_count, options.namespace,
-                   options.timeout)
+  sys.exit(initial_reparent(options.keyspace, options.master_cell,
+                            options.shard_count, options.namespace,
+                            options.timeout))
 
 
 if __name__ == '__main__':

test/cluster/sandbox/sandbox_utils.py

@@ -5,26 +5,6 @@
 import random
 
 
-def fix_shard_name(shard_name):
-  """Kubernetes doesn't allow '-' in the beginning or end of attributes.
-
-  Instead, replace them with an x.
-
-  Example: -80 becomes x80, 80- becomes 80x.
-
-  Args:
-    shard_name: string, A standard shard name (like -80).
-
-  Returns:
-    A fixed shard name suitable for kubernetes (string).
-  """
-  if shard_name.startswith('-'):
-    return 'x%s' % shard_name[1:]
-  if shard_name.endswith('-'):
-    return '%sx' % shard_name[:-1]
-  return shard_name
-
-
 def create_log_file(log_dir, filename):
   """Create a log file.
 

test/cluster/sandbox/subprocess_component.py

@@ -28,13 +28,14 @@ def start(self):
           self.log_dir, '%s.INFO' % self.name)
       errorfile = sandbox_utils.create_log_file(
           self.log_dir, '%s.ERROR' % self.name)
-      subprocess.call(['./%s' % self.script] + script_args, stdout=infofile,
-                      stderr=errorfile)
+      subprocess.check_call(
+          ['./%s' % self.script] + script_args, stdout=infofile,
+          stderr=errorfile)
       logging.info('Done.')
     except subprocess.CalledProcessError as error:
       raise sandbox.SandboxError(
-          'Subprocess %s returned errorcode %d, result %s.' % (
-              self.script, error.returncode, error.output))
+          'Subprocess %s returned errorcode %d, find log at %s.' % (
+              self.script, error.returncode, errorfile.name))
     finally:
       if infofile:
         infofile.close()

test/cluster/sandbox/vitess_kubernetes_sandbox.py

@@ -11,7 +11,6 @@
 from vttest import sharding_utils
 
 import sandbox
-import sandbox_utils
 import sandlet
 import subprocess_component
 
@@ -92,7 +91,6 @@ def _generate_helm_keyspaces(self):
 
       for shard_index, shard_name in enumerate(
           sharding_utils.get_shard_names(ks['shard_count'])):
-        shard_name = sandbox_utils.fix_shard_name(shard_name)
         shard = dict(
             name=shard_name,
             tablets=[dict(
@@ -221,7 +219,7 @@ def generate_helm_sandlet(self):
           'wait_for_mysql_%s' % name, self.name, 'wait_for_mysql.py',
           self.log_dir, namespace=self.name,
           cells=','.join(self.app_options.cells),
-          tablet_count=(shard_count * (
+          tablet_count=(shard_count * len(self.app_options.cells) * (
               keyspace['replica_count'] + keyspace['rdonly_count'])))
       wait_for_mysql_subprocess.dependencies = ['helm']
       initial_reparent_subprocess = subprocess_component.Subprocess(

test/cluster/sandbox/wait_for_mysql.py

@@ -4,6 +4,7 @@
 import logging
 import optparse
 import re
+import sys
 import time
 import vtctl_sandbox
 
@@ -30,6 +31,8 @@ def main():
   parser.add_option('-c', '--cells', help='Comma separated list of cells')
   parser.add_option('-t', '--tablet_count',
                     help='Total number of expected tablets', type=int)
+  parser.add_option('-w', '--wait', help='Max wait time (s)', type=int,
+                    default=300)
   logging.getLogger().setLevel(logging.INFO)
 
   options, _ = parser.parse_args()
@@ -42,17 +45,18 @@ def main():
 
   # Do this in a loop as the output of ListAllTablets may not be parseable
   # until all tablets have been started.
-  while time.time() - start_time < 300 and len(tablets) < options.tablet_count:
+  while (time.time() - start_time < options.wait and
+         len(tablets) < options.tablet_count):
     tablets = get_all_tablets(options.cells, options.namespace)
     logging.info('Expecting %d tablets, found %d tablets',
                  options.tablet_count, len(tablets))
 
   start_time = time.time()
-  while time.time() - start_time < 300:
+  while time.time() - start_time < options.wait:
     for tablet in [t for t in tablets if t not in good_tablets]:
       _, success = vtctl_sandbox.execute_vtctl_command(
           ['ExecuteFetchAsDba', tablet, 'show databases'],
-          namespace=options.namespace)
+          namespace=options.namespace, timeout_s=1)
       if success:
         good_tablets.append(tablet)
     logging.info('%d of %d tablets healthy.', len(good_tablets), len(tablets))
@@ -62,6 +66,7 @@ def main():
       break
   else:
     logging.warn('Timed out waiting for tablets to be ready.')
+    sys.exit(1)
 
 
 if __name__ == '__main__':