return 400 status code in case consolidateBy uses invalid argument

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")

render/data/targets.go

@@ -131,32 +131,47 @@ TableLoop:
 	return fmt.Errorf("data tables is not specified for %v", tt.List[0])
 }
 
-func (tt *Targets) GetRequestedAggregation(target string) string {
+func (tt *Targets) GetRequestedAggregation(target string) (string, error) {
 	if ffs, ok := tt.filteringFunctionsByTarget[target]; !ok {
-		return ""
+		return "", nil
 	} else {
 		for _, filteringFunc := range ffs {
 			ffName := filteringFunc.GetName()
 			ffArgs := filteringFunc.GetArguments()
-			if ffName == graphiteConsolidationFunction && len(ffArgs) > 0 {
-				switch ffArgs[0] {
-				// 'last' in graphite == clickhouse aggregate function 'anyLast'
-				case "last":
-					return "anyLast"
-				// 'first' in graphite == clickhouse aggregate function 'any'
-				case "first":
-					return "any"
-				// Graphite standard supports both average and avg.
-				// It is the only aggregation that has two aliases.
-				// https://graphite.readthedocs.io/en/latest/functions.html#graphite.render.functions.consolidateBy
-				case "average":
-					return "avg"
-				// avg, sum, max, min have the same name in clickhouse
-				default:
-					return ffArgs[0]
-				}
+
+			if ffName != graphiteConsolidationFunction {
+				continue
+			}
+
+			if len(ffArgs) < 1 {
+				return "", fmt.Errorf("no argumets were provided to consolidateBy function")
+			}
+
+			switch ffArgs[0] {
+			// 'last' in graphite == clickhouse aggregate function 'anyLast'
+			case "last":
+				return "anyLast", nil
+			// 'first' in graphite == clickhouse aggregate function 'any'
+			case "first":
+				return "any", nil
+			// Graphite standard supports both average and avg.
+			// It is the only aggregation that has two aliases.
+			// https://graphite.readthedocs.io/en/latest/functions.html#graphite.render.functions.consolidateBy
+			case "average":
+				return "avg", nil
+			// avg, sum, max, min have the same name in clickhouse
+			case "avg", "sum", "max", "min":
+				return ffArgs[0], nil
+			default:
+				return "",
+					fmt.Errorf(
+						"unknown \"%s\" argument function (allowed argumets are: 'avg', 'average', 'sum', 'max', 'min', 'last', 'first'): recieved %s",
+						graphiteConsolidationFunction,
+						ffArgs[0],
+					)
+
 			}
 		}
 	}
-	return ""
+	return "", nil
 }

tests/consolidateBy/test.toml

@@ -302,4 +302,4 @@ targets = [
 filtering_functions = [
     "consolidateBy('invalid')"
 ]
-error_regexp = "^500: Storage error"
+error_regexp = "^400: failed to choose appropriate aggregation"