// keyspacesWithServedFrom returns all the keyspaces that have ServedFrom set
// to one value.
490553
func keyspacesWithServedFrom(ctx context.Context, wr *wrangler.Wrangler) ([]string, error) {
	shortCtx, cancel := context.WithTimeout(ctx, *remoteActionsTimeout)
	keyspaces, err := wr.TopoServer().GetKeyspaces(shortCtx)
	cancel()
	if err != nil {
		return nil, vterrors.Wrap(err, "failed to get list of keyspaces")
	}

	wg := sync.WaitGroup{}
	mu := sync.Mutex{} // protects result
	result := make([]string, 0, len(keyspaces))
	rec := concurrency.AllErrorRecorder{}
	for _, keyspace := range keyspaces {
		wg.Add(1)
		go func(keyspace string) {
			defer wg.Done()
			shortCtx, cancel := context.WithTimeout(ctx, *remoteActionsTimeout)
			ki, err := wr.TopoServer().GetKeyspace(shortCtx, keyspace)
			cancel()
			if err != nil {
				rec.RecordError(vterrors.Wrapf(err, "failed to get details for keyspace '%v'", keyspace))
				return
			}
			if len(ki.ServedFroms) > 0 {
				mu.Lock()
				result = append(result, keyspace)
				mu.Unlock()
			}
		}(keyspace)
	}
	wg.Wait()

	if rec.HasErrors() {
		return nil, rec.Error()
	}
	if len(result) == 0 {
		return nil, fmt.Errorf("there are no keyspaces with ServedFrom")
	}
	return result, nil
}
522216
func (nm *NamespaceController) syncNamespaceFromKey(key string) (err error) {
	startTime := time.Now()
	defer func() {
		klog.V(4).Infof("Finished syncing namespace %q (%v)", key, time.Since(startTime))
	}()

	namespace, err := nm.lister.Get(key)
	if errors.IsNotFound(err) {
		klog.Infof("Namespace has been deleted %v", key)
		return nil
	}
	if err != nil {
		utilruntime.HandleError(fmt.Errorf("Unable to retrieve namespace %v from store: %v", key, err))
		return err
	}
	return nm.namespacedResourcesDeleter.Delete(namespace.Name)
}
1249557
func (wr *Wrangler) MigrateServedFrom(ctx context.Context, keyspace, shard string, servedType topodatapb.TabletType, cells []string, reverse bool, filteredReplicationWaitTime time.Duration) (err error) {
	// read the destination keyspace, check it
	ki, err := wr.ts.GetKeyspace(ctx, keyspace)
	if err != nil {
		return err
	}
	if len(ki.ServedFroms) == 0 {
		return fmt.Errorf("destination keyspace %v is not a vertical split target", keyspace)
	}

	// read the destination shard, check it
	si, err := wr.ts.GetShard(ctx, keyspace, shard)
	if err != nil {
		return err
	}
	if len(si.SourceShards) != 1 || len(si.SourceShards[0].Tables) == 0 {
		return fmt.Errorf("destination shard %v/%v is not a vertical split target", keyspace, shard)
	}

	// check the migration is valid before locking (will also be checked
	// after locking to be sure)
	sourceKeyspace := si.SourceShards[0].Keyspace
	if err := ki.CheckServedFromMigration(servedType, cells, sourceKeyspace, !reverse); err != nil {
		return err
	}

	// lock the keyspaces, source first.
	ctx, unlock, lockErr := wr.ts.LockKeyspace(ctx, sourceKeyspace, fmt.Sprintf("MigrateServedFrom(%v)", servedType))
	if lockErr != nil {
		return lockErr
	}
	defer unlock(&err)
	ctx, unlock, lockErr = wr.ts.LockKeyspace(ctx, keyspace, fmt.Sprintf("MigrateServedFrom(%v)", servedType))
	if lockErr != nil {
		return lockErr
	}
	defer unlock(&err)

	// execute the migration
	err = wr.migrateServedFromLocked(ctx, ki, si, servedType, cells, reverse, filteredReplicationWaitTime)

	// rebuild the keyspace serving graph if there was no error
	if err == nil {
		err = topotools.RebuildKeyspaceLocked(ctx, wr.logger, wr.ts, keyspace, cells)
	}

	return err
}
829747
func (vc *vcursorImpl) DefaultKeyspace() (*vindexes.Keyspace, error) {
	if vc.keyspace == "" {
		return nil, errNoKeyspace
	}
	ks, ok := vc.executor.VSchema().Keyspaces[vc.keyspace]
	if !ok {
		return nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "keyspace %s not found in vschema", vc.keyspace)
	}
	return ks.Keyspace, nil
}
36888
func checkTrustedKeyString(keys string) []string {
	tks := strings.Fields(keys)
	if len(tks) == 0 {
		return nil
	}
	// Walk all the keys and make sure they are valid.
	for _, key := range tks {
		if !nkeys.IsValidPublicOperatorKey(key) {
			return nil
		}
	}
	return tks
}
477764
func (itr *combinedIterator) serveEndKeyIfNeeded() (statedb.QueryResult, error) {
	if !itr.includeEndKey || itr.endKeyServed {
		logger.Debugf("Endkey not to be served. Returning nil... [toInclude=%t, alreadyServed=%t]",
			itr.includeEndKey, itr.endKeyServed)
		return nil, nil
	}
	logger.Debug("Serving the endKey")
	var vv *statedb.VersionedValue
	var err error
	vv = itr.updates.Get(itr.ns, itr.endKey)
	logger.Debugf("endKey value from updates:%s", vv)
	if vv == nil {
		if vv, err = itr.db.GetState(itr.ns, itr.endKey); err != nil {
			return nil, err
		}
		logger.Debugf("endKey value from stateDB:%s", vv)
	}
	itr.endKeyServed = true
	if vv == nil {
		return nil, nil
	}
	vkv := &statedb.VersionedKV{
		CompositeKey:   statedb.CompositeKey{Namespace: itr.ns, Key: itr.endKey},
		VersionedValue: statedb.VersionedValue{Value: vv.Value, Version: vv.Version}}

	if isDelete(vkv) {
		return nil, nil
	}
	return vkv, nil
}
157274
func (config *DeferredLoadingClientConfig) Namespace() (string, bool, error) {
	mergedKubeConfig, err := config.createClientConfig()
	if err != nil {
		return "", false, err
	}

	ns, overridden, err := mergedKubeConfig.Namespace()
	// if we get an error and it is not empty config, or if the merged config defined an explicit namespace, or
	// if in-cluster config is not possible, return immediately
	if (err != nil && !IsEmptyConfig(err)) || overridden || !config.icc.Possible() {
		// return on any error except empty config
		return ns, overridden, err
	}

	if len(ns) > 0 {
		// if we got a non-default namespace from the kubeconfig, use it
		if ns != "default" {
			return ns, false, nil
		}

		// if we got a default namespace, determine whether it was explicit or implicit
		if raw, err := mergedKubeConfig.RawConfig(); err == nil {
			// determine the current context
			currentContext := raw.CurrentContext
			if config.overrides != nil && len(config.overrides.CurrentContext) > 0 {
				currentContext = config.overrides.CurrentContext
			}
			if context := raw.Contexts[currentContext]; context != nil && len(context.Namespace) > 0 {
				return ns, false, nil
			}
		}
	}

	klog.V(4).Infof("Using in-cluster namespace")

	// allow the namespace from the service account token directory to be used.
	return config.icc.Namespace()
}
875063
func (e *Executor) IsKeyspaceRangeBasedSharded(keyspace string) bool {
	vschema := e.VSchema()
	ks, ok := vschema.Keyspaces[keyspace]
	if !ok {
		return false
	}
	if ks.Keyspace == nil {
		return false
	}
	return ks.Keyspace.Sharded
}
189070
func (s *Filters) SetExtendedKeyUsage(v []*string) *Filters {
	s.ExtendedKeyUsage = v
	return s
}
861291
func (server *ResilientServer) GetSrvKeyspace(ctx context.Context, cell, keyspace string) (*topodatapb.SrvKeyspace, error) {
	entry := server.getSrvKeyspaceEntry(cell, keyspace)

	// If the watch is already running, return the value
	entry.mutex.RLock()
	if entry.watchState == watchStateRunning {
		v, e := entry.value, entry.lastError
		entry.mutex.RUnlock()
		return v, e
	}
	entry.mutex.RUnlock()

	entry.mutex.Lock()
	defer entry.mutex.Unlock()

	// If the watch is already running (now that we have the write lock),
	// return the value
	if entry.watchState == watchStateRunning {
		return entry.value, entry.lastError
	}

	// Watch is not running. Start a new one if it is time to use it and if
	// there isn't one already one  in the process of being started.
	shouldRefresh := time.Since(entry.lastErrorTime) > server.cacheRefresh
	if shouldRefresh && (entry.watchState == watchStateIdle) {
		entry.watchState = watchStateStarting
		entry.watchStartingChan = make(chan struct{})
		go server.watchSrvKeyspace(ctx, entry, cell, keyspace)
	}

	// If the cached value is still valid, use it. Otherwise wait
	// for the watch attempt to complete to get a more up to date
	// response.
	//
	// In the event that the topo service is slow or unresponsive either
	// on the initial fetch or if the cache TTL expires, then several
	// requests could be blocked waiting for the response to come back.
	cacheValid := entry.value != nil && time.Since(entry.lastValueTime) < server.cacheTTL
	if cacheValid {
		server.counts.Add(cachedCategory, 1)
		return entry.value, nil
	}

	if entry.watchState == watchStateStarting {
		watchStartingChan := entry.watchStartingChan
		entry.mutex.Unlock()
		select {
		case <-watchStartingChan:
		case <-ctx.Done():
			entry.mutex.Lock()
			return nil, fmt.Errorf("timed out waiting for keyspace")
		}
		entry.mutex.Lock()
	}

	if entry.value != nil {
		return entry.value, nil
	}

	return nil, entry.lastError
}
478434
func keyFromSub(sub *subscription) string {
	var _rkey [1024]byte
	var key []byte

	if sub.queue != nil {
		// Just make the key subject spc group, e.g. 'foo bar'
		key = _rkey[:0]
		key = append(key, sub.subject...)
		key = append(key, byte(' '))
		key = append(key, sub.queue...)
	} else {
		key = sub.subject
	}
	return string(key)
}
227199
func (ts *Server) WatchSrvKeyspace(ctx context.Context, cell, keyspace string) (*WatchSrvKeyspaceData, <-chan *WatchSrvKeyspaceData, CancelFunc) {
	conn, err := ts.ConnForCell(ctx, cell)
	if err != nil {
		return &WatchSrvKeyspaceData{Err: err}, nil, nil
	}

	filePath := srvKeyspaceFileName(keyspace)
	current, wdChannel, cancel := conn.Watch(ctx, filePath)
	if current.Err != nil {
		return &WatchSrvKeyspaceData{Err: current.Err}, nil, nil
	}
	value := &topodatapb.SrvKeyspace{}
	if err := proto.Unmarshal(current.Contents, value); err != nil {
		// Cancel the watch, drain channel.
		cancel()
		for range wdChannel {
		}
		return &WatchSrvKeyspaceData{Err: vterrors.Wrapf(err, "error unpacking initial SrvKeyspace object")}, nil, nil
	}

	changes := make(chan *WatchSrvKeyspaceData, 10)

	// The background routine reads any event from the watch channel,
	// translates it, and sends it to the caller.
	// If cancel() is called, the underlying Watch() code will
	// send an ErrInterrupted and then close the channel. We'll
	// just propagate that back to our caller.
	go func() {
		defer close(changes)

		for wd := range wdChannel {
			if wd.Err != nil {
				// Last error value, we're done.
				// wdChannel will be closed right after
				// this, no need to do anything.
				changes <- &WatchSrvKeyspaceData{Err: wd.Err}
				return
			}

			value := &topodatapb.SrvKeyspace{}
			if err := proto.Unmarshal(wd.Contents, value); err != nil {
				cancel()
				for range wdChannel {
				}
				changes <- &WatchSrvKeyspaceData{Err: vterrors.Wrapf(err, "error unpacking SrvKeyspace object")}
				return
			}

			changes <- &WatchSrvKeyspaceData{Value: value}
		}
	}()

	return &WatchSrvKeyspaceData{Value: value}, changes, cancel
}
756976
func (r *v3Resolver) keyspaceID(row []sqltypes.Value) ([]byte, error) {
	v := row[r.shardingColumnIndex]
	destinations, err := r.vindex.Map(nil, []sqltypes.Value{v})
	if err != nil {
		return nil, err
	}
	if len(destinations) != 1 {
		return nil, vterrors.Errorf(vtrpc.Code_INTERNAL, "mapping row to keyspace id returned an invalid array of keyspace ids: %v", key.DestinationsString(destinations))
	}
	ksid, ok := destinations[0].(key.DestinationKeyspaceID)
	if !ok || len(ksid) == 0 {
		return nil, vterrors.Errorf(vtrpc.Code_INTERNAL, "could not map %v to a keyspace id, got destination %v", v, destinations[0])
	}
	return ksid, nil
}
1115746
func (s *ListPoliciesGrantingServiceAccessInput) SetServiceNamespaces(v []*string) *ListPoliciesGrantingServiceAccessInput {
	s.ServiceNamespaces = v
	return s
}
359805
func (ts *Server) UpdateKeyspace(ctx context.Context, ki *KeyspaceInfo) error {
	// make sure it is locked first
	if err := CheckKeyspaceLocked(ctx, ki.keyspace); err != nil {
		return err
	}

	data, err := proto.Marshal(ki.Keyspace)
	if err != nil {
		return err
	}
	keyspacePath := path.Join(KeyspacesPath, ki.keyspace, KeyspaceFile)
	version, err := ts.globalCell.Update(ctx, keyspacePath, data, ki.version)
	if err != nil {
		return err
	}
	ki.version = version

	event.Dispatch(&events.KeyspaceChange{
		KeyspaceName: ki.keyspace,
		Keyspace:     ki.Keyspace,
		Status:       "updated",
	})
	return nil
}
224776
func keyspaceShardsToSets(list string) (map[string]bool, map[string]bool) {
	keyspaces := make(map[string]bool)
	shards := make(map[string]bool)
	if list == "" {
		return keyspaces, shards
	}

	for _, item := range strings.Split(list, ",") {
		if strings.Contains(item, "/") {
			shards[item] = true
		} else {
			keyspaces[item] = true
		}
	}
	return keyspaces, shards
}
1164807
func (vtg *VTGate) GetSrvKeyspace(ctx context.Context, keyspace string) (*topodatapb.SrvKeyspace, error) {
	return vtg.resolver.toposerv.GetSrvKeyspace(ctx, vtg.resolver.cell, keyspace)
}