// doShardedQueryStat launches a batch of queries, at different shard indices.
558791
func doShardedQueryStat(c context.Context, ns string, stat queryStat) error {
	results := &queryResult{}
	if err := parallel.FanOutIn(func(ch chan<- func() error) {
		for i := int64(0); i < totalShards; i++ {
			i := i
			ch <- func() error {
				result, err := doQueryStat(c, ns, stat, i)
				if err != nil {
					return errors.Annotate(err, "while launching index %d: %s", i, result).Err()
				}
				logging.Infof(c, "For index %d got %s", i, result)
				results.add(result)
				return nil
			}
		}
	}); err != nil {
		return err
	}

	// Report
	stat.metric.Set(c, results.notArchived, ns, "not_archived")
	stat.metric.Set(c, results.archiveTasked, ns, "archive_tasked")
	stat.metric.Set(c, results.archivedPartial, ns, "archived_partial")
	stat.metric.Set(c, results.archivedComplete, ns, "archived_complete")
	logging.Infof(c, "Stat %s Project %s stat: %s", stat.metric.Info().Name, ns, results)
	return nil
}
32125
func (f *fsClient) Stat(isIncomplete, isFetchMeta bool, sse encrypt.ServerSide) (content *clientContent, err *probe.Error) {
	st, err := f.fsStat(isIncomplete)
	if err != nil {
		return nil, err.Trace(f.PathURL.String())
	}

	content = &clientContent{}
	content.URL = *f.PathURL
	content.Size = st.Size()
	content.Time = st.ModTime()
	content.Type = st.Mode()
	content.Metadata = map[string]string{
		"Content-Type": guessURLContentType(f.PathURL.Path),
	}

	// isFetchMeta is true only in the case of mc stat command which lists any extended attributes
	// present for this object.
	if isFetchMeta {
		path := f.PathURL.String()
		metaData, pErr := getAllXattrs(path)
		if pErr != nil {
			return content, nil
		}
		for k, v := range metaData {
			content.Metadata[k] = v
		}
	}

	return content, nil
}
1072013
func (s *VolumeServer) Stats(
	ctx context.Context,
	req *api.SdkVolumeStatsRequest,
) (*api.SdkVolumeStatsResponse, error) {
	if s.cluster() == nil || s.driver(ctx) == nil {
		return nil, status.Error(codes.Unavailable, "Resource has not been initialized")
	}

	if len(req.GetVolumeId()) == 0 {
		return nil, status.Error(codes.InvalidArgument, "Must supply volume id")
	}

	// Get access rights
	if err := s.checkAccessForVolumeId(ctx, req.GetVolumeId(), api.Ownership_Read); err != nil {
		return nil, err
	}

	stats, err := s.driver(ctx).Stats(req.GetVolumeId(), !req.GetNotCumulative())
	if err != nil {
		return nil, status.Errorf(
			codes.Internal,
			"Failed to obtain stats for volume %s: %v",
			req.GetVolumeId(),
			err.Error())
	}

	return &api.SdkVolumeStatsResponse{
		Stats: stats,
	}, nil
}
1296667
func (e *executor) executeDifferenceShard(ctx context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeDifferenceShard")
	defer span.Finish()

	var other *Row
	if len(c.Children) == 0 {
		return nil, fmt.Errorf("empty Difference query is currently not supported")
	}
	for i, input := range c.Children {
		row, err := e.executeBitmapCallShard(ctx, index, input, shard)
		if err != nil {
			return nil, err
		}

		if i == 0 {
			other = row
		} else {
			other = other.Difference(row)
		}
	}
	other.invalidateCount()
	return other, nil
}
1373050
func NewShardReplicationInfo(sr *topodatapb.ShardReplication, cell, keyspace, shard string) *ShardReplicationInfo {
	return &ShardReplicationInfo{
		ShardReplication: sr,
		cell:             cell,
		keyspace:         keyspace,
		shard:            shard,
	}
}
402617
func (c *cluster) MoveShard(ctx context.Context, col Collection, shard ShardID, fromServer, toServer ServerID) error {
	req, err := c.conn.NewRequest("POST", "_admin/cluster/moveShard")
	if err != nil {
		return WithStack(err)
	}
	input := moveShardRequest{
		Database:   col.Database().Name(),
		Collection: col.Name(),
		Shard:      shard,
		FromServer: fromServer,
		ToServer:   toServer,
	}
	if _, err := req.SetBody(input); err != nil {
		return WithStack(err)
	}
	applyContextSettings(ctx, req)
	resp, err := c.conn.Do(ctx, req)
	if err != nil {
		return WithStack(err)
	}
	if err := resp.CheckStatus(202); err != nil {
		return WithStack(err)
	}
	return nil
}
476611
func NewStaticShardManager(
	serverAddrs []string,
	shardFunc func(key string, numShard int) (shard int),
	options net2.ConnectionOptions) ShardManager {

	manager := &StaticShardManager{}
	manager.Init(
		shardFunc,
		func(err error) { log.Print(err) },
		log.Print,
		options)

	shardStates := make([]ShardState, len(serverAddrs), len(serverAddrs))
	for i, addr := range serverAddrs {
		shardStates[i].Address = addr
		shardStates[i].State = ActiveServer
	}

	manager.UpdateShardStates(shardStates)

	return manager
}
357403
func (db *RedisDB) ssetScore(key, member string) float64 {
	ss := db.sortedsetKeys[key]
	return ss[member]
}
944767
func (h *HostStatsCollector) toDiskStats(usage *disk.UsageStat, partitionStat *disk.PartitionStat) *DiskStats {
	ds := DiskStats{
		Size:              usage.Total,
		Used:              usage.Used,
		Available:         usage.Free,
		UsedPercent:       usage.UsedPercent,
		InodesUsedPercent: usage.InodesUsedPercent,
	}
	if math.IsNaN(ds.UsedPercent) {
		ds.UsedPercent = 0.0
	}
	if math.IsNaN(ds.InodesUsedPercent) {
		ds.InodesUsedPercent = 0.0
	}

	if partitionStat != nil {
		ds.Device = partitionStat.Device
		ds.Mountpoint = partitionStat.Mountpoint
	}

	return &ds
}
654515
func (shardSwap *shardSchemaSwap) readShardMetadata(metadata *shardSwapMetadata, waitGroup *sync.WaitGroup) {
	defer waitGroup.Done()

	tablet, err := shardSwap.getMasterTablet()
	if err != nil {
		metadata.err = err
		return
	}
	query := fmt.Sprintf(
		"SELECT name, value FROM _vt.shard_metadata WHERE db_name = '%s' and name in ('%s', '%s', '%s')",
		topoproto.TabletDbName(tablet), lastStartedMetadataName, lastFinishedMetadataName, currentSQLMetadataName)
	queryResult, err := shardSwap.executeAdminQuery(tablet, query, 3 /* maxRows */)
	if err != nil {
		metadata.err = err
		return
	}
	for _, row := range queryResult.Rows {
		switch row[0].ToString() {
		case lastStartedMetadataName:
			swapID, err := sqltypes.ToUint64(row[1])
			if err != nil {
				log.Warningf("Could not parse value of last started schema swap id %v, ignoring the value: %v", row[1], err)
			} else {
				metadata.lastStartedSwap = swapID
			}
		case lastFinishedMetadataName:
			swapID, err := sqltypes.ToUint64(row[1])
			if err != nil {
				log.Warningf("Could not parse value of last finished schema swap id %v, ignoring the value: %v", row[1], err)
			} else {
				metadata.lastFinishedSwap = swapID
			}
		case currentSQLMetadataName:
			metadata.currentSQL = row[1].ToString()
		}
	}
}
516529
func (ms *managerShard) runLocked(c context.Context) error {
	workers := ms.cache.Parallel
	if workers <= 0 {
		workers = 1
	}

	// NOTE: This does not currently restrain itself to the current shard. This
	// can be done using a "__key__" inequality filter to bound the query.
	prototype := entry{
		CacheName: ms.cache.Name,
	}
	q := datastore.NewQuery(prototype.kind())

	var (
		totalEntries   = 0
		totalRefreshed = 0
		totalPruned    = 0
		totalErrors    = int32(0)

		entries   = make([]*entry, 0, ms.queryBatchSize)
		putBuf    = make([]*entry, ms.queryBatchSize)
		deleteBuf = make([]*entry, ms.queryBatchSize)
	)

	// Calculate our pruning threshold. If an entry's "LastAccessed" is <= this
	// threshold, it is candidate for pruning.
	var pruneThreshold time.Time
	if pi := ms.cache.pruneInterval(); pi > 0 {
		pruneThreshold = ms.now.Add(-pi)
	}

	// handleEntries refreshes the accumulated entries. It is used as a callback
	// in between query batches, as well as a finalizer after the queries
	// complete.
	//
	// handleEntries is, itself, not goroutine-safe.
	//
	// It will refresh in parallel, adding entries to "putBuf" or "deleteBuf" as
	// appropriate. At the end of its operation, all deferred datastore
	// operations will execute, and the accumulated entries list will be purged
	// for next round.
	handleEntries := func(c context.Context) error {
		if len(entries) == 0 {
			return nil
		}

		// Use atomic-friendly int32 values to index the put/delete buffers. This
		// will let our parallel goroutines safely add entries with very low
		// overhead.
		var putIdx, deleteIdx int32
		putEntry := func(e *entry) { putBuf[atomic.AddInt32(&putIdx, 1)-1] = e }
		deleteEntry := func(e *entry) { deleteBuf[atomic.AddInt32(&deleteIdx, 1)-1] = e }

		// Process each entry in parallel.
		//
		// Each task will return a nil error, so the error result does not need to
		// be observed.
		_ = parallel.WorkPool(workers, func(taskC chan<- func() error) {
			for _, e := range entries {
				e := e

				taskC <- func() error {
					// Is this entry candidate for pruning?
					if !(pruneThreshold.IsZero() || e.LastAccessed.After(pruneThreshold)) {
						log.Fields{
							"key":            e.keyHash(),
							"lastRefresh":    e.LastRefreshed,
							"lastAccessed":   e.LastAccessed,
							"pruneThreshold": pruneThreshold,
						}.Infof(c, "Pruning expired cache entry.")
						deleteEntry(e)
						return nil
					}

					// Is this cache entry candidate for refresh?
					if ms.h != nil {
						refreshInterval := ms.h.RefreshInterval(e.Key)
						if refreshInterval > 0 && !e.LastRefreshed.After(ms.now.Add(-refreshInterval)) {
							// Call our Handler's Refresh function. We leave our cache namespace
							// first.
							switch value, delta, err := doRefresh(c, ms.h, e, ""); err {
							case nil:
								// Refresh successful! Update the entry.
								//
								// Even if he data hasn't changed, the LastRefreshed time has, and
								// the cost of the "Put" is the same either way.
								e.LastRefreshed = ms.now
								e.LastRefreshDelta = int64(delta)
								e.loadValue(value)
								putEntry(e)
								return nil

							case ErrDeleteCacheEntry:
								log.Fields{
									"key": e.keyHash(),
								}.Debugf(c, "Refresh requested entry deletion.")
								deleteEntry(e)
								return nil

							default:
								log.Fields{
									log.ErrorKey:  err,
									"key":         e.keyHash(),
									"lastRefresh": e.LastRefreshed,
								}.Errorf(c, "Failed to refresh cache entry.")
								atomic.AddInt32(&totalErrors, 1)
							}
						}
					}
					return nil
				}
			}
		})

		// Clear our entries buffer for next round.
		entries = entries[:0]

		// Flush our put/delete buffers. Accumulate errors (best effort) and return
		// them as a batch.
		//
		// A failure here is a datastore failure, and will halt processing by
		// propagating from the callback to the query return value.
		_ = parallel.FanOutIn(func(taskC chan<- func() error) {
			if putIdx > 0 {
				taskC <- func() error {
					if err := datastore.Put(c, putBuf[:putIdx]); err != nil {
						log.Fields{
							log.ErrorKey: err,
							"size":       putIdx,
						}.Errorf(c, "Failed to Put batch.")
						atomic.AddInt32(&totalErrors, 1)
					} else {
						totalRefreshed += int(putIdx)
					}
					return nil
				}
			}

			if deleteIdx > 0 {
				taskC <- func() error {
					if err := datastore.Delete(c, deleteBuf[:deleteIdx]); err != nil {
						log.Fields{
							log.ErrorKey: err,
							"size":       deleteIdx,
						}.Errorf(c, "Failed to Delete batch.")
						atomic.AddInt32(&totalErrors, 1)
					} else {
						totalPruned += int(deleteIdx)
					}
					return nil
				}
			}
		})
		return nil
	}

	err := datastore.RunBatch(c, ms.queryBatchSize, q, func(e *entry) error {
		totalEntries++
		ms.observeEntry()
		entries = append(entries, e)

		if len(entries) >= int(ms.queryBatchSize) {
			// Hit the end of a query batch. Process entries.
			handleEntries(c)
		}
		return nil
	})
	if err != nil {
		return errors.Annotate(err, "failed to run entry query").Err()
	}

	// Flush any outstanding entries (ignore error, will always be nil).
	_ = handleEntries(c)
	if totalErrors > 0 {
		ms.observeErrors(totalErrors)
	}

	log.Fields{
		"entries":   totalEntries,
		"errors":    totalErrors,
		"refreshed": totalRefreshed,
		"pruned":    totalPruned,
	}.Infof(c, "Successfully updated cache entries.")
	return nil
}
845306
func (s *EnableEnhancedMonitoringInput) SetShardLevelMetrics(v []*string) *EnableEnhancedMonitoringInput {
	s.ShardLevelMetrics = v
	return s
}
899035
func (db *MemDB) Stats() map[string]string {
	db.mtx.Lock()
	defer db.mtx.Unlock()

	stats := make(map[string]string)
	stats["database.type"] = "memDB"
	stats["database.size"] = fmt.Sprintf("%d", len(db.db))
	return stats
}
1152457
func CompareShards(ctx context.Context, fromTS, toTS *topo.Server) error {
	keyspaces, err := fromTS.GetKeyspaces(ctx)
	if err != nil {
		return vterrors.Wrapf(err, "fromTS.GetKeyspaces")
	}

	for _, keyspace := range keyspaces {
		shards, err := fromTS.GetShardNames(ctx, keyspace)
		if err != nil {
			return vterrors.Wrapf(err, "GetShardNames(%v)", keyspace)
		}

		for _, shard := range shards {
			fromSi, err := fromTS.GetShard(ctx, keyspace, shard)
			if err != nil {
				return vterrors.Wrapf(err, "GetShard(%v, %v)", keyspace, shard)
			}
			toSi, err := toTS.GetShard(ctx, keyspace, shard)
			if err != nil {
				return vterrors.Wrapf(err, "GetShard(%v, %v)", keyspace, shard)
			}

			if !reflect.DeepEqual(fromSi.Shard, toSi.Shard) {
				return vterrors.Errorf(vtrpc.Code_FAILED_PRECONDITION, "Shard %v for keyspace: %v does not match between from and to topology", shard, keyspace)
			}
		}
	}
	return nil
}
141469
func (m *BaseShardManager) GetShardsForItems(
	items []*Item) map[int]*ShardMapping {

	m.rwMutex.RLock()
	defer m.rwMutex.RUnlock()

	numShards := len(m.shardStates)
	results := make(map[int]*ShardMapping)

	for _, item := range items {
		shardId := m.getShardId(item.Key, numShards)

		entry, inMap := results[shardId]
		if !inMap {
			entry = &ShardMapping{}
			if shardId != -1 {
				state := m.shardStates[shardId]
				if state.State == ActiveServer {
					m.fillEntryWithConnection(state.Address, entry)
				} else {
					connSkippedByAddr.Add(state.Address, 1)
				}
			}
			entry.Items = make([]*Item, 0, 1)
			entry.Keys = make([]string, 0, 1)
			results[shardId] = entry
		}
		entry.Items = append(entry.Items, item)
		entry.Keys = append(entry.Keys, item.Key)
	}

	return results
}
1246608
func (d *ddl) ShardRowID(ctx sessionctx.Context, tableIdent ast.Ident, uVal uint64) error {
	schema, t, err := d.getSchemaAndTableByIdent(ctx, tableIdent)
	if err != nil {
		return errors.Trace(err)
	}
	ok, _ := hasAutoIncrementColumn(t.Meta())
	if ok && uVal != 0 {
		return errUnsupportedShardRowIDBits
	}
	if uVal == t.Meta().ShardRowIDBits {
		// Nothing need to do.
		return nil
	}
	err = verifyNoOverflowShardBits(d.sessPool, t, uVal)
	if err != nil {
		return err
	}
	job := &model.Job{
		Type:       model.ActionShardRowID,
		SchemaID:   schema.ID,
		TableID:    t.Meta().ID,
		BinlogInfo: &model.HistoryInfo{},
		Args:       []interface{}{uVal},
	}
	err = d.doDDLJob(ctx, job)
	err = d.callHookOnChanged(err)
	return errors.Trace(err)
}
980128
func (b *broker) findNewShards() {
	b.shardMu.Lock()
	defer b.shardMu.Unlock()

	b.logger.Log("[BROKER]", "fetching shards")

	shards, err := b.listShards()
	if err != nil {
		b.logger.Log("[BROKER]", err)
		return
	}

	for _, shard := range shards {
		if _, ok := b.shards[*shard.ShardId]; ok {
			continue
		}
		b.shards[*shard.ShardId] = shard
		b.shardc <- shard
	}
}