func()

in custom-metrics-stackdriver-adapter/pkg/adapter/translator/query_builder.go [678:772]
94 lines of code
28 McCabe index (conditional complexity)

func (t *Translator) filterForSelector(metricSelector labels.Selector, allowedLabelPrefixes []string, allowedFullLabelNames []string) (string, string, error) {
	requirements, selectable := metricSelector.Requirements()
	if !selectable {
		return "", "", apierr.NewBadRequest(fmt.Sprintf("Label selector is impossible to match: %s", metricSelector))
	}
	filters := []string{}
	var reducer string
	for _, req := range requirements {
		if req.Key() == "reducer" {
			if req.Operator() != selection.Equals && req.Operator() != selection.DoubleEquals {
				return "", "", NewLabelNotAllowedError(fmt.Sprintf("Reducer must use '=' or '==': You used %s", req.Operator()))
			}
			if req.Values().Len() != 1 {
				return "", "", NewLabelNotAllowedError("Reducer must select a single value")
			}
			r, found := req.Values().PopAny()
			if !found {
				return "", "", NewLabelNotAllowedError("Reducer must specify a value")
			}
			if !allowedReducers[r] {
				return "", "", NewLabelNotAllowedError("Specified reducer is not supported: " + r)
			}
			reducer = r
			continue
		}
		l := req.Values().List()
		switch req.Operator() {
		case selection.Equals, selection.DoubleEquals:
			if isAllowedLabelName(req.Key(), allowedLabelPrefixes, allowedFullLabelNames) {
				filters = append(filters, fmt.Sprintf("%s = %q", req.Key(), l[0]))
			} else {
				return "", "", NewLabelNotAllowedError(req.Key())
			}
		case selection.NotEquals:
			if isAllowedLabelName(req.Key(), allowedLabelPrefixes, allowedFullLabelNames) {
				filters = append(filters, fmt.Sprintf("%s != %q", req.Key(), l[0]))
			} else {
				return "", "", NewLabelNotAllowedError(req.Key())
			}
		case selection.In:
			if isAllowedLabelName(req.Key(), allowedLabelPrefixes, allowedFullLabelNames) {
				if len(l) == 1 {
					filters = append(filters, fmt.Sprintf("%s = %s", req.Key(), l[0]))
				} else {
					filters = append(filters, fmt.Sprintf("%s = one_of(%s)", req.Key(), strings.Join(quoteAll(l), ",")))
				}
			} else {
				return "", "", NewLabelNotAllowedError(req.Key())
			}
		case selection.NotIn:
			if isAllowedLabelName(req.Key(), allowedLabelPrefixes, allowedFullLabelNames) {
				if len(l) == 1 {
					filters = append(filters, fmt.Sprintf("%s != %s", req.Key(), l[0]))
				} else {
					filters = append(filters, fmt.Sprintf("NOT %s = one_of(%s)", req.Key(), strings.Join(quoteAll(l), ",")))
				}
			} else {
				return "", "", NewLabelNotAllowedError(req.Key())
			}
		case selection.Exists:
			prefix, suffix, err := splitMetricLabel(req.Key(), allowedLabelPrefixes)
			if err == nil {
				filters = append(filters, fmt.Sprintf("%s : %s", prefix, suffix))
			} else {
				return "", "", NewLabelNotAllowedError(req.Key())
			}
		case selection.DoesNotExist:
			// DoesNotExist is not allowed due to Stackdriver filtering syntax limitation
			return "", "", apierr.NewBadRequest("Label selector with operator DoesNotExist is not allowed")
		case selection.GreaterThan:
			if isAllowedLabelName(req.Key(), allowedLabelPrefixes, allowedFullLabelNames) {
				value, err := strconv.ParseInt(l[0], 10, 64)
				if err != nil {
					return "", "", apierr.NewInternalError(fmt.Errorf("Unexpected error: value %s could not be parsed to integer", l[0]))
				}
				filters = append(filters, fmt.Sprintf("%s > %v", req.Key(), value))
			} else {
				return "", "", NewLabelNotAllowedError(req.Key())
			}
		case selection.LessThan:
			if isAllowedLabelName(req.Key(), allowedLabelPrefixes, allowedFullLabelNames) {
				value, err := strconv.ParseInt(l[0], 10, 64)
				if err != nil {
					return "", "", apierr.NewInternalError(fmt.Errorf("Unexpected error: value %s could not be parsed to integer", l[0]))
				}
				filters = append(filters, fmt.Sprintf("%s < %v", req.Key(), value))
			} else {
				return "", "", NewLabelNotAllowedError(req.Key())
			}
		default:
			return "", "", NewOperationNotSupportedError(fmt.Sprintf("Selector with operator %q", req.Operator()))
		}
	}
	return strings.Join(filters, " AND "), reducer, nil
}