func (zkts *Server) UpdateKeyspace(ki *topo.KeyspaceInfo) error {
	keyspacePath := path.Join(globalKeyspacesPath, ki.KeyspaceName())
	data := jscfg.ToJson(ki.Keyspace)
	_, err := zkts.zconn.Set(keyspacePath, data, -1)
	if err != nil {
		if zookeeper.IsError(err, zookeeper.ZNONODE) {
			// The code should be:
			//   err = topo.ErrNoNode
			// Temporary code until we have Keyspace object
			// everywhere:
			_, err = zkts.zconn.Create(keyspacePath, data, 0, zookeeper.WorldACL(zookeeper.PERM_ALL))
			if err != nil {
				if zookeeper.IsError(err, zookeeper.ZNONODE) {
					// the directory doesn't even exist
					err = topo.ErrNoNode
				}
				return err
			}

			event.Dispatch(&events.KeyspaceChange{
				KeyspaceInfo: *ki,
				Status:       "updated (had to create Keyspace object)",
			})
			return nil
		}
		return err
	}

	event.Dispatch(&events.KeyspaceChange{
		KeyspaceInfo: *ki,
		Status:       "updated",
	})
	return nil
}

func (zkts *Server) CreateKeyspace(keyspace string, value *topo.Keyspace) error {
	keyspacePath := path.Join(globalKeyspacesPath, keyspace)
	pathList := []string{
		keyspacePath,
		path.Join(keyspacePath, "action"),
		path.Join(keyspacePath, "actionlog"),
		path.Join(keyspacePath, "shards"),
	}

	alreadyExists := false
	for i, zkPath := range pathList {
		c := ""
		if i == 0 {
			c = jscfg.ToJson(value)
		}
		_, err := zk.CreateRecursive(zkts.zconn, zkPath, c, 0, zookeeper.WorldACL(zookeeper.PERM_ALL))
		if err != nil {
			if zookeeper.IsError(err, zookeeper.ZNODEEXISTS) {
				alreadyExists = true
			} else {
				return fmt.Errorf("error creating keyspace: %v %v", zkPath, err)
			}
		}
	}
	if alreadyExists {
		return topo.ErrNodeExists
	}

	event.Dispatch(&events.KeyspaceChange{
		KeyspaceInfo: *topo.NewKeyspaceInfo(keyspace, value),
		Status:       "created",
	})
	return nil
}

// DeleteTablet is part of the topo.Server interface
func (zkts *Server) DeleteTablet(ctx context.Context, alias topo.TabletAlias) error {
	// We need to find out the keyspace and shard names because
	// those are required in the TabletChange event.
	ti, tiErr := zkts.GetTablet(ctx, alias)

	zkTabletPath := TabletPathForAlias(alias)
	err := zk.DeleteRecursive(zkts.zconn, zkTabletPath, -1)
	if err != nil {
		if zookeeper.IsError(err, zookeeper.ZNONODE) {
			err = topo.ErrNoNode
		}
		return err
	}

	// Only try to log if we have the required information.
	if tiErr == nil {
		// We only want to copy the identity info for the tablet (alias, etc.).
		// The rest has just been deleted, so it should be blank.
		event.Dispatch(&events.TabletChange{
			Tablet: topo.Tablet{
				Alias:    ti.Tablet.Alias,
				Keyspace: ti.Tablet.Keyspace,
				Shard:    ti.Tablet.Shard,
			},
			Status: "deleted",
		})
	}
	return nil
}

// DeleteTablet implements topo.Server.
func (s *Server) DeleteTablet(ctx context.Context, tabletAlias topo.TabletAlias) error {
	cell, err := s.getCell(tabletAlias.Cell)
	if err != nil {
		return err
	}

	// Get the keyspace and shard names for the TabletChange event.
	ti, tiErr := s.GetTablet(ctx, tabletAlias)

	_, err = cell.Delete(tabletDirPath(tabletAlias.String()), true /* recursive */)
	if err != nil {
		return convertError(err)
	}

	// Only try to log if we have the required info.
	if tiErr == nil {
		// Only copy the identity info for the tablet. The rest has been deleted.
		event.Dispatch(&events.TabletChange{
			Tablet: topo.Tablet{
				Alias:    ti.Tablet.Alias,
				Keyspace: ti.Tablet.Keyspace,
				Shard:    ti.Tablet.Shard,
			},
			Status: "deleted",
		})
	}
	return nil
}

// CreateShard implements topo.Server.
func (s *Server) CreateShard(ctx context.Context, keyspace, shard string, value *topo.Shard) error {
	data := jscfg.ToJSON(value)
	global := s.getGlobal()

	resp, err := global.Create(shardFilePath(keyspace, shard), data, 0 /* ttl */)
	if err != nil {
		return convertError(err)
	}
	if err := initLockFile(global, shardDirPath(keyspace, shard)); err != nil {
		return err
	}

	// We don't return ErrBadResponse in this case because the Create() suceeeded
	// and we don't really need the version to satisfy our contract - we're only
	// logging it.
	version := int64(-1)
	if resp.Node != nil {
		version = int64(resp.Node.ModifiedIndex)
	}
	event.Dispatch(&events.ShardChange{
		ShardInfo: *topo.NewShardInfo(keyspace, shard, value, version),
		Status:    "created",
	})
	return nil
}

// CreateKeyspace implements topo.Server.
func (s *Server) CreateKeyspace(ctx context.Context, keyspace string, value *pb.Keyspace) error {
	data, err := json.MarshalIndent(value, "", "  ")
	if err != nil {
		return err
	}
	global := s.getGlobal()

	resp, err := global.Create(keyspaceFilePath(keyspace), string(data), 0 /* ttl */)
	if err != nil {
		return convertError(err)
	}
	if err := initLockFile(global, keyspaceDirPath(keyspace)); err != nil {
		return err
	}

	// We don't return ErrBadResponse in this case because the Create() suceeeded
	// and we don't really need the version to satisfy our contract - we're only
	// logging it.
	version := int64(-1)
	if resp.Node != nil {
		version = int64(resp.Node.ModifiedIndex)
	}
	event.Dispatch(&events.KeyspaceChange{
		KeyspaceInfo: *topo.NewKeyspaceInfo(keyspace, value, version),
		Status:       "created",
	})
	return nil
}

// DeleteKeyspaceShards implements topo.Server.
func (s *Server) DeleteKeyspaceShards(ctx context.Context, keyspace string) error {
	shards, err := s.GetShardNames(ctx, keyspace)
	if err != nil {
		return err
	}

	wg := sync.WaitGroup{}
	rec := concurrency.AllErrorRecorder{}
	global := s.getGlobal()
	for _, shard := range shards {
		wg.Add(1)
		go func(shard string) {
			defer wg.Done()
			_, err := global.Delete(shardDirPath(keyspace, shard), true /* recursive */)
			rec.RecordError(convertError(err))
		}(shard)
	}
	wg.Wait()

	if err = rec.Error(); err != nil {
		return err
	}

	event.Dispatch(&events.KeyspaceChange{
		KeyspaceInfo: *topo.NewKeyspaceInfo(keyspace, nil, -1),
		Status:       "deleted all shards",
	})
	return nil
}

// UpdateShard updates the shard data, with the right version.
// It also creates a span, and dispatches the event.
func (ts Server) UpdateShard(ctx context.Context, si *ShardInfo) error {
	span := trace.NewSpanFromContext(ctx)
	span.StartClient("TopoServer.UpdateShard")
	span.Annotate("keyspace", si.keyspace)
	span.Annotate("shard", si.shardName)
	defer span.Finish()

	var version int64 = -1
	if si.version != 0 {
		version = si.version
	}

	newVersion, err := ts.Impl.UpdateShard(ctx, si.keyspace, si.shardName, si.Shard, version)
	if err != nil {
		return err
	}
	si.version = newVersion

	event.Dispatch(&events.ShardChange{
		KeyspaceName: si.Keyspace(),
		ShardName:    si.ShardName(),
		Shard:        si.Shard,
		Status:       "updated",
	})
	return nil
}

// UpdateTablet implements topo.Server.
func (s *Server) UpdateTablet(ctx context.Context, ti *topo.TabletInfo, existingVersion int64) (int64, error) {
	cell, err := s.getCell(ti.Alias.Cell)
	if err != nil {
		return -1, err
	}

	data, err := json.MarshalIndent(ti.Tablet, "", "  ")
	if err != nil {
		return -1, err
	}
	resp, err := cell.CompareAndSwap(tabletFilePath(ti.Alias),
		string(data), 0 /* ttl */, "" /* prevValue */, uint64(existingVersion))
	if err != nil {
		return -1, convertError(err)
	}
	if resp.Node == nil {
		return -1, ErrBadResponse
	}

	event.Dispatch(&events.TabletChange{
		Tablet: *ti.Tablet,
		Status: "updated",
	})
	return int64(resp.Node.ModifiedIndex), nil
}

func (zkts *Server) CreateTablet(tablet *topo.Tablet) error {
	zkTabletPath := TabletPathForAlias(tablet.Alias)

	// Create /zk/<cell>/vt/tablets/<uid>
	_, err := zk.CreateRecursive(zkts.zconn, zkTabletPath, tablet.Json(), 0, zookeeper.WorldACL(zookeeper.PERM_ALL))
	if err != nil {
		if zookeeper.IsError(err, zookeeper.ZNODEEXISTS) {
			err = topo.ErrNodeExists
		}
		return err
	}

	// Create /zk/<cell>/vt/tablets/<uid>/action
	tap := path.Join(zkTabletPath, "action")
	_, err = zkts.zconn.Create(tap, "", 0, zookeeper.WorldACL(zookeeper.PERM_ALL))
	if err != nil {
		return err
	}

	// Create /zk/<cell>/vt/tablets/<uid>/actionlog
	talp := path.Join(zkTabletPath, "actionlog")
	_, err = zkts.zconn.Create(talp, "", 0, zookeeper.WorldACL(zookeeper.PERM_ALL))
	if err != nil {
		return err
	}

	event.Dispatch(&events.TabletChange{
		Tablet: *tablet,
		Status: "created",
	})
	return nil
}

// CreateShard is part of the topo.Server interface
func (zkts *Server) CreateShard(ctx context.Context, keyspace, shard string, value *pb.Shard) error {
	shardPath := path.Join(globalKeyspacesPath, keyspace, "shards", shard)
	pathList := []string{
		shardPath,
		path.Join(shardPath, "action"),
		path.Join(shardPath, "actionlog"),
	}

	alreadyExists := false
	for i, zkPath := range pathList {
		c := ""
		if i == 0 {
			c = jscfg.ToJSON(value)
		}
		_, err := zk.CreateRecursive(zkts.zconn, zkPath, c, 0, zookeeper.WorldACL(zookeeper.PERM_ALL))
		if err != nil {
			if zookeeper.IsError(err, zookeeper.ZNODEEXISTS) {
				alreadyExists = true
			} else {
				return fmt.Errorf("error creating shard: %v %v", zkPath, err)
			}
		}
	}
	if alreadyExists {
		return topo.ErrNodeExists
	}

	event.Dispatch(&events.ShardChange{
		ShardInfo: *topo.NewShardInfo(keyspace, shard, value, -1),
		Status:    "created",
	})
	return nil
}

// UpdateTabletFields implements topo.Server.
func (s *Server) UpdateTabletFields(ctx context.Context, tabletAlias topo.TabletAlias, updateFunc func(*topo.Tablet) error) error {
	var ti *topo.TabletInfo
	var err error

	for {
		if ti, err = s.GetTablet(ctx, tabletAlias); err != nil {
			return err
		}
		if err = updateFunc(ti.Tablet); err != nil {
			return err
		}
		if _, err = s.UpdateTablet(ctx, ti, ti.Version()); err != topo.ErrBadVersion {
			break
		}
	}
	if err != nil {
		return err
	}

	event.Dispatch(&events.TabletChange{
		Tablet: *ti.Tablet,
		Status: "updated",
	})
	return nil
}

func TestSyslog(t *testing.T) {
	ev := new(TestEvent)
	event.Dispatch(ev)

	if !ev.triggered {
		t.Errorf("Syslog() was not called on event that implements Syslogger")
	}
}

// updateState will use the provided tablet record as a base, the current
// tablet record as the new one, run changeCallback, and dispatch the event.
func (agent *ActionAgent) updateState(ctx context.Context, oldTablet *topodatapb.Tablet, reason string) {
	newTablet := agent.Tablet()
	log.Infof("Running tablet callback because: %v", reason)
	agent.changeCallback(ctx, oldTablet, newTablet)
	event.Dispatch(&events.StateChange{
		OldTablet: *oldTablet,
		NewTablet: *newTablet,
		Reason:    reason,
	})
}

func TestInvalidSeverity(t *testing.T) {
	fw := &fakeWriter{}
	writer = fw

	event.Dispatch(&TestEvent{priority: syslog.Priority(123), message: "log me"})

	if fw.message == "log me" {
		t.Errorf("message was logged despite invalid severity")
	}
}

// TestBadWriter checks that we don't panic when the connection fails.
func TestBadWriter(t *testing.T) {
	writer = nil

	ev := new(TestEvent)
	event.Dispatch(ev)

	if ev.triggered {
		t.Errorf("passed nil writer to client")
	}
}

// DeleteKeyspace wraps the underlying Impl.DeleteKeyspace
// and dispatches the event.
func (ts Server) DeleteKeyspace(ctx context.Context, keyspace string) error {
	if err := ts.Impl.DeleteKeyspace(ctx, keyspace); err != nil {
		return err
	}
	event.Dispatch(&events.KeyspaceChange{
		KeyspaceName: keyspace,
		Keyspace:     nil,
		Status:       "deleted",
	})
	return nil
}

// CreateKeyspace wraps the underlying Impl.DeleteKeyspaceShards
// and dispatches the event.
func (ts Server) CreateKeyspace(ctx context.Context, keyspace string, value *pb.Keyspace) error {
	if err := ts.Impl.CreateKeyspace(ctx, keyspace, value); err != nil {
		return err
	}
	event.Dispatch(&events.KeyspaceChange{
		KeyspaceName: keyspace,
		Keyspace:     value,
		Status:       "created",
	})
	return nil
}

// DeleteShard wraps the underlying Impl.DeleteShard
// and dispatches the event.
func (ts Server) DeleteShard(ctx context.Context, keyspace, shard string) error {
	if err := ts.Impl.DeleteShard(ctx, keyspace, shard); err != nil {
		return err
	}
	event.Dispatch(&events.ShardChange{
		KeyspaceName: keyspace,
		ShardName:    shard,
		Shard:        nil,
		Status:       "deleted",
	})
	return nil
}

// DeleteKeyspaceShards is part of the topo.Server interface
func (zkts *Server) DeleteKeyspaceShards(ctx context.Context, keyspace string) error {
	shardsPath := path.Join(globalKeyspacesPath, keyspace, "shards")
	if err := zk.DeleteRecursive(zkts.zconn, shardsPath, -1); err != nil && !zookeeper.IsError(err, zookeeper.ZNONODE) {
		return err
	}

	event.Dispatch(&events.KeyspaceChange{
		KeyspaceInfo: *topo.NewKeyspaceInfo(keyspace, nil, -1),
		Status:       "deleted all shards",
	})
	return nil
}