diff --git a/.github/workflows/main.yaml b/.github/workflows/main.yaml index f90b482..3ad6192 100644 --- a/.github/workflows/main.yaml +++ b/.github/workflows/main.yaml @@ -8,6 +8,9 @@ on: branches: - "master" +permissions: + contents: "read" + jobs: quality-check: runs-on: "ubuntu-24.04" @@ -24,13 +27,15 @@ jobs: restore-keys: | ${{ runner.os }}-go- - run: | - go install honnef.co/go/tools/cmd/staticcheck@latest - go install golang.org/x/tools/cmd/goimports@latest - go install github.com/golangci/golangci-lint/cmd/golangci-lint@latest + go install honnef.co/go/tools/cmd/staticcheck@v0.7.0 + go install golang.org/x/tools/cmd/goimports@v0.48.0 + go install github.com/golangci/golangci-lint/v2/cmd/golangci-lint@v2.12.2 + go install golang.org/x/vuln/cmd/govulncheck@v1.6.0 - run: "go vet ./..." - run: "staticcheck ./..." - run: "golangci-lint run ./..." + - run: "govulncheck ./..." - run: | diff=$(goimports -d .) if [ -n "$diff" ]; then @@ -74,4 +79,4 @@ jobs: GRANT ALL PRIVILEGES ON DATABASE kit_test TO kit; EOF PGPASSWORD=kit psql -h localhost -U kit -d kit_test -c "ALTER SCHEMA public OWNER TO kit;" - - run: "go test -v -count=1 ./..." + - run: "go test -race -v -count=1 ./..." diff --git a/CHANGELOG.md b/CHANGELOG.md index dce0631..2495157 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- Harden PostgreSQL pool cleanup, metrics registration, TLS validation, and migration locking. +- Prevent HTTP SSRF proxy bypasses and redact sensitive telemetry and error responses. +- Remove connection-level HTTP spans that incorrectly parent unrelated keep-alive requests. +- Propagate shutdown cancellation, bound shutdown and HTTP waits, and recover worker panics. +- Use secure telemetry defaults and redact secrets from printed configuration. + +### Security + +- Omit SQL query text from traces by default and reduce attacker-controlled telemetry cardinality. +- Pin CI tools, run race-enabled tests, and scan dependencies with `govulncheck`. + ## [0.11.0] - 2026-05-25 ### Added @@ -27,7 +40,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added -- **httpclient**: Retry round tripper with exponential backoff and jitter for transient failures (5xx, 429, and connection errors). Enabled via `WithMaxRetries`, `WithRetryMinWait`, and `WithRetryMaxWait` options. - **pg**: `WithMaxConnLifetimeJitter` option to smear connection recycle events and avoid synchronized reconnect storms. Defaults to 5 minutes; pass `0` to disable. - **pg**: `WithHealthCheckPeriod` option to tune how often the pool checks idle connections and enforces MinConns/MaxConnIdleTime/MaxConnLifetime. A zero value leaves the pgx default (1 minute) in place. diff --git a/GNUmakefile b/GNUmakefile index e80433a..3640e54 100644 --- a/GNUmakefile +++ b/GNUmakefile @@ -13,6 +13,7 @@ # PERFORMANCE OF THIS SOFTWARE. GO ?= go +GOFMT ?= gofmt GOIMPORTS ?= goimports GOLANGCI_LINT ?= golangci-lint @@ -46,6 +47,10 @@ lint-goimports: exit 1; \ fi +.PHONY: go-fmt +go-fmt: + $(GO) fmt $(PACKAGES) + .PHONY: test test: $(GO) test -count=1 $(PACKAGES) diff --git a/README.md b/README.md index 4b5a3cb..7de1444 100644 --- a/README.md +++ b/README.md @@ -24,7 +24,7 @@ Kit provides foundational packages for building production-grade Go services wit ## Requirements -- Go 1.25+ +- Go 1.26+ ## Installation @@ -36,6 +36,14 @@ go get go.gearno.de/kit See the [Go package documentation](https://pkg.go.dev/go.gearno.de/kit) for API reference and examples. +## Production safety + +- Enable `httpclient.WithSSRFProtection` whenever users can influence URLs. +- Use `pg.WithTLS` and `pg.WithRequireTLS` for non-local databases. +- SQL query text is excluded from traces by default. +- Metrics bind to loopback and OTLP/HTTP uses verified TLS by default. +- Outbound requests have bounded response-header and total timeouts. + ## Contributing Contributions are welcome. Please open an issue to discuss proposed changes before submitting a pull request. diff --git a/go.mod b/go.mod index a14412d..e9ede22 100644 --- a/go.mod +++ b/go.mod @@ -1,19 +1,19 @@ module go.gearno.de/kit -go 1.26 +go 1.26.0 require ( github.com/fatih/color v1.19.0 github.com/go-chi/chi/v5 v5.2.5 - github.com/jackc/pgx/v5 v5.9.1 + github.com/jackc/pgx/v5 v5.11.0 github.com/prometheus/client_golang v1.23.2 - github.com/stretchr/testify v1.11.1 + github.com/stretchr/testify v1.12.1 go.gearno.de/crypto/uuid v0.1.0 go.gearno.de/x/panicf v0.1.1 - go.opentelemetry.io/otel v1.42.0 - go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.42.0 - go.opentelemetry.io/otel/sdk v1.42.0 - go.opentelemetry.io/otel/trace v1.42.0 + go.opentelemetry.io/otel v1.47.0 + go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.47.0 + go.opentelemetry.io/otel/sdk v1.47.0 + go.opentelemetry.io/otel/trace v1.47.0 sigs.k8s.io/yaml v1.6.0 ) @@ -21,11 +21,10 @@ require ( github.com/beorn7/perks v1.0.1 // indirect github.com/cenkalti/backoff/v5 v5.0.3 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect - github.com/davecgh/go-spew v1.1.1 // indirect - github.com/go-logr/logr v1.4.3 // indirect + github.com/go-logr/logr v1.4.4 // indirect github.com/go-logr/stdr v1.2.2 // indirect github.com/google/uuid v1.6.0 // indirect - github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 // indirect + github.com/grpc-ecosystem/grpc-gateway/v2 v2.30.0 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect github.com/jackc/puddle/v2 v2.2.2 // indirect @@ -33,23 +32,23 @@ require ( github.com/mattn/go-colorable v0.1.14 // indirect github.com/mattn/go-isatty v0.0.20 // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect - github.com/pmezard/go-difflib v1.0.0 // indirect github.com/prometheus/client_model v0.6.2 // indirect github.com/prometheus/common v0.67.5 // indirect github.com/prometheus/procfs v0.20.1 // indirect - github.com/stretchr/objx v0.5.2 // indirect + github.com/stretchr/objx v0.5.3 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect - go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.42.0 // indirect - go.opentelemetry.io/otel/metric v1.42.0 // indirect - go.opentelemetry.io/proto/otlp v1.10.0 // indirect + go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.47.0 // indirect + go.opentelemetry.io/otel/log v1.47.0 // indirect + go.opentelemetry.io/otel/metric v1.47.0 // indirect + go.opentelemetry.io/proto/otlp v1.11.0 // indirect go.yaml.in/yaml/v2 v2.4.4 // indirect - golang.org/x/net v0.52.0 // indirect - golang.org/x/sync v0.20.0 // indirect - golang.org/x/sys v0.42.0 // indirect - golang.org/x/text v0.35.0 // indirect - google.golang.org/genproto/googleapis/api v0.0.0-20260401024825-9d38bb4040a9 // indirect - google.golang.org/genproto/googleapis/rpc v0.0.0-20260401024825-9d38bb4040a9 // indirect - google.golang.org/grpc v1.80.0 // indirect - google.golang.org/protobuf v1.36.11 // indirect - gopkg.in/yaml.v3 v3.0.1 // indirect + go.yaml.in/yaml/v3 v3.0.5 // indirect + golang.org/x/net v0.61.0 // indirect + golang.org/x/sync v0.24.0 // indirect + golang.org/x/sys v0.49.0 // indirect + golang.org/x/text v0.43.0 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20260928230214-8a89bd6388cc // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260928230214-8a89bd6388cc // indirect + google.golang.org/grpc v1.84.0 // indirect + google.golang.org/protobuf v1.36.12 // indirect ) diff --git a/go.sum b/go.sum index 1d4de07..6ab96ba 100644 --- a/go.sum +++ b/go.sum @@ -5,19 +5,13 @@ github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F9 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= -github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= -github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= -github.com/fatih/color v1.18.0 h1:S8gINlzdQ840/4pfAwic/ZE0djQEH3wM94VfqLTZcOM= -github.com/fatih/color v1.18.0/go.mod h1:4FelSpRwEGDpQ12mAdzqdOukCy4u8WUtOY6lkT/6HfU= github.com/fatih/color v1.19.0 h1:Zp3PiM21/9Ld6FzSKyL5c/BULoe/ONr9KlbYVOfG8+w= github.com/fatih/color v1.19.0/go.mod h1:zNk67I0ZUT1bEGsSGyCZYZNrHuTkJJB+r6Q9VuMi0LE= -github.com/go-chi/chi/v5 v5.2.4 h1:WtFKPHwlywe8Srng8j2BhOD9312j9cGUxG1SP4V2cR4= -github.com/go-chi/chi/v5 v5.2.4/go.mod h1:X7Gx4mteadT3eDOMTsXzmI4/rwUpOwBHLpAfupzFJP0= github.com/go-chi/chi/v5 v5.2.5 h1:Eg4myHZBjyvJmAFjFvWgrqDTXFyOzjj7YIm3L3mu6Ug= github.com/go-chi/chi/v5 v5.2.5/go.mod h1:X7Gx4mteadT3eDOMTsXzmI4/rwUpOwBHLpAfupzFJP0= github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= -github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= -github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/logr v1.4.4 h1:tG4xh9yMsRCAiodLVTxyrkzSZ9+o0L1Kg/+cPVcbP/8= +github.com/go-logr/logr v1.4.4/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= @@ -26,26 +20,18 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.7 h1:X+2YciYSxvMQK0UZ7sg45ZVabVZBeBuvMkmuI2V3Fak= -github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.7/go.mod h1:lW34nIZuQ8UDPdkon5fmfp2l3+ZkQ2me/+oecHYLOII= -github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 h1:HWRh5R2+9EifMyIHV7ZV+MIZqgz+PMpZ14Jynv3O2Zs= -github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0/go.mod h1:JfhWUomR1baixubs02l85lZYYOm7LV6om4ceouMv45c= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.30.0 h1:/Tnpcb2E0Pz/tN9s3bfEY2Q8ePCEX9iuS+cneUwncnw= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.30.0/go.mod h1:zOBXOsUaBSjKgmH4OGzV1esUpR3oUSCPYVd2cUBjKYY= github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= -github.com/jackc/pgx/v5 v5.8.0 h1:TYPDoleBBme0xGSAX3/+NujXXtpZn9HBONkQC7IEZSo= -github.com/jackc/pgx/v5 v5.8.0/go.mod h1:QVeDInX2m9VyzvNeiCJVjCkNFqzsNb43204HshNSZKw= -github.com/jackc/pgx/v5 v5.9.1 h1:uwrxJXBnx76nyISkhr33kQLlUqjv7et7b9FjCen/tdc= -github.com/jackc/pgx/v5 v5.9.1/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= +github.com/jackc/pgx/v5 v5.11.0 h1:IzBBtyK9AHqf98cctWFifYSci2hgQR/cd56wB4p+ogg= +github.com/jackc/pgx/v5 v5.11.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= github.com/klauspost/compress v1.18.2 h1:iiPHWW0YrcFgpBYhsA6D1+fqHssJscY/Tm/y2Uqnapk= github.com/klauspost/compress v1.18.2/go.mod h1:R0h/fSBs8DE4ENlcrlib3PsXS61voFxhIs2DeRhCvJ4= -github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= -github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= -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/mattn/go-colorable v0.1.14 h1:9A9LHSqF/7dyVVX6g0U9cwm9pG3kP9gSzcuIPHPsaIE= @@ -54,7 +40,6 @@ github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWE github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= 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/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= 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= @@ -62,103 +47,65 @@ github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNw 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/procfs v0.19.2 h1:zUMhqEW66Ex7OXIiDkll3tl9a1ZdilUOd/F6ZXw4Vws= -github.com/prometheus/procfs v0.19.2/go.mod h1:M0aotyiemPhBCM0z5w87kL22CxfcH05ZpYlu+b4J7mw= github.com/prometheus/procfs v0.20.1 h1:XwbrGOIplXW/AU3YhIhLODXMJYyC1isLFfYCsTEycfc= github.com/prometheus/procfs v0.20.1/go.mod h1:o9EMBZGRyvDrSPH1RqdxhojkuXstoe4UlK79eF5TGGo= -github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= -github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= -github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY= -github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA= +github.com/stretchr/objx v0.5.3 h1:jmXUvGomnU1o3W/V5h2VEradbpJDwGrzugQQvL0POH4= +github.com/stretchr/objx v0.5.3/go.mod h1:rDQraq+vQZU7Fde9LOZLr8Tax6zZvy4kuNKF+QYS+U0= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= -github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= -github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE= +github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg= go.gearno.de/crypto/uuid v0.1.0 h1:94BYg7GYItJ6yYZ1GJayb3VYhI9/FjxuR1nFaduR4hE= go.gearno.de/crypto/uuid v0.1.0/go.mod h1:fnIIvKO9QnsyLO3ZJLJT3r8KZv/p0FOeT5eZKilYWXg= go.gearno.de/x/panicf v0.1.1 h1:E3Cr9NB8Ry2EsvEG/1eHr7kplP3tEjTf5d56dTX64VQ= go.gearno.de/x/panicf v0.1.1/go.mod h1:VnB8oF0UefMZcYeD4v+Wk4U5Z1uza7PHLlhT2CbNEbU= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= -go.opentelemetry.io/otel v1.39.0 h1:8yPrr/S0ND9QEfTfdP9V+SiwT4E0G7Y5MO7p85nis48= -go.opentelemetry.io/otel v1.39.0/go.mod h1:kLlFTywNWrFyEdH0oj2xK0bFYZtHRYUdv1NklR/tgc8= -go.opentelemetry.io/otel v1.42.0 h1:lSQGzTgVR3+sgJDAU/7/ZMjN9Z+vUip7leaqBKy4sho= -go.opentelemetry.io/otel v1.42.0/go.mod h1:lJNsdRMxCUIWuMlVJWzecSMuNjE7dOYyWlqOXWkdqCc= -go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.39.0 h1:f0cb2XPmrqn4XMy9PNliTgRKJgS5WcL/u0/WRYGz4t0= -go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.39.0/go.mod h1:vnakAaFckOMiMtOIhFI2MNH4FYrZzXCYxmb1LlhoGz8= -go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.42.0 h1:THuZiwpQZuHPul65w4WcwEnkX2QIuMT+UFoOrygtoJw= -go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.42.0/go.mod h1:J2pvYM5NGHofZ2/Ru6zw/TNWnEQp5crgyDeSrYpXkAw= -go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.39.0 h1:Ckwye2FpXkYgiHX7fyVrN1uA/UYd9ounqqTuSNAv0k4= -go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.39.0/go.mod h1:teIFJh5pW2y+AN7riv6IBPX2DuesS3HgP39mwOspKwU= -go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.42.0 h1:uLXP+3mghfMf7XmV4PkGfFhFKuNWoCvvx5wP/wOXo0o= -go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.42.0/go.mod h1:v0Tj04armyT59mnURNUJf7RCKcKzq+lgJs6QSjHjaTc= -go.opentelemetry.io/otel/metric v1.39.0 h1:d1UzonvEZriVfpNKEVmHXbdf909uGTOQjA0HF0Ls5Q0= -go.opentelemetry.io/otel/metric v1.39.0/go.mod h1:jrZSWL33sD7bBxg1xjrqyDjnuzTUB0x1nBERXd7Ftcs= -go.opentelemetry.io/otel/metric v1.42.0 h1:2jXG+3oZLNXEPfNmnpxKDeZsFI5o4J+nz6xUlaFdF/4= -go.opentelemetry.io/otel/metric v1.42.0/go.mod h1:RlUN/7vTU7Ao/diDkEpQpnz3/92J9ko05BIwxYa2SSI= -go.opentelemetry.io/otel/sdk v1.39.0 h1:nMLYcjVsvdui1B/4FRkwjzoRVsMK8uL/cj0OyhKzt18= -go.opentelemetry.io/otel/sdk v1.39.0/go.mod h1:vDojkC4/jsTJsE+kh+LXYQlbL8CgrEcwmt1ENZszdJE= -go.opentelemetry.io/otel/sdk v1.42.0 h1:LyC8+jqk6UJwdrI/8VydAq/hvkFKNHZVIWuslJXYsDo= -go.opentelemetry.io/otel/sdk v1.42.0/go.mod h1:rGHCAxd9DAph0joO4W6OPwxjNTYWghRWmkHuGbayMts= -go.opentelemetry.io/otel/sdk/metric v1.39.0 h1:cXMVVFVgsIf2YL6QkRF4Urbr/aMInf+2WKg+sEJTtB8= -go.opentelemetry.io/otel/sdk/metric v1.39.0/go.mod h1:xq9HEVH7qeX69/JnwEfp6fVq5wosJsY1mt4lLfYdVew= -go.opentelemetry.io/otel/sdk/metric v1.42.0 h1:D/1QR46Clz6ajyZ3G8SgNlTJKBdGp84q9RKCAZ3YGuA= -go.opentelemetry.io/otel/trace v1.39.0 h1:2d2vfpEDmCJ5zVYz7ijaJdOF59xLomrvj7bjt6/qCJI= -go.opentelemetry.io/otel/trace v1.39.0/go.mod h1:88w4/PnZSazkGzz/w84VHpQafiU4EtqqlVdxWy+rNOA= -go.opentelemetry.io/otel/trace v1.42.0 h1:OUCgIPt+mzOnaUTpOQcBiM/PLQ/Op7oq6g4LenLmOYY= -go.opentelemetry.io/otel/trace v1.42.0/go.mod h1:f3K9S+IFqnumBkKhRJMeaZeNk9epyhnCmQh/EysQCdc= -go.opentelemetry.io/proto/otlp v1.9.0 h1:l706jCMITVouPOqEnii2fIAuO3IVGBRPV5ICjceRb/A= -go.opentelemetry.io/proto/otlp v1.9.0/go.mod h1:xE+Cx5E/eEHw+ISFkwPLwCZefwVjY+pqKg1qcK03+/4= -go.opentelemetry.io/proto/otlp v1.10.0 h1:IQRWgT5srOCYfiWnpqUYz9CVmbO8bFmKcwYxpuCSL2g= -go.opentelemetry.io/proto/otlp v1.10.0/go.mod h1:/CV4QoCR/S9yaPj8utp3lvQPoqMtxXdzn7ozvvozVqk= +go.opentelemetry.io/otel v1.47.0 h1:j7ALJ/zgkS7Z6aeJW09p8VC9804bC+PpeTfCD4XPnOM= +go.opentelemetry.io/otel v1.47.0/go.mod h1:8wS9O2qfXrYrzp6hIF/HOYJJf/wIhFPhR2xLuP+iXQU= +go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.47.0 h1:julhjPeUH/q/7hinbSdDdqt5h7Zw9YWmRlWRhI0jd54= +go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.47.0/go.mod h1:Ao2mz688LH/tFf0yMAenidq6k2YNSx6SIY2q6jDACck= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.47.0 h1:aXZXsZ012wOgVkqaiQv+Z/d8V0xKjmg70F6m4CE94lc= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.47.0/go.mod h1:ziu1gUIJhAaxM3EV1e3Ralk5AAyL1UOYz3lJJ4VSI38= +go.opentelemetry.io/otel/log v1.47.0 h1:cOTS1CcLbSQeZKanGJ+0JpF/+t4PELi3O3bbl2lqCcI= +go.opentelemetry.io/otel/log v1.47.0/go.mod h1:9byitSQ5pLC6PpqwGXjqdMKya6ZTswHRZh2vvXT33nw= +go.opentelemetry.io/otel/metric v1.47.0 h1:4PptaldXx3Eat1XjMZ68pPJEs5wrhlemctZE9a3UdWY= +go.opentelemetry.io/otel/metric v1.47.0/go.mod h1:ADGSXxRrXM6bjbvLo535EstVFlPpPYZm4LBKixjDHwU= +go.opentelemetry.io/otel/sdk v1.47.0 h1:zWXEr4j2lFefG87TU6Yg8a7ngfohIKFZHKp0Hf5hC6I= +go.opentelemetry.io/otel/sdk v1.47.0/go.mod h1:VUc24kiOeoGsxG8G9ULx3fWKvB7jMhnGE8Oi607lgR0= +go.opentelemetry.io/otel/sdk/metric v1.47.0 h1:lfISg2j93VT6yqdk9OfUaZmw/GfcZqCCV3jdXtsPnKw= +go.opentelemetry.io/otel/sdk/metric v1.47.0/go.mod h1:ypLp+mW1Nt2x+Szt3b5/i1syodyts49lMOwxpDI3VGw= +go.opentelemetry.io/otel/trace v1.47.0 h1:JOjX/Oci8K94QHddo+bbfya/Ai/nf6/dt9ZfrFNWSrM= +go.opentelemetry.io/otel/trace v1.47.0/go.mod h1:jNaSLa2PZEYFG6fRjJABAu+bw4FS08uDmPg28lTghu0= +go.opentelemetry.io/proto/otlp v1.11.0 h1:5rrYs0Ykyj50sdU/JU0x8etU+LubXWb+gED6TbEdMIk= +go.opentelemetry.io/proto/otlp v1.11.0/go.mod h1:SmVizdCOAm3XBtG1g1NnOdhW6jtddT72hLMhv8VwA8E= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= -go.yaml.in/yaml/v2 v2.4.3 h1:6gvOSjQoTB3vt1l+CU+tSyi/HOjfOjRLJ4YwYZGwRO0= -go.yaml.in/yaml/v2 v2.4.3/go.mod h1:zSxWcmIDjOzPXpjlTTbAsKokqkDNAVtZO0WOMiT90s8= 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/net v0.49.0 h1:eeHFmOGUTtaaPSGNmjBKpbng9MulQsJURQUAfUwY++o= -golang.org/x/net v0.49.0/go.mod h1:/ysNB2EvaqvesRkuLAyjI1ycPZlQHM3q01F02UY/MV8= -golang.org/x/net v0.52.0 h1:He/TN1l0e4mmR3QqHMT2Xab3Aj3L9qjbhRm78/6jrW0= -golang.org/x/net v0.52.0/go.mod h1:R1MAz7uMZxVMualyPXb+VaqGSa3LIaUqk0eEt3w36Sw= -golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4= -golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= -golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= -golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw= +go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg= +golang.org/x/net v0.61.0 h1:RR+6j/BTBCrggOLugxa+66R8FvoN6MHlzDlZIbN9YFw= +golang.org/x/net v0.61.0/go.mod h1:WgDCOTH8iwtB66T4/GfI94AHo/hhdb7Hah9NDseQfiI= +golang.org/x/sync v0.24.0 h1:iqiDrGoDgyuvAKrhLQthUEzU/R378N78gXbBjxed7oA= +golang.org/x/sync v0.24.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.40.0 h1:DBZZqJ2Rkml6QMQsZywtnjnnGvHza6BTfYFWY9kjEWQ= -golang.org/x/sys v0.40.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= -golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo= -golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= -golang.org/x/text v0.33.0 h1:B3njUFyqtHDUI5jMn1YIr5B0IE2U0qck04r6d4KPAxE= -golang.org/x/text v0.33.0/go.mod h1:LuMebE6+rBincTi9+xWTY8TztLzKHc/9C1uBCG27+q8= -golang.org/x/text v0.35.0 h1:JOVx6vVDFokkpaq1AEptVzLTpDe9KGpj5tR4/X+ybL8= -golang.org/x/text v0.35.0/go.mod h1:khi/HExzZJ2pGnjenulevKNX1W67CUy0AsXcNubPGCA= -gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk= -gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E= +golang.org/x/sys v0.49.0 h1:XbzkgYJHdqh/8m2Uu0W/dQv8nktxx4BFHp1M0gROTXA= +golang.org/x/sys v0.49.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= +golang.org/x/text v0.43.0 h1:1QivrlhwAsnMOcqgOdl9my7s9OyZl8O5/UGUTJVXoWc= +golang.org/x/text v0.43.0/go.mod h1:hbSIYA/amXcRXTFSZqh8tWHEoqRvvbKPJ2uz/KU+a+s= gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= -google.golang.org/genproto/googleapis/api v0.0.0-20260128011058-8636f8732409 h1:merA0rdPeUV3YIIfHHcH4qBkiQAc1nfCKSI7lB4cV2M= -google.golang.org/genproto/googleapis/api v0.0.0-20260128011058-8636f8732409/go.mod h1:fl8J1IvUjCilwZzQowmw2b7HQB2eAuYBabMXzWurF+I= -google.golang.org/genproto/googleapis/api v0.0.0-20260401024825-9d38bb4040a9 h1:VPWxll4HlMw1Vs/qXtN7BvhZqsS9cdAittCNvVENElA= -google.golang.org/genproto/googleapis/api v0.0.0-20260401024825-9d38bb4040a9/go.mod h1:7QBABkRtR8z+TEnmXTqIqwJLlzrZKVfAUm7tY3yGv0M= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260128011058-8636f8732409 h1:H86B94AW+VfJWDqFeEbBPhEtHzJwJfTbgE2lZa54ZAQ= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260128011058-8636f8732409/go.mod h1:j9x/tPzZkyxcgEFkiKEEGxfvyumM01BEtsW8xzOahRQ= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260401024825-9d38bb4040a9 h1:m8qni9SQFH0tJc1X0vmnpw/0t+AImlSvp30sEupozUg= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260401024825-9d38bb4040a9/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= -google.golang.org/grpc v1.78.0 h1:K1XZG/yGDJnzMdd/uZHAkVqJE+xIDOcmdSFZkBUicNc= -google.golang.org/grpc v1.78.0/go.mod h1:I47qjTo4OKbMkjA/aOOwxDIiPSBofUtQUI5EfpWvW7U= -google.golang.org/grpc v1.80.0 h1:Xr6m2WmWZLETvUNvIUmeD5OAagMw3FiKmMlTdViWsHM= -google.golang.org/grpc v1.80.0/go.mod h1:ho/dLnxwi3EDJA4Zghp7k2Ec1+c2jqup0bFkw07bwF4= -google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= -google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= +google.golang.org/genproto/googleapis/api v0.0.0-20260928230214-8a89bd6388cc h1:rZHRz0ogq4Pid1IKtA/ivHi8XONVF5tqUQXA2TtgpkE= +google.golang.org/genproto/googleapis/api v0.0.0-20260928230214-8a89bd6388cc/go.mod h1:K/n0XLazCJHjD73YuMyNJsN38BPPy1kfFnKxUXaThFU= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260928230214-8a89bd6388cc h1:4bNTbnb44EqGVy9HaQxSY2jnafSUdgV3qHtedvNpDKg= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260928230214-8a89bd6388cc/go.mod h1:OaIUM3+LpYcK2GXM4FTmhWoIq371Owdr+Cc7/BsYHHc= +google.golang.org/grpc v1.84.0 h1:soMyaPJ8pAak5PIQ0DGBUir0XRo2fRoMqhNWMLlLxO0= +google.golang.org/grpc v1.84.0/go.mod h1:ljCht0DrxQrXBDRTZp52Qxh3Ffk8CdYm2sj4O2QN2C0= +google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc= +google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= -gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= -gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= -gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= -gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= sigs.k8s.io/yaml v1.6.0 h1:G8fkbMSAFqgEFgh4b1wmtzDnioxFCUgTZhlbj5P9QYs= sigs.k8s.io/yaml v1.6.0/go.mod h1:796bPqUfzR/0jLAl6XjHl3Ck7MiyVv8dbTdyT3/pMf4= diff --git a/httpclient/httpclient.go b/httpclient/httpclient.go index 5c8f935..8f14239 100644 --- a/httpclient/httpclient.go +++ b/httpclient/httpclient.go @@ -45,8 +45,10 @@ type ( logger *log.Logger registerer prometheus.Registerer - ssrfProtection bool - ssrfAllowLoopback bool + ssrfProtection bool + ssrfAllowLoopback bool + clientTimeout time.Duration + responseHeaderTimeout time.Duration } ) @@ -100,6 +102,9 @@ func WithRegisterer(r prometheus.Registerer) Option { // differs from the original request, blocking redirect-based // pivots. // +// Environment-configured HTTP proxies are disabled because a proxy +// hides the destination peer address from the protected dialer. +// // Use this option whenever the destination URL is influenced by // untrusted input (for example, customer-supplied webhook // endpoints or OAuth2 token URLs). @@ -119,6 +124,14 @@ func WithSSRFAllowLoopback() Option { } } +func WithTimeout(timeout time.Duration) Option { + return func(o *Options) { o.clientTimeout = timeout } +} + +func WithResponseHeaderTimeout(timeout time.Duration) Option { + return func(o *Options) { o.responseHeaderTimeout = timeout } +} + // DefaultTransport returns a new http.Transport with similar default // values to http.DefaultTransport, but with idle connections and // keepalives disabled. @@ -167,7 +180,7 @@ func DefaultPooledClient(options ...Option) *http.Client { } func buildClient(rt http.RoundTripper, opts *Options) *http.Client { - c := &http.Client{Transport: rt} + c := &http.Client{Transport: rt, Timeout: opts.clientTimeout} if opts.ssrfProtection { c.CheckRedirect = noCrossOriginRedirects } @@ -180,14 +193,17 @@ func createBaseTransport(opts *Options) *http.Transport { KeepAlive: 30 * time.Second, DualStack: true, } + proxy := http.ProxyFromEnvironment if opts.ssrfProtection { dial.Control = makeSSRFDialControl(opts.ssrfAllowLoopback) + proxy = nil } return &http.Transport{ - Proxy: http.ProxyFromEnvironment, + Proxy: proxy, DialContext: dial.DialContext, IdleConnTimeout: 90 * time.Second, + ResponseHeaderTimeout: opts.responseHeaderTimeout, TLSHandshakeTimeout: 10 * time.Second, ExpectContinueTimeout: 1 * time.Second, ForceAttemptHTTP2: true, @@ -196,9 +212,11 @@ func createBaseTransport(opts *Options) *http.Transport { func configureOptions(options []Option) *Options { opts := &Options{ - logger: log.NewLogger(log.WithOutput(io.Discard)), - tracerProvider: otel.GetTracerProvider(), - registerer: prometheus.DefaultRegisterer, + logger: log.NewLogger(log.WithOutput(io.Discard)), + tracerProvider: otel.GetTracerProvider(), + registerer: prometheus.DefaultRegisterer, + clientTimeout: 30 * time.Second, + responseHeaderTimeout: 15 * time.Second, } for _, o := range options { diff --git a/httpclient/roundtripper.go b/httpclient/roundtripper.go index d1ecf05..b88b69c 100644 --- a/httpclient/roundtripper.go +++ b/httpclient/roundtripper.go @@ -20,6 +20,7 @@ import ( "errors" "fmt" "net/http" + "net/url" "strconv" "time" @@ -125,7 +126,7 @@ func NewTelemetryRoundTripper( tracer: tp.Tracer( tracerName, trace.WithInstrumentationVersion( - version.New(0).Alpha(1), + version.Current(), ), ), requestsTotal: requestsTotal, @@ -144,6 +145,7 @@ func (rt *TelemetryRoundTripper) RoundTrip(r *http.Request) (*http.Response, err ctx = r2.Context() start = time.Now() requestID = r2.Header.Get("x-request-id") + safeURL = sanitizedURL(r2.URL) ) if requestID == "" { @@ -156,53 +158,47 @@ func (rt *TelemetryRoundTripper) RoundTrip(r *http.Request) (*http.Response, err } r2.Header.Set("x-request-id", requestID) - var ( - rootSpan = trace.SpanFromContext(ctx) - span trace.Span - logger = rt.logger.With( - log.String("http_request_method", r2.Method), - log.String("http_request_scheme", r2.URL.Scheme), - log.String("http_request_host", r2.URL.Host), - log.String("http_request_path", r2.URL.Path), - log.String("http_request_flavor", r2.Proto), - log.String("http_request_user_agent", r2.UserAgent()), - log.String("http_request_client_ip", r2.RemoteAddr), - log.String("http_request_id", requestID), - ) + logger := rt.logger.With( + log.String("http_request_method", r2.Method), + log.String("http_request_scheme", r2.URL.Scheme), + log.String("http_request_host", r2.URL.Host), + log.String("http_request_path", r2.URL.Path), + log.String("http_request_flavor", r2.Proto), + log.String("http_request_user_agent", r2.UserAgent()), + log.String("http_request_client_ip", r2.RemoteAddr), + log.String("http_request_id", requestID), ) - if rootSpan.IsRecording() { - spanName := fmt.Sprintf("%s %s %s", r2.Method, r2.URL.Host, r2.URL.Path) - ctx, span = rt.tracer.Start( - ctx, - spanName, - trace.WithSpanKind(trace.SpanKindClient), - trace.WithAttributes( - semconv.NetworkPeerAddress(r2.URL.Host), - semconv.NetworkPeerPort(atoi(r2.URL.Port())), - semconv.URLScheme(r2.URL.Scheme), - attribute.String("http.method", r.Method), - attribute.String("http.url", r2.URL.String()), - attribute.String("http.target", r2.URL.Path), - attribute.String("http.host", r2.URL.Host), - attribute.String("http.scheme", r2.URL.Scheme), - attribute.String("http.flavor", r2.Proto), - attribute.String("http.client_ip", r2.RemoteAddr), - attribute.String("http.user_agent", r2.UserAgent()), - attribute.String("http.request_id", requestID), - ), - ) - defer span.End() + spanName := fmt.Sprintf("%s %s", r2.Method, r2.URL.Hostname()) + ctx, span := rt.tracer.Start( + ctx, + spanName, + trace.WithSpanKind(trace.SpanKindClient), + trace.WithAttributes( + semconv.NetworkPeerAddress(r2.URL.Hostname()), + semconv.NetworkPeerPort(atoi(r2.URL.Port())), + semconv.URLScheme(r2.URL.Scheme), + attribute.String("http.method", r.Method), + attribute.String("http.url", safeURL), + attribute.String("http.target", r2.URL.Path), + attribute.String("http.host", r2.URL.Host), + attribute.String("http.scheme", r2.URL.Scheme), + attribute.String("http.flavor", r2.Proto), + attribute.String("http.client_ip", r2.RemoteAddr), + attribute.String("http.user_agent", r2.UserAgent()), + attribute.String("http.request_id", requestID), + ), + ) + defer span.End() - propagator := otel.GetTextMapPropagator() - propagator.Inject(ctx, propagation.HeaderCarrier(r2.Header)) - } + propagator := otel.GetTextMapPropagator() + propagator.Inject(ctx, propagation.HeaderCarrier(r2.Header)) resp, err := rt.next.RoundTrip(r2) if err != nil { rt.logger.ErrorCtx(ctx, "cannot execute http transaction", log.Error(err)) - if rootSpan.IsRecording() { + if span.IsRecording() { span.RecordError(err) span.SetStatus(codes.Error, err.Error()) } @@ -210,11 +206,14 @@ func (rt *TelemetryRoundTripper) RoundTrip(r *http.Request) (*http.Response, err return nil, err } - if rootSpan.IsRecording() { + if span.IsRecording() { span.SetAttributes( attribute.Int("http.status_code", resp.StatusCode), attribute.String("http.status_text", resp.Status), ) + if resp.StatusCode >= http.StatusInternalServerError { + span.SetStatus(codes.Error, resp.Status) + } } duration := time.Since(start) @@ -231,7 +230,7 @@ func (rt *TelemetryRoundTripper) RoundTrip(r *http.Request) (*http.Response, err rt.requestDurationSeconds.With(metricLabels).Observe(duration.Seconds()) logLevel := log.LevelInfo - logMessage := fmt.Sprintf("%s %s %d %s", r2.Method, r.URL.String(), resp.StatusCode, duration) + logMessage := fmt.Sprintf("%s %s %d %s", r2.Method, safeURL, resp.StatusCode, duration) if resp.StatusCode >= http.StatusInternalServerError { logLevel = log.LevelError } @@ -241,6 +240,16 @@ func (rt *TelemetryRoundTripper) RoundTrip(r *http.Request) (*http.Response, err return resp, nil } +func sanitizedURL(u *url.URL) string { + safe := *u + safe.User = nil + safe.RawQuery = "" + safe.ForceQuery = false + safe.Fragment = "" + safe.RawFragment = "" + return safe.String() +} + func atoi(s string) int { v, err := strconv.Atoi(s) if err != nil { diff --git a/httpclient/roundtripper_test.go b/httpclient/roundtripper_test.go index 12944b0..a2f06ac 100644 --- a/httpclient/roundtripper_test.go +++ b/httpclient/roundtripper_test.go @@ -23,6 +23,7 @@ import ( "net/http/httptest" "net/url" "testing" + "time" "github.com/prometheus/client_golang/prometheus" "github.com/stretchr/testify/assert" @@ -60,6 +61,14 @@ func TestNewTelemetryRoundTripper(t *testing.T) { assert.NotNil(t, tr) } +func TestDefaultClientUsesBoundedTimeouts(t *testing.T) { + client := DefaultClient(WithRegisterer(NoopRegisterer{})) + assert.Equal(t, 30*time.Second, client.Timeout) + telemetry := client.Transport.(*TelemetryRoundTripper) + transport := telemetry.next.(*http.Transport) + assert.Equal(t, 15*time.Second, transport.ResponseHeaderTimeout) +} + func TestRoundTrip(t *testing.T) { mockRT := new(MockRoundTripper) logger := log.NewLogger(log.WithOutput(io.Discard)) @@ -101,3 +110,9 @@ func TestRoundTrip(t *testing.T) { assert.Equal(t, http.StatusOK, response.StatusCode) mockRT.AssertExpectations(t) } + +func TestSanitizedURL(t *testing.T) { + u, err := url.Parse("https://user:secret@example.com/path?token=secret#fragment") + assert.NoError(t, err) + assert.Equal(t, "https://example.com/path", sanitizedURL(u)) +} diff --git a/httpclient/ssrf_test.go b/httpclient/ssrf_test.go index 9eb542b..054f1a4 100644 --- a/httpclient/ssrf_test.go +++ b/httpclient/ssrf_test.go @@ -116,6 +116,11 @@ func TestSSRFDialControl_RejectsLoopback(t *testing.T) { assert.ErrorIs(t, err, ErrBlockedAddress) } +func TestSSRFProtectionDisablesEnvironmentProxy(t *testing.T) { + transport := createBaseTransport(configureOptions([]Option{WithSSRFProtection()})) + assert.Nil(t, transport.Proxy) +} + func TestSSRFDialControl_AllowsPublicByDefault(t *testing.T) { t.Parallel() diff --git a/httpserver/handler_wrapper.go b/httpserver/handler_wrapper.go index 35055d9..10d1b1a 100644 --- a/httpserver/handler_wrapper.go +++ b/httpserver/handler_wrapper.go @@ -20,7 +20,9 @@ import ( "context" "encoding/json" "fmt" + "net" "net/http" + "net/url" "runtime" "strconv" "time" @@ -77,7 +79,6 @@ func newHandlerWrapper( ) *handlerWrapper { metricLabels := []string{ "method", - "host", "flavor", "status_code", "path", @@ -156,7 +157,7 @@ func newHandlerWrapper( tracer: tp.Tracer( tracerName, trace.WithInstrumentationVersion( - version.New(0).Alpha(1), + version.Current(), ), ), requestsTotal: requestsTotal, @@ -205,6 +206,8 @@ func (hw *handlerWrapper) ServeHTTP(w http.ResponseWriter, r *http.Request) { id, err := uuid.NewV7() if err != nil { logger.ErrorCtx(ctx, "cannot generate request id", log.Error(err)) + RenderJSON(ww, http.StatusInternalServerError, internalErrorResponse) + return } requestID = id.String() @@ -213,59 +216,59 @@ func (hw *handlerWrapper) ServeHTTP(w http.ResponseWriter, r *http.Request) { ww.Header().Set("x-request-id", requestID) logger = logger.With(log.String("http_request_id", requestID)) - var ( - rootSpan = trace.SpanFromContext(ctx) - span trace.Span + propagator := otel.GetTextMapPropagator() + ctx = propagator.Extract(ctx, propagation.HeaderCarrier(r2.Header)) + peerAddress, peerPort := splitAddress(r2.RemoteAddr) + spanName := fmt.Sprintf("%s server", r2.Method) + ctx, span := hw.tracer.Start( + ctx, + spanName, + trace.WithSpanKind(trace.SpanKindServer), + trace.WithAttributes( + semconv.NetworkPeerAddress(peerAddress), + semconv.NetworkPeerPort(peerPort), + semconv.URLScheme(r2.URL.Scheme), + attribute.String("http.method", r.Method), + attribute.String("http.url", sanitizedURL(r2.URL)), + attribute.String("http.target", r2.URL.Path), + attribute.String("http.host", r2.Host), + attribute.String("http.flavor", r2.Proto), + attribute.String("http.client_ip", r2.RemoteAddr), + attribute.String("http.user_agent", r2.UserAgent()), + attribute.String("http.request_id", requestID), + ), ) - - if rootSpan.IsRecording() { - propagator := otel.GetTextMapPropagator() - ctx = propagator.Extract(ctx, propagation.HeaderCarrier(r2.Header)) - - spanName := fmt.Sprintf("%s %s %s", r2.Method, r2.URL.Host, r2.URL.Path) - ctx, span = hw.tracer.Start( - ctx, - spanName, - trace.WithSpanKind(trace.SpanKindServer), - trace.WithAttributes( - semconv.NetworkPeerAddress(r2.URL.Host), - semconv.NetworkPeerPort(atoi(r2.URL.Port())), - semconv.URLScheme(r2.URL.Scheme), - attribute.String("http.method", r.Method), - attribute.String("http.url", r2.URL.String()), - attribute.String("http.target", r2.URL.Path), - attribute.String("http.host", r2.Host), - attribute.String("http.flavor", r2.Proto), - attribute.String("http.client_ip", r2.RemoteAddr), - attribute.String("http.user_agent", r2.UserAgent()), - attribute.String("http.request_id", requestID), - ), - ) - defer span.End() - } + defer span.End() // Hack to get route pattern from Chi. As today using the STD // router will require to much works to have proper sub router // support, a task for later. - ctx = context.WithValue(ctx, chi.RouteCtxKey, chi.NewRouteContext()) + if chi.RouteContext(ctx) == nil { + ctx = context.WithValue(ctx, chi.RouteCtxKey, chi.NewRouteContext()) + } ctx = context.WithValue(ctx, loggerRequestContextKey, logger) defer func() { duration := time.Since(start) hasPanic := false + pathPattern := chi.RouteContext(ctx).RoutePattern() + if pathPattern == "" { + pathPattern = "unknown" + } + span.SetName(fmt.Sprintf("%s %s", r2.Method, pathPattern)) rvr := recover() if rvr != nil { hasPanic = true if err, ok := rvr.(error); ok { - if rootSpan.IsRecording() { + if span.IsRecording() { span.RecordError(err) span.SetStatus(codes.Error, err.Error()) } } else { - if rootSpan.IsRecording() { + if span.IsRecording() { span.SetStatus(codes.Error, fmt.Sprintf("%v", rvr)) } } @@ -286,10 +289,9 @@ func (hw *handlerWrapper) ServeHTTP(w http.ResponseWriter, r *http.Request) { metricLabels := prometheus.Labels{ "method": r2.Method, - "host": r2.Host, "flavor": r2.Proto, "status_code": strconv.Itoa(ww.Status()), - "path": chi.RouteContext(ctx).RoutePattern(), + "path": pathPattern, } hw.requestsTotal.With(metricLabels).Inc() @@ -320,10 +322,10 @@ func (hw *handlerWrapper) ServeHTTP(w http.ResponseWriter, r *http.Request) { logger = logger.With( log.Int("http_response_size", ww.BytesWritten()), log.Int("http_response_status", ww.Status()), - log.String("http_request_path_pattern", chi.RouteContext(ctx).RoutePattern()), + log.String("http_request_path_pattern", pathPattern), ) - if ww.Status() > 499 && !hasPanic && rootSpan.IsRecording() { + if ww.Status() > 499 && !hasPanic && span.IsRecording() { span.SetStatus(codes.Error, fmt.Sprintf("%d status code", ww.Status())) } @@ -346,6 +348,24 @@ func atoi(s string) int { return v } +func splitAddress(address string) (string, int) { + host, port, err := net.SplitHostPort(address) + if err != nil { + return address, 0 + } + return host, atoi(port) +} + +func sanitizedURL(u *url.URL) string { + safe := *u + safe.User = nil + safe.RawQuery = "" + safe.ForceQuery = false + safe.Fragment = "" + safe.RawFragment = "" + return safe.String() +} + func estimateRequestSize(r *http.Request) float64 { s := 0 if r.URL != nil { diff --git a/httpserver/httpserver.go b/httpserver/httpserver.go index d1496cd..3f556a7 100644 --- a/httpserver/httpserver.go +++ b/httpserver/httpserver.go @@ -17,19 +17,14 @@ package httpserver import ( - "context" "io" stdlog "log" - "net" "net/http" - "sync" "time" "github.com/prometheus/client_golang/prometheus" - "go.gearno.de/kit/internal/version" "go.gearno.de/kit/log" "go.opentelemetry.io/otel" - "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/trace" ) @@ -103,15 +98,6 @@ func NewServer(addr string, h http.Handler, options ...Option) *http.Server { opts.registerer, ) - tracer := opts.tracerProvider.Tracer( - "go.gearno.de/kit/httpserver", - trace.WithInstrumentationVersion( - version.New(0).Alpha(1), - ), - ) - - connSpans := &sync.Map{} - return &http.Server{ Addr: addr, Handler: handler, @@ -120,22 +106,5 @@ func NewServer(addr string, h http.Handler, options ...Option) *http.Server { ReadTimeout: opts.readTimeout, WriteTimeout: opts.writeTimeout, IdleTimeout: 15 * time.Second, - ConnContext: func(ctx context.Context, c net.Conn) context.Context { - ctx, span := tracer.Start(ctx, "http.connection", - trace.WithSpanKind(trace.SpanKindServer), - trace.WithAttributes( - attribute.String("net.peer.addr", c.RemoteAddr().String()), - ), - ) - connSpans.Store(c, span) - return ctx - }, - ConnState: func(c net.Conn, state http.ConnState) { - if state == http.StateClosed || state == http.StateHijacked { - if span, ok := connSpans.LoadAndDelete(c); ok { - span.(trace.Span).End() - } - } - }, } } diff --git a/httpserver/httpserver_test.go b/httpserver/httpserver_test.go index 0c2e354..68133de 100644 --- a/httpserver/httpserver_test.go +++ b/httpserver/httpserver_test.go @@ -112,6 +112,17 @@ func TestHTTPServer_BasicOperation(t *testing.T) { } } +func TestHTTPServer_DoesNotCreateConnectionSpans(t *testing.T) { + server := NewServer( + ":8080", + http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}), + WithRegisterer(prometheus.NewRegistry()), + ) + + assert.Nil(t, server.ConnContext) + assert.Nil(t, server.ConnState) +} + // mockPanicHandler is a handler that always panics type mockPanicHandler struct{} @@ -284,7 +295,7 @@ func TestHTTPServer_Metrics(t *testing.T) { require.NoError(t, err) _, err = io.Copy(io.Discard, resp.Body) require.NoError(t, err) - resp.Body.Close() + require.NoError(t, resp.Body.Close()) } // Check metrics @@ -335,14 +346,14 @@ func TestHTTPServer_Logging(t *testing.T) { require.NoError(t, err) _, err = io.Copy(io.Discard, resp.Body) require.NoError(t, err) - resp.Body.Close() + require.NoError(t, resp.Body.Close()) // Make an error request resp, err = http.Get(ts.URL + "/error") require.NoError(t, err) _, err = io.Copy(io.Discard, resp.Body) require.NoError(t, err) - resp.Body.Close() + require.NoError(t, resp.Body.Close()) // Check logs logOutput := logBuf.String() diff --git a/httpserver/render.go b/httpserver/render.go index 3c705d1..4b3db12 100644 --- a/httpserver/render.go +++ b/httpserver/render.go @@ -42,9 +42,13 @@ func RenderText(w http.ResponseWriter, statusCode int, v string) { } func RenderError(w http.ResponseWriter, statusCode int, err error) { + message := err.Error() + if statusCode >= http.StatusInternalServerError { + message = http.StatusText(statusCode) + } response := map[string]string{ "error": strings.ReplaceAll(strings.ToLower(http.StatusText(statusCode)), " ", "_"), - "message": err.Error(), + "message": message, } RenderJSON(w, statusCode, response) diff --git a/httpserver/render_test.go b/httpserver/render_test.go new file mode 100644 index 0000000..71ecb94 --- /dev/null +++ b/httpserver/render_test.go @@ -0,0 +1,28 @@ +package httpserver + +import ( + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestRenderError(t *testing.T) { + t.Run("preserves client error", func(t *testing.T) { + response := httptest.NewRecorder() + RenderError(response, http.StatusBadRequest, errors.New("invalid email")) + var body map[string]string + require.NoError(t, json.NewDecoder(response.Body).Decode(&body)) + assert.Equal(t, "invalid email", body["message"]) + }) + + t.Run("hides internal detail", func(t *testing.T) { + response := httptest.NewRecorder() + RenderError(response, http.StatusInternalServerError, errors.New("password=secret")) + assert.NotContains(t, response.Body.String(), "secret") + }) +} diff --git a/httpserver/wrap_writter.go b/httpserver/wrap_writter.go index 60b107d..22ff12c 100644 --- a/httpserver/wrap_writter.go +++ b/httpserver/wrap_writter.go @@ -211,12 +211,12 @@ type httpFancyWriter struct { func (f *httpFancyWriter) Flush() { f.wroteHeader = true - fl := f.basicWriter.ResponseWriter.(http.Flusher) + fl := f.ResponseWriter.(http.Flusher) fl.Flush() } func (f *httpFancyWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) { - hj := f.basicWriter.ResponseWriter.(http.Hijacker) + hj := f.ResponseWriter.(http.Hijacker) return hj.Hijack() } @@ -225,14 +225,14 @@ func (f *http2FancyWriter) Push(target string, opts *http.PushOptions) error { } func (f *httpFancyWriter) ReadFrom(r io.Reader) (int64, error) { - if f.basicWriter.tee != nil { + if f.tee != nil { n, err := io.Copy(&f.basicWriter, r) return n, err } - rf := f.basicWriter.ResponseWriter.(io.ReaderFrom) - f.basicWriter.maybeWriteHeader() + rf := f.ResponseWriter.(io.ReaderFrom) + f.maybeWriteHeader() n, err := rf.ReadFrom(r) - f.basicWriter.bytes += int(n) + f.bytes += int(n) return n, err } @@ -251,7 +251,7 @@ type http2FancyWriter struct { func (f *http2FancyWriter) Flush() { f.wroteHeader = true - fl := f.basicWriter.ResponseWriter.(http.Flusher) + fl := f.ResponseWriter.(http.Flusher) fl.Flush() } diff --git a/internal/version/version.go b/internal/version/version.go index eca68e0..f60599c 100644 --- a/internal/version/version.go +++ b/internal/version/version.go @@ -15,26 +15,32 @@ // CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. package version -import "fmt" +import "runtime/debug" -type ( - V struct { - v uint - } -) - -func New(v uint) *V { - return &V{v} -} - -func (v V) Alpha(r uint) string { - return fmt.Sprintf("%d.alpha-%d", v.v, r) -} +const modulePath = "go.gearno.de/kit" -func (v V) Beta(r uint) string { - return fmt.Sprintf("%d.beta-%d", v.v, r) +func Current() string { + info, ok := debug.ReadBuildInfo() + if !ok { + return "devel" + } + if info.Main.Path == modulePath { + return normalizedVersion(info.Main) + } + for _, dep := range info.Deps { + if dep.Path == modulePath { + if dep.Replace != nil { + return normalizedVersion(*dep.Replace) + } + return normalizedVersion(*dep) + } + } + return "devel" } -func (v V) Stable(r uint) string { - return fmt.Sprintf("%d.stable-%d", v.v, r) +func normalizedVersion(module debug.Module) string { + if module.Version == "" || module.Version == "(devel)" { + return "devel" + } + return module.Version } diff --git a/internal/version/version_test.go b/internal/version/version_test.go new file mode 100644 index 0000000..5d4e79e --- /dev/null +++ b/internal/version/version_test.go @@ -0,0 +1,14 @@ +package version + +import ( + "runtime/debug" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestNormalizedVersion(t *testing.T) { + assert.Equal(t, "devel", normalizedVersion(debug.Module{})) + assert.Equal(t, "devel", normalizedVersion(debug.Module{Version: "(devel)"})) + assert.Equal(t, "v1.2.3", normalizedVersion(debug.Module{Version: "v1.2.3"})) +} diff --git a/log/buffer.go b/log/buffer.go index 91a6b42..b987ad0 100644 --- a/log/buffer.go +++ b/log/buffer.go @@ -16,5 +16,9 @@ func getBuffer() *bytes.Buffer { } func freeBuffer(bf *bytes.Buffer) { + if bf.Cap() > 64*1024 { + return + } + bf.Reset() bufPool.Put(bf) } diff --git a/log/log.go b/log/log.go index d590f79..7eb2624 100644 --- a/log/log.go +++ b/log/log.go @@ -53,8 +53,8 @@ type ( Format = string // Match is a function that determines whether a log message - // should be logged. Return true to log the message, false to - // skip it. + // should be skipped. Return true to skip the message and false + // to log it. Match func(level Level, msg string, attrs []Attr) bool ) @@ -107,8 +107,8 @@ func WithFormat(format Format) Option { } } -// WithMatch sets a match function that determines whether a log -// message should be logged. If the match returns false, the message +// SkipMatch sets a match function that determines whether a log +// message should be skipped. If the match returns true, the message // is skipped. func SkipMatch(match Match) Option { return func(l *Logger) { diff --git a/log/pretty_handler.go b/log/pretty_handler.go index 6c210eb..271ca4b 100644 --- a/log/pretty_handler.go +++ b/log/pretty_handler.go @@ -108,7 +108,7 @@ func (h *PrettyHandler) Handle(_ context.Context, r slog.Record) error { fmt.Fprint(bf, " ") for i, g := range h.groups { fmt.Fprint(bf, color.New(color.FgWhite).Sprint(g)) - if i != len(h.groups) { + if i != len(h.groups)-1 { fmt.Fprint(bf, color.New(color.FgWhite).Sprint(".")) } } diff --git a/log/pretty_handler_test.go b/log/pretty_handler_test.go index 87d40e6..f22e6a6 100644 --- a/log/pretty_handler_test.go +++ b/log/pretty_handler_test.go @@ -214,4 +214,6 @@ func TestBufferPool(t *testing.T) { buf2 := getBuffer() assert.NotNil(t, buf2) + assert.Empty(t, buf2.String()) + freeBuffer(buf2) } diff --git a/migrator/migrator.go b/migrator/migrator.go index fe2a879..5cd6b23 100644 --- a/migrator/migrator.go +++ b/migrator/migrator.go @@ -71,21 +71,15 @@ func (m *Migrator) Run(ctx context.Context, dirname string) error { return nil } - err := m.pg.WithAdvisoryLock( + err := m.pg.WithAdvisoryLockTx( ctx, MigrationAdvisoryLock, - func(conn pg.Querier) error { - err := m.pg.WithConn( - ctx, - func(connCtx context.Context, conn pg.Querier) error { - return createIfNotExistVersionsTable(connCtx, conn) - }, - ) - if err != nil { + func(txCtx context.Context, tx pg.Tx) error { + if err := createIfNotExistVersionsTable(txCtx, tx); err != nil { return fmt.Errorf("cannot create schema version table: %w", err) } - appliedVersions, err := loadSchemaVersions(ctx, conn) + appliedVersions, err := loadSchemaVersions(txCtx, tx) if err != nil { return fmt.Errorf("cannot load schema versions: %w", err) } @@ -97,10 +91,10 @@ func (m *Migrator) Run(ctx context.Context, dirname string) error { m.logger.Info("applying migration", log.String("version", migration.Version)) - err := m.pg.WithTx( - ctx, - func(txCtx context.Context, tx pg.Tx) error { - return migration.Apply(txCtx, tx) + err := tx.Savepoint( + txCtx, + func(savepointCtx context.Context, savepoint pg.Tx) error { + return migration.Apply(savepointCtx, savepoint) }, ) if err != nil { diff --git a/migrator/migrator_test.go b/migrator/migrator_test.go index 6dfd9d7..6e1d217 100644 --- a/migrator/migrator_test.go +++ b/migrator/migrator_test.go @@ -15,17 +15,19 @@ import ( "go.gearno.de/kit/pg" ) -func newTestPGClient(t *testing.T) *pg.Client { +func newTestPGClient(t *testing.T, extra ...pg.Option) *pg.Client { t.Helper() - client, err := pg.NewClient( + options := []pg.Option{ pg.WithAddr("localhost:5432"), pg.WithUser("kit"), pg.WithPassword("kit"), pg.WithDatabase("kit_test"), pg.WithLogger(log.NewLogger(log.WithOutput(io.Discard))), pg.WithRegisterer(prometheus.NewRegistry()), - ) + } + options = append(options, extra...) + client, err := pg.NewClient(options...) if err != nil { t.Skipf("skipping: cannot create PostgreSQL client: %v", err) } @@ -47,6 +49,21 @@ func newTestPGClient(t *testing.T) *pg.Client { return client } +func TestMigrator_RunWithSingleConnection(t *testing.T) { + client := newTestPGClient(t, pg.WithPoolSize(1), pg.WithMinPoolSize(1)) + dropTables(t, client, "schema_versions", "test_mig_single_connection") + t.Cleanup(func() { dropTables(t, client, "schema_versions", "test_mig_single_connection") }) + disk := fstest.MapFS{ + "migrations/001_create.sql": &fstest.MapFile{ + Data: []byte("CREATE TABLE test_mig_single_connection (id serial PRIMARY KEY);"), + }, + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + m := migrator.NewMigrator(client, disk, log.NewLogger(log.WithOutput(io.Discard))) + require.NoError(t, m.Run(ctx, "migrations")) +} + func dropTables(t *testing.T, client *pg.Client, tables ...string) { t.Helper() ctx := context.Background() diff --git a/pg/client.go b/pg/client.go index 807cae2..fb0e018 100644 --- a/pg/client.go +++ b/pg/client.go @@ -54,7 +54,9 @@ type ( database string applicationName string - debug bool + debug bool + traceQueryText bool + requireTLS bool poolSize int32 minPoolSize int32 @@ -376,6 +378,22 @@ func WithDebug() Option { } } +// WithTraceQueryText records complete SQL statements on database +// spans. It is disabled by default because statements can contain +// credentials or personal data. +func WithTraceQueryText() Option { + return func(c *Client) { + c.traceQueryText = true + } +} + +// WithRequireTLS rejects client configuration without verified TLS. +func WithRequireTLS() Option { + return func(c *Client) { + c.requireTLS = true + } +} + // NewClient creates a new database client with customizable options // for logging, tracing, TLS, and Prometheus metrics. // @@ -416,7 +434,24 @@ func NewClient(options ...Option) (*Client, error) { return nil, fmt.Errorf("invalid port: %w", err) } - config, _ := pgxpool.ParseConfig("") + if c.tlsConfig != nil && !c.tlsConfig.InsecureSkipVerify { + c.tlsConfig = c.tlsConfig.Clone() + c.tlsConfig.ServerName = host + } + if c.poolSize < 1 { + return nil, fmt.Errorf("pool size must be greater than zero: %d", c.poolSize) + } + if c.minPoolSize < 0 || c.minPoolSize > c.poolSize { + return nil, fmt.Errorf("minimum pool size must be between zero and pool size: %d", c.minPoolSize) + } + if c.requireTLS && (c.tlsConfig == nil || c.tlsConfig.InsecureSkipVerify) { + return nil, errors.New("verified TLS is required") + } + + config, err := pgxpool.ParseConfig("") + if err != nil { + return nil, fmt.Errorf("cannot create default connection pool config: %w", err) + } config.ConnConfig.Host = host config.ConnConfig.Port = uint16(port) config.ConnConfig.User = c.user @@ -456,12 +491,12 @@ func NewClient(options ...Option) (*Client, error) { c.tracer = c.tracerProvider.Tracer( tracerName, trace.WithInstrumentationVersion( - version.New(0).Alpha(1), + version.Current(), ), ) tracers := []pgx.QueryTracer{ - &tracer{c.tracer}, + &tracer{tracer: c.tracer, recordQueryText: c.traceQueryText}, } if c.debug { @@ -481,16 +516,18 @@ func NewClient(options ...Option) (*Client, error) { return nil, fmt.Errorf("cannot create connection pool from config: %w", err) } - c.registerer.MustRegister( - newCollector( - pool, - map[string]string{ - "database": c.database, - "user": c.user, - "addr": c.addr, - }, - ), + collector := newCollector( + pool, + map[string]string{ + "database": c.database, + "user": c.user, + "addr": c.addr, + }, ) + if err := c.registerer.Register(collector); err != nil { + pool.Close() + return nil, fmt.Errorf("cannot register connection pool metrics: %w", err) + } c.pool = pool @@ -670,6 +707,25 @@ func (c *Client) WithAdvisoryLock( ctx context.Context, id AdvisoryLock, f func(Querier) error, +) error { + return c.withAdvisoryLockTx(ctx, id, func(_ context.Context, tx Tx) error { + return f(tx) + }) +} + +// WithAdvisoryLockTx executes f in the transaction that owns the lock. +func (c *Client) WithAdvisoryLockTx( + ctx context.Context, + id AdvisoryLock, + f ExecFunc[Tx], +) error { + return c.withAdvisoryLockTx(ctx, id, f) +} + +func (c *Client) withAdvisoryLockTx( + ctx context.Context, + id AdvisoryLock, + f ExecFunc[Tx], ) error { var ( rootSpan = trace.SpanFromContext(ctx) @@ -703,7 +759,7 @@ func (c *Client) WithAdvisoryLock( return err } - err = f(tx) + err = f(txCtx, tx) if err != nil { if rootSpan.IsRecording() { span.SetStatus(codes.Error, err.Error()) @@ -720,12 +776,16 @@ func (c *Client) WithAdvisoryLock( func (c *Client) RefreshTypes(ctx context.Context) error { conns := c.pool.AcquireAllIdle(ctx) + var refreshErr error for _, conn := range conns { if err := conn.Conn().Close(ctx); err != nil { - return fmt.Errorf("cannot refresh postgresql type: %w", err) + refreshErr = errors.Join( + refreshErr, + fmt.Errorf("cannot refresh postgresql type: %w", err), + ) } conn.Release() } - return nil + return refreshErr } diff --git a/pg/client_test.go b/pg/client_test.go index 9446c83..daa0b4b 100644 --- a/pg/client_test.go +++ b/pg/client_test.go @@ -154,6 +154,33 @@ func TestNewClient(t *testing.T) { }, ) + t.Run("invalid pool sizes", func(t *testing.T) { + _, err := pg.NewClient(pg.WithPoolSize(0), pg.WithRegisterer(prometheus.NewRegistry())) + require.Error(t, err) + _, err = pg.NewClient( + pg.WithPoolSize(1), pg.WithMinPoolSize(2), + pg.WithRegisterer(prometheus.NewRegistry()), + ) + require.Error(t, err) + }) + + t.Run("require verified TLS", func(t *testing.T) { + _, err := pg.NewClient(pg.WithRequireTLS(), pg.WithRegisterer(prometheus.NewRegistry())) + require.Error(t, err) + _, err = pg.NewClient( + pg.WithRequireTLS(), pg.WithUnsecureTLS(), + pg.WithRegisterer(prometheus.NewRegistry()), + ) + require.Error(t, err) + client, err := pg.NewClient( + pg.WithRequireTLS(), + pg.WithTLS([]*x509.Certificate{selfSignedCert(t)}), + pg.WithRegisterer(prometheus.NewRegistry()), + ) + require.NoError(t, err) + client.Close() + }) + t.Run( "with TLS option", func(t *testing.T) { @@ -282,6 +309,18 @@ func TestNewClient(t *testing.T) { } }, ) + + t.Run("duplicate metrics return error", func(t *testing.T) { + reg := prometheus.NewRegistry() + first, err := pg.NewClient(pg.WithRegisterer(reg)) + require.NoError(t, err) + first.Close() + assert.NotPanics(t, func() { + second, err := pg.NewClient(pg.WithRegisterer(reg)) + require.Error(t, err) + assert.Nil(t, second) + }) + }) } // --------------------------------------------------------------------------- diff --git a/pg/tracer.go b/pg/tracer.go index 16b250b..b2db54b 100644 --- a/pg/tracer.go +++ b/pg/tracer.go @@ -33,7 +33,8 @@ import ( type ( tracer struct { - tracer trace.Tracer + tracer trace.Tracer + recordQueryText bool } ) @@ -60,7 +61,7 @@ const ( // SQLStateKey represents PostgreSQL error code, // see https://www.postgresql.org/docs/current/errcodes-appendix.html. - SQLStateKey = attribute.Key("db.response.status_code") + SQLStateKey = attribute.Key("db.postgresql.sql_state") ) func connectionConfigAttributes(config *pgx.ConnConfig) []trace.SpanStartOption { @@ -104,6 +105,14 @@ func recordError(span trace.Span, err error) { } } +func (t *tracer) queryAttributes(sql string) []attribute.KeyValue { + attrs := []attribute.KeyValue{semconv.DBOperationName(sqlOperationName(sql))} + if t.recordQueryText { + attrs = append(attrs, semconv.DBQueryText(sql)) + } + return attrs +} + func (t *tracer) TraceQueryStart( ctx context.Context, conn *pgx.Conn, @@ -113,13 +122,9 @@ func (t *tracer) TraceQueryStart( return ctx } - operationName := sqlOperationName(data.SQL) opts := []trace.SpanStartOption{ trace.WithSpanKind(trace.SpanKindClient), - trace.WithAttributes( - semconv.DBOperationName(operationName), - semconv.DBQueryText(data.SQL), - ), + trace.WithAttributes(t.queryAttributes(data.SQL)...), } if conn != nil { @@ -195,13 +200,9 @@ func (t *tracer) TraceBatchQuery( return } - operationName := sqlOperationName(data.SQL) opts := []trace.SpanStartOption{ trace.WithSpanKind(trace.SpanKindClient), - trace.WithAttributes( - semconv.DBOperationName(operationName), - semconv.DBQueryText(data.SQL), - ), + trace.WithAttributes(t.queryAttributes(data.SQL)...), } if conn != nil { @@ -288,13 +289,9 @@ func (t *tracer) TracePrepareStart( return ctx } - operationName := sqlOperationName(data.SQL) opts := []trace.SpanStartOption{ trace.WithSpanKind(trace.SpanKindClient), - trace.WithAttributes( - semconv.DBOperationName(operationName), - semconv.DBQueryText(data.SQL), - ), + trace.WithAttributes(t.queryAttributes(data.SQL)...), } if conn != nil { diff --git a/pg/tracer_test.go b/pg/tracer_test.go new file mode 100644 index 0000000..feab690 --- /dev/null +++ b/pg/tracer_test.go @@ -0,0 +1,19 @@ +package pg + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestQueryAttributesRedactSQLByDefault(t *testing.T) { + attrs := (&tracer{}).queryAttributes("SELECT 'secret'") + assert.Len(t, attrs, 1) + assert.Equal(t, "SELECT", attrs[0].Value.AsString()) +} + +func TestQueryAttributesCanIncludeSQL(t *testing.T) { + attrs := (&tracer{recordQueryText: true}).queryAttributes("SELECT 'secret'") + assert.Len(t, attrs, 2) + assert.Equal(t, "SELECT 'secret'", attrs[1].Value.AsString()) +} diff --git a/unit/otel_error_handler.go b/unit/otel_error_handler.go index 6127e84..9b0a23a 100644 --- a/unit/otel_error_handler.go +++ b/unit/otel_error_handler.go @@ -25,12 +25,11 @@ import ( type ( otelErrorHandler struct { logger *log.Logger - ctx context.Context } ) func (h *otelErrorHandler) Handle(err error) { if err != nil { - h.logger.ErrorCtx(h.ctx, "open telemetry error", log.Error(err)) + h.logger.ErrorCtx(context.Background(), "open telemetry error", log.Error(err)) } } diff --git a/unit/unit.go b/unit/unit.go index b4c799d..a098e7f 100644 --- a/unit/unit.go +++ b/unit/unit.go @@ -28,6 +28,7 @@ import ( "net/http" "os" "os/signal" + "strings" "sync" "syscall" "time" @@ -66,8 +67,9 @@ type ( } Config struct { - Metrics MetricsConfig `json:"metrics"` - Tracing TracingConfig `json:"tracing"` + Metrics MetricsConfig `json:"metrics"` + Tracing TracingConfig `json:"tracing"` + ShutdownTimeout int `json:"shutdown-timeout"` } MetricsConfig struct { @@ -75,12 +77,14 @@ type ( } TracingConfig struct { - Addr string `json:"addr"` - MaxBatchSize int `json:"max-batch-size"` - BatchTimeout int `json:"batch-timeout"` - ExportTimeout int `json:"export-timeout"` - MaxQueueSize int `json:"max-queue-size"` - SamplingRatio float64 `json:"sampling-ratio"` + Addr string `json:"addr"` + Insecure bool `json:"insecure"` + Headers map[string]string `json:"headers,omitempty"` + MaxBatchSize int `json:"max-batch-size"` + BatchTimeout int `json:"batch-timeout"` + ExportTimeout int `json:"export-timeout"` + MaxQueueSize int `json:"max-queue-size"` + SamplingRatio float64 `json:"sampling-ratio"` } ) @@ -91,16 +95,17 @@ func NewUnit(main Runnable, name, version, environment string) *Unit { environment: environment, main: main, config: &Config{ + ShutdownTimeout: 30, Metrics: MetricsConfig{ - Addr: ":9090", + Addr: "127.0.0.1:9090", }, Tracing: TracingConfig{ - Addr: ":4317", + Addr: "localhost:4318", MaxBatchSize: 1024, BatchTimeout: 10, ExportTimeout: 15, MaxQueueSize: 5000, - SamplingRatio: 1.0, + SamplingRatio: 0.1, }, }, } @@ -135,16 +140,29 @@ func (u *Unit) RunContext(parentCtx context.Context) error { } } + if err := u.config.Validate(); err != nil { + return fmt.Errorf("invalid unit configuration: %w", err) + } + switch log.Format(*format) { + case log.FormatJSON, log.FormatPretty, log.FormatText: + default: + return fmt.Errorf("unsupported log format %q", *format) + } + if *printCfg { config := map[string]any{"unit": u.config} if configurable, ok := u.main.(Configurable); ok { config[u.name] = configurable.GetConfiguration() } + redacted, err := redactConfiguration(config) + if err != nil { + return fmt.Errorf("cannot redact configuration: %w", err) + } encoder := json.NewEncoder(os.Stdout) encoder.SetIndent("", "\t") - if err := encoder.Encode(config); err != nil { + if err := encoder.Encode(redacted); err != nil { return fmt.Errorf("cannot encode configuration: %w", err) } @@ -165,8 +183,8 @@ func (u *Unit) RunContext(parentCtx context.Context) error { defer cancel(context.Canceled) wg := sync.WaitGroup{} - metricsInitialized := make(chan prometheus.Registerer) - tracingInitialized := make(chan trace.TracerProvider) + metricsInitialized := make(chan prometheus.Registerer, 1) + tracingInitialized := make(chan trace.TracerProvider, 1) metricsServerCtx, stopMetricsServer := context.WithCancel(context.Background()) defer stopMetricsServer() @@ -223,10 +241,20 @@ func (u *Unit) RunContext(parentCtx context.Context) error { <-signalCtx.Done() + cancel(context.Canceled) stopMetricsServer() stopTracingExporter() - wg.Wait() + shutdownComplete := make(chan struct{}) + go func() { + wg.Wait() + close(shutdownComplete) + }() + select { + case <-shutdownComplete: + case <-time.After(time.Duration(u.config.ShutdownTimeout) * time.Second): + return fmt.Errorf("shutdown timed out after %d seconds", u.config.ShutdownTimeout) + } if cause := context.Cause(runCtx); cause != nil { return cause @@ -306,12 +334,11 @@ func (u *Unit) runTracingExporter(ctx context.Context, initialized chan<- trace. logger.InfoCtx(ctx, "starting traces exporter", log.String("addr", config.Addr)) - otel.SetErrorHandler(&otelErrorHandler{logger: logger, ctx: ctx}) + otel.SetErrorHandler(&otelErrorHandler{logger: logger}) - exporter := otlptracehttp.NewUnstarted( + exporterOptions := []otlptracehttp.Option{ otlptracehttp.WithCompression(otlptracehttp.GzipCompression), otlptracehttp.WithEndpoint(config.Addr), - otlptracehttp.WithInsecure(), otlptracehttp.WithRetry( otlptracehttp.RetryConfig{ Enabled: true, @@ -320,8 +347,15 @@ func (u *Unit) runTracingExporter(ctx context.Context, initialized chan<- trace. MaxElapsedTime: 5 * time.Minute, }, ), - otlptracehttp.WithTimeout(15*time.Second), - ) + otlptracehttp.WithTimeout(15 * time.Second), + } + if config.Insecure { + exporterOptions = append(exporterOptions, otlptracehttp.WithInsecure()) + } + if len(config.Headers) > 0 { + exporterOptions = append(exporterOptions, otlptracehttp.WithHeaders(config.Headers)) + } + exporter := otlptracehttp.NewUnstarted(exporterOptions...) if err := exporter.Start(ctx); err != nil { return fmt.Errorf("cannot create otel exporter: %w", err) @@ -411,7 +445,10 @@ func (u *Unit) loadConfigurationFromFile(filename string) error { } if _, ok := config["unit"]; ok { - encoded, _ := json.Marshal(config["unit"]) + encoded, err := json.Marshal(config["unit"]) + if err != nil { + return fmt.Errorf("cannot encode %q config section: %w", "unit", err) + } if err := json.Unmarshal(encoded, u.config); err != nil { return fmt.Errorf("cannot decode %q config section: %w", "unit", err) } @@ -419,7 +456,10 @@ func (u *Unit) loadConfigurationFromFile(filename string) error { if configurable, ok := u.main.(Configurable); ok { if _, ok := config[u.name]; ok { - encoded, _ := json.Marshal(config[u.name]) + encoded, err := json.Marshal(config[u.name]) + if err != nil { + return fmt.Errorf("cannot encode %q config section: %w", u.name, err) + } if err := json.Unmarshal(encoded, configurable.GetConfiguration()); err != nil { return fmt.Errorf("cannot decode %q config section: %w", u.name, err) } @@ -428,3 +468,71 @@ func (u *Unit) loadConfigurationFromFile(filename string) error { return nil } + +func (c *Config) Validate() error { + if c == nil { + return errors.New("configuration is nil") + } + if c.ShutdownTimeout < 1 { + return errors.New("shutdown-timeout must be greater than zero") + } + if _, _, err := net.SplitHostPort(c.Metrics.Addr); err != nil { + return fmt.Errorf("invalid metrics address %q: %w", c.Metrics.Addr, err) + } + if _, _, err := net.SplitHostPort(c.Tracing.Addr); err != nil { + return fmt.Errorf("invalid tracing address %q: %w", c.Tracing.Addr, err) + } + if c.Tracing.MaxBatchSize < 1 { + return errors.New("tracing max-batch-size must be greater than zero") + } + if c.Tracing.MaxQueueSize < c.Tracing.MaxBatchSize { + return errors.New("tracing max-queue-size must be greater than or equal to max-batch-size") + } + if c.Tracing.BatchTimeout < 1 || c.Tracing.ExportTimeout < 1 { + return errors.New("tracing timeouts must be greater than zero") + } + if c.Tracing.SamplingRatio < 0 || c.Tracing.SamplingRatio > 1 { + return errors.New("tracing sampling-ratio must be between zero and one") + } + return nil +} + +func redactConfiguration(config any) (any, error) { + encoded, err := json.Marshal(config) + if err != nil { + return nil, err + } + var normalized any + if err := json.Unmarshal(encoded, &normalized); err != nil { + return nil, err + } + redactValue(normalized) + return normalized, nil +} + +func redactValue(value any) { + switch value := value.(type) { + case map[string]any: + for key, child := range value { + if sensitiveConfigKey(key) { + value[key] = "[REDACTED]" + } else { + redactValue(child) + } + } + case []any: + for _, child := range value { + redactValue(child) + } + } +} + +func sensitiveConfigKey(key string) bool { + key = strings.ToLower(key) + for _, marker := range []string{"authorization", "cookie", "credential", "password", "private-key", "secret", "token"} { + if strings.Contains(key, marker) { + return true + } + } + return false +} diff --git a/unit/unit_test.go b/unit/unit_test.go new file mode 100644 index 0000000..1219f1f --- /dev/null +++ b/unit/unit_test.go @@ -0,0 +1,44 @@ +package unit + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func validConfig() *Config { + return &Config{ + ShutdownTimeout: 30, + Metrics: MetricsConfig{Addr: "127.0.0.1:9090"}, + Tracing: TracingConfig{ + Addr: "collector.example.com:4318", MaxBatchSize: 512, + BatchTimeout: 5, ExportTimeout: 10, MaxQueueSize: 2048, + SamplingRatio: 0.1, + }, + } +} + +func TestConfigValidate(t *testing.T) { + require.NoError(t, validConfig().Validate()) + config := validConfig() + config.Tracing.SamplingRatio = 2 + assert.Error(t, config.Validate()) +} + +func TestNewUnitUsesProductionSafeDefaults(t *testing.T) { + u := NewUnit(nil, "test", "v1", "test") + require.NoError(t, u.config.Validate()) + assert.Equal(t, "127.0.0.1:9090", u.config.Metrics.Addr) + assert.False(t, u.config.Tracing.Insecure) +} + +func TestRedactConfiguration(t *testing.T) { + redacted, err := redactConfiguration(map[string]any{ + "database": map[string]any{"password": "secret", "host": "db"}, + }) + require.NoError(t, err) + database := redacted.(map[string]any)["database"].(map[string]any) + assert.Equal(t, "[REDACTED]", database["password"]) + assert.Equal(t, "db", database["host"]) +} diff --git a/worker/periodic.go b/worker/periodic.go index d666eee..c52c18f 100644 --- a/worker/periodic.go +++ b/worker/periodic.go @@ -18,6 +18,8 @@ package worker import ( "context" + "fmt" + "runtime" "sync" "time" @@ -142,7 +144,7 @@ func NewPeriodic(name string, handler PeriodicHandler, logger *log.Logger, opts tracer: o.tracerProvider.Tracer( tracerName, trace.WithInstrumentationVersion( - version.New(0).Alpha(1), + version.Current(), ), ), interval: o.interval, @@ -188,10 +190,8 @@ func (w *PeriodicWorker) Run(ctx context.Context) error { } func (w *PeriodicWorker) run(ctx context.Context) { - nonCancelableCtx := context.WithoutCancel(ctx) - runCtx, span := w.tracer.Start( - nonCancelableCtx, + ctx, "periodic_worker.run", trace.WithSpanKind(trace.SpanKindInternal), trace.WithAttributes( @@ -201,10 +201,30 @@ func (w *PeriodicWorker) run(ctx context.Context) { defer span.End() start := time.Now() + status := "succeeded" + defer func() { + duration := time.Since(start) + if rvr := recover(); rvr != nil { + status = "panicked" + err := fmt.Errorf("periodic worker panicked: %v", rvr) + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) + stack := make([]byte, 4096) + length := runtime.Stack(stack, false) + w.logger.ErrorCtx( + context.WithoutCancel(runCtx), + "run panicked", + log.Any("panic", rvr), + log.String("stacktrace", string(stack[:length])), + log.Duration("duration", duration), + ) + } + w.runsTotal.WithLabelValues(w.name, status).Inc() + w.runDuration.WithLabelValues(w.name, status).Observe(duration.Seconds()) + }() + err := w.handler.Run(runCtx) duration := time.Since(start) - - status := "succeeded" if err != nil { status = "failed" span.RecordError(err) @@ -222,7 +242,4 @@ func (w *PeriodicWorker) run(ctx context.Context) { log.Duration("duration", duration), ) } - - w.runsTotal.WithLabelValues(w.name, status).Inc() - w.runDuration.WithLabelValues(w.name, status).Observe(duration.Seconds()) } diff --git a/worker/periodic_test.go b/worker/periodic_test.go index 86bdd35..cda2e57 100644 --- a/worker/periodic_test.go +++ b/worker/periodic_test.go @@ -169,3 +169,36 @@ func TestPeriodicWorkerContinuesAfterError(t *testing.T) { cancel() require.ErrorIs(t, <-done, context.Canceled) } + +func TestPeriodicWorkerContinuesAfterPanic(t *testing.T) { + var calls atomic.Int64 + h := &periodicHandler{runFn: func(context.Context) error { + if calls.Add(1) == 1 { + panic("transient panic") + } + return nil + }} + w := worker.NewPeriodic("test", h, testLogger(), testWorkerOpts()...) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- w.Run(ctx) }() + require.Eventually(t, func() bool { return calls.Load() >= 2 }, 2*time.Second, 5*time.Millisecond) + cancel() + require.ErrorIs(t, <-done, context.Canceled) +} + +func TestPeriodicWorkerCancelsInFlightRun(t *testing.T) { + started := make(chan struct{}) + h := &periodicHandler{runFn: func(ctx context.Context) error { + close(started) + <-ctx.Done() + return context.Cause(ctx) + }} + w := worker.NewPeriodic("test", h, testLogger(), testWorkerOpts()...) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- w.Run(ctx) }() + <-started + cancel() + require.ErrorIs(t, <-done, context.Canceled) +} diff --git a/worker/worker.go b/worker/worker.go index 67fcc10..533c2a7 100644 --- a/worker/worker.go +++ b/worker/worker.go @@ -249,7 +249,7 @@ func New[T any](name string, handler Handler[T], logger *log.Logger, opts ...Opt tracer: o.tracerProvider.Tracer( tracerName, trace.WithInstrumentationVersion( - version.New(0).Alpha(1), + version.Current(), ), ), interval: o.interval, @@ -282,16 +282,18 @@ func (w *Worker[T]) Run(ctx context.Context) error { case <-ticker.C: w.pollCyclesTotal.WithLabelValues(w.name).Inc() - nonCancelableCtx := context.WithoutCancel(ctx) if sr, ok := w.handler.(StaleRecoverer); ok { - w.recoverStaleTasks(nonCancelableCtx, sr) + w.recoverStaleTasks(ctx, sr) } for { + if ctx.Err() != nil { + return context.Cause(ctx) + } if err := w.processNext(ctx, sem, &wg); err != nil { if !errors.Is(err, ErrNoTask) { w.claimErrorsTotal.WithLabelValues(w.name).Inc() - w.logger.ErrorCtx(nonCancelableCtx, "cannot claim task", log.Error(err)) + w.logger.ErrorCtx(context.WithoutCancel(ctx), "cannot claim task", log.Error(err)) } break } @@ -319,7 +321,7 @@ func (w *Worker[T]) processNext(ctx context.Context, sem chan struct{}, wg *sync }() claimStart := time.Now() - task, err := w.handler.Claim(nonCancelableCtx) + task, err := w.handler.Claim(ctx) w.claimDuration.WithLabelValues(w.name).Observe(time.Since(claimStart).Seconds()) if err != nil { <-sem diff --git a/worker/worker_test.go b/worker/worker_test.go index 712699a..daacee5 100644 --- a/worker/worker_test.go +++ b/worker/worker_test.go @@ -252,10 +252,10 @@ func TestWorkerContinuesAfterClaimPanic(t *testing.T) { h := &testHandler{ claimFn: func(ctx context.Context) (int, error) { n := claimCalls.Add(1) - switch { - case n == 1: + switch n { + case 1: panic("database driver bug") - case n == 2: + case 2: return 42, nil default: return 0, worker.ErrNoTask @@ -327,10 +327,10 @@ func TestWorkerContinuesAfterClaimError(t *testing.T) { h := &testHandler{ claimFn: func(ctx context.Context) (int, error) { n := claimCalls.Add(1) - switch { - case n == 1: + switch n { + case 1: return 0, errors.New("database connection lost") - case n == 2: + case 2: return 42, nil default: return 0, worker.ErrNoTask