// RebuildKeyspace rebuilds the serving graph data while locking out other changes.
430354
func RebuildKeyspace(ctx context.Context, log logutil.Logger, ts *topo.Server, keyspace string, cells []string) (err error) {
	ctx, unlock, lockErr := ts.LockKeyspace(ctx, keyspace, "RebuildKeyspace")
	if lockErr != nil {
		return lockErr
	}
	defer unlock(&err)

	return RebuildKeyspaceLocked(ctx, log, ts, keyspace, cells)
}
1089136
func (profile *MonitorProfile) ReBuild(timeSpan string, tags map[string][]string) (*MonitorProfile, error) {
	newProfile := MonitorProfile{
		MetricConfigs:    make([]MetricConfig, len(profile.MetricConfigs)),
		GraphConfigs:     make([]GraphConfig, len(profile.GraphConfigs)),
		ThresholdConfigs: make([]ThresholdConfig, len(profile.ThresholdConfigs)),
	}

	build, err := NewMetricConfigBuilder("/metrics/api/performance/query", "POST")
	if err != nil {
		return nil, err
	}

	//rebuild metrics
	for i := range profile.MetricConfigs {
		metricGroup := &profile.MetricConfigs[i]
		for j := range metricGroup.Metrics {
			metric := metricGroup.Metrics[j]
			build.Metric(metric).SetTags(tags)
		}

		config, err := build.Config(metricGroup.ID, metricGroup.Name, metricGroup.Description, timeSpan)
		if err != nil {
			return nil, err
		}

		newProfile.MetricConfigs[i] = *config
	}

	//rebuild graphs
	for i := range profile.GraphConfigs {
		newProfile.GraphConfigs[i] = profile.GraphConfigs[i]
		newProfile.GraphConfigs[i].Tags = tags
	}

	//rebuild thresholds
	for i := range profile.ThresholdConfigs {
		newProfile.ThresholdConfigs[i] = profile.ThresholdConfigs[i]
	}

	//build a map of metric -> metricSource -> units
	unitMap := map[string]map[string]string{}
	for i := range newProfile.MetricConfigs {
		metricGroup := &newProfile.MetricConfigs[i]
		for j := range metricGroup.Metrics {
			metric := &metricGroup.Metrics[j]
			if metric.Unit != "" {
				if _, ok := unitMap[metricGroup.ID]; !ok {
					unitMap[metricGroup.ID] = map[string]string{}
				}
				unitMap[metricGroup.ID][metric.ID] = metric.Unit
			}
		}
	}

	//set graph units
	for i := range newProfile.GraphConfigs {
		graph := &newProfile.GraphConfigs[i]
		if graph.Units == "" {
			for j := range graph.DataPoints {
				dataPoint := &graph.DataPoints[j]
				if _, ok := unitMap[dataPoint.MetricSource]; !ok {
					continue
				}
				if _, ok := unitMap[dataPoint.MetricSource][dataPoint.Metric]; !ok {
					continue
				}
				graph.Units = unitMap[dataPoint.MetricSource][dataPoint.Metric]
				break
			}
		}
	}

	return &newProfile, nil
}
749096
func (idx *index) rebuild() {
	// initialize trees
	if idx.less != nil {
		idx.btr = btree.New(btreeDegrees, idx)
	}
	if idx.rect != nil {
		idx.rtr = rtree.New(idx)
	}
	// iterate through all keys and fill the index
	idx.db.keys.Ascend(func(item btree.Item) bool {
		dbi := item.(*dbItem)
		if !idx.match(dbi.key) {
			// does not match the pattern, conintue
			return true
		}
		if idx.less != nil {
			idx.btr.ReplaceOrInsert(dbi)
		}
		if idx.rect != nil {
			idx.rtr.Insert(dbi)
		}
		return true
	})
}
1160996
func (ts *Server) FindAllShardsInKeyspace(ctx context.Context, keyspace string) (map[string]*ShardInfo, error) {
	shards, err := ts.GetShardNames(ctx, keyspace)
	if err != nil {
		return nil, vterrors.Wrapf(err, "failed to get list of shards for keyspace '%v'", keyspace)
	}

	result := make(map[string]*ShardInfo, len(shards))
	wg := sync.WaitGroup{}
	mu := sync.Mutex{}
	rec := concurrency.FirstErrorRecorder{}
	for _, shard := range shards {
		wg.Add(1)
		go func(shard string) {
			defer wg.Done()
			si, err := ts.GetShard(ctx, keyspace, shard)
			if err != nil {
				if IsErrType(err, NoNode) {
					log.Warningf("GetShard(%v, %v) returned ErrNoNode, consider checking the topology.", keyspace, shard)
				} else {
					rec.RecordError(vterrors.Wrapf(err, "GetShard(%v, %v) failed", keyspace, shard))
				}
				return
			}
			mu.Lock()
			result[shard] = si
			mu.Unlock()
		}(shard)
	}
	wg.Wait()
	if rec.HasErrors() {
		return nil, rec.Error()
	}
	return result, nil
}
1112015
func CreateKeyspaceWithNetworkTopologyStrategy(c *gocql.ClusterConfig, ks string, rfmap map[string]uint) error {
	kis, err := GetKeyspaces(c)
	if err != nil {
		return err
	}
	kss := set.NewStringSet(kis)
	if !kss.Contains(ks) {
		log.Printf("Creating keyspace: %v\n", ks)
		c.Keyspace = ""
		s, err := c.CreateSession()
		if err != nil {
			return err
		}
		defer s.Close()
		q := fmt.Sprintf("CREATE KEYSPACE IF NOT EXISTS %v WITH REPLICATION = {'class': '%v', ", ks, "NetworkTopologyStrategy")
		rfsl := []string{}
		for dc, rf := range rfmap {
			rfsl = append(rfsl, fmt.Sprintf("'%v' : %v", dc, rf))
		}
		q = fmt.Sprintf("%v%v};", q, strings.Join(rfsl, ", "))
		err = s.Query(q).Exec()
		if err != nil {
			return err
		}
	}
	c.Keyspace = ks
	return nil
}
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
}
790833
func newKeyspaceIDResolverFactoryV3(ctx context.Context, ts *topo.Server, keyspace string, cell string) (keyspaceIDResolverFactory, error) {
	srvVSchema, err := ts.GetSrvVSchema(ctx, cell)
	if err != nil {
		return nil, err
	}
	kschema, ok := srvVSchema.Keyspaces[keyspace]
	if !ok {
		return nil, fmt.Errorf("SrvVSchema has no entry for keyspace %v", keyspace)
	}
	keyspaceSchema, err := vindexes.BuildKeyspaceSchema(kschema, keyspace)
	if err != nil {
		return nil, fmt.Errorf("cannot build vschema for keyspace %v: %v", keyspace, err)
	}
	return func(table *schema.Table) (int, keyspaceIDResolver, error) {
		// Find the v3 schema.
		tableSchema, ok := keyspaceSchema.Tables[table.Name.String()]
		if !ok {
			return -1, nil, fmt.Errorf("no vschema definition for table %v", table.Name)
		}

		// use the lowest cost unique vindex as the sharding key
		colVindex, err := vindexes.FindVindexForSharding(table.Name.String(), tableSchema.ColumnVindexes)
		if err != nil {
			return -1, nil, err
		}

		// TODO @rafael - when rewriting the mapping function, this will need to change.
		// for now it's safe to assume the sharding key will be always on index 0.
		shardingColumnName := colVindex.Columns[0].String()
		for i, col := range table.Columns {
			if col.Name.EqualString(shardingColumnName) {
				// We found the column.
				return i, &keyspaceIDResolverFactoryV3{
					vindex: colVindex.Vindex,
				}, nil
			}
		}
		// The column was not found.
		return -1, nil, fmt.Errorf("cannot find column %v in table %v", shardingColumnName, table.Name)
	}, nil
}
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
}
877500
func (q *Query) Keyspace() string {
	if q.session == nil {
		return ""
	}
	// TODO(chbannis): this should be parsed from the query or we should let
	// this be set by users.
	return q.session.cfg.Keyspace
}
487972
func (ts *Server) UpdateSrvKeyspace(ctx context.Context, cell, keyspace string, srvKeyspace *topodatapb.SrvKeyspace) error {
	conn, err := ts.ConnForCell(ctx, cell)
	if err != nil {
		return err
	}

	nodePath := srvKeyspaceFileName(keyspace)
	data, err := proto.Marshal(srvKeyspace)
	if err != nil {
		return err
	}
	_, err = conn.Update(ctx, nodePath, data, nil)
	return err
}
474076
func (wr *Wrangler) ReloadSchemaKeyspace(ctx context.Context, keyspace string, concurrency *sync2.Semaphore, includeMaster bool) error {
	shards, err := wr.ts.GetShardNames(ctx, keyspace)
	if err != nil {
		return fmt.Errorf("GetShardNames(%v) failed: %v", keyspace, err)
	}

	for _, shard := range shards {
		wr.ReloadSchemaShard(ctx, keyspace, shard, "" /* waitPosition */, concurrency, includeMaster)
	}
	return nil
}
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
}
618487
func VtctldSrvKeyspace(cell, keyspace string) template.HTML {
	return MakeVtctldRedirect(keyspace, map[string]string{
		"type":     "srv_keyspace",
		"cell":     cell,
		"keyspace": keyspace,
	})
}
599380
func (agent *ActionAgent) maybeRebuildKeyspace(cell, keyspace string) {
	_, err := agent.TopoServer.GetSrvKeyspace(agent.batchCtx, cell, keyspace)
	switch {
	case err == nil:
		// SrvKeyspace exists, we're done
		log.Infof("SrvKeyspace(%v,%v) exists, not building it", cell, keyspace)
		return
	case topo.IsErrType(err, topo.NoNode):
		log.Infof("SrvKeyspace(%v,%v) doesn't exist, rebuilding it", cell, keyspace)
		// SrvKeyspace doesn't exist, we'll try to rebuild it
	default:
		log.Warningf("Cannot read SrvKeyspace(%v,%v) (may need to run 'vtctl RebuildKeyspaceGraph %v'), skipping rebuild: %v", cell, keyspace, keyspace, err)
		return
	}

	if err := topotools.RebuildKeyspace(agent.batchCtx, logutil.NewConsoleLogger(), agent.TopoServer, keyspace, []string{cell}); err != nil {
		log.Warningf("RebuildKeyspace(%v,%v) failed: %v, may need to run 'vtctl RebuildKeyspaceGraph %v')", cell, keyspace, err, keyspace)
		return
	}

	if err := topotools.RebuildVSchema(agent.batchCtx, logutil.NewConsoleLogger(), agent.TopoServer, []string{cell}); err != nil {
		log.Warningf("RebuildVSchema(%v) failed: %v, may need to run 'vtctl RebuildVSchemaGraph --cells %v", cell, err, cell)
	}
}
757323
func RebuildVSchema(ctx context.Context, log logutil.Logger, ts *topo.Server, cells []string) error {
	// get the actual list of cells
	if len(cells) == 0 {
		var err error
		cells, err = ts.GetKnownCells(ctx)
		if err != nil {
			return fmt.Errorf("GetKnownCells failed: %v", err)
		}
	}

	// get the keyspaces
	keyspaces, err := ts.GetKeyspaces(ctx)
	if err != nil {
		return fmt.Errorf("GetKeyspaces failed: %v", err)
	}

	// build the SrvVSchema in parallel, protected by mu
	wg := sync.WaitGroup{}
	mu := sync.Mutex{}
	var finalErr error
	srvVSchema := &vschemapb.SrvVSchema{
		Keyspaces: map[string]*vschemapb.Keyspace{},
	}
	for _, keyspace := range keyspaces {
		wg.Add(1)
		go func(keyspace string) {
			defer wg.Done()

			k, err := ts.GetVSchema(ctx, keyspace)
			if topo.IsErrType(err, topo.NoNode) {
				err = nil
				k = &vschemapb.Keyspace{}
			}

			mu.Lock()
			defer mu.Unlock()
			if err != nil {
				log.Errorf2(err, "GetVSchema(%v) failed", keyspace)
				finalErr = err
				return
			}
			srvVSchema.Keyspaces[keyspace] = k
		}(keyspace)
	}
	wg.Wait()
	if finalErr != nil {
		return finalErr
	}

	rr, err := ts.GetRoutingRules(ctx)
	if err != nil {
		return fmt.Errorf("GetRoutingRules failed: %v", err)
	}
	srvVSchema.RoutingRules = rr

	// now save the SrvVSchema in all cells in parallel
	for _, cell := range cells {
		wg.Add(1)
		go func(cell string) {
			defer wg.Done()
			if err := ts.UpdateSrvVSchema(ctx, cell, srvVSchema); err != nil {
				log.Errorf2(err, "UpdateSrvVSchema(%v) failed", cell)
				mu.Lock()
				finalErr = err
				mu.Unlock()
			}
		}(cell)
	}
	wg.Wait()

	return finalErr
}
221657
func (api *API) ReloadSpaces() error {
	if api.facade.BestAPIVersion() < 3 {
		return errors.NewNotSupported(nil, "Controller does not support reloading spaces")
	}
	err := api.facade.FacadeCall("ReloadSpaces", nil, nil)
	if params.IsCodeNotSupported(err) {
		return errors.NewNotSupported(nil, err.Error())
	}
	return err
}
1151133
func (k Key) BaseNamespace() string {
	n := k.Namespaces()
	return n[len(n)-1]
}