chore: improve

Author: StarpTech

router/demo.config.yaml

@@ -5,6 +5,24 @@
 
 version: "1"
 
+graphql_metrics:
+  enabled: false
+
+traffic_shaping:
+  all:
+    retry:
+      enabled: false
+
+telemetry:
+  metrics:
+    otlp:
+      enabled: false
+    prometheus:
+      schema_usage:
+        enabled: true
+        exporter:
+          interval: 3s
+
 events:
   providers:
     nats:
@@ -19,4 +37,4 @@ events:
     redis:
       - id: my-redis
         urls:
-        - "redis://localhost:6379/2"
+          - "redis://localhost:6379/2"

router/internal/graphqlmetrics/exporter.go

@@ -228,11 +228,13 @@ func (e *Exporter[T]) exportBatch(batch []T) error {
 }
 
 // prepareAndSendBatch starts a goroutine to export the batch with retry logic.
+// The goroutine takes ownership of the batch slice and will return it to the pool when done.
 func (e *Exporter[T]) prepareAndSendBatch(batch []T) {
 	e.logger.Debug("Preparing to send batch", zap.Int("batch_size", len(batch)))
 	e.inflightBatches.Inc()
 	go func() {
 		defer e.inflightBatches.Dec()
+		defer e.putBatchBuffer(batch) // Return buffer to pool after export completes
 		e.exportBatchWithRetry(batch)
 	}()
 }
@@ -328,17 +330,15 @@ func (e *Exporter[T]) start() {
 			e.logger.Debug("Tick: flushing buffer", zap.Int("buffer_size", len(buffer)))
 			if len(buffer) > 0 {
 				e.prepareAndSendBatch(buffer)
-				// Return buffer to pool after sending
-				e.putBatchBuffer(buffer)
+				// Ownership transferred to goroutine, get a new buffer
 				buffer = nil
 			}
 		case item := <-e.queue:
 			buffer = append(buffer, item)
 			if len(buffer) == e.settings.BatchSize {
 				e.logger.Debug("Buffer full, sending batch", zap.Int("batch_size", len(buffer)))
 				e.prepareAndSendBatch(buffer)
-				// Return buffer to pool after sending
-				e.putBatchBuffer(buffer)
+				// Ownership transferred to goroutine, get a new buffer
 				buffer = nil
 			}
 		case <-e.shutdownSignal:
@@ -360,15 +360,13 @@ func (e *Exporter[T]) drainQueue(buffer []T) {
 			buffer = append(buffer, item)
 			if len(buffer) == e.settings.BatchSize {
 				e.prepareAndSendBatch(buffer)
-				// Return buffer to pool and get a new one
-				e.putBatchBuffer(buffer)
+				// Ownership transferred to goroutine, get a new buffer
 				buffer = e.getBatchBuffer()
 			}
 		default:
 			if len(buffer) > 0 {
 				e.prepareAndSendBatch(buffer)
-				// Return buffer to pool before returning
-				e.putBatchBuffer(buffer)
+				// Ownership transferred to goroutine
 			}
 			e.logger.Debug("Queue drained", zap.Int("drained_items", drainedItems))
 			return

router/internal/graphqlmetrics/exporter_test.go

@@ -175,6 +175,10 @@ func TestExportBatchesWithUniqueSchemaUsages(t *testing.T) {
 			SchemaInfo: &graphqlmetricsv1.SchemaInfo{
 				Version: "1",
 			},
+			RequestInfo: &graphqlmetricsv1.RequestInfo{
+				Error:      false,
+				StatusCode: http.StatusOK,
+			},
 			Attributes: map[string]string{},
 		}
 
@@ -246,6 +250,10 @@ func TestForceFlushSync(t *testing.T) {
 			SchemaInfo: &graphqlmetricsv1.SchemaInfo{
 				Version: "1",
 			},
+			RequestInfo: &graphqlmetricsv1.RequestInfo{
+				Error:      false,
+				StatusCode: http.StatusOK,
+			},
 			Attributes: map[string]string{},
 		}
 
@@ -291,6 +299,10 @@ func TestForceFlushSync(t *testing.T) {
 			SchemaInfo: &graphqlmetricsv1.SchemaInfo{
 				Version: "1",
 			},
+			RequestInfo: &graphqlmetricsv1.RequestInfo{
+				Error:      false,
+				StatusCode: http.StatusOK,
+			},
 			Attributes: map[string]string{},
 		}
 
@@ -360,6 +372,10 @@ func TestExportBatchInterval(t *testing.T) {
 			SchemaInfo: &graphqlmetricsv1.SchemaInfo{
 				Version: "1",
 			},
+			RequestInfo: &graphqlmetricsv1.RequestInfo{
+				Error:      false,
+				StatusCode: http.StatusOK,
+			},
 			Attributes: map[string]string{},
 		}
 

router/internal/graphqlmetrics/prometheus_sink.go

@@ -82,7 +82,7 @@ func (s *PrometheusSink) Export(ctx context.Context, batch []*graphqlmetrics.Sch
 			ctx,
 			int64(count),
 			[]attribute.KeyValue{},
-			otelmetric.WithAttributeSet(attribute.NewSet(allAttrs...)),
+			otelmetric.WithAttributes(allAttrs...),
 		)
 	}
 

router/pkg/config/config.go

@@ -121,7 +121,7 @@ type PrometheusSchemaFieldUsage struct {
 type PrometheusSchemaFieldUsageExporter struct {
 	BatchSize     int           `yaml:"batch_size" envDefault:"4096" env:"BATCH_SIZE"`
 	QueueSize     int           `yaml:"queue_size" envDefault:"12800" env:"QUEUE_SIZE"`
-	Interval      time.Duration `yaml:"interval" envDefault:"1s" env:"INTERVAL"`
+	Interval      time.Duration `yaml:"interval" envDefault:"2s" env:"INTERVAL"`
 	ExportTimeout time.Duration `yaml:"export_timeout" envDefault:"10s" env:"EXPORT_TIMEOUT"`
 }
 

router/pkg/config/config.schema.json

@@ -1264,7 +1264,7 @@
                         },
                         "interval": {
                           "type": "string",
-                          "default": "1s",
+                          "default": "2s",
                           "duration": {
                             "minimum": "100ms"
                           },

router/pkg/config/config_test.go

@@ -702,7 +702,7 @@ telemetry:
 		exporter := c.Config.Telemetry.Metrics.Prometheus.SchemaFieldUsage.Exporter
 		require.Equal(t, 4096, exporter.BatchSize)
 		require.Equal(t, 12800, exporter.QueueSize)
-		require.Equal(t, 10*time.Second, exporter.Interval)
+		require.Equal(t, 2*time.Second, exporter.Interval)
 		require.Equal(t, 10*time.Second, exporter.ExportTimeout)
 	})
 

router/pkg/config/testdata/config_defaults.json

@@ -66,7 +66,7 @@
           "Exporter": {
             "BatchSize": 4096,
             "QueueSize": 12800,
-            "Interval": 1000000000,
+            "Interval": 2000000000,
             "ExportTimeout": 10000000000
           }
         }

router/pkg/config/testdata/config_full.json

@@ -96,7 +96,7 @@
           "Exporter": {
             "BatchSize": 4096,
             "QueueSize": 12800,
-            "Interval": 1000000000,
+            "Interval": 2000000000,
             "ExportTimeout": 10000000000
           }
         }

router/pkg/graphqlschemausage/schemausage.go

@@ -27,6 +27,7 @@ type TypeFieldMetrics []*TypeFieldUsageInfo
 
 // IntoGraphQLMetrics converts the TypeFieldMetrics into a []*graphqlmetrics.TypeFieldUsageInfo
 func (t TypeFieldMetrics) IntoGraphQLMetrics() []*graphqlmetrics.TypeFieldUsageInfo {
+	// Pre-allocate slice with exact capacity
 	metrics := make([]*graphqlmetrics.TypeFieldUsageInfo, len(t))
 	for i, info := range t {
 		metrics[i] = info.IntoGraphQLMetrics()