Merged in feature/common-metrics (pull request #73)

Add common metrics

* add common metrics

and sample for using

* updates

* attempt integration

* wip

* working

stubs generate

* docs

* comments

* Merge branch 'main' of bitbucket.org:aarete/query-orchestration into feature/common-metrics

* runner metrics all

* exclude generated

from the coverage

* fix lint

* refactor metrics

* change .gen filename

* new gen

* Merge branch 'main' of bitbucket.org:aarete/query-orchestration into feature/common-metrics

* docs

* main merge

* refactor

* add to generate

* merge main
This commit is contained in:
Jay Brown
2025-03-05 16:59:01 +00:00
parent 0af049e96f
commit ab28cf30db
13 changed files with 865 additions and 84 deletions
+16
View File
@@ -37,5 +37,21 @@ func main() {
os.Exit(1)
}
// optional add shutdown metrics
defer func() {
err = cfg.GetMetrics().RecordLifecycleEvent(1.0, "shutdown", "main")
if err != nil {
slog.Error("Failed to record shutdown metric ", "error", err)
}
}()
// start and block until server is done
server.Listen(ctx)
// optional shutdown
metrics := cfg.GetMetrics()
if err := metrics.Shutdown(ctx); err != nil {
slog.Error("Error shutting down metrics server", "error", err)
}
}
+110 -77
View File
@@ -27,99 +27,73 @@ import (
)
func main() {
// Create a context that we can cancel
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
// Define metrics configuration
// This example shows all supported metric types and their configurations
standardDefinitions := prometheus.NewStandardMetricDefinitions()
// Define metrics configuration using StandardMetricDefinitions
config := prometheus.MetricsConfig{
Port: 8080,
Namespace: "myapp",
Subsystem: "api",
// When ServeHttp is true, the package will start its own HTTP server
// If you want to use your own HTTP server (e.g., with Echo middleware),
// set this to false and configure your server to handle the /metrics endpoint
ServeHttp: true,
Definitions: []prometheus.MetricDefinition{
{
// Counter: Only increases, never decreases
Type: prometheus.Counter,
Name: "api_requests_total",
Help: "Total number of API requests",
Labels: []string{"method", "endpoint", "status"},
},
{
// Gauge: Can increase and decrease
Type: prometheus.Gauge,
Name: "active_users",
Help: "Number of currently active users",
Labels: []string{"user_type"},
},
{
// Histogram: Tracks distribution of values
Type: prometheus.Histogram,
Name: "request_duration_seconds",
Help: "Request duration distribution",
Labels: []string{"endpoint"},
Buckets: []float64{0.1, 0.5, 1, 2, 5}, // Define your bucket boundaries
},
{
// Summary: Calculates quantiles over a sliding time window
Type: prometheus.Summary,
Name: "response_size_bytes",
Help: "Response size distribution",
Labels: []string{"endpoint"},
Objectives: map[float64]float64{
0.5: 0.05, // 50th percentile with 5% error
0.9: 0.01, // 90th percentile with 1% error
0.99: 0.001, // 99th percentile with 0.1% error
// Start with standard metrics and append our custom one
Definitions: append(standardDefinitions,
prometheus.MetricDefinition{
Type: prometheus.Counter,
Name: "test_custom_events_total",
Help: "Total number of custom test events",
Labels: []string{
"event_type",
"priority",
},
},
},
),
}
// Create new metrics instance with context
metrics, err := prometheus.New(ctx, config)
if err != nil {
log.Fatalf("Failed to create metrics: %v", err)
}
// Start a goroutine to simulate metric recording
go simulateMetrics(ctx, metrics)
// Set up graceful shutdown
sigChan := make(chan os.Signal, 1)
signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
fmt.Println("Running... Press Ctrl+C to exit")
// Wait for interrupt signal
<-sigChan
fmt.Println("\nShutting down...")
// Cancel context to stop metric simulation
cancel()
// Create a timeout context for shutdown
shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second)
defer shutdownCancel()
// Gracefully shutdown the metrics server
if err := metrics.Shutdown(shutdownCtx); err != nil {
log.Printf("Error shutting down metrics server: %v", err)
}
}
// simulateMetrics demonstrates how to record different types of metrics
// in a concurrent environment. It takes a context for cancellation and
// the metrics instance.
func simulateMetrics(ctx context.Context, metrics *prometheus.Metrics) {
// Sample data for simulation
endpoints := []string{"/api/users", "/api/products", "/api/orders"}
methods := []string{"GET", "POST", "PUT", "DELETE"}
statuses := []string{"200", "400", "500"}
userTypes := []string{"free", "premium", "enterprise"}
errorTypes := []string{"none", "validation", "database", "timeout"}
dependencies := []string{"database", "cache", "auth-service", "email-service"}
cacheOperations := []string{"hit", "miss", "set", "eviction"}
taskTypes := []string{"cleanup", "indexing", "backup", "sync"}
taskStatuses := []string{"running", "queued", "failed"}
results := []string{"success", "failure", "timeout"}
components := []string{"api", "database", "cache", "auth"}
queueNames := []string{"main-queue", "background-queue", "notification-queue"}
resourceTypes := []string{"cpu", "memory", "disk", "network"}
// Custom metric sample data
eventTypes := []string{"scheduled", "triggered", "manual"}
priorities := []string{"low", "medium", "high"}
ticker := time.NewTicker(2 * time.Second)
defer ticker.Stop()
@@ -129,46 +103,105 @@ func simulateMetrics(ctx context.Context, metrics *prometheus.Metrics) {
case <-ctx.Done():
return
case <-ticker.C:
// Simulate request counter (Counter type)
// Simulate standard metrics
// 1. Requests total
endpoint := endpoints[rand.Intn(len(endpoints))]
method := methods[rand.Intn(len(methods))]
status := statuses[rand.Intn(len(statuses))]
errorType := errorTypes[rand.Intn(len(errorTypes))]
err := metrics.RecordMetric("api_requests_total", 1, map[string]string{
"method": method,
"endpoint": endpoint,
"status": status,
})
err := metrics.RecordRequestsTotal(1.0, endpoint, method, endpoint, status, errorType)
if err != nil {
log.Printf("Error recording request metric: %v", err)
log.Printf("Error recording requests metric: %v", err)
}
// Simulate active users (Gauge type)
userType := userTypes[rand.Intn(len(userTypes))]
activeUsers := rand.Float64() * 100
err = metrics.RecordMetric("active_users", activeUsers, map[string]string{
"user_type": userType,
})
if err != nil {
log.Printf("Error recording active users metric: %v", err)
}
// Simulate request duration (Histogram type)
// 2. Request duration
duration := rand.Float64() * 3 // Random duration between 0 and 3 seconds
err = metrics.RecordMetric("request_duration_seconds", duration, map[string]string{
"endpoint": endpoint,
})
err = metrics.RecordRequestDurationSeconds(duration, endpoint, method, endpoint, status)
if err != nil {
log.Printf("Error recording duration metric: %v", err)
}
// Simulate response size (Summary type)
responseSize := rand.Float64() * 5000 // Random size between 0 and 5000 bytes
err = metrics.RecordMetric("response_size_bytes", responseSize, map[string]string{
"endpoint": endpoint,
// 3. Active operations
activeOps := rand.Float64() * 50
err = metrics.RecordActiveOperations(activeOps, method)
if err != nil {
log.Printf("Error recording active operations metric: %v", err)
}
// 4. Resource usage
resourceType := resourceTypes[rand.Intn(len(resourceTypes))]
err = metrics.RecordResourceUsage(rand.Float64()*100, resourceType, "main-instance")
if err != nil {
log.Printf("Error recording resource usage metric: %v", err)
}
// 5. Dependency health
dep := dependencies[rand.Intn(len(dependencies))]
health := rand.Float64() // Random health score between 0 and 1
err = metrics.RecordDependencyHealth(health, dep, "external_service")
if err != nil {
log.Printf("Error recording dependency health metric: %v", err)
}
// 6. Cache operations
cacheOp := cacheOperations[rand.Intn(len(cacheOperations))]
result := results[rand.Intn(len(results))]
err = metrics.RecordCacheOperationsTotal(1.0, cacheOp, "main-cache", result)
if err != nil {
log.Printf("Error recording cache operations metric: %v", err)
}
// 7. Cache operation duration
err = metrics.RecordCacheOperationDurationSeconds(rand.Float64()*0.1, cacheOp, result)
if err != nil {
log.Printf("Error recording cache duration metric: %v", err)
}
// 8. Background tasks
taskType := taskTypes[rand.Intn(len(taskTypes))]
taskStatus := taskStatuses[rand.Intn(len(taskStatuses))]
err = metrics.RecordBackgroundTasks(rand.Float64()*10, taskType, taskStatus)
if err != nil {
log.Printf("Error recording background tasks metric: %v", err)
}
// 9. Active connections
err = metrics.RecordActiveConnections(rand.Float64()*100, "http")
if err != nil {
log.Printf("Error recording active connections metric: %v", err)
}
// 10. Queue length
queueName := queueNames[rand.Intn(len(queueNames))]
err = metrics.RecordQueueLength(rand.Float64()*50, queueName)
if err != nil {
log.Printf("Error recording queue length metric: %v", err)
}
// 11. Errors total
component := components[rand.Intn(len(components))]
err = metrics.RecordErrorsTotal(1.0, component, errorType)
if err != nil {
log.Printf("Error recording errors total metric: %v", err)
}
// 12. Items in progress
err = metrics.RecordItemsInProgress(rand.Float64()*20, endpoint, component)
if err != nil {
log.Printf("Error recording items in progress metric: %v", err)
}
// Custom metric (still using RecordMetric since it's not part of the generated stubs)
eventType := eventTypes[rand.Intn(len(eventTypes))]
priority := priorities[rand.Intn(len(priorities))]
err = metrics.RecordMetric("test_custom_events_total", 1, map[string]string{
"event_type": eventType,
"priority": priority,
})
if err != nil {
log.Printf("Error recording response size metric: %v", err)
log.Printf("Error recording custom event metric: %v", err)
}
}
}
+1 -1
View File
@@ -148,7 +148,7 @@ require (
golang.org/x/crypto v0.32.0 // indirect
golang.org/x/net v0.34.0 // indirect
golang.org/x/sys v0.30.0 // indirect
golang.org/x/text v0.21.0 // indirect
golang.org/x/text v0.21.0
google.golang.org/genproto/googleapis/rpc v0.0.0-20250127172529-29210b9bc287 // indirect
gopkg.in/yaml.v3 v3.0.1
)
+7 -3
View File
@@ -3,11 +3,10 @@ package runner
import (
"context"
"errors"
"github.com/aws/aws-sdk-go-v2/service/sqs/types"
"queryorchestration/internal/server"
"queryorchestration/internal/serviceconfig/aws"
"github.com/aws/aws-sdk-go-v2/service/sqs/types"
"queryorchestration/internal/serviceconfig/observability"
)
type Controller interface {
@@ -17,6 +16,7 @@ type Controller interface {
type ListenerConfig interface {
server.Config
aws.ConfigProvider
observability.ConfigProvider
RegisterController() error
GetController() Controller
GetQueueURL() string
@@ -58,6 +58,10 @@ func New(ctx context.Context, cfg ListenerConfig) (*Server, error) {
return nil, err
}
if err := observability.InitializeMetrics(ctx, cfg); err != nil {
return nil, err
}
err = cfg.PingQueue(ctx)
if err != nil {
return nil, err
@@ -3,19 +3,35 @@ package observability
import (
"context"
"log/slog"
"queryorchestration/internal/serviceconfig/observability/prometheus"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
sdktrace "go.opentelemetry.io/otel/sdk/trace"
)
const (
SystemName = "doczy"
)
type ConfigProvider interface {
IsOtelEnabled() bool
SetOtel(context.Context) func() error
GetMetrics() *prometheus.Metrics
SetMetrics(metrics *prometheus.Metrics)
}
type ObsConfig struct {
EnableOtel bool `env:"ENABLE_OTEL" envDefault:"false"`
metrics *prometheus.Metrics
}
func (o *ObsConfig) GetMetrics() *prometheus.Metrics {
return o.metrics
}
func (o *ObsConfig) SetMetrics(metrics *prometheus.Metrics) {
o.metrics = metrics
}
func (o *ObsConfig) IsOtelEnabled() bool {
@@ -47,3 +63,34 @@ func (o *ObsConfig) SetOtel(ctx context.Context) func() error {
return nil
}
}
const (
SubSystem = "RunnerService"
)
func InitializeMetrics(ctx context.Context, cfg ConfigProvider) error {
standardDefinitions := prometheus.NewStandardMetricDefinitions()
config := prometheus.MetricsConfig{
Port: 8080,
Namespace: SystemName,
Subsystem: SubSystem,
ServeHttp: true,
Definitions: standardDefinitions,
}
metrics, err := prometheus.New(ctx, config)
if err != nil {
slog.Error("Failed to create metrics object", "error", err)
return err
}
slog.Info("Metrics server started on port 8080")
cfg.SetMetrics(metrics)
err = cfg.GetMetrics().RecordLifecycleEvent(1.0, "startup", "main")
if err != nil {
slog.Error("Failed to create metrics at startup", "error", err)
return err
}
return nil
}
@@ -0,0 +1,188 @@
//go:generate go run ./generator/main.go
package prometheus
// NewStandardMetricDefinitions provides a base set of metric definitions that can be used
// across all services. Services can use these directly and append their own
// specific metrics to this slice.
// These base metrics are just suggestions/guesses and are expected to evolve as we implement them.
func NewStandardMetricDefinitions() []MetricDefinition {
return []MetricDefinition{
{
Type: Counter,
Name: "requests_total",
Help: "Total number of requests processed",
Labels: []string{
"operation",
"method",
"path", // Added from Second
"status",
"error_type",
},
},
{
Type: Histogram,
Name: "request_duration_seconds",
Help: "Request duration distribution in seconds",
Labels: []string{
"operation",
"method",
"path", // Added from Second
"status", // Added from Second
},
Buckets: []float64{0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10}, // Merged buckets
},
{
Type: Gauge,
Name: "active_operations",
Help: "Number of currently active operations",
Labels: []string{
"operation_type",
},
},
{
Type: Gauge,
Name: "resource_usage",
Help: "Current resource usage metrics",
Labels: []string{
"resource_type",
"resource_name",
},
},
{
Type: Summary,
Name: "operation_size_bytes",
Help: "Size of operation results in bytes",
Labels: []string{
"operation",
"type",
},
Objectives: map[float64]float64{
0.5: 0.05,
0.9: 0.01,
0.99: 0.001,
},
},
{
Type: Gauge,
Name: "dependency_health",
Help: "Health status of service dependencies",
Labels: []string{
"dependency_name",
"dependency_type",
},
},
{
Type: Counter,
Name: "rate_limited_total",
Help: "Total number of rate limited operations",
Labels: []string{
"operation",
"limit_type",
},
},
{
Type: Counter,
Name: "cache_operations_total",
Help: "Total number of cache operations",
Labels: []string{
"operation",
"cache_name",
"result", // Added from Second
},
},
{
Type: Gauge,
Name: "background_tasks",
Help: "Status and count of background tasks",
Labels: []string{
"task_type",
"status",
},
},
{
Type: Counter,
Name: "business_operations_total",
Help: "Total number of business operations",
Labels: []string{
"operation",
"category",
"status",
"result", // Added from Second
},
},
// Added from Second
{
Type: Histogram,
Name: "operation_duration_seconds",
Help: "Duration of business operations in seconds",
Labels: []string{
"operation",
"result",
},
Buckets: []float64{0.001, 0.01, 0.1, 1, 10, 60},
},
{
Type: Gauge,
Name: "active_connections",
Help: "Current number of active connections/requests being processed",
Labels: []string{
"type",
},
},
{
Type: Gauge,
Name: "queue_length",
Help: "Current length of pending work items in queues",
Labels: []string{
"queue_name",
},
},
{
Type: Counter,
Name: "errors_total",
Help: "Total number of errors encountered",
Labels: []string{
"component",
"error_type",
},
},
{
Type: Counter,
Name: "retries_total",
Help: "Total number of retry attempts made",
Labels: []string{
"operation",
"component",
},
},
{
Type: Counter,
Name: "lifecycle_event",
Help: "major lifecycle events",
Labels: []string{
"operation",
"component",
},
},
{
Type: Histogram,
Name: "cache_operation_duration_seconds",
Help: "Duration of cache operations in seconds",
Labels: []string{
"operation",
"result",
},
Buckets: []float64{0.0001, 0.001, 0.005, 0.01, 0.1},
},
{
Type: Gauge,
Name: "items_in_progress",
Help: "Current number of items being processed",
Labels: []string{
"operation",
"component",
},
},
}
}
@@ -0,0 +1,72 @@
# Echo Custom Metrics integration
In March of 2025 we looked into integration the metrics that we get from the echo server prometheus
middleware into the custom metrics that we get from the framework for non echo servers. (as a sizing exercise)
Attempting to do a simple integration will fail because the metrics for each appear to be collected in different repositories.
The final solution for getting this to work will probably require the following:
- Obtain a shared registry for the prometheus metrics. (see example 1 below)
- Change InitializeMetrics to use the shared registry. (see example 3 below)
- Initialize echo with the custom registry. (see example 2 below)
- Debug the metrics to see if they are being collected correctly.
Notes:
Code like the examples below is only partially working. The custom metrics still fail to show up with the
prometheus metrics. So it may also be necessary to expose a /metrics endpoint
manually like we do in the non echo case to get past this. Either way further investigation is needed.
### Example 1 of getting a custom shared registry for the prometheus metrics.
```go
var (
singletonCustomRegistry *prometheus.Registry
onceRegistry sync.Once
)
func GetSingletonCustomRegistry(namespace string) *prometheus.Registry {
if namespace == "" {
namespace = "doczy"
}
onceRegistry.Do(func() {
singletonCustomRegistry = prometheus.NewRegistry()
singletonCustomRegistry.MustRegister(collectors.NewGoCollector())
singletonCustomRegistry.MustRegister(collectors.NewProcessCollector(collectors.ProcessCollectorOpts{
Namespace: namespace,
}))
})
return singletonCustomRegistry
}
```
### Example 2. Initialize echo with the custom registry
```go
e := echo.New()
registry := p2.GetSingletonCustomRegistry("")
//// Configure Echo Prometheus middleware with the custom registry
e.Use(echoprometheus.NewMiddlewareWithConfig(echoprometheus.MiddlewareConfig{
Registerer: registry,
}))
//
e.GET("/metrics", echo.WrapHandler(
promhttp.HandlerFor(registry, promhttp.HandlerOpts{}),
))
// end experimental
// Other middlewares
e.Use(getSlogCustomEchoMiddleware(cfg.GetLogger()))
e.Use(middleware.Recover())
e.Use(middleware.CORS())
```
### Example 3. Initialize metrics with the custom registry
```go
func InitializeMetrics(ctx context.Context, cfg ListenerConfig) error {
standardDefinitions := prometheus.NewStandardMetricDefinitions()
registry := prometheus.GetSingletonCustomRegistry("")
...
```
@@ -0,0 +1,153 @@
//ignore go:build ignore
package main
import (
"bytes"
"fmt"
"go/format"
"golang.org/x/text/cases"
"golang.org/x/text/language"
"log"
"os"
"queryorchestration/internal/serviceconfig/observability/prometheus"
"strings"
"text/template"
)
const outputFilename = "./metric_stubs.gen.go"
// funcTemplate is the template for generating each metric recording function
const funcTemplate = `
// Record{{.FuncName}} records a {{.Name}} metric
{{.Comment}}
func (m *Metrics) Record{{.FuncName}}({{.Params}}) error {
return m.RecordMetric("{{.Name}}", {{.DefaultValue}}, map[string]string{ {{.LabelMap}} })
}`
// templateData holds the data needed for the template
type templateData struct {
FuncName string
Name string
Params string
DefaultValue string
LabelMap string
Comment string
}
// reservedKeywords contains Go reserved keywords that can't be used as parameter names
var reservedKeywords = map[string]bool{
"type": true,
"range": true,
"map": true,
"func": true,
"return": true,
"goto": true,
"break": true,
}
// toCamelCase function
func toCamelCase(s string) string {
parts := strings.Split(s, "_")
casesTitle := cases.Title(language.English)
for i := range parts {
parts[i] = casesTitle.String(parts[i])
}
return strings.Join(parts, "")
}
// getSafeParamName returns a safe parameter name that avoids Go reserved keywords
func getSafeParamName(label string) string {
paramName := strings.Replace(label, "-", "", -1)
if reservedKeywords[paramName] {
return paramName + "Value"
}
return paramName
}
func generateFunctionData(metric prometheus.MetricDefinition) templateData {
// Generate function name
funcName := toCamelCase(metric.Name)
// Prepare parameters
var params []string
// Always add value parameter since RecordMetric always needs it
params = append(params, "value float64")
// Add label parameters
for _, label := range metric.Labels {
paramName := getSafeParamName(label)
params = append(params, fmt.Sprintf("%s string", paramName))
}
// Generate label map
var labelPairs []string
for _, label := range metric.Labels {
paramName := getSafeParamName(label)
labelPairs = append(labelPairs, fmt.Sprintf(`"%s": %s`, label, paramName))
}
// Generate comment with parameter documentation
var commentLines []string
if metric.Help != "" {
commentLines = append(commentLines, fmt.Sprintf("// %s", metric.Help))
}
commentLines = append(commentLines, "//")
commentLines = append(commentLines, "// Parameters:")
commentLines = append(commentLines, "// value: The value to record")
for _, label := range metric.Labels {
paramName := getSafeParamName(label)
commentLines = append(commentLines, fmt.Sprintf("// %s: The %s label value", paramName, label))
}
return templateData{
FuncName: funcName,
Name: metric.Name,
Params: strings.Join(params, ", "),
DefaultValue: "value", // Always use the value parameter
LabelMap: strings.Join(labelPairs, ", "),
Comment: strings.Join(commentLines, "\n"),
}
}
func main() {
// Get metric definitions from the package above
metrics := prometheus.NewStandardMetricDefinitions()
// Create the template
tmpl, err := template.New("function").Parse(funcTemplate)
if err != nil {
log.Fatalf("Error creating template: %v", err)
}
// Generate the output file
var buf bytes.Buffer
// Write package declaration and imports
buf.WriteString("// Code generated by metric-generator; DO NOT EDIT.\n\n")
buf.WriteString("package prometheus\n\n")
// Generate functions for each metric
for _, metric := range metrics {
data := generateFunctionData(metric)
if err := tmpl.Execute(&buf, data); err != nil {
log.Fatalf("Error executing template for %s: %v", metric.Name, err)
}
buf.WriteString("\n\n") // Add spacing between functions
}
// Format the generated code
formattedCode, err := format.Source(buf.Bytes())
if err != nil {
log.Fatalf("Error formatting code: %v", err)
}
// Write to file
if err := os.WriteFile(outputFilename, formattedCode, 0644); err != nil {
log.Fatalf("Error writing output file: %v", err)
}
fmt.Println("Successfully generated " + outputFilename)
}
@@ -0,0 +1,7 @@
# Generator for metrics stubs
This main.go is called from `go generate ./...` and will generate the
stubs located inj the `generated_metrics_stubs.go` file in the prometheus package.
These stubs should be use exclusively by any code that is pushing metrics.
TODO: We may even want to make the underlying raw calls private. (to discuss)
@@ -0,0 +1,224 @@
// Code generated by metric-generator; DO NOT EDIT.
package prometheus
// RecordRequestsTotal records a requests_total metric
// Total number of requests processed
//
// Parameters:
//
// value: The value to record
// operation: The operation label value
// method: The method label value
// path: The path label value
// status: The status label value
// error_type: The error_type label value
func (m *Metrics) RecordRequestsTotal(value float64, operation string, method string, path string, status string, error_type string) error {
return m.RecordMetric("requests_total", value, map[string]string{"operation": operation, "method": method, "path": path, "status": status, "error_type": error_type})
}
// RecordRequestDurationSeconds records a request_duration_seconds metric
// Request duration distribution in seconds
//
// Parameters:
//
// value: The value to record
// operation: The operation label value
// method: The method label value
// path: The path label value
// status: The status label value
func (m *Metrics) RecordRequestDurationSeconds(value float64, operation string, method string, path string, status string) error {
return m.RecordMetric("request_duration_seconds", value, map[string]string{"operation": operation, "method": method, "path": path, "status": status})
}
// RecordActiveOperations records a active_operations metric
// Number of currently active operations
//
// Parameters:
//
// value: The value to record
// operation_type: The operation_type label value
func (m *Metrics) RecordActiveOperations(value float64, operation_type string) error {
return m.RecordMetric("active_operations", value, map[string]string{"operation_type": operation_type})
}
// RecordResourceUsage records a resource_usage metric
// Current resource usage metrics
//
// Parameters:
//
// value: The value to record
// resource_type: The resource_type label value
// resource_name: The resource_name label value
func (m *Metrics) RecordResourceUsage(value float64, resource_type string, resource_name string) error {
return m.RecordMetric("resource_usage", value, map[string]string{"resource_type": resource_type, "resource_name": resource_name})
}
// RecordOperationSizeBytes records a operation_size_bytes metric
// Size of operation results in bytes
//
// Parameters:
//
// value: The value to record
// operation: The operation label value
// typeValue: The type label value
func (m *Metrics) RecordOperationSizeBytes(value float64, operation string, typeValue string) error {
return m.RecordMetric("operation_size_bytes", value, map[string]string{"operation": operation, "type": typeValue})
}
// RecordDependencyHealth records a dependency_health metric
// Health status of service dependencies
//
// Parameters:
//
// value: The value to record
// dependency_name: The dependency_name label value
// dependency_type: The dependency_type label value
func (m *Metrics) RecordDependencyHealth(value float64, dependency_name string, dependency_type string) error {
return m.RecordMetric("dependency_health", value, map[string]string{"dependency_name": dependency_name, "dependency_type": dependency_type})
}
// RecordRateLimitedTotal records a rate_limited_total metric
// Total number of rate limited operations
//
// Parameters:
//
// value: The value to record
// operation: The operation label value
// limit_type: The limit_type label value
func (m *Metrics) RecordRateLimitedTotal(value float64, operation string, limit_type string) error {
return m.RecordMetric("rate_limited_total", value, map[string]string{"operation": operation, "limit_type": limit_type})
}
// RecordCacheOperationsTotal records a cache_operations_total metric
// Total number of cache operations
//
// Parameters:
//
// value: The value to record
// operation: The operation label value
// cache_name: The cache_name label value
// result: The result label value
func (m *Metrics) RecordCacheOperationsTotal(value float64, operation string, cache_name string, result string) error {
return m.RecordMetric("cache_operations_total", value, map[string]string{"operation": operation, "cache_name": cache_name, "result": result})
}
// RecordBackgroundTasks records a background_tasks metric
// Status and count of background tasks
//
// Parameters:
//
// value: The value to record
// task_type: The task_type label value
// status: The status label value
func (m *Metrics) RecordBackgroundTasks(value float64, task_type string, status string) error {
return m.RecordMetric("background_tasks", value, map[string]string{"task_type": task_type, "status": status})
}
// RecordBusinessOperationsTotal records a business_operations_total metric
// Total number of business operations
//
// Parameters:
//
// value: The value to record
// operation: The operation label value
// category: The category label value
// status: The status label value
// result: The result label value
func (m *Metrics) RecordBusinessOperationsTotal(value float64, operation string, category string, status string, result string) error {
return m.RecordMetric("business_operations_total", value, map[string]string{"operation": operation, "category": category, "status": status, "result": result})
}
// RecordOperationDurationSeconds records a operation_duration_seconds metric
// Duration of business operations in seconds
//
// Parameters:
//
// value: The value to record
// operation: The operation label value
// result: The result label value
func (m *Metrics) RecordOperationDurationSeconds(value float64, operation string, result string) error {
return m.RecordMetric("operation_duration_seconds", value, map[string]string{"operation": operation, "result": result})
}
// RecordActiveConnections records a active_connections metric
// Current number of active connections/requests being processed
//
// Parameters:
//
// value: The value to record
// typeValue: The type label value
func (m *Metrics) RecordActiveConnections(value float64, typeValue string) error {
return m.RecordMetric("active_connections", value, map[string]string{"type": typeValue})
}
// RecordQueueLength records a queue_length metric
// Current length of pending work items in queues
//
// Parameters:
//
// value: The value to record
// queue_name: The queue_name label value
func (m *Metrics) RecordQueueLength(value float64, queue_name string) error {
return m.RecordMetric("queue_length", value, map[string]string{"queue_name": queue_name})
}
// RecordErrorsTotal records a errors_total metric
// Total number of errors encountered
//
// Parameters:
//
// value: The value to record
// component: The component label value
// error_type: The error_type label value
func (m *Metrics) RecordErrorsTotal(value float64, component string, error_type string) error {
return m.RecordMetric("errors_total", value, map[string]string{"component": component, "error_type": error_type})
}
// RecordRetriesTotal records a retries_total metric
// Total number of retry attempts made
//
// Parameters:
//
// value: The value to record
// operation: The operation label value
// component: The component label value
func (m *Metrics) RecordRetriesTotal(value float64, operation string, component string) error {
return m.RecordMetric("retries_total", value, map[string]string{"operation": operation, "component": component})
}
// RecordLifecycleEvent records a lifecycle_event metric
// major lifecycle events
//
// Parameters:
//
// value: The value to record
// operation: The operation label value
// component: The component label value
func (m *Metrics) RecordLifecycleEvent(value float64, operation string, component string) error {
return m.RecordMetric("lifecycle_event", value, map[string]string{"operation": operation, "component": component})
}
// RecordCacheOperationDurationSeconds records a cache_operation_duration_seconds metric
// Duration of cache operations in seconds
//
// Parameters:
//
// value: The value to record
// operation: The operation label value
// result: The result label value
func (m *Metrics) RecordCacheOperationDurationSeconds(value float64, operation string, result string) error {
return m.RecordMetric("cache_operation_duration_seconds", value, map[string]string{"operation": operation, "result": result})
}
// RecordItemsInProgress records a items_in_progress metric
// Current number of items being processed
//
// Parameters:
//
// value: The value to record
// operation: The operation label value
// component: The component label value
func (m *Metrics) RecordItemsInProgress(value float64, operation string, component string) error {
return m.RecordMetric("items_in_progress", value, map[string]string{"operation": operation, "component": component})
}
@@ -16,14 +16,29 @@ Prometheus is a simplified wrapper around the Prometheus client library for Go,
- Context-based graceful shutdown
- Clean server lifecycle management
## Generating client stubs for the metrics schema
The metrics schema for the package is defined in `common.go`.
To generate the client stubs for the metrics schema, run the following command:
```
go generate ./...
```
This will place the generated stubs in the prometheus package as
`generated_metrics_stubs.go`.
## Example cmd (metricsExample_test)
See `cmd/metricsExample_test` for a sample usage in a working main.go.
The Readme.md in that directory has more details on how to use the harness.
## Running tests with prometheus server
See [prometheus.md](./prometheus.md) for prometheus related instructions and testing instructions.
## Quick Start (with your own http server)
This example (and the `metricsExample_test`) shows how to use the harness to create a simple http server that exposes metrics.
If you are using this with an API server you can indicate in your config that you already have a http server and it will not start a http server for you.
(e.g. Use the one that we automatically setup as Echo middleware)
@@ -75,11 +90,11 @@ func main() {
<-sigChan
// Cancel context
cancel()
// Give the server up to 5 seconds to shutdown gracefully
shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second)
defer shutdownCancel()
if err := metrics.Shutdown(shutdownCtx); err != nil {
log.Printf("Error shutting down metrics server: %v", err)
}
@@ -126,6 +141,7 @@ type MetricDefinition struct {
## Metric Types
### Counter
Used for cumulative metrics that only increase in value.
```go
@@ -138,6 +154,7 @@ MetricDefinition{
```
### Gauge
Used for metrics that can increase and decrease.
```go
@@ -150,6 +167,7 @@ MetricDefinition{
```
### Histogram
Used for measurements like request duration or response size.
```go
@@ -163,6 +181,7 @@ MetricDefinition{
```
### Summary
Similar to histogram but calculates quantiles over time.
```go
@@ -187,6 +206,7 @@ err := metrics.RecordMetric("metric_name", value, map[string]string{
```
The method is thread-safe and will return an error if:
- The metric doesn't exist
- The label values don't match the metric's label names
- The metric type is invalid
@@ -200,6 +220,7 @@ When `ServeHttp` is set to `true` in the configuration, the package automaticall
The library provides two ways to manage the lifecycle of the metrics server:
1. Context-based shutdown:
```go
ctx, cancel := context.WithCancel(context.Background())
metrics, err := prometheus.New(ctx, config)
@@ -208,6 +229,7 @@ cancel()
```
2. Manual shutdown:
```go
shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second)
defer shutdownCancel()
@@ -223,6 +245,7 @@ All operations in the package are thread-safe, allowing for concurrent metric re
## Error Handling
The package returns errors in the following cases:
- Invalid metric type specified in configuration
- Failed metric registration (e.g., duplicate metrics)
- Recording to non-existent metric
@@ -232,6 +255,7 @@ The package returns errors in the following cases:
## Best Practices
1. Always provide a context when creating new metrics:
```go
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
@@ -239,6 +263,7 @@ metrics, err := prometheus.New(ctx, config)
```
2. Implement graceful shutdown in your application:
```go
shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second)
defer shutdownCancel()
@@ -248,6 +273,7 @@ if err := metrics.Shutdown(shutdownCtx); err != nil {
```
3. Handle server errors appropriately:
```go
if err := metrics.Shutdown(ctx); err != nil {
// Log the error but don't panic - the application can continue
@@ -256,9 +282,12 @@ if err := metrics.Shutdown(ctx); err != nil {
```
4. Use appropriate timeouts for shutdown:
```go
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
metrics.Shutdown(ctx)
```
## Research into echo integration with custom metrics
See [echocustommetrics.md](./echocustommetrics.md) for details on how to integrate custom metrics with echo. (for future use)
+4
View File
@@ -11,6 +11,7 @@ tasks:
deps:
- generate:server
- generate:client
- generate:metrics:stubs
generate:client:
cmds:
- |
@@ -47,3 +48,6 @@ tasks:
$file
fi
done
generate:metrics:stubs:
cmds:
- go generate ./...
+5 -1
View File
@@ -9,6 +9,8 @@ vars:
API: "./api/..."
PKG: "./pkg/..."
CMD: "./cmd/..."
# yamllint disable-line rule:line-length
EXCLUDED_FILES: ".gen.go|internal/serviceconfig/observability/prometheus/generator/main.go"
tasks:
mocks:generate:
@@ -46,7 +48,9 @@ tasks:
TMP_FILE: "{{.OUT_DIR}}/coverage.tmp"
FILTER_COVERAGE_FILE: "{{.OUT_DIR}}/filtered_coverage.out"
cmds:
- grep -v ".gen.go" {{.COVERAGE_FILE}} > {{.FILTER_COVERAGE_FILE}}
- |
grep -v -E "{{.EXCLUDED_FILES}}" {{.COVERAGE_FILE}} \
> {{.FILTER_COVERAGE_FILE}}
- go tool cover -func={{.FILTER_COVERAGE_FILE}} > {{.TMP_FILE}}
- |
COVERAGE=$(grep total: {{.TMP_FILE}} \