Add more metrics generation settings (#11)

Makefile

@@ -22,7 +22,7 @@ clean:
 
 .PHONY: build-configuration
 build-configuration:
-	go build -o _output/ndc-prometheus-configuration .
+	go build -o _output/ndc-prometheus ./configuration
 
 .PHONY: build-jsonschema
 build-jsonschema:

configuration/update.go

@@ -23,17 +23,23 @@ import (
 var clientTracer = otel.Tracer("PrometheusClient")
 var bannedLabels = []string{"__name__"}
 
+type ExcludeLabels struct {
+	Regex  *regexp.Regexp
+	Labels []string
+}
+
 type updateCommand struct {
-	Client    *client.Client
-	OutputDir string
-	Config    *metadata.Configuration
-	Include   []*regexp.Regexp
-	Exclude   []*regexp.Regexp
+	Client        *client.Client
+	OutputDir     string
+	Config        *metadata.Configuration
+	Include       []*regexp.Regexp
+	Exclude       []*regexp.Regexp
+	ExcludeLabels []ExcludeLabels
 }
 
 func introspectSchema(ctx context.Context, args *UpdateArguments) error {
 	start := time.Now()
-	slog.Info("introspect metrics metadata...", slog.String("dir", args.Dir))
+	slog.Info("introspecting metadata", slog.String("dir", args.Dir))
 	originalConfig, err := metadata.ReadConfiguration(args.Dir)
 	if err != nil {
 		if !os.IsNotExist(err) {
@@ -56,6 +62,24 @@ func introspectSchema(ctx context.Context, args *UpdateArguments) error {
 	}
 
 	if originalConfig.Generator.Metrics.Enabled {
+		slog.Info("introspecting metrics",
+			slog.String("behavior", string(originalConfig.Generator.Metrics.Behavior)),
+			slog.Any("include", originalConfig.Generator.Metrics.Include),
+			slog.Any("exclude", originalConfig.Generator.Metrics.Exclude),
+		)
+		for _, el := range originalConfig.Generator.Metrics.ExcludeLabels {
+			if len(el.Labels) == 0 {
+				continue
+			}
+			rg, err := regexp.Compile(el.Pattern)
+			if err != nil {
+				return fmt.Errorf("invalid exclude_labels pattern `%s`: %s", el.Pattern, err)
+			}
+			cmd.ExcludeLabels = append(cmd.ExcludeLabels, ExcludeLabels{
+				Regex:  rg,
+				Labels: el.Labels,
+			})
+		}
 		if err := cmd.updateMetricsMetadata(ctx); err != nil {
 			return err
 		}
@@ -78,6 +102,15 @@ func (uc *updateCommand) updateMetricsMetadata(ctx context.Context) error {
 	}
 
 	newMetrics := map[string]metadata.MetricInfo{}
+	if uc.Config.Generator.Metrics.Behavior == metadata.MetricsGenerationMerge {
+		for key, metric := range uc.Config.Metadata.Metrics {
+			if (len(uc.Include) > 0 && !validateRegularExpressions(uc.Include, key)) || validateRegularExpressions(uc.Exclude, key) {
+				continue
+			}
+			newMetrics[key] = metric
+		}
+	}
+
 	for key, info := range metricsInfo {
 		if len(info) == 0 {
 			continue
@@ -98,27 +131,35 @@ func (uc *updateCommand) updateMetricsMetadata(ctx context.Context) error {
 			Labels:      labels,
 		}
 	}
-
 	uc.Config.Metadata.Metrics = newMetrics
 	return nil
 }
 
 func (uc *updateCommand) getAllLabelsOfMetric(ctx context.Context, name string) (map[string]metadata.LabelInfo, error) {
-	labels, warnings, err := uc.Client.LabelNames(ctx, []string{name}, time.Time{}, time.Now())
+	labels, warnings, err := uc.Client.Series(ctx, []string{name}, uc.Config.Generator.Metrics.StartAt, time.Now(), 1)
 	if err != nil {
 		return nil, err
 	}
 
 	if len(warnings) > 0 {
-		slog.Warn(fmt.Sprintf("warning when fetching labels for metric `%s`", name), slog.Any("warnings", warnings))
+		slog.Debug(fmt.Sprintf("warning when fetching labels for metric `%s`", name), slog.Any("warnings", warnings))
 	}
-
 	results := make(map[string]metadata.LabelInfo)
-	for _, label := range labels {
-		if slices.Contains(bannedLabels, label) {
+	if len(labels) == 0 {
+		return results, nil
+	}
+	excludedLabels := bannedLabels
+	for _, el := range uc.ExcludeLabels {
+		if el.Regex.MatchString(name) {
+			excludedLabels = append(excludedLabels, el.Labels...)
+		}
+	}
+	for key := range labels[0] {
+		if !key.IsValid() || slices.Contains(excludedLabels, string(key)) {
 			continue
 		}
-		results[label] = metadata.LabelInfo{}
+
+		results[string(key)] = metadata.LabelInfo{}
 	}
 	return results, nil
 }
@@ -157,7 +198,7 @@ func (uc *updateCommand) writeConfigFile() error {
 	var buf bytes.Buffer
 	writer := bufio.NewWriter(&buf)
 
-	_, _ = writer.WriteString("# yaml-language-server: $schema=../../jsonschema/configuration.json\n")
+	_, _ = writer.WriteString("# yaml-language-server: $schema=https://raw.githubusercontent.com/hasura/ndc-prometheus/main/jsonschema/configuration.json\n")
 	encoder := yaml.NewEncoder(writer)
 	encoder.SetIndent(2)
 	if err := encoder.Encode(uc.Config); err != nil {
@@ -174,9 +215,12 @@ var defaultConfiguration = metadata.Configuration{
 	},
 	Generator: metadata.GeneratorSettings{
 		Metrics: metadata.MetricsGeneratorSettings{
-			Enabled: true,
-			Include: []string{},
-			Exclude: []string{},
+			Enabled:       true,
+			Behavior:      metadata.MetricsGenerationMerge,
+			Include:       []string{},
+			Exclude:       []string{},
+			ExcludeLabels: []metadata.ExcludeLabelsSetting{},
+			StartAt:       time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC),
 		},
 	},
 	Metadata: metadata.Metadata{

connector-definition/configuration.yaml

@@ -5,8 +5,11 @@ connection_settings:
 generator:
   metrics:
     enabled: true
+    behavior: merge
     include: []
     exclude: []
+    exclude_labels: []
+    start_at: "2024-01-01T00:00:00Z"
 metadata:
   metrics: {}
   native_operations:

connector/api/series.go

@@ -25,7 +25,7 @@ type PrometheusSeriesArguments struct {
 }
 
 // Validate validates arguments and options
-func (psa *PrometheusSeriesArguments) Validate(state *metadata.State, span trace.Span) (*PrometheusSeriesArguments, []v1.Option, error) {
+func (psa PrometheusSeriesArguments) Validate(state *metadata.State, span trace.Span) (*PrometheusSeriesArguments, []v1.Option, error) {
 	endTime := time.Now()
 	arguments := PrometheusSeriesArguments{
 		Match: psa.Match,
@@ -66,11 +66,15 @@ func FunctionPrometheusSeries(ctx context.Context, state *metadata.State, argume
 	ctx, span := state.Tracer.Start(ctx, "Prometheus Series")
 	defer span.End()
 
-	args, opts, err := arguments.Validate(state, span)
+	args, _, err := arguments.Validate(state, span)
 	if err != nil {
 		return nil, err
 	}
-	labelSets, warnings, err := state.Client.Series(ctx, args.Match, *args.Start, *args.End, opts...)
+	var limit uint64
+	if args.Limit != nil {
+		limit = *args.Limit
+	}
+	labelSets, warnings, err := state.Client.Series(ctx, args.Match, *args.Start, *args.End, limit)
 	if len(warnings) > 0 {
 		span.SetAttributes(attribute.StringSlice("warnings", warnings))
 	}

connector/client/client.go

@@ -2,6 +2,7 @@ package client
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"log/slog"
@@ -82,6 +83,51 @@ func (c *Client) ApplyOptions(span trace.Span, timeout any) ([]v1.Option, error)
 	return options, nil
 }
 
+func (c *Client) do(ctx context.Context, req *http.Request) (*http.Response, []byte, v1.Warnings, error) {
+	resp, body, err := c.client.Do(ctx, req)
+	if err != nil {
+		return resp, body, nil, err
+	}
+
+	code := resp.StatusCode
+
+	if code/100 != 2 && !apiError(code) {
+		errorType, errorMsg := errorTypeAndMsgFor(resp)
+		return resp, body, nil, &v1.Error{
+			Type:   errorType,
+			Msg:    errorMsg,
+			Detail: string(body),
+		}
+	}
+
+	var result apiResponse
+
+	if http.StatusNoContent != code {
+		if jsonErr := json.Unmarshal(body, &result); jsonErr != nil {
+			return resp, body, nil, &v1.Error{
+				Type: v1.ErrBadResponse,
+				Msg:  jsonErr.Error(),
+			}
+		}
+	}
+
+	if apiError(code) && result.Status == "success" {
+		err = &v1.Error{
+			Type: v1.ErrBadResponse,
+			Msg:  "inconsistent body for response code",
+		}
+	}
+
+	if result.Status == "error" {
+		err = &v1.Error{
+			Type: result.ErrorType,
+			Msg:  result.Error,
+		}
+	}
+
+	return resp, []byte(result.Data), result.Warnings, err
+}
+
 // wrap the prometheus client with trace context
 type httpClient struct {
 	api.Client
@@ -111,7 +157,7 @@ func (ac *httpClient) Do(ctx context.Context, req *http.Request) (*http.Response
 		if err != nil {
 			attrs = append(attrs, slog.String("error", err.Error()))
 		}
-		slog.Debug(fmt.Sprintf("%s %s", strings.ToUpper(req.Method), req.RequestURI), attrs...)
+		slog.Debug(fmt.Sprintf("%s %s", strings.ToUpper(req.Method), req.URL.String()), attrs...)
 	}
 	return r, bs, err
 }