Kiu/VictoriaMetrics
Kiu/VictoriaMetrics
1
0
Fork 0
mirror of https://github.com/VictoriaMetrics/VictoriaMetrics.git synced 2024-12-16 00:41:24 +01:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

lib/vmselectapi: rename deleteMetrics to more correct deleteSeries

Browse Source
This commit is contained in:
Aliaksandr Valialkin 2022-07-05 23:56:31 +03:00
parent 5afa54e845
commit 78eeca6f0d
No known key found for this signature in database
GPG Key ID: A72BEC6CD3D0DED1
6 changed files with 29 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
app/vmselect/netstorage/netstorage.go
View File

@ -657,7 +657,7 @@ func DeleteSeries(qt *querytracer.Tracer, at *auth.Token, sq *storage.SearchQuer
} }
snr := startStorageNodesRequest(qt, true, func(qt *querytracer.Tracer, idx int, sn *storageNode) interface{} { snr := startStorageNodesRequest(qt, true, func(qt *querytracer.Tracer, idx int, sn *storageNode) interface{} {
sn.deleteSeriesRequests.Inc() sn.deleteSeriesRequests.Inc()
deletedCount, err := sn.deleteMetrics(qt, requestData, deadline) deletedCount, err := sn.deleteSeries(qt, requestData, deadline)
if err != nil { if err != nil {
sn.deleteSeriesErrors.Inc() sn.deleteSeriesErrors.Inc()
} }
@ -1497,17 +1497,17 @@ func (sn *storageNode) registerMetricNames(qt *querytracer.Tracer, mrs []storage
return sn.execOnConnWithPossibleRetry(qt, "registerMetricNames_v3", f, deadline) return sn.execOnConnWithPossibleRetry(qt, "registerMetricNames_v3", f, deadline)
} }
func (sn *storageNode) deleteMetrics(qt *querytracer.Tracer, requestData []byte, deadline searchutils.Deadline) (int, error) { func (sn *storageNode) deleteSeries(qt *querytracer.Tracer, requestData []byte, deadline searchutils.Deadline) (int, error) {
var deletedCount int var deletedCount int
f := func(bc *handshake.BufferedConn) error { f := func(bc *handshake.BufferedConn) error {
n, err := sn.deleteMetricsOnConn(bc, requestData) n, err := sn.deleteSeriesOnConn(bc, requestData)
if err != nil { if err != nil {
return err return err
} }
deletedCount = n deletedCount = n
return nil return nil
} }
if err := sn.execOnConnWithPossibleRetry(qt, "deleteMetrics_v5", f, deadline); err != nil { if err := sn.execOnConnWithPossibleRetry(qt, "deleteSeries_v5", f, deadline); err != nil {
return 0, err return 0, err
} }
return deletedCount, nil return deletedCount, nil
@ -1784,13 +1784,13 @@ func (sn *storageNode) registerMetricNamesOnConn(bc *handshake.BufferedConn, mrs
return nil return nil
} }
func (sn *storageNode) deleteMetricsOnConn(bc *handshake.BufferedConn, requestData []byte) (int, error) { func (sn *storageNode) deleteSeriesOnConn(bc *handshake.BufferedConn, requestData []byte) (int, error) {
// Send the request to sn // Send the request to sn
if err := writeBytes(bc, requestData); err != nil { if err := writeBytes(bc, requestData); err != nil {
return 0, fmt.Errorf("cannot send deleteMetrics request to conn: %w", err) return 0, fmt.Errorf("cannot send deleteSeries request to conn: %w", err)
} }
if err := bc.Flush(); err != nil { if err := bc.Flush(); err != nil {
return 0, fmt.Errorf("cannot flush deleteMetrics request to conn: %w", err) return 0, fmt.Errorf("cannot flush deleteSeries request to conn: %w", err)
} }
// Read response error. // Read response error.

4
app/vmstorage/servers/vmselect.go
View File

@ -86,8 +86,8 @@ func (api *vmstorageAPI) TSDBStatus(qt *querytracer.Tracer, accountID, projectID
return api.s.GetTSDBStatus(qt, accountID, projectID, tfss, date, focusLabel, topN, maxMetrics, deadline) return api.s.GetTSDBStatus(qt, accountID, projectID, tfss, date, focusLabel, topN, maxMetrics, deadline)
} }
func (api *vmstorageAPI) DeleteMetrics(qt *querytracer.Tracer, tfss []*storage.TagFilters, maxMetrics int, deadline uint64) (int, error) { func (api *vmstorageAPI) DeleteSeries(qt *querytracer.Tracer, tfss []*storage.TagFilters, maxMetrics int, deadline uint64) (int, error) {
return api.s.DeleteMetrics(qt, tfss) return api.s.DeleteSeries(qt, tfss)
} }
func (api *vmstorageAPI) RegisterMetricNames(qt *querytracer.Tracer, mrs []storage.MetricRow) error { func (api *vmstorageAPI) RegisterMetricNames(qt *querytracer.Tracer, mrs []storage.MetricRow) error {

6
lib/storage/storage.go
View File

@ -1346,10 +1346,10 @@ func (s *Storage) prefetchMetricNames(qt *querytracer.Tracer, tsids []TSID, dead
// ErrDeadlineExceeded is returned when the request times out. // ErrDeadlineExceeded is returned when the request times out.
var ErrDeadlineExceeded = fmt.Errorf("deadline exceeded") var ErrDeadlineExceeded = fmt.Errorf("deadline exceeded")
// DeleteMetrics deletes all the metrics matching the given tfss. // DeleteSeries deletes all the series matching the given tfss.
// //
// Returns the number of metrics deleted. // Returns the number of metrics deleted.
func (s *Storage) DeleteMetrics(qt *querytracer.Tracer, tfss []*TagFilters) (int, error) { func (s *Storage) DeleteSeries(qt *querytracer.Tracer, tfss []*TagFilters) (int, error) {
deletedCount, err := s.idb().DeleteTSIDs(qt, tfss) deletedCount, err := s.idb().DeleteTSIDs(qt, tfss)
if err != nil { if err != nil {
return deletedCount, fmt.Errorf("cannot delete tsids: %w", err) return deletedCount, fmt.Errorf("cannot delete tsids: %w", err)
@ -1889,7 +1889,7 @@ func (s *Storage) add(rows []rawRow, dstMrs []*MetricRow, mrs []MetricRow, preci
// Fast path - the TSID for the given MetricNameRaw has been found in cache and isn't deleted. // Fast path - the TSID for the given MetricNameRaw has been found in cache and isn't deleted.
// There is no need in checking whether r.TSID.MetricID is deleted, since tsidCache doesn't // There is no need in checking whether r.TSID.MetricID is deleted, since tsidCache doesn't
// contain MetricName->TSID entries for deleted time series. // contain MetricName->TSID entries for deleted time series.
// See Storage.DeleteMetrics code for details. // See Storage.DeleteSeries code for details.
prevTSID = r.TSID prevTSID = r.TSID
prevMetricNameRaw = mr.MetricNameRaw prevMetricNameRaw = mr.MetricNameRaw

16
lib/storage/storage_test.go
View File

@ -531,8 +531,8 @@ func testStorageRandTimestamps(s *Storage) error {
return nil return nil
} }
func TestStorageDeleteMetrics(t *testing.T) { func TestStorageDeleteSeries(t *testing.T) {
path := "TestStorageDeleteMetrics" path := "TestStorageDeleteSeries"
s, err := OpenStorage(path, 0, 0, 0) s, err := OpenStorage(path, 0, 0, 0)
if err != nil { if err != nil {
t.Fatalf("cannot open storage: %s", err) t.Fatalf("cannot open storage: %s", err)
@ -549,7 +549,7 @@ func TestStorageDeleteMetrics(t *testing.T) {
t.Run("serial", func(t *testing.T) { t.Run("serial", func(t *testing.T) {
for i := 0; i < 3; i++ { for i := 0; i < 3; i++ {
if err = testStorageDeleteMetrics(s, 0); err != nil { if err = testStorageDeleteSeries(s, 0); err != nil {
t.Fatalf("unexpected error on iteration %d: %s", i, err) t.Fatalf("unexpected error on iteration %d: %s", i, err)
} }
@ -569,7 +569,7 @@ func TestStorageDeleteMetrics(t *testing.T) {
go func(workerNum int) { go func(workerNum int) {
var err error var err error
for j := 0; j < 2; j++ { for j := 0; j < 2; j++ {
err = testStorageDeleteMetrics(s, workerNum) err = testStorageDeleteSeries(s, workerNum)
if err != nil { if err != nil {
break break
} }
@ -604,7 +604,7 @@ func TestStorageDeleteMetrics(t *testing.T) {
} }
} }
func testStorageDeleteMetrics(s *Storage, workerNum int) error { func testStorageDeleteSeries(s *Storage, workerNum int) error {
const rowsPerMetric = 100 const rowsPerMetric = 100
const metricsCount = 30 const metricsCount = 30
@ -694,7 +694,7 @@ func testStorageDeleteMetrics(s *Storage, workerNum int) error {
if n := metricBlocksCount(tfs); n == 0 { if n := metricBlocksCount(tfs); n == 0 {
return fmt.Errorf("expecting non-zero number of metric blocks for tfs=%s", tfs) return fmt.Errorf("expecting non-zero number of metric blocks for tfs=%s", tfs)
} }
deletedCount, err := s.DeleteMetrics(nil, []*TagFilters{tfs}) deletedCount, err := s.DeleteSeries(nil, []*TagFilters{tfs})
if err != nil { if err != nil {
return fmt.Errorf("cannot delete metrics: %w", err) return fmt.Errorf("cannot delete metrics: %w", err)
} }
@ -702,11 +702,11 @@ func testStorageDeleteMetrics(s *Storage, workerNum int) error {
return fmt.Errorf("expecting non-zero number of deleted metrics on iteration %d", i) return fmt.Errorf("expecting non-zero number of deleted metrics on iteration %d", i)
} }
if n := metricBlocksCount(tfs); n != 0 { if n := metricBlocksCount(tfs); n != 0 {
return fmt.Errorf("expecting zero metric blocks after DeleteMetrics call for tfs=%s; got %d blocks", tfs, n) return fmt.Errorf("expecting zero metric blocks after DeleteSeries call for tfs=%s; got %d blocks", tfs, n)
} }
// Try deleting empty tfss // Try deleting empty tfss
deletedCount, err = s.DeleteMetrics(nil, nil) deletedCount, err = s.DeleteSeries(nil, nil)
if err != nil { if err != nil {
return fmt.Errorf("cannot delete empty tfss: %w", err) return fmt.Errorf("cannot delete empty tfss: %w", err)
} }

4
lib/vmselectapi/api.go
View File

@ -30,8 +30,8 @@ type API interface {
// TSDBStatus returns tsdb status for the given sq. // TSDBStatus returns tsdb status for the given sq.
TSDBStatus(qt *querytracer.Tracer, accountID, projectID uint32, tfss []*storage.TagFilters, date uint64, focusLabel string, topN, maxMetrics int, deadline uint64) (*storage.TSDBStatus, error) TSDBStatus(qt *querytracer.Tracer, accountID, projectID uint32, tfss []*storage.TagFilters, date uint64, focusLabel string, topN, maxMetrics int, deadline uint64) (*storage.TSDBStatus, error)
// DeleteMetrics deletes metrics matching the given tfss. // DeleteSeries deletes series matching the given tfss.
DeleteMetrics(qt *querytracer.Tracer, tfss []*storage.TagFilters, maxMetrics int, deadline uint64) (int, error) DeleteSeries(qt *querytracer.Tracer, tfss []*storage.TagFilters, maxMetrics int, deadline uint64) (int, error)
// RegisterMetricNames registers the given mrs in the storage. // RegisterMetricNames registers the given mrs in the storage.
RegisterMetricNames(qt *querytracer.Tracer, mrs []storage.MetricRow) error RegisterMetricNames(qt *querytracer.Tracer, mrs []storage.MetricRow) error

14
lib/vmselectapi/server.go
View File

@ -51,7 +51,7 @@ type Server struct {
indexSearchDuration *metrics.Histogram indexSearchDuration *metrics.Histogram
registerMetricNamesRequests *metrics.Counter registerMetricNamesRequests *metrics.Counter
deleteMetricsRequests *metrics.Counter deleteSeriesRequests *metrics.Counter
labelNamesRequests *metrics.Counter labelNamesRequests *metrics.Counter
labelValuesRequests *metrics.Counter labelValuesRequests *metrics.Counter
tagValueSuffixesRequests *metrics.Counter tagValueSuffixesRequests *metrics.Counter
@ -98,7 +98,7 @@ func NewServer(addr string, api API, limits Limits, disableResponseCompression b
indexSearchDuration: metrics.NewHistogram(fmt.Sprintf(`vm_index_search_duration_seconds{addr=%q}`, addr)), indexSearchDuration: metrics.NewHistogram(fmt.Sprintf(`vm_index_search_duration_seconds{addr=%q}`, addr)),
registerMetricNamesRequests: metrics.NewCounter(fmt.Sprintf(`vm_vmselect_rpc_requests_total{action="registerMetricNames",addr=%q}`, addr)), registerMetricNamesRequests: metrics.NewCounter(fmt.Sprintf(`vm_vmselect_rpc_requests_total{action="registerMetricNames",addr=%q}`, addr)),
deleteMetricsRequests: metrics.NewCounter(fmt.Sprintf(`vm_vmselect_rpc_requests_total{action="deleteMetrics",addr=%q}`, addr)), deleteSeriesRequests: metrics.NewCounter(fmt.Sprintf(`vm_vmselect_rpc_requests_total{action="deleteSeries",addr=%q}`, addr)),
labelNamesRequests: metrics.NewCounter(fmt.Sprintf(`vm_vmselect_rpc_requests_total{action="labelNames",addr=%q}`, addr)), labelNamesRequests: metrics.NewCounter(fmt.Sprintf(`vm_vmselect_rpc_requests_total{action="labelNames",addr=%q}`, addr)),
labelValuesRequests: metrics.NewCounter(fmt.Sprintf(`vm_vmselect_rpc_requests_total{action="labelValues",addr=%q}`, addr)), labelValuesRequests: metrics.NewCounter(fmt.Sprintf(`vm_vmselect_rpc_requests_total{action="labelValues",addr=%q}`, addr)),
tagValueSuffixesRequests: metrics.NewCounter(fmt.Sprintf(`vm_vmselect_rpc_requests_total{action="tagValueSuffixes",addr=%q}`, addr)), tagValueSuffixesRequests: metrics.NewCounter(fmt.Sprintf(`vm_vmselect_rpc_requests_total{action="tagValueSuffixes",addr=%q}`, addr)),
@ -488,8 +488,8 @@ func (s *Server) processRPC(ctx *vmselectRequestCtx, rpcName string) error {
return s.processSeriesCount(ctx) return s.processSeriesCount(ctx)
case "tsdbStatus_v5": case "tsdbStatus_v5":
return s.processTSDBStatus(ctx) return s.processTSDBStatus(ctx)
case "deleteMetrics_v5": case "deleteSeries_v5":
return s.processDeleteMetrics(ctx) return s.processDeleteSeries(ctx)
case "registerMetricNames_v3": case "registerMetricNames_v3":
return s.processRegisterMetricNames(ctx) return s.processRegisterMetricNames(ctx)
default: default:
@ -537,8 +537,8 @@ func (s *Server) processRegisterMetricNames(ctx *vmselectRequestCtx) error {
return nil return nil
} }
func (s *Server) processDeleteMetrics(ctx *vmselectRequestCtx) error { func (s *Server) processDeleteSeries(ctx *vmselectRequestCtx) error {
s.deleteMetricsRequests.Inc() s.deleteSeriesRequests.Inc()
// Read request // Read request
if err := ctx.readSearchQuery(); err != nil { if err := ctx.readSearchQuery(); err != nil {
@ -555,7 +555,7 @@ func (s *Server) processDeleteMetrics(ctx *vmselectRequestCtx) error {
if err != nil { if err != nil {
return ctx.writeErrorMessage(err) return ctx.writeErrorMessage(err)
} }
deletedCount, err := s.api.DeleteMetrics(ctx.qt, tfss, maxMetrics, ctx.deadline) deletedCount, err := s.api.DeleteSeries(ctx.qt, tfss, maxMetrics, ctx.deadline)
if err != nil { if err != nil {
return ctx.writeErrorMessage(err) return ctx.writeErrorMessage(err)
} }