Tagged delete

api/ccache.go

@@ -65,7 +65,7 @@ func (s *Server) ccacheDelete(ctx *middleware.Context, req models.CCacheDelete)
 				res.AddError(err)
 				code = http.StatusBadRequest
 			} else {
-				query, err := tagquery.NewQuery(expressions, 0)
+				query, err := tagquery.NewQuery(expressions, 0, 0)
 				if err != nil {
 					res.AddError(err)
 					code = http.StatusInternalServerError

api/cluster.go

@@ -209,7 +209,7 @@ func (s *Server) indexAutoCompleteTags(ctx *middleware.Context, req models.Index
 		expressions = append(expressions, "__tag^="+req.Prefix)
 	}
 
-	query, err := tagquery.NewQueryFromStrings(expressions, 0)
+	query, err := tagquery.NewQueryFromStrings(expressions, 0, 0)
 	if err != nil {
 		response.Write(ctx, response.NewError(http.StatusBadRequest, err.Error()))
 		return
@@ -240,7 +240,7 @@ func (s *Server) indexAutoCompleteTagValues(ctx *middleware.Context, req models.
 		expressions = append(expressions, req.Tag+"^="+req.Prefix)
 	}
 
-	query, err := tagquery.NewQueryFromStrings(expressions, 0)
+	query, err := tagquery.NewQueryFromStrings(expressions, 0, 0)
 	if err != nil {
 		response.Write(ctx, response.NewError(http.StatusBadRequest, err.Error()))
 		return
@@ -280,7 +280,7 @@ func (s *Server) indexTagDelSeries(ctx *middleware.Context, request models.Index
 			builder.Reset()
 		}
 
-		query, err := tagquery.NewQuery(expressions, 0)
+		query, err := tagquery.NewQuery(expressions, 0, 0)
 		if err != nil {
 			response.Write(ctx, response.WrapErrorForTagDB(err))
 			return
@@ -297,14 +297,48 @@ func (s *Server) indexTagDelSeries(ctx *middleware.Context, request models.Index
 	response.Write(ctx, response.NewMsgp(200, res))
 }
 
+func (s *Server) localTagDelByQuery(request models.IndexTagDelByQuery) (int, error) {
+	query, err := tagquery.NewQueryFromStrings(request.Expr, 0, request.OlderThan)
+	if err != nil {
+		return 0, err
+	}
+
+	if request.Execute {
+		deleted, err := s.MetricIndex.DeleteTagged(request.OrgId, query)
+		return len(deleted), err
+	}
+	matched, _ := s.MetricIndex.FindTerms(request.OrgId, []string{}, query)
+	return int(matched), nil
+}
+
+func (s *Server) IndexTagDelByQuery(ctx *middleware.Context, request models.IndexTagDelByQuery) {
+
+	res := models.IndexTagDelByQueryResp{}
+
+	// nothing to do on query nodes.
+	if s.MetricIndex == nil {
+		response.Write(ctx, response.NewMsgp(200, res))
+		return
+	}
+
+	var err error
+	res.Count, err = s.localTagDelByQuery(request)
+	if err != nil {
+		response.Write(ctx, response.WrapErrorForTagDB(err))
+		return
+	}
+
+	response.Write(ctx, response.NewMsgp(200, res))
+}
+
 func (s *Server) IndexTagTerms(ctx *middleware.Context, req models.IndexTagTerms) {
 	// query nodes don't own any data.
 	if s.MetricIndex == nil {
 		response.Write(ctx, response.NewMsgp(200, &models.GraphiteTagTermsResp{}))
 		return
 	}
 
-	query, err := tagquery.NewQueryFromStrings(req.Expr, 0)
+	query, err := tagquery.NewQueryFromStrings(req.Expr, 0, 0)
 	if err != nil {
 		response.Write(ctx, response.NewError(http.StatusBadRequest, err.Error()))
 		return
@@ -323,7 +357,7 @@ func (s *Server) indexFindByTag(ctx *middleware.Context, req models.IndexFindByT
 		return
 	}
 
-	query, err := tagquery.NewQueryFromStrings(req.Expr, req.From)
+	query, err := tagquery.NewQueryFromStrings(req.Expr, req.From, 0)
 	if err != nil {
 		response.Write(ctx, response.NewError(http.StatusBadRequest, err.Error()))
 		return

api/graphite.go

@@ -1412,7 +1412,7 @@ func (s *Server) graphiteTagDelSeries(ctx *middleware.Context, request models.Gr
 				builder.Reset()
 			}
 
-			query, err := tagquery.NewQuery(expressions, 0)
+			query, err := tagquery.NewQuery(expressions, 0, 0)
 			if err != nil {
 				response.Write(ctx, response.WrapErrorForTagDB(err))
 				return
@@ -1451,6 +1451,44 @@ func (s *Server) graphiteTagDelSeries(ctx *middleware.Context, request models.Gr
 	response.Write(ctx, response.NewJson(200, res, ""))
 }
 
+func (s *Server) graphiteTagDelByQuery(ctx *middleware.Context, request models.GraphiteTagDelByQuery) {
+	res := models.GraphiteTagDelByQueryResp{}
+
+	data := models.IndexTagDelByQuery{OrgId: ctx.OrgId, Expr: request.Expr, OlderThan: request.OlderThan, Execute: request.Execute}
+	responses, errors := s.queryAllPeers(ctx.Req.Context(), data, "clusterTagDelByQuery,", "/index/tags/delByQuery")
+
+	// nothing to do locally on query nodes.
+	if s.MetricIndex != nil {
+		matched, err := s.localTagDelByQuery(data)
+		if err != nil {
+			response.Write(ctx, response.WrapErrorForTagDB(err))
+			return
+		}
+		res.Count += matched
+		res.Peers[cluster.Manager.ThisNode().GetName()] = matched
+	}
+
+	// if there are any errors, write one of them and return
+	for _, err := range errors {
+		response.Write(ctx, response.WrapErrorForTagDB(err))
+		return
+	}
+
+	res.Peers = make(map[string]int, len(responses))
+	peerResp := models.IndexTagDelSeriesResp{}
+	for peer, resp := range responses {
+		_, err := peerResp.UnmarshalMsg(resp.buf)
+		if err != nil {
+			response.Write(ctx, response.WrapErrorForTagDB(err))
+			return
+		}
+		res.Count += peerResp.Count
+		res.Peers[peer] = peerResp.Count
+	}
+
+	response.Write(ctx, response.NewJson(200, res, ""))
+}
+
 // showPlan attempts to create a Plan given a /render target query.
 // If the Plan creation is successful it returns 200, JSON marshaling of Plan.
 // Otherwise, it returns 400, error details.

api/models/cluster.go

@@ -53,3 +53,8 @@ type IndexFindByTagResp struct {
 type IndexTagDelSeriesResp struct {
 	Count int
 }
+
+//go:generate msgp
+type IndexTagDelByQueryResp struct {
+	Count int
+}