refactor tag count querying, use it in autocomplete

Author: mchrome

autocomplete/autocomplete.go

@@ -71,13 +71,28 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-func (h *Handler) requestExpr(r *http.Request) (*where.Where, *where.Where, map[string]bool, error) {
-	f := r.Form["expr"]
-	expr := make([]string, 0, len(f))
+func getTagCountQuerier(config *config.Config, opts clickhouse.Options) *finder.TagCountQuerier {
+	var tcq *finder.TagCountQuerier = nil
+	if config.ClickHouse.TagsCountTable != "" {
+		tcq = finder.NewTagCountQuerier(
+			config.ClickHouse.URL,
+			config.ClickHouse.TagsCountTable,
+			opts,
+			config.FeatureFlags.UseCarbonBehavior,
+			config.FeatureFlags.DontMatchMissingTags,
+			config.ClickHouse.TaggedUseDaily,
+		)
+	}
+	return tcq
+}
 
-	for i := 0; i < len(f); i++ {
-		if f[i] != "" {
-			expr = append(expr, f[i])
+func (h *Handler) requestExpr(r *http.Request, tcq *finder.TagCountQuerier, from, until int64) (*where.Where, *where.Where, map[string]bool, error) {
+	formExpr := r.Form["expr"]
+	expr := make([]string, 0, len(formExpr))
+
+	for i := 0; i < len(formExpr); i++ {
+		if formExpr[i] != "" {
+			expr = append(expr, formExpr[i])
 		}
 	}
 
@@ -95,6 +110,15 @@ func (h *Handler) requestExpr(r *http.Request) (*where.Where, *where.Where, map[
 		return wr, pw, usedTags, err
 	}
 
+	if tcq != nil {
+		tagValuesCosts, err := tcq.GetCostsFromCountTable(r.Context(), terms, from, until)
+		if err != nil {
+			return wr, pw, usedTags, err
+		}
+		finder.SetCosts(terms, tagValuesCosts)
+	}
+	finder.SortTaggedTermsByCost(terms)
+
 	wr, pw, err = finder.TaggedWhere(terms, h.config.FeatureFlags.UseCarbonBehavior, h.config.FeatureFlags.DontMatchMissingTags)
 	if err != nil {
 		return wr, pw, usedTags, err
@@ -214,6 +238,7 @@ func (h *Handler) ServeTags(w http.ResponseWriter, r *http.Request) {
 		queueFail     bool
 		queueDuration time.Duration
 		findCache     bool
+		opts          clickhouse.Options
 	)
 
 	username := r.Header.Get("X-Forwarded-User")
@@ -290,7 +315,21 @@ func (h *Handler) ServeTags(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 
-	wr, pw, usedTags, err := h.requestExpr(r)
+	opts = clickhouse.Options{
+		TLSConfig:               h.config.ClickHouse.TLSConfig,
+		Timeout:                 h.config.ClickHouse.IndexTimeout,
+		ConnectTimeout:          h.config.ClickHouse.ConnectTimeout,
+		CheckRequestProgress:    h.config.FeatureFlags.LogQueryProgress,
+		ProgressSendingInterval: h.config.ClickHouse.ProgressSendingInterval,
+	}
+
+	wr, pw, usedTags, err := h.requestExpr(
+		r,
+		getTagCountQuerier(h.config, opts),
+		start.AddDate(0, 0, -h.config.ClickHouse.TaggedAutocompleDays).Unix(),
+		start.Unix(),
+	)
+
 	if err != nil {
 		status = http.StatusBadRequest
 		http.Error(w, err.Error(), status)
@@ -366,13 +405,7 @@ func (h *Handler) ServeTags(w http.ResponseWriter, r *http.Request) {
 			scope.WithTable(r.Context(), h.config.ClickHouse.TaggedTable),
 			h.config.ClickHouse.URL,
 			sql,
-			clickhouse.Options{
-				TLSConfig:               h.config.ClickHouse.TLSConfig,
-				Timeout:                 h.config.ClickHouse.IndexTimeout,
-				ConnectTimeout:          h.config.ClickHouse.ConnectTimeout,
-				CheckRequestProgress:    h.config.FeatureFlags.LogQueryProgress,
-				ProgressSendingInterval: h.config.ClickHouse.ProgressSendingInterval,
-			},
+			opts,
 			nil,
 		)
 
@@ -490,6 +523,7 @@ func (h *Handler) ServeValues(w http.ResponseWriter, r *http.Request) {
 		queueFail     bool
 		queueDuration time.Duration
 		findCache     bool
+		opts          clickhouse.Options
 	)
 
 	username := r.Header.Get("X-Forwarded-User")
@@ -567,8 +601,23 @@ func (h *Handler) ServeValues(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 
+	opts = clickhouse.Options{
+		TLSConfig:               h.config.ClickHouse.TLSConfig,
+		Timeout:                 h.config.ClickHouse.IndexTimeout,
+		ConnectTimeout:          h.config.ClickHouse.ConnectTimeout,
+		CheckRequestProgress:    h.config.FeatureFlags.LogQueryProgress,
+		ProgressSendingInterval: h.config.ClickHouse.ProgressSendingInterval,
+	}
+
 	if !findCache {
-		wr, pw, usedTags, err := h.requestExpr(r)
+
+		wr, pw, usedTags, err := h.requestExpr(
+			r,
+			getTagCountQuerier(h.config, opts),
+			start.AddDate(0, 0, -h.config.ClickHouse.TaggedAutocompleDays).Unix(),
+			start.Unix(),
+		)
+
 		if err == finder.ErrCostlySeriesByTag {
 			status = http.StatusForbidden
 			http.Error(w, err.Error(), status)
@@ -640,13 +689,7 @@ func (h *Handler) ServeValues(w http.ResponseWriter, r *http.Request) {
 			scope.WithTable(r.Context(), h.config.ClickHouse.TaggedTable),
 			h.config.ClickHouse.URL,
 			sql,
-			clickhouse.Options{
-				TLSConfig:               h.config.ClickHouse.TLSConfig,
-				Timeout:                 h.config.ClickHouse.IndexTimeout,
-				ConnectTimeout:          h.config.ClickHouse.ConnectTimeout,
-				CheckRequestProgress:    h.config.FeatureFlags.LogQueryProgress,
-				ProgressSendingInterval: h.config.ClickHouse.ProgressSendingInterval,
-			},
+			opts,
 			nil,
 		)
 

deploy/doc/config.md

@@ -225,4 +225,4 @@ Here we additionally create a materialized view to automatically save the quanti
 graphite-clickhouse will query this table when it tries to decide which tag should be used when querying graphite_tagged table.
 Overall using this parameter will somewhat increase writing load but can improve reading tagged metrics greatly in some cases.
 
-Note that this option only works for terms with '=' operator in them.
+Note that this option only works for terms with '=' operator in them. Using it will also override tag costs that were set manually with tagged-costs option.

doc/config.md

@@ -228,7 +228,7 @@ Here we additionally create a materialized view to automatically save the quanti
 graphite-clickhouse will query this table when it tries to decide which tag should be used when querying graphite_tagged table.
 Overall using this parameter will somewhat increase writing load but can improve reading tagged metrics greatly in some cases.
 
-Note that this option only works for terms with '=' operator in them.
+Note that this option only works for terms with '=' operator in them. Using it will also override tag costs that were set manually with tagged-costs option.
 
 ```toml
 [common]

finder/tagged.go

@@ -6,7 +6,6 @@ import (
 	"fmt"
 	"net/http"
 	"sort"
-	"strconv"
 	"strings"
 
 	"github.com/lomik/graphite-clickhouse/config"
@@ -79,10 +78,10 @@ func (s TaggedTermList) Less(i, j int) bool {
 type TaggedFinder struct {
 	url                  string                   // clickhouse dsn
 	table                string                   // graphite_tag table
-	tag1CountTable       string                   // table that helps to choose the most optimal Tag1
+	tcq                  *TagCountQuerier         // An object for querying tag weights from clickhouse. See doc/config.md for details.
 	absKeepEncoded       bool                     // Abs returns url encoded value. For queries from prometheus
 	opts                 clickhouse.Options       // clickhouse query timeout
-	taggedCosts          map[string]*config.Costs // costs for taggs (sor tune index search)
+	configuredTagCosts   map[string]*config.Costs // costs for taggs (sor tune index search)
 	dailyEnabled         bool
 	useCarbonBehavior    bool
 	dontMatchMissingTags bool
@@ -92,19 +91,30 @@ type TaggedFinder struct {
 }
 
 func NewTagged(url string, table, tag1CountTable string, dailyEnabled, useCarbonBehavior, dontMatchMissingTags, absKeepEncoded bool, opts clickhouse.Options, taggedCosts map[string]*config.Costs) *TaggedFinder {
-	return &TaggedFinder{
+	fnd := &TaggedFinder{
 		url:                  url,
 		table:                table,
-		tag1CountTable:       tag1CountTable,
+		tcq:                  nil,
 		absKeepEncoded:       absKeepEncoded,
 		opts:                 opts,
-		taggedCosts:          taggedCosts,
+		configuredTagCosts:   taggedCosts,
 		dailyEnabled:         dailyEnabled,
 		useCarbonBehavior:    useCarbonBehavior,
 		dontMatchMissingTags: dontMatchMissingTags,
 		metricMightExists:    true,
 		stats:                make([]metrics.FinderStat, 0),
 	}
+	if tag1CountTable != "" {
+		fnd.tcq = NewTagCountQuerier(
+			url,
+			tag1CountTable,
+			opts,
+			useCarbonBehavior,
+			dontMatchMissingTags,
+			dailyEnabled,
+		)
+	}
+	return fnd
 }
 
 func (term *TaggedTerm) concat() string {
@@ -606,17 +616,18 @@ func (t *TaggedFinder) PrepareTaggedTerms(ctx context.Context, cfg *config.Confi
 		return nil, err
 	}
 
-	if t.tag1CountTable != "" {
-		err = t.SetCostsFromCountTable(ctx, terms, from, until)
+	var tagCounts map[string]*config.Costs = nil
+	if t.tcq != nil {
+		tagCounts, err = t.tcq.GetCostsFromCountTable(ctx, terms, from, until)
 		if err != nil {
 			return nil, err
 		}
 	}
 
-	// Set costs from count table only if it has all tag-value pairs contained in the seriesByTag query (t.metricMightExist == true)
-	// or if tagged costs were set in the config file and t.metricMightExist == false
-	if t.metricMightExists || len(t.taggedCosts) != 0 {
-		SetCosts(terms, t.taggedCosts)
+	if tagCounts != nil {
+		SetCosts(terms, tagCounts)
+	} else if len(t.configuredTagCosts) != 0 {
+		SetCosts(terms, t.configuredTagCosts)
 	}
 
 	SortTaggedTermsByCost(terms)
@@ -659,126 +670,10 @@ func SortTaggedTermsByCost(terms []TaggedTerm) {
 	})
 }
 
-func (t *TaggedFinder) SetCostsFromCountTable(ctx context.Context, terms []TaggedTerm, from int64, until int64) error {
-	w := where.New()
-	eqTermCount := 0
-
-	for i := 0; i < len(terms); i++ {
-		if terms[i].Op == TaggedTermEq && !terms[i].HasWildcard && terms[i].Value != "" {
-			sqlTerm, err := TaggedTermWhere1(&terms[i], t.useCarbonBehavior, t.dontMatchMissingTags)
-			if err != nil {
-				return err
-			}
-
-			w.Or(sqlTerm)
-
-			eqTermCount++
-		}
-	}
-
-	if w.SQL() == "" {
-		return nil
-	}
-
-	if t.dailyEnabled {
-		w.Andf(
-			"Date >= '%s' AND Date <= '%s'",
-			date.FromTimestampToDaysFormat(from),
-			date.UntilTimestampToDaysFormat(until),
-		)
-	} else {
-		w.Andf(
-			"Date >= '%s'",
-			date.FromTimestampToDaysFormat(from),
-		)
-	}
-
-	sql := fmt.Sprintf("SELECT Tag1, sum(Count) as cnt FROM %s %s GROUP BY Tag1 FORMAT TabSeparatedRaw", t.tag1CountTable, w.SQL())
-
-	var err error
-
-	t.stats = append(t.stats, metrics.FinderStat{})
-	stat := &t.stats[len(t.stats)-1]
-	stat.Table = t.tag1CountTable
-
-	t.body, stat.ChReadRows, stat.ChReadBytes, err = clickhouse.Query(scope.WithTable(ctx, t.tag1CountTable), t.url, sql, t.opts, nil)
-	if err != nil {
-		return err
-	}
-
-	rows := t.List()
-
-	// create cost var to validate CH response without writing to t.taggedCosts
-	var costs map[string]*config.Costs
-
-	costs, err = chResultToCosts(rows)
-	if err != nil {
-		return err
-	}
-
-	// The metric does not exist if the response has less rows
-	// than there were tags with '=' op in the initial request
-	// This is due to each tag-value pair of a metric being written
-	// exactly one time as Tag1
-	if len(rows) < eqTermCount {
-		t.body = []byte{}
-		t.metricMightExists = false
-
-		return nil
-	}
-
-	t.taggedCosts = costs
-
-	return nil
-}
-
 func SetCosts(terms []TaggedTerm, costs map[string]*config.Costs) {
 	for i := 0; i < len(terms); i++ {
 		if cost, ok := costs[terms[i].Key]; ok {
 			setCost(&terms[i], cost)
 		}
 	}
 }
-
-func chResultToCosts(body [][]byte) (map[string]*config.Costs, error) {
-	costs := make(map[string]*config.Costs, 0)
-
-	for i := 0; i < len(body); i++ {
-		s := stringutils.UnsafeString(body[i])
-
-		tag, val, count, err := parseTag1CountRow(s)
-		if err != nil {
-			return nil, fmt.Errorf("failed to parse result from clickhouse while querying for tag costs: %s", err.Error())
-		}
-
-		if costs[tag] == nil {
-			costs[tag] = &config.Costs{Cost: nil, ValuesCost: make(map[string]int, 0)}
-		}
-
-		costs[tag].ValuesCost[val] = count
-	}
-
-	return costs, nil
-}
-
-func parseTag1CountRow(s string) (string, string, int, error) {
-	var (
-		tag1, count, tag, val string
-		cnt, n                int
-		err                   error
-	)
-
-	if tag1, count, n = stringutils.Split2(s, "\t"); n != 2 {
-		return "", "", 0, fmt.Errorf("no tag count")
-	}
-
-	if tag, val, n = stringutils.Split2(tag1, "="); n != 2 {
-		return "", "", 0, fmt.Errorf("no '=' in Tag1")
-	}
-
-	if cnt, err = strconv.Atoi(count); err != nil {
-		return "", "", 0, fmt.Errorf("can't convert count to int")
-	}
-
-	return tag, val, cnt, nil
-}