Merge remote-tracking branch 'upstream/master' into 2-prometheus-label-filters

Author: Hedius

cmd/e2e-test/carbon-clickhouse.go

@@ -28,7 +28,7 @@ type CarbonClickhouse struct {
 
 func (c *CarbonClickhouse) Start(testDir, clickhouseURL string) (string, error) {
 	if len(c.Version) == 0 {
-		c.Version = "0.11.4"
+		c.Version = "latest"
 	}
 	if len(c.DockerImage) == 0 {
 		c.DockerImage = "ghcr.io/go-graphite/carbon-clickhouse"

cmd/e2e-test/checks.go

@@ -11,6 +11,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/go-graphite/protocol/carbonapi_v3_pb"
 	"github.com/lomik/graphite-clickhouse/helper/client"
 	"github.com/lomik/graphite-clickhouse/helper/datetime"
 	"github.com/lomik/graphite-clickhouse/helper/tests/compare"
@@ -271,6 +272,24 @@ func compareRender(errors *[]string, name, url string, actual, expected []client
 	}
 }
 
+func parseFilteringFunctions(strFilteringFuncs []string) ([]*carbonapi_v3_pb.FilteringFunction, error) {
+	res := make([]*carbonapi_v3_pb.FilteringFunction, 0, len(strFilteringFuncs))
+	for _, strFF := range strFilteringFuncs {
+		strFFSplit := strings.Split(strFF, "(")
+		if len(strFFSplit) != 2 {
+			return nil, fmt.Errorf("could not parse filtering function: %s", strFF)
+		}
+		name := strFFSplit[0]
+		args := strings.Split(strFFSplit[1], ",")
+		for i := range args {
+			args[i] = strings.TrimSpace(args[i])
+			args[i] = strings.Trim(args[i], ")'")
+		}
+		res = append(res, &carbonapi_v3_pb.FilteringFunction{Name: name, Arguments: args})
+	}
+	return res, nil
+}
+
 func verifyRender(ch *Clickhouse, gch *GraphiteClickhouse, check *RenderCheck, defaultPreision time.Duration) []string {
 	var errors []string
 	httpClient := http.Client{
@@ -280,7 +299,18 @@ func verifyRender(ch *Clickhouse, gch *GraphiteClickhouse, check *RenderCheck, d
 	from := datetime.TimestampTruncate(check.from, defaultPreision)
 	until := datetime.TimestampTruncate(check.until, defaultPreision)
 	for _, format := range check.Formats {
-		if url, result, respHeader, err := client.Render(&httpClient, address, format, check.Targets, from, until); err == nil {
+
+		var filteringFunctions []*carbonapi_v3_pb.FilteringFunction
+		if format == client.FormatPb_v3 {
+			var err error
+			filteringFunctions, err = parseFilteringFunctions(check.FilteringFunctions)
+			if err != nil {
+				errors = append(errors, err.Error())
+				continue
+			}
+		}
+
+		if url, result, respHeader, err := client.Render(&httpClient, address, format, check.Targets, filteringFunctions, check.MaxDataPoints, from, until); err == nil {
 			id := requestId(respHeader)
 			name := ""
 			if check.ErrorRegexp != "" {
@@ -303,7 +333,7 @@ func verifyRender(ch *Clickhouse, gch *GraphiteClickhouse, check *RenderCheck, d
 			if check.CacheTTL > 0 && check.ErrorRegexp == "" {
 				// second query must be find-cached
 				name = "cache"
-				if url, result, respHeader, err = client.Render(&httpClient, address, format, check.Targets, from, until); err == nil {
+				if url, result, respHeader, err = client.Render(&httpClient, address, format, check.Targets, filteringFunctions, check.MaxDataPoints, from, until); err == nil {
 					compareRender(&errors, name, url, result, check.result, true, respHeader, check.CacheTTL)
 				} else {
 					errStr := strings.TrimRight(err.Error(), "\n")

cmd/e2e-test/e2etesting.go

@@ -59,13 +59,15 @@ type Metric struct {
 }
 
 type RenderCheck struct {
-	Name        string              `toml:"name"`
-	Formats     []client.FormatType `toml:"formats"`
-	From        string              `toml:"from"`
-	Until       string              `toml:"until"`
-	Targets     []string            `toml:"targets"`
-	Timeout     time.Duration       `toml:"timeout"`
-	DumpIfEmpty []string            `toml:"dump_if_empty"`
+	Name               string              `toml:"name"`
+	Formats            []client.FormatType `toml:"formats"`
+	From               string              `toml:"from"`
+	Until              string              `toml:"until"`
+	Targets            []string            `toml:"targets"`
+	MaxDataPoints      int64               `toml:"max_data_points"`
+	FilteringFunctions []string            `toml:"filtering_functions"`
+	Timeout            time.Duration       `toml:"timeout"`
+	DumpIfEmpty        []string            `toml:"dump_if_empty"`
 
 	Optimize []string `toml:"optimize"` // optimize tables before run tests
 
@@ -338,6 +340,7 @@ func verifyGraphiteClickhouse(test *TestSchema, gch *GraphiteClickhouse, clickho
 						zap.String("clickhouse config", clickhouseDir),
 						zap.String("graphite-clickhouse config", gch.ConfigTpl),
 						zap.Strings("targets", check.Targets),
+						zap.Strings("filtering_functions", check.FilteringFunctions),
 						zap.String("from_raw", check.From),
 						zap.String("until_raw", check.Until),
 						zap.Int64("from", check.from),
@@ -361,6 +364,7 @@ func verifyGraphiteClickhouse(test *TestSchema, gch *GraphiteClickhouse, clickho
 				zap.String("clickhouse config", clickhouseDir),
 				zap.String("graphite-clickhouse config", gch.ConfigTpl),
 				zap.Strings("targets", check.Targets),
+				zap.Strings("filtering_functions", check.FilteringFunctions),
 				zap.String("from_raw", check.From),
 				zap.String("until_raw", check.Until),
 				zap.Int64("from", check.from),
@@ -377,6 +381,7 @@ func verifyGraphiteClickhouse(test *TestSchema, gch *GraphiteClickhouse, clickho
 				zap.String("clickhouse config", clickhouseDir),
 				zap.String("graphite-clickhouse config", gch.ConfigTpl),
 				zap.Strings("targets", check.Targets),
+				zap.Strings("filtering_functions", check.FilteringFunctions),
 				zap.String("from_raw", check.From),
 				zap.String("until_raw", check.Until),
 				zap.Int64("from", check.from),

cmd/graphite-clickhouse-client/main.go

@@ -5,9 +5,11 @@ import (
 	"fmt"
 	"net/http"
 	"os"
+	"strconv"
 	"strings"
 	"time"
 
+	"github.com/go-graphite/protocol/carbonapi_v3_pb"
 	"github.com/lomik/graphite-clickhouse/helper/client"
 	"github.com/lomik/graphite-clickhouse/helper/datetime"
 )
@@ -31,6 +33,7 @@ func main() {
 	address := flag.String("address", "http://127.0.0.1:9090", "Address of graphite-clickhouse server")
 	fromStr := flag.String("from", "0", "from")
 	untilStr := flag.String("until", "", "until")
+	maxDataPointsStr := flag.String("maxDataPoints", "1048576", "Maximum amount of datapoints in response")
 
 	metricsFind := flag.String("find", "", "Query for /metrics/find/ , valid formats are carbonapi_v3_pb. protobuf, pickle")
 
@@ -71,6 +74,11 @@ func main() {
 		fmt.Printf("invalid until: %s\n", *untilStr)
 		os.Exit(1)
 	}
+	maxDataPoints, err := strconv.ParseInt(*maxDataPointsStr, 10, 64)
+	if err != nil {
+		fmt.Printf("invalid maxDataPoints: %s\n", *maxDataPointsStr)
+		os.Exit(1)
+	}
 
 	httpClient := http.Client{
 		Timeout: *timeout,
@@ -182,7 +190,7 @@ func main() {
 		if formatRender == client.FormatDefault {
 			formatRender = client.FormatPb_v3
 		}
-		queryRaw, r, respHeader, err := client.Render(&httpClient, *address, formatRender, targets, int64(from), int64(until))
+		queryRaw, r, respHeader, err := client.Render(&httpClient, *address, formatRender, targets, []*carbonapi_v3_pb.FilteringFunction{}, maxDataPoints, int64(from), int64(until))
 		if respHeader != nil {
 			fmt.Printf("Responce header: %+v\n", respHeader)
 		}

finder/tagged.go

@@ -291,7 +291,10 @@ func ParseTaggedConditions(conditions []string, config *config.Config, autocompl
 		case "=":
 			terms[i].Op = TaggedTermEq
 			terms[i].HasWildcard = where.HasWildcard(terms[i].Value)
-			if !terms[i].HasWildcard {
+			// special case when using useCarbonBehaviour = true
+			// which matches everything that does not have that tag
+			emptyValue := config.FeatureFlags.UseCarbonBehavior && terms[i].Value == ""
+			if !terms[i].HasWildcard && !emptyValue {
 				nonWildcards++
 			}
 		case "!=":

helper/clickhouse/clickhouse.go

@@ -72,7 +72,10 @@ func extractClickhouseError(e string) (int, string) {
 			return http.StatusServiceUnavailable, "Storage configuration error"
 		}
 	}
-	return http.StatusInternalServerError, "Storage error"
+	if strings.HasPrefix(e, "clickhouse response status 404: Code: 60. DB::Exception: Table default.") {
+		return http.StatusServiceUnavailable, "Storage default tables damaged"
+	}
+	return http.StatusServiceUnavailable, "Storage unavailable"
 }
 
 func HandleError(w http.ResponseWriter, err error) (status int, queueFail bool) {

helper/clickhouse/clickhouse_test.go

@@ -34,10 +34,15 @@ func Test_extractClickhouseError(t *testing.T) {
 			wantStatus:  http.StatusForbidden,
 			wantMessage: "Storage read limit for memory",
 		},
+		{
+			errStr:      "clickhouse response status 404: Code: 60. DB::Exception: Table default.graphite_index does not exist. (UNKNOWN_TABLE) (version 23.12.6.19 (official build))\n",
+			wantStatus:  http.StatusServiceUnavailable,
+			wantMessage: "Storage default tables damaged",
+		},
 		{
 			errStr:      "Other error",
-			wantStatus:  http.StatusInternalServerError,
-			wantMessage: "Storage error",
+			wantStatus:  http.StatusServiceUnavailable,
+			wantMessage: "Storage unavailable",
 		},
 	}
 	for _, tt := range tests {

helper/client/render.go

@@ -37,7 +37,7 @@ type Metric struct {
 
 // Render do /metrics/find/ request
 // Valid formats are carbonapi_v3_pb. protobuf, pickle, json
-func Render(client *http.Client, address string, format FormatType, targets []string, from, until int64) (string, []Metric, http.Header, error) {
+func Render(client *http.Client, address string, format FormatType, targets []string, filteringFunctions []*protov3.FilteringFunction, maxDataPoints, from, until int64) (string, []Metric, http.Header, error) {
 	rUrl := "/render/"
 	if format == FormatDefault {
 		format = FormatPb_v3
@@ -56,6 +56,7 @@ func Render(client *http.Client, address string, format FormatType, targets []st
 	}
 	fromStr := strconv.FormatInt(from, 10)
 	untilStr := strconv.FormatInt(until, 10)
+	maxDataPointsStr := strconv.FormatInt(maxDataPoints, 10)
 
 	u, err := url.Parse(address + rUrl)
 	if err != nil {
@@ -77,10 +78,12 @@ func Render(client *http.Client, address string, format FormatType, targets []st
 		}
 		for i, target := range targets {
 			r.Metrics[i] = protov3.FetchRequest{
-				Name:           target,
-				StartTime:      from,
-				StopTime:       until,
-				PathExpression: target,
+				Name:            target,
+				StartTime:       from,
+				StopTime:        until,
+				PathExpression:  target,
+				FilterFunctions: filteringFunctions,
+				MaxDataPoints:   maxDataPoints,
 			}
 		}
 
@@ -93,10 +96,11 @@ func Render(client *http.Client, address string, format FormatType, targets []st
 		}
 	case FormatPb_v2, FormatProtobuf, FormatPickle, FormatJSON:
 		v := url.Values{
-			"format": []string{format.String()},
-			"from":   []string{fromStr},
-			"until":  []string{untilStr},
-			"target": targets,
+			"format":        []string{format.String()},
+			"from":          []string{fromStr},
+			"until":         []string{untilStr},
+			"target":        targets,
+			"maxDataPoints": []string{maxDataPointsStr},
 		}
 		u.RawQuery = v.Encode()
 	default:

render/data/data.go

@@ -54,10 +54,14 @@ func (d *Data) GetAggregation(id uint32) (string, error) {
 	if err != nil {
 		return function, err
 	}
-	if function == "any" || function == "anyLast" {
+	switch function {
+	case "any":
+		return "first", nil
+	case "anyLast":
 		return "last", nil
+	default:
+		return function, nil
 	}
-	return function, nil
 }
 
 // data wraps Data and adds asynchronous processing of data

render/data/query.go

@@ -5,6 +5,7 @@ import (
 	"crypto/tls"
 	"errors"
 	"fmt"
+	"net/http"
 	"os"
 	"strings"
 	"sync"
@@ -15,6 +16,7 @@ import (
 
 	"github.com/lomik/graphite-clickhouse/config"
 	"github.com/lomik/graphite-clickhouse/helper/clickhouse"
+	"github.com/lomik/graphite-clickhouse/helper/errs"
 	"github.com/lomik/graphite-clickhouse/helper/rollup"
 	"github.com/lomik/graphite-clickhouse/metrics"
 	"github.com/lomik/graphite-clickhouse/pkg/dry"
@@ -145,7 +147,11 @@ func (q *query) getDataPoints(ctx context.Context, cond *conditions) error {
 	// carbonlink request
 	carbonlinkResponseRead := queryCarbonlink(ctx, carbonlink, cond.metricsUnreverse)
 
-	cond.prepareLookup()
+	err = cond.prepareLookup()
+	if err != nil {
+		logger.Error("prepare_lookup", zap.Error(err))
+		return errs.NewErrorWithCode(err.Error(), http.StatusBadRequest)
+	}
 	cond.setStep(q.cStep)
 	if cond.step < 1 {
 		return ErrSetStepTimeout
@@ -279,7 +285,7 @@ func (c *conditions) prepareMetricsLists() {
 	}
 }
 
-func (c *conditions) prepareLookup() {
+func (c *conditions) prepareLookup() error {
 	age := uint32(dry.Max(0, time.Now().Unix()-c.From))
 	c.aggregations = make(map[string][]string)
 	c.appliedFunctions = make(map[string][]string)
@@ -295,7 +301,11 @@ func (c *conditions) prepareLookup() {
 		// Currently it just finds the first target matching the metric
 		// to avoid making multiple request for every type of aggregation for a given metric.
 		for _, alias := range c.AM.Get(c.metricsUnreverse[i]) {
-			if requestedAgg := c.GetRequestedAggregation(alias.Target); requestedAgg != "" {
+			requestedAgg, err := c.GetRequestedAggregation(alias.Target)
+			if err != nil {
+				return fmt.Errorf("failed to choose appropriate aggregation for '%s': %s", alias.Target, err.Error())
+			}
+			if requestedAgg != "" {
 				agg = rollup.AggrMap[requestedAgg]
 				c.appliedFunctions[alias.Target] = []string{graphiteConsolidationFunction}
 				break
@@ -330,6 +340,7 @@ func (c *conditions) prepareLookup() {
 			mm.WriteString(c.metricsRequested[i] + "\n")
 		}
 	}
+	return nil
 }
 
 var ErrSetStepTimeout = errors.New("unexpected error, setStep timeout")