Merge pull request vitessio#2626 from thompsonja/sandbox_updates

Kubernetes sandbox reliability updates.

Author: thompsonja

helm/vitess/templates/_helpers.tpl

@@ -14,14 +14,17 @@
 {{- range . }}{{template "format-flags" .}}{{end -}}
 {{- end -}}
 
-# Format a shard name, making sure it starts and ends with [A-Za-z0-9].
-{{- define "format-shard-name" -}}
+# Clean labels, making sure it starts and ends with [A-Za-z0-9].
+# This is especially important for shard names, which can start or end with
+# '-' (like -80 or 80-), which would be an invalid kubernetes label.
+{{- define "clean-label" -}}
+{{- $replaced_label := . | replace "_" "-"}}
 {{- if hasPrefix "-" . -}}
-x{{.}}
+x{{$replaced_label}}
 {{- else if hasSuffix "-" . -}}
-{{.}}x
+{{$replaced_label}}x
 {{- else -}}
-{{.}}
+{{$replaced_label}}
 {{- end -}}
 {{- end -}}
 

helm/vitess/templates/_vttablet.tpl

@@ -176,9 +176,10 @@ volumes:
 {{- $tablet := index . 4 -}}
 {{- with $tablet.vttablet -}}
 {{- $0 := $.Values.vttablet -}}
-{{- $keyspaceClean := $keyspace.name | replace "_" "-" -}}
-{{- $shardClean := include "format-shard-name" $shard.name -}}
-{{- $setName := printf "%s-%s-%s" $keyspaceClean $shardClean $tablet.type | lower -}}
+{{- $cellClean := include "clean-label" $cell.name -}}
+{{- $keyspaceClean := include "clean-label" $keyspace.name -}}
+{{- $shardClean := include "clean-label" $shard.name -}}
+{{- $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 "clean-label" $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,30 @@ 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.error('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 +61,10 @@ 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
-  logging.fatal('Timed out waiting for initial reparent.')
+      return 0
+
+  logging.error('Timed out waiting for initial reparent.')
+  return 1
 
 
 def main():
@@ -73,9 +81,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(
@@ -187,12 +185,15 @@ def _generate_helm_values_config(self):
           keyspaces=copy.deepcopy(keyspaces),
       )
       # Each tablet's UID must be unique, so increment the uidBase for tablets
-      # by the cell epsilon value to ensure uniqueness. This logic will go away
-      # once StatefulSet is available.
+      # by the cell epsilon value to ensure uniqueness. Also convert the UID to
+      # a string, or else the parser will attempt to parse UID as a float, which
+      # causes issues when UID's are large. This logic will go away once
+      # StatefulSet is available.
       for keyspace in cell_dict['keyspaces']:
         for shard in keyspace['shards']:
           for tablets in shard['tablets']:
-            tablets['uidBase'] += index * self.cell_epsilon
+            tablets['uidBase'] = str(
+                tablets['uidBase'] + index * self.cell_epsilon)
       yaml_values['topology']['cells'].append(cell_dict)
 
       if index == 0:
@@ -214,24 +215,31 @@ def generate_helm_sandlet(self):
     helm_sandlet = sandlet.Sandlet('helm')
     helm_sandlet.components.add_component(kubernetes_components.HelmComponent(
         'helm', self.name, self._generate_helm_values_config()))
+
+    # Add a subprocess task to wait for all mysql instances to be healthy.
+    tablet_count = 0
+    for keyspace in self.app_options.keyspaces:
+      tablet_count += (keyspace['shard_count'] * len(self.app_options.cells) * (
+          keyspace['replica_count'] + keyspace['rdonly_count']))
+    wait_for_mysql_subprocess = subprocess_component.Subprocess(
+        'wait_for_mysql', self.name, 'wait_for_mysql.py',
+        self.log_dir, namespace=self.name,
+        cells=','.join(self.app_options.cells),
+        tablet_count=tablet_count)
+    wait_for_mysql_subprocess.dependencies = ['helm']
+    helm_sandlet.components.add_component(wait_for_mysql_subprocess)
+
+    # Add a subprocess task for each keyspace to perform the initial reparent.
     for keyspace in self.app_options.keyspaces:
       name = keyspace['name']
       shard_count = keyspace['shard_count']
-      wait_for_mysql_subprocess = subprocess_component.Subprocess(
-          '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 * (
-              keyspace['replica_count'] + keyspace['rdonly_count'])))
-      wait_for_mysql_subprocess.dependencies = ['helm']
       initial_reparent_subprocess = subprocess_component.Subprocess(
-          'initial_reparent_%s' % name, self.name,
+          'initial_reparent_%s_%d' % (name, shard_count), self.name,
           'initial_reparent.py', self.log_dir, namespace=self.name,
           keyspace=name, shard_count=shard_count,
           master_cell=self.app_options.cells[0])
       initial_reparent_subprocess.dependencies = [
           wait_for_mysql_subprocess.name]
-      helm_sandlet.components.add_component(wait_for_mysql_subprocess)
       helm_sandlet.components.add_component(initial_reparent_subprocess)
     self.sandlets.add_component(helm_sandlet)
 

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__':