diff --git a/cmd/config/config_env.go b/cmd/config/config_env.go index 4b0384fa..cb905cd4 100644 --- a/cmd/config/config_env.go +++ b/cmd/config/config_env.go @@ -171,11 +171,20 @@ func envToOtelConfig() (*otel.Config, error) { cfg := &otel.Config{} metricsEndpoint := viper.GetString("PGSTREAM_METRICS_ENDPOINT") - if metricsEndpoint != "" { - cfg.Metrics = &otel.MetricsConfig{ - Endpoint: metricsEndpoint, - CollectionInterval: viper.GetDuration("PGSTREAM_METRICS_COLLECTION_INTERVAL"), + prometheusEnabled := viper.GetBool("PGSTREAM_METRICS_PROMETHEUS_ENABLED") + if metricsEndpoint != "" || prometheusEnabled { + metricsCfg := &otel.MetricsConfig{} + if metricsEndpoint != "" { + metricsCfg.Endpoint = metricsEndpoint + metricsCfg.CollectionInterval = viper.GetDuration("PGSTREAM_METRICS_COLLECTION_INTERVAL") } + if prometheusEnabled { + metricsCfg.Prometheus = &otel.PrometheusConfig{ + Enabled: prometheusEnabled, + Endpoint: viper.GetString("PGSTREAM_METRICS_PROMETHEUS_ENDPOINT"), + } + } + cfg.Metrics = metricsCfg } tracesEndpoint := viper.GetString("PGSTREAM_TRACES_ENDPOINT") diff --git a/cmd/config/config_yaml.go b/cmd/config/config_yaml.go index 2373c93a..85eff41e 100644 --- a/cmd/config/config_yaml.go +++ b/cmd/config/config_yaml.go @@ -43,8 +43,14 @@ type HealthConfig struct { } type MetricsConfig struct { - Endpoint string `mapstructure:"endpoint" yaml:"endpoint"` - CollectionInterval int `mapstructure:"collection_interval" yaml:"collection_interval"` + Prometheus *PrometheusConfig `mapstructure:"prometheus" yaml:"prometheus"` + Endpoint string `mapstructure:"endpoint" yaml:"endpoint"` + CollectionInterval int `mapstructure:"collection_interval" yaml:"collection_interval"` +} + +type PrometheusConfig struct { + Enabled bool `mapstructure:"enabled" yaml:"enabled"` + Endpoint string `mapstructure:"endpoint" yaml:"endpoint"` } type TracesConfig struct { @@ -400,10 +406,18 @@ func (c *InstrumentationConfig) toHealthConfig() *health.Config { func (c *InstrumentationConfig) toOtelConfig() (*otel.Config, error) { cfg := &otel.Config{} if c.Metrics != nil { - cfg.Metrics = &otel.MetricsConfig{ + metricsCfg := &otel.MetricsConfig{ Endpoint: c.Metrics.Endpoint, CollectionInterval: time.Duration(c.Metrics.CollectionInterval) * time.Second, + Prometheus: nil, + } + if c.Metrics.Prometheus != nil { + metricsCfg.Prometheus = &otel.PrometheusConfig{ + Enabled: c.Metrics.Prometheus.Enabled, + Endpoint: c.Metrics.Prometheus.Endpoint, + } } + cfg.Metrics = metricsCfg } if c.Traces != nil { diff --git a/cmd/health_cmd.go b/cmd/health_cmd.go index 53f7a5a2..04afbb72 100644 --- a/cmd/health_cmd.go +++ b/cmd/health_cmd.go @@ -12,6 +12,7 @@ import ( "github.com/xataio/pgstream/internal/phase" pglib "github.com/xataio/pgstream/internal/postgres" loglib "github.com/xataio/pgstream/pkg/log" + "github.com/xataio/pgstream/pkg/otel" ) const ( @@ -40,6 +41,11 @@ func startHealthServer(ctx context.Context, logger loglib.Logger, sourcePostgres return func() {}, nil } + instrumentationCfg, err := config.ParseInstrumentationConfig() + if err != nil { + return nil, fmt.Errorf("parsing instrumentation config: %w", err) + } + opts := []health.Option{ health.WithLogger(logger), health.WithVersion(version()), @@ -60,7 +66,11 @@ func startHealthServer(ctx context.Context, logger loglib.Logger, sourcePostgres opts = append(opts, health.WithReadinessCheck(pool.Ping)) } - srv := health.NewServer(*cfg, opts...) + var metricsCfg *otel.MetricsConfig + if instrumentationCfg.Metrics != nil { + metricsCfg = instrumentationCfg.Metrics + } + srv := health.NewServer(*cfg, metricsCfg, opts...) if err := srv.Listen(); err != nil { if pool != nil { pool.Close(context.Background()) diff --git a/docs/Observability.md b/docs/Observability.md index f845145f..8027d1df 100644 --- a/docs/Observability.md +++ b/docs/Observability.md @@ -219,6 +219,32 @@ PGSTREAM_TRACES_ENDPOINT="http://localhost:4317" PGSTREAM_TRACES_SAMPLE_RATIO=0.5 ``` +### Prometheus Scraping (Pull-based) + +If you'd rather scrape metrics directly than deploy an OTel collector, enable the Prometheus exporter. It's served from the existing [health endpoint](configuration.md#instrumentation) server, so no separate port or process is needed: + +```yaml +instrumentation: + health: + enabled: true + address: "0.0.0.0:9910" # defaults to localhost:9910 + metrics: + prometheus: + enabled: true + endpoint: "/metrics" # defaults to /metrics +``` + +Or using environment variables: + +```sh +PGSTREAM_HEALTH_CHECK_ENABLED=true +PGSTREAM_HEALTH_CHECK_ADDRESS="0.0.0.0:9910" +PGSTREAM_METRICS_PROMETHEUS_ENABLED=true +PGSTREAM_METRICS_PROMETHEUS_ENDPOINT="/metrics" +``` + +This works standalone — `metrics.endpoint` is not required, so you get pgstream's metrics (including the [Go runtime metrics](#go-runtime-metrics)) without an OTLP collector in the loop. It can also be combined with `metrics.endpoint` to export to both a Prometheus scraper and an OTLP collector at the same time. + ## Monitoring Dashboards ### Pre-built pgstream Dashboard diff --git a/docs/configuration.md b/docs/configuration.md index 0320c200..625a7171 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -11,6 +11,9 @@ instrumentation: metrics: endpoint: "0.0.0.0:4317" collection_interval: 60 # collection interval for metrics in seconds. Defaults to 60s + prometheus: + enabled: true # exposes a Prometheus scrape endpoint on the health server. Defaults to false + endpoint: "/metrics" # path the Prometheus endpoint is served on. Defaults to /metrics traces: endpoint: "0.0.0.0:4317" sample_ratio: 0.5 # ratio of traces that will be sampled. Must be between 0.0-1.0, where 0 is no traces sampled, and 1 is all traces sampled. @@ -420,10 +423,12 @@ One of exponential/constant/disable retries retry policies can be provided for t
Metrics -| Environment Variable | Default | Required | Description | -| ------------------------------------ | ------- | -------- | ---------------------------------------------------------------------- | -| PGSTREAM_METRICS_ENDPOINT | N/A | No | Endpoint where the pgstream metrics will be exported to. | -| PGSTREAM_METRICS_COLLECTION_INTERVAL | 60s | No | Interval at which the pgstream metrics will be collected and exported. | +| Environment Variable | Default | Required | Description | +| --------------------------------------- | --------- | -------- | ------------------------------------------------------------------------------------------------ | +| PGSTREAM_METRICS_ENDPOINT | N/A | No | Endpoint where the pgstream metrics will be pushed to via OTLP. Not required to use Prometheus. | +| PGSTREAM_METRICS_COLLECTION_INTERVAL | 60s | No | Interval at which the pgstream metrics will be collected and exported via OTLP. | +| PGSTREAM_METRICS_PROMETHEUS_ENABLED | False | No | Exposes a Prometheus scrape endpoint on the health server. Requires the health endpoint enabled. | +| PGSTREAM_METRICS_PROMETHEUS_ENDPOINT | /metrics | No | Path the Prometheus endpoint is served on. |
@@ -442,6 +447,8 @@ One of exponential/constant/disable retries retry policies can be provided for t Exposes `/health` (liveness, always 200), `/ready` (readiness, pings the source postgres database when configured), and `/status` (current pipeline phase: `snapshot` or `replication`). Only the `run` and `snapshot` commands start the server. Responses are JSON. +When `instrumentation.metrics.prometheus.enabled` (`PGSTREAM_METRICS_PROMETHEUS_ENABLED`) is true, the server also exposes a Prometheus scrape endpoint (default path `/metrics`, configurable via `instrumentation.metrics.prometheus.endpoint` / `PGSTREAM_METRICS_PROMETHEUS_ENDPOINT`). The endpoint returns `404` when Prometheus is disabled. This lets you scrape pgstream's metrics directly, without deploying an OTel collector. + | Environment Variable | Default | Required | Description | | ------------------------------ | ---------------- | -------- | ------------------------------------------------------------------------------------------------------------ | | PGSTREAM_HEALTH_CHECK_ENABLED | False | No | Enable the health endpoint server. | diff --git a/go.mod b/go.mod index c59164dd..1d79c89e 100644 --- a/go.mod +++ b/go.mod @@ -26,6 +26,7 @@ require ( github.com/opensearch-project/opensearch-go v1.1.0 github.com/pgvector/pgvector-go v0.4.0 github.com/pgvector/pgvector-go/pgx v0.4.0 + github.com/prometheus/client_golang v1.23.2 github.com/pterm/pterm v0.12.83 github.com/rs/xid v1.6.0 github.com/rs/zerolog v1.35.1 @@ -47,6 +48,7 @@ require ( go.opentelemetry.io/otel v1.44.0 go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.44.0 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.44.0 + go.opentelemetry.io/otel/exporters/prometheus v0.66.0 go.opentelemetry.io/otel/metric v1.44.0 go.opentelemetry.io/otel/sdk v1.44.0 go.opentelemetry.io/otel/sdk/metric v1.44.0 @@ -71,6 +73,7 @@ require ( github.com/Masterminds/semver/v3 v3.3.1 // indirect github.com/Microsoft/go-winio v0.6.2 // indirect github.com/OneOfOne/xxhash v1.2.8 // indirect + github.com/beorn7/perks v1.0.1 // indirect github.com/bytedance/gopkg v0.1.3 // indirect github.com/bytedance/sonic/loader v0.5.1 // indirect github.com/cenkalti/backoff/v5 v5.0.3 // indirect @@ -132,6 +135,7 @@ require ( github.com/moby/sys/user v0.4.0 // indirect github.com/moby/sys/userns v0.1.0 // indirect github.com/moby/term v0.5.2 // indirect + github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/nexus-rpc/sdk-go v0.3.0 // indirect github.com/opencontainers/go-digest v1.0.0 // indirect github.com/opencontainers/image-spec v1.1.1 // indirect @@ -140,6 +144,10 @@ require ( github.com/pierrec/lz4/v4 v4.1.22 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect + github.com/prometheus/client_model v0.6.2 // indirect + github.com/prometheus/common v0.67.5 // indirect + github.com/prometheus/otlptranslator v1.0.0 // indirect + github.com/prometheus/procfs v0.20.1 // indirect github.com/redpanda-data/benthos/v4 v4.45.1 // indirect github.com/rivo/uniseg v0.4.7 // indirect github.com/robfig/cron v1.2.0 // indirect @@ -174,6 +182,7 @@ require ( go.opentelemetry.io/proto/otlp v1.10.0 // indirect go.temporal.io/api v1.45.0 // indirect go.temporal.io/sdk v1.33.0 // indirect + go.yaml.in/yaml/v2 v2.4.4 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect golang.org/x/arch v0.0.0-20210923205945-b76863e36670 // indirect golang.org/x/crypto v0.53.0 // indirect diff --git a/go.sum b/go.sum index c6daf804..affa072f 100644 --- a/go.sum +++ b/go.sum @@ -50,6 +50,8 @@ github.com/OneOfOne/xxhash v1.2.8/go.mod h1:eZbhyaAYD41SGSSsnmcpxVoRiQ/MPUTjUdII github.com/atomicgo/cursor v0.0.1/go.mod h1:cBON2QmmrysudxNBFthvMtN32r3jxVRIvzkUiF/RuIk= github.com/aws/aws-sdk-go v1.42.27/go.mod h1:OGr6lGMAKGlG9CVrYnWYDKIyb829c6EVBRjxqjmPepc= github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA= +github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= +github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= github.com/bytedance/gopkg v0.1.3 h1:TPBSwH8RsouGCBcMBktLt1AymVo2TVsBVCY4b6TnZ/M= github.com/bytedance/gopkg v0.1.3/go.mod h1:576VvJ+eJgyCzdjS+c4+77QF3p7ubbtiKARP3TxducM= github.com/bytedance/sonic v1.15.1 h1:nJD5PmM0vY7J8CT6MxoqbVAAMhkSmV2HgRAUrrpLoOw= @@ -165,8 +167,8 @@ github.com/gofrs/uuid/v5 v5.3.1 h1:aPx49MwJbekCzOyhZDjJVb0hx3A0KLjlbLx6p2gY0p0= github.com/gofrs/uuid/v5 v5.3.1/go.mod h1:CDOjlDMVAtN56jqyRUZh58JT31Tiw7/oQyEXZV+9bD8= github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q= github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= -github.com/golang-jwt/jwt/v5 v5.2.2 h1:Rl4B7itRWVtYIHFrSNd7vhTiz9UpLdi6gZhZ3wEeDy8= -github.com/golang-jwt/jwt/v5 v5.2.2/go.mod h1:pqrtFR0X4osieyHYxtmOUWsAWrfe1Q5UVIyoH402zdk= +github.com/golang-jwt/jwt/v5 v5.3.0 h1:pv4AsKCKKZuqlgs5sUmn4x8UlGa0kEVt/puTpKx9vvo= +github.com/golang-jwt/jwt/v5 v5.3.0/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE= github.com/golang-migrate/migrate/v4 v4.19.1 h1:OCyb44lFuQfYXYLx1SCxPZQGU7mcaZ7gH9yH4jSFbBA= github.com/golang-migrate/migrate/v4 v4.19.1/go.mod h1:CTcgfjxhaUtsLipnLoQRWCrjYXycRz/g5+RWDuYgPrE= github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q= @@ -271,6 +273,8 @@ github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= +github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= github.com/labstack/echo/v4 v4.15.4 h1:DL45vVYa+BWE+XuW+zZNd9H0YEdZ80UAWJGcTVW4EVs= github.com/labstack/echo/v4 v4.15.4/go.mod h1:CuMetKIRwsuO/qlAgMq+KTAalwGoB/h4tC+yPdrTj1g= github.com/labstack/gommon v0.5.0 h1:6VSQ2NOzsnEJ5W6+84E0RbcaDDmgB6NIAzWCczTEe6c= @@ -322,6 +326,8 @@ github.com/moby/term v0.5.2 h1:6qk3FJAFDs6i/q3W/pQ97SX192qKfZgGjCQqfCJkgzQ= github.com/moby/term v0.5.2/go.mod h1:d3djjFCrjnB+fl8NJux+EJzu0msscUP+f8it8hPkFLc= github.com/morikuni/aec v1.0.0 h1:nP9CBfwrvYnBRgY6qfDQkygYDmYwOilePFkwzv4dU8A= github.com/morikuni/aec v1.0.0/go.mod h1:BbKIizmSmc5MMPqRYbxO4ZU0S0+P200+tUnFx7PXmsc= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/nexus-rpc/sdk-go v0.3.0 h1:Y3B0kLYbMhd4C2u00kcYajvmOrfozEtTV/nHSnV57jA= github.com/nexus-rpc/sdk-go v0.3.0/go.mod h1:TpfkM2Cw0Rlk9drGkoiSMpFqflKTiQLWUNyKJjF8mKQ= github.com/nsf/jsondiff v0.0.0-20230430225905-43f6cf3098c1 h1:dOYG7LS/WK00RWZc8XGgcUTlTxpp3mKhdR2Q9z9HbXM= @@ -353,7 +359,17 @@ github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRI github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 h1:o4JXh1EVt9k/+g42oCprj/FisM4qX9L3sZB3upGN2ZU= github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE= +github.com/prometheus/client_golang v1.23.2 h1:Je96obch5RDVy3FDMndoUsjAhG5Edi49h0RJWRi/o0o= +github.com/prometheus/client_golang v1.23.2/go.mod h1:Tb1a6LWHB3/SPIzCoaDXI4I8UHKeFTEQ1YCr+0Gyqmg= github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA= +github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk= +github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE= +github.com/prometheus/common v0.67.5 h1:pIgK94WWlQt1WLwAC5j2ynLaBRDiinoAb86HZHTUGI4= +github.com/prometheus/common v0.67.5/go.mod h1:SjE/0MzDEEAyrdr5Gqc6G+sXI67maCxzaT3A2+HqjUw= +github.com/prometheus/otlptranslator v1.0.0 h1:s0LJW/iN9dkIH+EnhiD3BlkkP5QVIUVEoIwkU+A6qos= +github.com/prometheus/otlptranslator v1.0.0/go.mod h1:vRYWnXvI6aWGpsdY/mOT/cbeVRBlPWtBNDb7kGR3uKM= +github.com/prometheus/procfs v0.20.1 h1:XwbrGOIplXW/AU3YhIhLODXMJYyC1isLFfYCsTEycfc= +github.com/prometheus/procfs v0.20.1/go.mod h1:o9EMBZGRyvDrSPH1RqdxhojkuXstoe4UlK79eF5TGGo= github.com/pterm/pterm v0.12.27/go.mod h1:PhQ89w4i95rhgE+xedAoqous6K9X+r6aSOI2eFF7DZI= github.com/pterm/pterm v0.12.29/go.mod h1:WI3qxgvoQFFGKGjGnJR849gU0TsEOvKn5Q8LlY1U7lg= github.com/pterm/pterm v0.12.30/go.mod h1:MOqLIyMOgmTDz9yorcYbcw+HsgoZo3BQfg2wtl3HEFE= @@ -521,6 +537,8 @@ go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0 h1:4YsVu3B8+3qtWYYrsUY go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0/go.mod h1:+wnlSn0mD1ADVMe3v9Z/WIaiz6q6gL2J/ejaAmdmv80= go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.44.0 h1:qazEJlUOQzhCpzQpFETGby7EdqjI1wsd0W+6Gg1SCTU= go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.44.0/go.mod h1:fOD2Yefuxixkx3ahVNf0O/PERb6r4OlbxfATVnYvzCo= +go.opentelemetry.io/otel/exporters/prometheus v0.66.0 h1:vkrK8PAznv2NKt2r+kdu252ccGzkEqLc2aSXbQIALYQ= +go.opentelemetry.io/otel/exporters/prometheus v0.66.0/go.mod h1:V/UB6D3vMF/UBOL5igAsAYnk1nG/bzYYTzvsB16cy7o= go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc= go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo= go.opentelemetry.io/otel/metric/x v0.66.0 h1:YkCrx1zLOChi9ZcZ6euupOcsgzbVlec7D/xoEU1+cTA= @@ -543,6 +561,8 @@ go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU= go.uber.org/zap v1.18.1/go.mod h1:xg/QME4nWcxGxrpdeYfq7UvYrLh66cuVKdrbD1XF/NI= +go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ= +go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ= go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc= go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= golang.org/x/arch v0.0.0-20210923205945-b76863e36670 h1:18EFjUmQOcUvxNYSkA6jO9VAiXCnxFY6NyDX0bHDmkU= diff --git a/internal/health/health.go b/internal/health/health.go index 50e1921d..8e92e576 100644 --- a/internal/health/health.go +++ b/internal/health/health.go @@ -12,12 +12,15 @@ import ( "net/http" "time" + "github.com/prometheus/client_golang/prometheus/promhttp" "github.com/xataio/pgstream/internal/json" loglib "github.com/xataio/pgstream/pkg/log" + "github.com/xataio/pgstream/pkg/otel" ) const ( - DefaultAddress = "localhost:9910" + DefaultAddress = "localhost:9910" + DefaultPrometheusEndpoint = "/metrics" readinessCheckTimeout = 2 * time.Second readHeaderTimeout = 5 * time.Second @@ -38,6 +41,7 @@ type Server struct { phaseProvider func() string listener net.Listener httpServer *http.Server + metricsConfig *otel.MetricsConfig } var errNotListening = errors.New("health server not initialised: call Listen first") @@ -46,14 +50,15 @@ type Option func(*Server) // NewServer builds a health server from the given config. The server is not // started until Start() is called. -func NewServer(cfg Config, opts ...Option) *Server { +func NewServer(cfg Config, metricsCfg *otel.MetricsConfig, opts ...Option) *Server { address := cfg.Address if address == "" { address = DefaultAddress } s := &Server{ - logger: loglib.NewNoopLogger(), - address: address, + logger: loglib.NewNoopLogger(), + address: address, + metricsConfig: metricsCfg, } for _, opt := range opts { opt(s) @@ -114,6 +119,14 @@ func (s *Server) Listen() error { mux.HandleFunc("/ready", s.handleReady) mux.HandleFunc("/status", s.handleStatus) + if s.isPrometheusEnabled() { + endpoint := DefaultPrometheusEndpoint + if promEndpoint := s.metricsConfig.Prometheus.Endpoint; promEndpoint != "" { + endpoint = promEndpoint + } + mux.HandleFunc(endpoint, s.handlePrometheus) + } + s.listener = ln s.httpServer = &http.Server{ Handler: mux, @@ -200,3 +213,16 @@ func (s *Server) writeJSON(w http.ResponseWriter, code int, body any) { s.logger.Warn(err, "health server: writing response") } } + +func (s *Server) isPrometheusEnabled() bool { + return s.metricsConfig != nil && s.metricsConfig.Prometheus != nil && s.metricsConfig.Prometheus.Enabled + +} + +func (s *Server) handlePrometheus(w http.ResponseWriter, req *http.Request) { + if s.isPrometheusEnabled() { + promhttp.Handler().ServeHTTP(w, req) + } else { + s.writeJSON(w, http.StatusNotFound, map[string]any{}) + } +} diff --git a/internal/health/health_test.go b/internal/health/health_test.go index 51fbdcbb..e43312bb 100644 --- a/internal/health/health_test.go +++ b/internal/health/health_test.go @@ -13,23 +13,24 @@ import ( "time" "github.com/stretchr/testify/require" + "github.com/xataio/pgstream/pkg/otel" ) func TestNewServer_DefaultAddress(t *testing.T) { t.Parallel() - s := NewServer(Config{}) + s := NewServer(Config{}, nil) require.Equal(t, DefaultAddress, s.address) } func TestNewServer_CustomAddress(t *testing.T) { t.Parallel() - s := NewServer(Config{Address: "127.0.0.1:0"}) + s := NewServer(Config{Address: "127.0.0.1:0"}, nil) require.Equal(t, "127.0.0.1:0", s.address) } func TestHandleHealth(t *testing.T) { t.Parallel() - s := NewServer(Config{}, WithVersion("v1.2.3")) + s := NewServer(Config{}, nil, WithVersion("v1.2.3")) req := httptest.NewRequest(http.MethodGet, "/health", nil) rec := httptest.NewRecorder() @@ -46,7 +47,7 @@ func TestHandleHealth(t *testing.T) { func TestHandleReady_NoCheck(t *testing.T) { t.Parallel() - s := NewServer(Config{}, WithVersion("v1.2.3")) + s := NewServer(Config{}, nil, WithVersion("v1.2.3")) req := httptest.NewRequest(http.MethodGet, "/ready", nil) rec := httptest.NewRecorder() @@ -62,7 +63,7 @@ func TestHandleReady_NoCheck(t *testing.T) { func TestHandleReady_CheckPasses(t *testing.T) { t.Parallel() var called bool - s := NewServer(Config{}, WithReadinessCheck(func(_ context.Context) error { + s := NewServer(Config{}, nil, WithReadinessCheck(func(_ context.Context) error { called = true return nil })) @@ -77,7 +78,7 @@ func TestHandleReady_CheckPasses(t *testing.T) { func TestHandleReady_CheckFails(t *testing.T) { t.Parallel() - s := NewServer(Config{}, WithReadinessCheck(func(_ context.Context) error { + s := NewServer(Config{}, nil, WithReadinessCheck(func(_ context.Context) error { return errors.New("source unreachable") })) @@ -103,7 +104,7 @@ func TestHandleReady_CheckFails(t *testing.T) { func TestHandleStatus_NoProvider(t *testing.T) { t.Parallel() - s := NewServer(Config{}, WithVersion("v1.2.3")) + s := NewServer(Config{}, nil, WithVersion("v1.2.3")) req := httptest.NewRequest(http.MethodGet, "/status", nil) rec := httptest.NewRecorder() @@ -122,7 +123,7 @@ func TestHandleStatus_NoProvider(t *testing.T) { func TestHandleStatus_WithProvider(t *testing.T) { t.Parallel() phase := "snapshot" - s := NewServer(Config{}, WithVersion("v1.2.3"), WithPhaseProvider(func() string { + s := NewServer(Config{}, nil, WithVersion("v1.2.3"), WithPhaseProvider(func() string { return phase })) @@ -146,7 +147,7 @@ func TestHandleStatus_WithProvider(t *testing.T) { func TestListenServeShutdown(t *testing.T) { t.Parallel() - s := NewServer(Config{Address: "127.0.0.1:0"}) + s := NewServer(Config{Address: "127.0.0.1:0"}, nil) require.NoError(t, s.Listen()) @@ -164,22 +165,49 @@ func TestListenServeShutdown(t *testing.T) { func TestServe_BeforeListen(t *testing.T) { t.Parallel() - s := NewServer(Config{Address: "127.0.0.1:0"}) + s := NewServer(Config{Address: "127.0.0.1:0"}, nil) require.ErrorIs(t, s.Serve(), errNotListening) } func TestShutdown_BeforeListen(t *testing.T) { t.Parallel() - s := NewServer(Config{Address: "127.0.0.1:0"}) + s := NewServer(Config{Address: "127.0.0.1:0"}, nil) require.NoError(t, s.Shutdown(context.Background())) } func TestListen_BindError(t *testing.T) { t.Parallel() - first := NewServer(Config{Address: "127.0.0.1:0"}) + first := NewServer(Config{Address: "127.0.0.1:0"}, nil) require.NoError(t, first.Listen()) defer first.Shutdown(context.Background()) - second := NewServer(Config{Address: first.listener.Addr().String()}) + second := NewServer(Config{Address: first.listener.Addr().String()}, nil) require.Error(t, second.Listen()) } + +func makeMetricsConfig() *otel.MetricsConfig { + return &otel.MetricsConfig{ + Prometheus: &otel.PrometheusConfig{ + Enabled: true, + }, + } +} + +func TestHandlePrometheus(t *testing.T) { + t.Parallel() + s := NewServer(Config{}, makeMetricsConfig()) + rec := httptest.NewRecorder() + req := httptest.NewRequest("GET", DefaultPrometheusEndpoint, nil) + s.handlePrometheus(rec, req) + require.Equal(t, http.StatusOK, rec.Code) +} + +func TestHandlePrometheusDisabled(t *testing.T) { + t.Parallel() + s := NewServer(Config{}, nil) + req := httptest.NewRequest("GET", DefaultPrometheusEndpoint, nil) + rec := httptest.NewRecorder() + s.handlePrometheus(rec, req) + + require.Equal(t, http.StatusNotFound, rec.Code) +} diff --git a/pkg/otel/config.go b/pkg/otel/config.go index 2b73af08..c12b13a7 100644 --- a/pkg/otel/config.go +++ b/pkg/otel/config.go @@ -12,6 +12,12 @@ type Config struct { type MetricsConfig struct { Endpoint string CollectionInterval time.Duration + Prometheus *PrometheusConfig +} + +type PrometheusConfig struct { + Enabled bool + Endpoint string } type TracesConfig struct { diff --git a/pkg/otel/otel_provider.go b/pkg/otel/otel_provider.go index ddc4e5f9..c01acb48 100644 --- a/pkg/otel/otel_provider.go +++ b/pkg/otel/otel_provider.go @@ -11,6 +11,7 @@ import ( "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc" "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc" + "go.opentelemetry.io/otel/exporters/prometheus" "go.opentelemetry.io/otel/metric" metricnoop "go.opentelemetry.io/otel/metric/noop" sdkmetric "go.opentelemetry.io/otel/sdk/metric" @@ -79,25 +80,36 @@ func (o *Provider) initMeterProvider(ctx context.Context, metricsConfig *Metrics return nil } - metricsExporter, err := otlpmetricgrpc.New(ctx, - otlpmetricgrpc.WithTemporalitySelector(deltaSelector), - otlpmetricgrpc.WithInsecure(), - otlpmetricgrpc.WithEndpoint(metricsConfig.Endpoint)) - if err != nil { - return err + providerOpts := []sdkmetric.Option{ + sdkmetric.WithResource(newResource()), } - // periodic reader collects and exports metrics to the exporter at the - // defined interval (defaults to 60s) - reader := sdkmetric.NewPeriodicReader(metricsExporter, - sdkmetric.WithInterval(metricsConfig.collectionInterval()), - sdkmetric.WithProducer(runtime.NewProducer())) + if metricsConfig.Endpoint != "" { + metricsExporter, err := otlpmetricgrpc.New(ctx, + otlpmetricgrpc.WithTemporalitySelector(deltaSelector), + otlpmetricgrpc.WithInsecure(), + otlpmetricgrpc.WithEndpoint(metricsConfig.Endpoint)) + if err != nil { + return err + } - mp := sdkmetric.NewMeterProvider( - sdkmetric.WithResource(newResource()), - sdkmetric.WithReader(reader)) - o.shutdownFns = append(o.shutdownFns, mp.Shutdown) + // periodic reader collects and exports metrics to the exporter at the + // defined interval (defaults to 60s) + reader := sdkmetric.NewPeriodicReader(metricsExporter, + sdkmetric.WithInterval(metricsConfig.collectionInterval()), + sdkmetric.WithProducer(runtime.NewProducer())) + providerOpts = append(providerOpts, sdkmetric.WithReader(reader)) + } + if metricsConfig.Prometheus != nil && metricsConfig.Prometheus.Enabled { + promExporter, err := prometheus.New(prometheus.WithProducer(runtime.NewProducer())) + if err != nil { + return err + } + providerOpts = append(providerOpts, sdkmetric.WithReader(promExporter)) + } + mp := sdkmetric.NewMeterProvider(providerOpts...) + o.shutdownFns = append(o.shutdownFns, mp.Shutdown) o.meterProvider = mp otel.SetMeterProvider(o.meterProvider) diff --git a/pkg/stream/integration/helper_test.go b/pkg/stream/integration/helper_test.go index 1761209a..cad4a832 100644 --- a/pkg/stream/integration/helper_test.go +++ b/pkg/stream/integration/helper_test.go @@ -143,7 +143,7 @@ func startPhaseHealthServer(t *testing.T, tracker *phase.Tracker) (statusURL str addr := ln.Addr().String() require.NoError(t, ln.Close()) - srv := health.NewServer(health.Config{Address: addr}, + srv := health.NewServer(health.Config{Address: addr}, nil, health.WithVersion("test"), health.WithPhaseProvider(func() string { return string(tracker.Get())