Victoria Metrics源码: 水平扩展实现，查询拆分与时序数据打散写入

// app/vmselect/promql/eval.go: 644
sq := &storage.SearchQuery{
	AccountID:    ec.AuthToken.AccountID,
	ProjectID:    ec.AuthToken.ProjectID,
	MinTimestamp: minTimestamp,
	MaxTimestamp: ec.End,
	TagFilterss:  [][]storage.TagFilter{tfs},
}
rss, isPartial, err := netstorage.ProcessSearchQuery(ec.AuthToken, sq, true, ec.Deadline)

// app/vmselect/netstorage/netstorage.go:747 in func ProcessSearchQuery()
for _, sn := range storageNodes {
	go func(sn *storageNode) {
		sn.searchRequests.Inc()
		err := sn.processSearchQuery(tbfw, requestData, tr, fetchData, deadline)
		if err != nil {
			sn.searchRequestErrors.Inc()
			err = fmt.Errorf("cannot perform search on vmstorage %s: %s", sn.connPool.Addr(), err)
		}
		resultsCh <- err
	}(sn)
}

// ./app/vminsert/promremotewrite/request_handlers.go
func insertRows(at *auth.Token, timeseries []prompb.TimeSeries) error {
	ctx := netstorage.GetInsertCtx()
	defer netstorage.PutInsertCtx(ctx)

	ctx.Reset() // This line is required for initializing ctx internals.
	rowsTotal := 0
	for i := range timeseries {
		ts := &timeseries[i]
		storageNodeIdx := ctx.GetStorageNodeIdx(at, ts.Labels)
		ctx.MetricNameBuf = ctx.MetricNameBuf[:0]
		for i := range ts.Samples {
			r := &ts.Samples[i]
			if len(ctx.MetricNameBuf) == 0 {
				ctx.MetricNameBuf = storage.MarshalMetricNameRaw(ctx.MetricNameBuf[:0], at.AccountID, at.ProjectID, ts.Labels)
			}
			if err := ctx.WriteDataPointExt(at, storageNodeIdx, ctx.MetricNameBuf, r.Timestamp, r.Value); err != nil {
				return err
			}
		}
		rowsTotal += len(ts.Samples)
	}
	rowsInserted.Get(at).Add(rowsTotal)
	rowsPerInsert.Update(float64(rowsTotal))
	return ctx.FlushBufs()
}