add runtime settings

configuration/update.go

@@ -232,6 +232,13 @@ var defaultConfiguration = metadata.Configuration{
 		Metrics:          map[string]metadata.MetricInfo{},
 		NativeOperations: metadata.NativeOperations{},
 	},
+	Runtime: metadata.RuntimeSettings{
+		Flat: false,
+		Format: metadata.RuntimeFormatSettings{
+			Timestamp: metadata.TimestampUnix,
+			Value:     metadata.ValueFloat64,
+		},
+	},
 }
 
 func compileRegularExpressions(inputs []string) []*regexp.Regexp {

connector/connector.go

@@ -19,6 +19,7 @@ type PrometheusConnector struct {
 	capabilities *schema.RawCapabilitiesResponse
 	rawSchema    *schema.RawSchemaResponse
 	metadata     *metadata.Metadata
+	runtime      *metadata.RuntimeSettings
 	apiHandler   api.DataConnectorHandler
 }
 
@@ -55,6 +56,7 @@ func (c *PrometheusConnector) ParseConfiguration(ctx context.Context, configurat
 	}
 
 	c.metadata = &config.Metadata
+	c.runtime = &config.Runtime
 
 	return config, nil
 }

connector/internal/collection.go

@@ -25,6 +25,7 @@ var valueBinaryOperators = map[string]string{
 type QueryCollectionExecutor struct {
 	Client    *client.Client
 	Tracer    trace.Tracer
+	Runtime   *metadata.RuntimeSettings
 	Request   *schema.QueryRequest
 	Metric    metadata.MetricInfo
 	Variables map[string]any
@@ -118,7 +119,7 @@ func (qce *QueryCollectionExecutor) queryInstant(ctx context.Context, queryStrin
 		vector = vector[:*qce.Request.Query.Limit]
 	}
 
-	results := createQueryResultsFromVector(vector, qce.Metric.Labels)
+	results := createQueryResultsFromVector(vector, qce.Metric.Labels, qce.Runtime)
 	return results, nil
 }
 
@@ -156,7 +157,7 @@ func (qce *QueryCollectionExecutor) queryRange(ctx context.Context, queryString
 	if qce.Request.Query.Limit != nil && *qce.Request.Query.Limit < len(matrix) {
 		matrix = matrix[:*qce.Request.Query.Limit]
 	}
-	results := createQueryResultsFromMatrix(matrix, qce.Metric.Labels)
+	results := createQueryResultsFromMatrix(matrix, qce.Metric.Labels, qce.Runtime)
 
 	return results, nil
 }

connector/internal/native_query.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"strconv"
+	"time"
 
 	"github.com/hasura/ndc-prometheus/connector/client"
 	"github.com/hasura/ndc-prometheus/connector/metadata"
@@ -25,6 +26,7 @@ type nativeQueryParameters struct {
 type NativeQueryExecutor struct {
 	Client      *client.Client
 	Tracer      trace.Tracer
+	Runtime     *metadata.RuntimeSettings
 	Request     *schema.QueryRequest
 	NativeQuery *metadata.NativeQuery
 	Arguments   map[string]any
@@ -187,7 +189,7 @@ func (nqe *NativeQueryExecutor) queryInstant(ctx context.Context, queryString st
 	if err != nil {
 		return nil, schema.UnprocessableContentError(err.Error(), nil)
 	}
-	results := createQueryResultsFromVector(vector, nqe.NativeQuery.Labels)
+	results := createQueryResultsFromVector(vector, nqe.NativeQuery.Labels, nqe.Runtime)
 
 	return results, nil
 }
@@ -198,37 +200,49 @@ func (nqe *NativeQueryExecutor) queryRange(ctx context.Context, queryString stri
 		return nil, schema.UnprocessableContentError(err.Error(), nil)
 	}
 
-	results := createQueryResultsFromMatrix(matrix, nqe.NativeQuery.Labels)
+	results := createQueryResultsFromMatrix(matrix, nqe.NativeQuery.Labels, nqe.Runtime)
 
 	return results, nil
 }
 
-func createQueryResultsFromVector(vector model.Vector, labels map[string]metadata.LabelInfo) []map[string]any {
+func createQueryResultsFromVector(vector model.Vector, labels map[string]metadata.LabelInfo, runtime *metadata.RuntimeSettings) []map[string]any {
 	results := make([]map[string]any, len(vector))
 	for i, item := range vector {
+		ts := formatTimestamp(item.Timestamp, runtime.Format.Timestamp)
+		value := formatValue(item.Value, runtime.Format.Value)
 		r := map[string]any{
-			metadata.TimestampKey: item.Timestamp,
-			metadata.ValueKey:     item.Value.String(),
+			metadata.TimestampKey: ts,
+			metadata.ValueKey:     value,
 			metadata.LabelsKey:    item.Metric,
-			metadata.ValuesKey: []map[string]any{
-				{
-					metadata.TimestampKey: item.Timestamp,
-					metadata.ValueKey:     item.Value.String(),
-				},
-			},
 		}
 
 		for label := range labels {
 			r[label] = string(item.Metric[model.LabelName(label)])
 		}
+		if !runtime.Flat {
+			r[metadata.ValuesKey] = []map[string]any{
+				{
+					metadata.TimestampKey: ts,
+					metadata.ValueKey:     value,
+				},
+			}
+		}
 
 		results[i] = r
 	}
 
 	return results
 }
 
-func createQueryResultsFromMatrix(matrix model.Matrix, labels map[string]metadata.LabelInfo) []map[string]any {
+func createQueryResultsFromMatrix(matrix model.Matrix, labels map[string]metadata.LabelInfo, runtime *metadata.RuntimeSettings) []map[string]any {
+	if runtime.Flat {
+		return createFlatQueryResultsFromMatrix(matrix, labels, runtime)
+	}
+
+	return createGroupQueryResultsFromMatrix(matrix, labels, runtime)
+}
+
+func createGroupQueryResultsFromMatrix(matrix model.Matrix, labels map[string]metadata.LabelInfo, runtime *metadata.RuntimeSettings) []map[string]any {
 	results := make([]map[string]any, len(matrix))
 	for i, item := range matrix {
 		r := map[string]any{
@@ -242,13 +256,15 @@ func createQueryResultsFromMatrix(matrix model.Matrix, labels map[string]metadat
 		valuesLen := len(item.Values)
 		values := make([]map[string]any, valuesLen)
 		for i, value := range item.Values {
+			ts := formatTimestamp(value.Timestamp, runtime.Format.Timestamp)
+			v := formatValue(value.Value, runtime.Format.Value)
 			values[i] = map[string]any{
-				metadata.TimestampKey: value.Timestamp,
-				metadata.ValueKey:     value.Value.String(),
+				metadata.TimestampKey: ts,
+				metadata.ValueKey:     v,
 			}
 			if i == valuesLen-1 {
-				r[metadata.TimestampKey] = value.Timestamp
-				r[metadata.ValueKey] = value.Value.String()
+				r[metadata.TimestampKey] = ts
+				r[metadata.ValueKey] = v
 			}
 		}
 
@@ -258,3 +274,46 @@ func createQueryResultsFromMatrix(matrix model.Matrix, labels map[string]metadat
 
 	return results
 }
+
+func createFlatQueryResultsFromMatrix(matrix model.Matrix, labels map[string]metadata.LabelInfo, runtime *metadata.RuntimeSettings) []map[string]any {
+	results := []map[string]any{}
+
+	for _, item := range matrix {
+		for _, value := range item.Values {
+			ts := formatTimestamp(value.Timestamp, runtime.Format.Timestamp)
+			v := formatValue(value.Value, runtime.Format.Value)
+			r := map[string]any{
+				metadata.LabelsKey:    item.Metric,
+				metadata.TimestampKey: ts,
+				metadata.ValueKey:     v,
+				metadata.ValuesKey:    nil,
+			}
+
+			for label := range labels {
+				r[label] = string(item.Metric[model.LabelName(label)])
+			}
+
+			results = append(results, r)
+		}
+	}
+
+	return results
+}
+
+func formatTimestamp(ts model.Time, format metadata.TimestampFormat) any {
+	switch format {
+	case metadata.TimestampRFC3339:
+		return ts.Time().Format(time.RFC3339)
+	default:
+		return ts
+	}
+}
+
+func formatValue(value model.SampleValue, format metadata.ValueFormat) any {
+	switch format {
+	case metadata.ValueFloat64:
+		return float64(value)
+	default:
+		return value.String()
+	}
+}

connector/metadata/configuration.go

@@ -11,9 +11,14 @@ import (
 
 // Configuration the configuration of Prometheus connector
 type Configuration struct {
+	// Connection settings to connect the Prometheus server
 	ConnectionSettings client.ClientSettings `json:"connection_settings" yaml:"connection_settings"`
-	Generator          GeneratorSettings     `json:"generator" yaml:"generator"`
-	Metadata           Metadata              `json:"metadata" yaml:"metadata"`
+	// Settings to generate metrics metadata
+	Generator GeneratorSettings `json:"generator" yaml:"generator"`
+	// The metadata of metrics and native queries
+	Metadata Metadata `json:"metadata" yaml:"metadata"`
+	// Runtime settings
+	Runtime RuntimeSettings `json:"runtime" yaml:"runtime"`
 }
 
 // MetricsGenerationBehavior the behavior of metrics generation
@@ -53,8 +58,36 @@ type GeneratorSettings struct {
 	Metrics MetricsGeneratorSettings `json:"metrics" yaml:"metrics"`
 }
 
+// TimestampFormat the format for timestamp serialization
+type TimestampFormat string
+
+const (
+	TimestampRFC3339 TimestampFormat = "rfc3339"
+	TimestampUnix    TimestampFormat = "unix"
+)
+
+// ValueFormat the format for value serialization
+type ValueFormat string
+
+const (
+	ValueString  ValueFormat = "string"
+	ValueFloat64 ValueFormat = "float64"
+)
+
+// RuntimeFormatSettings format settings for timestamps and values in runtime
+type RuntimeFormatSettings struct {
+	// The serialization format for timestamp
+	Timestamp TimestampFormat `json:"timestamp" yaml:"timestamp" jsonschema:"enum=rfc3339,enum=unix,default=unix"`
+	// The serialization format for value
+	Value ValueFormat `json:"value" yaml:"value" jsonschema:"enum=string,enum=float64,default=string"`
+}
+
 // RuntimeSettings contain settings for the runtime engine
 type RuntimeSettings struct {
+	// Flatten value points to the root array
+	Flat bool `json:"flat" yaml:"flat"`
+	// The serialization format for response fields
+	Format RuntimeFormatSettings `json:"format" yaml:"format"`
 }
 
 // ReadConfiguration reads the configuration from file

connector/query.go

@@ -53,6 +53,7 @@ func (c *PrometheusConnector) execQuery(ctx context.Context, state *metadata.Sta
 		executor := &internal.NativeQueryExecutor{
 			Tracer:      state.Tracer,
 			Client:      state.Client,
+			Runtime:     c.runtime,
 			Request:     request,
 			NativeQuery: &metadata.NativeQuery{},
 			Arguments:   arguments,
@@ -82,6 +83,7 @@ func (c *PrometheusConnector) execQuery(ctx context.Context, state *metadata.Sta
 		executor := &internal.NativeQueryExecutor{
 			Tracer:      state.Tracer,
 			Client:      state.Client,
+			Runtime:     c.runtime,
 			Request:     request,
 			NativeQuery: &nativeQuery,
 			Arguments:   arguments,
@@ -115,6 +117,7 @@ func (c *PrometheusConnector) execQuery(ctx context.Context, state *metadata.Sta
 			executor := &internal.QueryCollectionExecutor{
 				Tracer:    state.Tracer,
 				Client:    state.Client,
+				Runtime:   c.runtime,
 				Request:   request,
 				Metric:    collection,
 				Arguments: arguments,

jsonschema/configuration.json

@@ -105,14 +105,18 @@
         },
         "metadata": {
           "$ref": "#/$defs/Metadata"
+        },
+        "runtime": {
+          "$ref": "#/$defs/RuntimeSettings"
         }
       },
       "additionalProperties": false,
       "type": "object",
       "required": [
         "connection_settings",
         "generator",
-        "metadata"
+        "metadata",
+        "runtime"
       ]
     },
     "EnvironmentValue": {
@@ -419,6 +423,48 @@
       },
       "type": "object"
     },
+    "RuntimeFormatSettings": {
+      "properties": {
+        "timestamp": {
+          "type": "string",
+          "enum": [
+            "rfc3339",
+            "unix"
+          ],
+          "default": "unix"
+        },
+        "value": {
+          "type": "string",
+          "enum": [
+            "string",
+            "float64"
+          ],
+          "default": "string"
+        }
+      },
+      "additionalProperties": false,
+      "type": "object",
+      "required": [
+        "timestamp",
+        "value"
+      ]
+    },
+    "RuntimeSettings": {
+      "properties": {
+        "flat": {
+          "type": "boolean"
+        },
+        "format": {
+          "$ref": "#/$defs/RuntimeFormatSettings"
+        }
+      },
+      "additionalProperties": false,
+      "type": "object",
+      "required": [
+        "flat",
+        "format"
+      ]
+    },
     "TLSConfig": {
       "properties": {
         "ca": {

tests/configuration/configuration.yaml

@@ -1,4 +1,4 @@
-# yaml-language-server: $schema=https://raw.githubusercontent.com/hasura/ndc-prometheus/main/jsonschema/configuration.json
+# yaml-language-server: $schema=../../jsonschema/configuration.json
 connection_settings:
   url:
     env: CONNECTION_URL
@@ -8,6 +8,11 @@ connection_settings:
         env: PROMETHEUS_USERNAME
       password:
         env: PROMETHEUS_PASSWORD
+runtime:
+  flat: true
+  format:
+    timestamp: rfc3339
+    value: float64
 generator:
   metrics:
     enabled: true