def test_service_switch(self):
        """tests the service switch from disable -> enable -> disable."""
        # make the replica spare
        utils.run_vtctl(["ChangeSlaveType", replica_tablet.tablet_alias, "spare"])
        utils.wait_for_tablet_type(replica_tablet.tablet_alias, "spare")

        # Check UpdateStreamState is disabled.
        v = utils.get_vars(replica_tablet.port)
        if v["UpdateStreamState"] != "Disabled":
            self.fail("Update stream service should be 'Disabled' but is '%s'" % v["UpdateStreamState"])

        # Make sure we can't start a new request.
        start_position = _get_repl_current_position()
        replica_conn = self._get_replica_stream_conn()
        try:
            for event in replica_conn.stream_update(
                "test_keyspace", "0", topodata_pb2.REPLICA, position=start_position
            ):
                self.assertFail("got event: %s" % str(event))
            self.assertFail("stream_update terminated with no exception")
        except dbexceptions.DatabaseError as e:
            self.assertIn("operation not allowed in state NOT_SERVING", str(e))

        # Go back to replica.
        utils.run_vtctl(["ChangeSlaveType", replica_tablet.tablet_alias, "replica"])
        utils.wait_for_tablet_type(replica_tablet.tablet_alias, "replica")

        # Check UpdateStreamState is enabled.
        v = utils.get_vars(replica_tablet.port)
        if v["UpdateStreamState"] != "Enabled":
            self.fail("Update stream service should be 'Enabled' but is '%s'" % v["UpdateStreamState"])

  def _check_query_service(self, tablet, serving, tablet_control_disabled):
    """_check_query_service will check that the query service is enabled
    or disabled on the tablet. It will also check if the tablet control
    status is the reason for being enabled / disabled.

    It will also run a remote RunHealthCheck to be sure it doesn't change
    the serving state.
    """
    tablet_vars = utils.get_vars(tablet.port)
    if serving:
      expected_state = 'SERVING'
    else:
      expected_state = 'NOT_SERVING'
    self.assertEqual(tablet_vars['TabletStateName'], expected_state, 'tablet %s is not in the right serving state: got %s expected %s' % (tablet.tablet_alias, tablet_vars['TabletStateName'], expected_state))

    status = tablet.get_status()
    if tablet_control_disabled:
      self.assertIn("Query Service disabled by TabletControl", status)
    else:
      self.assertNotIn("Query Service disabled by TabletControl", status)

    if tablet.tablet_type == 'rdonly':
      utils.run_vtctl(['RunHealthCheck', tablet.tablet_alias, 'rdonly'],
                      auto_log=True)

      tablet_vars = utils.get_vars(tablet.port)
      if serving:
        expected_state = 'SERVING'
      else:
        expected_state = 'NOT_SERVING'
      self.assertEqual(tablet_vars['TabletStateName'], expected_state, 'tablet %s is not in the right serving state after health check: got %s expected %s' % (tablet.tablet_alias, tablet_vars['TabletStateName'], expected_state))

  def wait_for_vttablet_state(self, expected, timeout=60.0, port=None):
    # wait for zookeeper PID just to be sure we have it
    if environment.topo_server_implementation == 'zookeeper':
      if not self.checked_zk_pid:
        utils.run(environment.binary_args('zk') + ['wait', '-e', self.zk_pid],
                  stdout=utils.devnull)
        self.checked_zk_pid = True

    while True:
      v = utils.get_vars(port or self.port)
      if v == None:
        logging.debug(
            '  vttablet %s not answering at /debug/vars, waiting...',
            self.tablet_alias)
      else:
        if 'Voltron' not in v:
          logging.debug(
              '  vttablet %s not exporting Voltron, waiting...',
              self.tablet_alias)
        else:
          s = v['TabletStateName']
          if s != expected:
            logging.debug(
                '  vttablet %s in state %s != %s', self.tablet_alias, s,
                expected)
          else:
            break
      timeout = utils.wait_step('waiting for state %s' % expected, timeout,
                                sleep_time=0.1)

def run_test_zkocc_qps():
  _populate_zk()

  # preload the test_nj cell
  zkocc_14850 = utils.zkocc_start()

  qpser = utils.run_bg(utils.vtroot+'/bin/zkclient2 -server localhost:%u -mode qps /zk/test_nj/zkocc1/data1 /zk/test_nj/zkocc1/data2' % utils.zkocc_port_base)
  time.sleep(10)
  utils.kill_sub_process(qpser)

  # get the zkocc vars, make sure we have what we need
  v = utils.get_vars(utils.zkocc_port_base)
  if v['ZkReader']['test_nj']['State']['Current'] != 'Connected':
    raise utils.TestError('invalid zk global state: ', v['ZkReader']['test_nj']['State']['Current'])
  if v['ZkReader']['test_nj']['State']['DurationConnected'] < 9e9:
    raise utils.TestError('not enough time in Connected state', v['ZkReader']['test_nj']['State']['DurationConnected'])

  # some checks on performance / stats
  # a typical workstation will do 15k QPS, check we have more than 3k
  rpcCalls = v['ZkReader']['RpcCalls']
  if rpcCalls < 30000:
    raise utils.TestError('QPS is too low: %u < 30000', rpcCalls / 10)
  cacheReads = v['ZkReader']['test_nj']['CacheReads']
  if cacheReads < 30000:
    raise utils.TestError('Cache QPS is too low: %u < 30000', cacheReads / 10)
  totalCacheReads = v['ZkReader']['total']['CacheReads']
  if cacheReads != totalCacheReads:
    raise utils.TestError('Rollup stats are wrong: %u != %u', cacheReads,
                          totalCacheReads)
  if v['ZkReader']['UnknownCellErrors'] != 0:
    raise utils.TestError('unexpected UnknownCellErrors', v['ZkReader']['UnknownCellErrors'])

  utils.zkocc_kill(zkocc_14850)

  def wait_for_binlog_player_count(self, expected, timeout=30.0):
    """Wait for a tablet to have binlog players.

    Args:
      expected: number of expected binlog players to wait for.
      timeout: how long to wait.
    """
    while True:
      v = utils.get_vars(self.port)
      if v is None:
        if self.proc.poll() is not None:
          raise utils.TestError(
              'vttablet died while test waiting for binlog count %s' %
              expected)
        logging.debug('  vttablet not answering at /debug/vars, waiting...')
      else:
        if 'BinlogPlayerMapSize' not in v:
          logging.debug(
              '  vttablet not exporting BinlogPlayerMapSize, waiting...')
        else:
          s = v['BinlogPlayerMapSize']
          if s != expected:
            logging.debug("  vttablet's binlog player map has count %d != %d",
                          s, expected)
          else:
            break
      timeout = utils.wait_step(
          'waiting for binlog player count %d' % expected,
          timeout, sleep_time=0.5)
    logging.debug('tablet %s binlog player has %d players',
                  self.tablet_alias, expected)

  def check_binlog_player_vars(self, tablet_obj, source_shards,
                               seconds_behind_master_max=0):
    """Checks the binlog player variables are correctly exported.

    Args:
      tablet_obj: the tablet to check.
      source_shards: the shards to check we are replicating from.
      seconds_behind_master_max: if non-zero, the lag should be smaller than
                                 this value.
    """
    v = utils.get_vars(tablet_obj.port)
    self.assertIn('BinlogPlayerMapSize', v)
    self.assertEquals(v['BinlogPlayerMapSize'], len(source_shards))
    self.assertIn('BinlogPlayerSecondsBehindMaster', v)
    self.assertIn('BinlogPlayerSecondsBehindMasterMap', v)
    self.assertIn('BinlogPlayerSourceShardNameMap', v)
    shards = v['BinlogPlayerSourceShardNameMap'].values()
    self.assertEquals(sorted(shards), sorted(source_shards))
    self.assertIn('BinlogPlayerSourceTabletAliasMap', v)
    for i in xrange(len(source_shards)):
      self.assertIn('%d' % i, v['BinlogPlayerSourceTabletAliasMap'])
    if seconds_behind_master_max != 0:
      self.assertTrue(
          v['BinlogPlayerSecondsBehindMaster'] <
          seconds_behind_master_max,
          'BinlogPlayerSecondsBehindMaster is too high: %d > %d' % (
              v['BinlogPlayerSecondsBehindMaster'],
              seconds_behind_master_max))
      for i in xrange(len(source_shards)):
        self.assertTrue(
            v['BinlogPlayerSecondsBehindMasterMap']['%d' % i] <
            seconds_behind_master_max,
            'BinlogPlayerSecondsBehindMasterMap is too high: %d > %d' % (
                v['BinlogPlayerSecondsBehindMasterMap']['%d' % i],
                seconds_behind_master_max))

  def test_zkocc_qps(self):
    # preload the test_nj cell
    zkocc_14850 = utils.zkocc_start()

    qpser = utils.run_bg(utils.vtroot+'/bin/zkclient2 -server localhost:%u -mode qps /zk/test_nj/vt/zkocc1/data1 /zk/test_nj/vt/zkocc1/data2' % utils.zkocc_port_base)
    time.sleep(10)
    utils.kill_sub_process(qpser)

    # get the zkocc vars, make sure we have what we need
    v = utils.get_vars(utils.zkocc_port_base)
    if v['ZkReader']['test_nj']['State'] != 'Connected':
      raise utils.TestError('invalid zk global state: ', v['ZkReader']['test_nj']['State'])

    # some checks on performance / stats
    # a typical workstation will do 45-47k QPS, check we have more than 15k
    rpcCalls = v['ZkReader']['RpcCalls']
    if rpcCalls < 150000:
      self.fail('QPS is too low: %u < 15000' % (rpcCalls / 10))
    else:
      logging.debug("Recorded qps: %u", rpcCalls / 10)
    cacheReads = v['ZkReader']['test_nj']['CacheReads']
    if cacheReads < 150000:
      self.fail('Cache QPS is too low: %u < 15000' % (cacheReads / 10))
    totalCacheReads = v['ZkReader']['total']['CacheReads']
    self.assertEqual(cacheReads, totalCacheReads, 'Rollup stats are wrong')
    self.assertEqual(v['ZkReader']['UnknownCellErrors'], 0, 'unexpected UnknownCellErrors')
    utils.zkocc_kill(zkocc_14850)

  def check_binlog_server_vars(self, tablet_obj, horizontal=True,
                               min_statements=0, min_transactions=0):
    """Checks the binlog server variables are correctly exported.

    Args:
      tablet_obj: the tablet to check.
      horizontal: true if horizontal split, false for vertical split.
      min_statements: check the statement count is greater or equal to this.
      min_transactions: check the transaction count is greater or equal to this.
    """
    v = utils.get_vars(tablet_obj.port)
    if horizontal:
      skey = 'UpdateStreamKeyRangeStatements'
      tkey = 'UpdateStreamKeyRangeTransactions'
    else:
      skey = 'UpdateStreamTablesStatements'
      tkey = 'UpdateStreamTablesTransactions'

    self.assertIn(skey, v)
    self.assertIn(tkey, v)
    if min_statements > 0:
      self.assertTrue(v[skey] >= min_statements,
                      'only got %d < %d statements' % (v[skey], min_statements))
    if min_transactions > 0:
      self.assertTrue(v[tkey] >= min_transactions,
                      'only got %d < %d transactions' % (v[tkey],
                                                         min_transactions))

 def wait_for_vtocc_state(self, expected, timeout=60.0, port=None):
   while True:
     v = utils.get_vars(port or self.port)
     last_seen_state = "?"
     if v == None:
       logging.debug(
           '  vttablet %s not answering at /debug/vars, waiting...',
           self.tablet_alias)
     else:
       if 'TabletStateName' not in v:
         logging.debug(
             '  vttablet %s not exporting TabletStateName, waiting...',
             self.tablet_alias)
       else:
         s = v['TabletStateName']
         last_seen_state = s
         if s != expected:
           logging.debug(
               '  vttablet %s in state %s != %s', self.tablet_alias, s,
               expected)
         else:
           break
     timeout = utils.wait_step('waiting for state %s (last seen state: %s)' % (expected, last_seen_state),
                               timeout,
                               sleep_time=0.1)

  def test_zkocc_qps(self):
    # preload the test_nj cell
    zkocc_14850 = utils.zkocc_start()

    qpser = utils.run_bg(environment.binary_argstr('zkclient2')+' -server localhost:%u -mode qps /zk/test_nj/vt/zkocc1/data1 /zk/test_nj/vt/zkocc1/data2' % environment.topo_server().zkocc_port_base)
    qpser.wait()

    # get the zkocc vars, make sure we have what we need
    v = utils.get_vars(environment.topo_server().zkocc_port_base)
    if v['ZkReader']['test_nj']['State'] != 'Connected':
      self.fail('invalid zk global state: ' + v['ZkReader']['test_nj']['State'])

    # some checks on performance / stats
    rpcCalls = v['ZkReader']['RpcCalls']
    if rpcCalls < MIN_QPS * 10:
      self.fail('QPS is too low: %u < %u' % (rpcCalls / 10, MIN_QPS))
    else:
      logging.debug("Recorded qps: %u", rpcCalls / 10)
    cacheReads = v['ZkReader']['test_nj']['CacheReads']
    if cacheReads < MIN_QPS * 10:
      self.fail('Cache QPS is too low: %u < %u' % (cacheReads, MIN_QPS * 10))
    totalCacheReads = v['ZkReader']['total']['CacheReads']
    self.assertEqual(cacheReads, totalCacheReads, 'Rollup stats are wrong')
    self.assertEqual(v['ZkReader']['UnknownCellErrors'], 0, 'unexpected UnknownCellErrors')
    utils.zkocc_kill(zkocc_14850)

  def test_vtgate_qps(self):
    # create the topology
    utils.run_vtctl('CreateKeyspace test_keyspace')
    t = tablet.Tablet(tablet_uid=1, cell="nj")
    t.init_tablet("master", "test_keyspace", "0")
    t.update_addrs()
    utils.run_vtctl('RebuildKeyspaceGraph test_keyspace', auto_log=True)

    # start vtgate and the qps-er
    vtgate_proc, vtgate_port = utils.vtgate_start(
        extra_args=['-cpu_profile', os.path.join(environment.tmproot,
                                                 'vtgate.pprof')])
    qpser = utils.run_bg(environment.binary_args('zkclient2') + [
        '-server', 'localhost:%u' % vtgate_port,
        '-mode', 'qps',
        '-zkclient_cpu_profile', os.path.join(environment.tmproot, 'zkclient2.pprof'),
        'test_nj', 'test_keyspace'])
    qpser.wait()

    # get the vtgate vars, make sure we have what we need
    v = utils.get_vars(vtgate_port)

    # some checks on performance / stats
    rpcCalls = v['TopoReaderRpcQueryCount']['test_nj']
    if rpcCalls < MIN_QPS * 10:
      self.fail('QPS is too low: %u < %u' % (rpcCalls / 10, MIN_QPS))
    else:
      logging.debug("Recorded qps: %u", rpcCalls / 10)
    utils.vtgate_kill(vtgate_proc)

  def verify_reconciliation_counters(self, worker_port, online_or_offline,
                                     table, inserts, updates, deletes, equal):
    """Checks that the reconciliation Counters have the expected values."""
    worker_vars = utils.get_vars(worker_port)

    i = worker_vars['Worker' + online_or_offline + 'InsertsCounters']
    if inserts == 0:
      self.assertNotIn(table, i)
    else:
      self.assertEqual(i[table], inserts)

    u = worker_vars['Worker' + online_or_offline + 'UpdatesCounters']
    if updates == 0:
      self.assertNotIn(table, u)
    else:
      self.assertEqual(u[table], updates)

    d = worker_vars['Worker' + online_or_offline + 'DeletesCounters']
    if deletes == 0:
      self.assertNotIn(table, d)
    else:
      self.assertEqual(d[table], deletes)

    e = worker_vars['Worker' + online_or_offline + 'EqualRowsCounters']
    if equal == 0:
      self.assertNotIn(table, e)
    else:
      self.assertEqual(e[table], equal)

  def check_stream_health_equals_binlog_player_vars(self, tablet_obj, count):
    """Checks the variables exported by streaming health check match vars.

    Args:
      tablet_obj: the tablet to check.
      count: number of binlog players to expect.
    """

    blp_stats = utils.get_vars(tablet_obj.port)
    self.assertEqual(blp_stats['BinlogPlayerMapSize'], count)

    # Enforce health check because it's not running by default as
    # tablets may not be started with it, or may not run it in time.
    utils.run_vtctl(['RunHealthCheck', tablet_obj.tablet_alias])
    stream_health = utils.run_vtctl_json(['VtTabletStreamHealth',
                                          '-count', '1',
                                          tablet_obj.tablet_alias])
    logging.debug('Got health: %s', str(stream_health))
    self.assertNotIn('serving', stream_health)
    self.assertIn('realtime_stats', stream_health)
    self.assertNotIn('health_error', stream_health['realtime_stats'])
    self.assertIn('binlog_players_count', stream_health['realtime_stats'])
    self.assertEqual(blp_stats['BinlogPlayerMapSize'],
                     stream_health['realtime_stats']['binlog_players_count'])
    self.assertEqual(blp_stats['BinlogPlayerSecondsBehindMaster'],
                     stream_health['realtime_stats'].get(
                         'seconds_behind_master_filtered_replication', 0))

 def wait_for_vttablet_state(self, expected, timeout=60.0, port=None):
   expr = re.compile('^' + expected + '$')
   while True:
     v = utils.get_vars(port or self.port)
     last_seen_state = '?'
     if v is None:
       if self.proc.poll() is not None:
         raise utils.TestError(
             'vttablet died while test waiting for state %s' % expected)
       logging.debug(
           '  vttablet %s not answering at /debug/vars, waiting...',
           self.tablet_alias)
     else:
       if 'TabletStateName' not in v:
         logging.debug(
             '  vttablet %s not exporting TabletStateName, waiting...',
             self.tablet_alias)
       else:
         s = v['TabletStateName']
         last_seen_state = s
         if expr.match(s):
           break
         else:
           logging.debug(
               '  vttablet %s in state %s != %s', self.tablet_alias, s,
               expected)
     timeout = utils.wait_step(
         'waiting for %s state %s (last seen state: %s)' %
         (self.tablet_alias, expected, last_seen_state),
         timeout, sleep_time=0.1)

  def wait_for_binlog_server_state(self, expected, timeout=30.0):
    """Wait for the tablet's binlog server to be in the provided state.

    Args:
      expected: the state to wait for.
      timeout: how long to wait before error.
    """
    while True:
      v = utils.get_vars(self.port)
      if v is None:
        if self.proc.poll() is not None:
          raise utils.TestError(
              'vttablet died while test waiting for binlog state %s' %
              expected)
        logging.debug('  vttablet not answering at /debug/vars, waiting...')
      else:
        if 'UpdateStreamState' not in v:
          logging.debug(
              '  vttablet not exporting BinlogServerState, waiting...')
        else:
          s = v['UpdateStreamState']
          if s != expected:
            logging.debug("  vttablet's binlog server in state %s != %s", s,
                          expected)
          else:
            break
      timeout = utils.wait_step(
          'waiting for binlog server state %s' % expected,
          timeout, sleep_time=0.5)
    logging.debug('tablet %s binlog service is in state %s',
                  self.tablet_alias, expected)

  def test_vtgate_qps(self):
    # create the topology
    utils.run_vtctl('CreateKeyspace test_keyspace')
    t = tablet.Tablet(tablet_uid=1, cell="nj")
    t.init_tablet("master", "test_keyspace", "0")
    t.update_addrs()
    utils.run_vtctl('RebuildShardGraph test_keyspace/0', auto_log=True)
    utils.run_vtctl('RebuildKeyspaceGraph test_keyspace', auto_log=True)

    # start vtgate and the qps-er
    vtgate_proc, vtgate_port = utils.vtgate_start()
    qpser = utils.run_bg(utils.vtroot+'/bin/zkclient2 -server localhost:%u -mode qps2 test_nj test_keyspace' % vtgate_port)
    time.sleep(10)
    utils.kill_sub_process(qpser)

    # get the vtgate vars, make sure we have what we need
    v = utils.get_vars(vtgate_port)

    # some checks on performance / stats
    # a typical workstation will do 38-40k QPS, check we have more than 15k
    rpcCalls = v['TopoReaderRpcQueryCount']['test_nj']
    if rpcCalls < 150000:
      self.fail('QPS is too low: %u < 15000' % (rpcCalls / 10))
    else:
      logging.debug("Recorded qps: %u", rpcCalls / 10)
    utils.vtgate_kill(vtgate_proc)

 def _check_stats(self):
   v = utils.get_vars(self.vtgate_port)
   self.assertEqual(v['VttabletCall']['Histograms']['Execute.source_keyspace.0.replica']['Count'], 2, "unexpected value for VttabletCall(Execute.source_keyspace.0.replica) inside %s" % str(v))
   self.assertEqual(v['VtgateApi']['Histograms']['ExecuteKeyRanges.destination_keyspace.master']['Count'], 6, "unexpected value for VtgateApi(ExecuteKeyRanges.destination_keyspace.master) inside %s" % str(v))
   self.assertEqual(len(v['VtgateApiErrorCounts']), 0, "unexpected errors for VtgateApiErrorCounts inside %s" % str(v))
   self.assertEqual(v['EndpointCount']['test_nj.source_keyspace.0.master'], 1, "unexpected EndpointCount inside %s" % str(v))
   self.assertEqual(v['DegradedEndpointCount']['test_nj.source_keyspace.0.master'], 0, "unexpected DegradedEndpointCount inside %s" % str(v))

  def test_restart_during_action(self):
    # Start up a master mysql and vttablet
    utils.run_vtctl(['CreateKeyspace', 'test_keyspace'])

    tablet_62344.init_tablet('master', 'test_keyspace', '0')
    utils.run_vtctl(['RebuildShardGraph', 'test_keyspace/0'])
    utils.validate_topology()
    srvShard = utils.run_vtctl_json(['GetSrvShard', 'test_nj', 'test_keyspace/0'])
    self.assertEqual(srvShard['MasterCell'], 'test_nj')
    tablet_62344.create_db('vt_test_keyspace')
    tablet_62344.start_vttablet()

    utils.run_vtctl(['Ping', tablet_62344.tablet_alias])

    # schedule long action
    utils.run_vtctl(['-no-wait', 'Sleep', tablet_62344.tablet_alias, '15s'], stdout=utils.devnull)
    # ping blocks until the sleep finishes unless we have a schedule race
    action_path, _ = utils.run_vtctl(['-no-wait', 'Ping', tablet_62344.tablet_alias], trap_output=True)
    action_path = action_path.strip()

    # kill agent leaving vtaction running
    tablet_62344.kill_vttablet()

    # restart agent
    tablet_62344.start_vttablet()

    # we expect this action with a short wait time to fail. this isn't the best
    # and has some potential for flakiness.
    utils.run_vtctl(['-wait-time', '2s', 'WaitForAction', action_path],
                    expect_fail=True)

    # wait until the background sleep action is done, otherwise there will be
    # a leftover vtaction whose result may overwrite running actions
    # NOTE(alainjobart): Yes, I've seen it happen, it's a pain to debug:
    # the zombie Sleep clobbers the Clone command in the following tests
    utils.run_vtctl(['-wait-time', '20s', 'WaitForAction', action_path],
                    auto_log=True)

    if environment.topo_server_implementation == 'zookeeper':
      # extra small test: we ran for a while, get the states we were in,
      # make sure they're accounted for properly
      # first the query engine States
      v = utils.get_vars(tablet_62344.port)
      logging.debug("vars: %s" % str(v))

      # then the Zookeeper connections
      if v['ZkMetaConn']['test_nj']['Current'] != 'Connected':
        self.fail('invalid zk test_nj state: %s' %
                  v['ZkMetaConn']['test_nj']['Current'])
      if v['ZkMetaConn']['global']['Current'] != 'Connected':
        self.fail('invalid zk global state: %s' %
                  v['ZkMetaConn']['global']['Current'])
      if v['ZkMetaConn']['test_nj']['DurationConnected'] < 10e9:
        self.fail('not enough time in Connected state: %u',
                  v['ZkMetaConn']['test_nj']['DurationConnected'])
      if v['TabletType'] != 'master':
        self.fail('TabletType not exported correctly')

    tablet_62344.kill_vttablet()

 def test_purge_cache(self):
   utils.debug("===========test_purge_cache=========")
   cache_counters = framework.MultiDict(utils.get_vars(replica_tablet.port))['CacheCounters']
   utils.debug("cache counters %s" % cache_counters)
   try:
     purge_cache_counter = cache_counters['PurgeCache']
   except KeyError, e:
     purge_cache_counter = 0

def wait_for_vars(var, key, value):
  timeout = 20.0
  while True:
    v = utils.get_vars(utils.vtgate.port)
    if v and var in v and key in v[var] and v[var][key] == value:
      break
    timeout = utils.wait_step(
        'waiting for /debug/vars of %s/%s' % (var, key),
        timeout)