feat(api): serve buoy telemetry
This commit is contained in:
parent
af050c9751
commit
6c1329c2a9
11 changed files with 1193 additions and 0 deletions
11
apps/api/context.go
Normal file
11
apps/api/context.go
Normal file
|
|
@ -0,0 +1,11 @@
|
||||||
|
package main
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"net/http"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
func contextWithTimeout(r *http.Request, timeout time.Duration) (context.Context, context.CancelFunc) {
|
||||||
|
return context.WithTimeout(r.Context(), timeout)
|
||||||
|
}
|
||||||
45
apps/api/go.mod
Normal file
45
apps/api/go.mod
Normal file
|
|
@ -0,0 +1,45 @@
|
||||||
|
module git.fiwlabs.dev/fiwdev/nereus/apps/api
|
||||||
|
|
||||||
|
go 1.26
|
||||||
|
|
||||||
|
require (
|
||||||
|
github.com/go-chi/chi/v5 v5.2.3
|
||||||
|
github.com/jackc/pgx/v5 v5.10.0
|
||||||
|
github.com/prometheus/client_golang v1.23.2
|
||||||
|
go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.68.0
|
||||||
|
go.opentelemetry.io/otel v1.43.0
|
||||||
|
go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.43.0
|
||||||
|
go.opentelemetry.io/otel/sdk v1.43.0
|
||||||
|
go.opentelemetry.io/otel/trace v1.43.0
|
||||||
|
)
|
||||||
|
|
||||||
|
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/felixge/httpsnoop v1.0.4 // indirect
|
||||||
|
github.com/go-logr/logr v1.4.3 // 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/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
|
||||||
|
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
|
||||||
|
github.com/prometheus/client_model v0.6.2 // indirect
|
||||||
|
github.com/prometheus/common v0.66.1 // indirect
|
||||||
|
github.com/prometheus/procfs v0.16.1 // indirect
|
||||||
|
go.opentelemetry.io/auto/sdk v1.2.1 // indirect
|
||||||
|
go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.43.0 // indirect
|
||||||
|
go.opentelemetry.io/otel/metric v1.43.0 // indirect
|
||||||
|
go.opentelemetry.io/proto/otlp v1.10.0 // indirect
|
||||||
|
go.yaml.in/yaml/v2 v2.4.2 // indirect
|
||||||
|
golang.org/x/net v0.53.0 // indirect
|
||||||
|
golang.org/x/sync v0.21.0 // indirect
|
||||||
|
golang.org/x/sys v0.43.0 // indirect
|
||||||
|
golang.org/x/text v0.39.0 // indirect
|
||||||
|
google.golang.org/genproto/googleapis/api v0.0.0-20260414002931-afd174a4e478 // indirect
|
||||||
|
google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478 // indirect
|
||||||
|
google.golang.org/grpc v1.82.1 // indirect
|
||||||
|
google.golang.org/protobuf v1.36.11 // indirect
|
||||||
|
)
|
||||||
109
apps/api/go.sum
Normal file
109
apps/api/go.sum
Normal file
|
|
@ -0,0 +1,109 @@
|
||||||
|
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/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM=
|
||||||
|
github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw=
|
||||||
|
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/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg=
|
||||||
|
github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
|
||||||
|
github.com/go-chi/chi/v5 v5.2.3 h1:WQIt9uxdsAbgIYgid+BpYc+liqQZGMHRaUwp0JUcvdE=
|
||||||
|
github.com/go-chi/chi/v5 v5.2.3/go.mod h1:L2yAIGWB3H+phAw1NxKwWM+7eUH/lU8pOMm5hHcoops=
|
||||||
|
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/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=
|
||||||
|
github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
|
||||||
|
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.28.0 h1:HWRh5R2+9EifMyIHV7ZV+MIZqgz+PMpZ14Jynv3O2Zs=
|
||||||
|
github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0/go.mod h1:JfhWUomR1baixubs02l85lZYYOm7LV6om4ceouMv45c=
|
||||||
|
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.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0=
|
||||||
|
github.com/jackc/pgx/v5 v5.10.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.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
|
||||||
|
github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
|
||||||
|
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/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=
|
||||||
|
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.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9ZoGs=
|
||||||
|
github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA=
|
||||||
|
github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzMyRg=
|
||||||
|
github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is=
|
||||||
|
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/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=
|
||||||
|
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/contrib/instrumentation/net/http/otelhttp v0.68.0 h1:CqXxU8VOmDefoh0+ztfGaymYbhdB/tT3zs79QaZTNGY=
|
||||||
|
go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.68.0/go.mod h1:BuhAPThV8PBHBvg8ZzZ/Ok3idOdhWIodywz2xEcRbJo=
|
||||||
|
go.opentelemetry.io/otel v1.43.0 h1:mYIM03dnh5zfN7HautFE4ieIig9amkNANT+xcVxAj9I=
|
||||||
|
go.opentelemetry.io/otel v1.43.0/go.mod h1:JuG+u74mvjvcm8vj8pI5XiHy1zDeoCS2LB1spIq7Ay0=
|
||||||
|
go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.43.0 h1:88Y4s2C8oTui1LGM6bTWkw0ICGcOLCAI5l6zsD1j20k=
|
||||||
|
go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.43.0/go.mod h1:Vl1/iaggsuRlrHf/hfPJPvVag77kKyvrLeD10kpMl+A=
|
||||||
|
go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.43.0 h1:RAE+JPfvEmvy+0LzyUA25/SGawPwIUbZ6u0Wug54sLc=
|
||||||
|
go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.43.0/go.mod h1:AGmbycVGEsRx9mXMZ75CsOyhSP6MFIcj/6dnG+vhVjk=
|
||||||
|
go.opentelemetry.io/otel/metric v1.43.0 h1:d7638QeInOnuwOONPp4JAOGfbCEpYb+K6DVWvdxGzgM=
|
||||||
|
go.opentelemetry.io/otel/metric v1.43.0/go.mod h1:RDnPtIxvqlgO8GRW18W6Z/4P462ldprJtfxHxyKd2PY=
|
||||||
|
go.opentelemetry.io/otel/sdk v1.43.0 h1:pi5mE86i5rTeLXqoF/hhiBtUNcrAGHLKQdhg4h4V9Dg=
|
||||||
|
go.opentelemetry.io/otel/sdk v1.43.0/go.mod h1:P+IkVU3iWukmiit/Yf9AWvpyRDlUeBaRg6Y+C58QHzg=
|
||||||
|
go.opentelemetry.io/otel/sdk/metric v1.43.0 h1:S88dyqXjJkuBNLeMcVPRFXpRw2fuwdvfCGLEo89fDkw=
|
||||||
|
go.opentelemetry.io/otel/sdk/metric v1.43.0/go.mod h1:C/RJtwSEJ5hzTiUz5pXF1kILHStzb9zFlIEe85bhj6A=
|
||||||
|
go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09nk+3A=
|
||||||
|
go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0=
|
||||||
|
go.opentelemetry.io/proto/otlp v1.10.0 h1:IQRWgT5srOCYfiWnpqUYz9CVmbO8bFmKcwYxpuCSL2g=
|
||||||
|
go.opentelemetry.io/proto/otlp v1.10.0/go.mod h1:/CV4QoCR/S9yaPj8utp3lvQPoqMtxXdzn7ozvvozVqk=
|
||||||
|
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.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI=
|
||||||
|
go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU=
|
||||||
|
golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA=
|
||||||
|
golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs=
|
||||||
|
golang.org/x/sync v0.21.0 h1:HLII4xRRTtCRkxYp4HNFF0Js/Og6q2i++KXbg0gHCwM=
|
||||||
|
golang.org/x/sync v0.21.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
|
||||||
|
golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI=
|
||||||
|
golang.org/x/sys v0.43.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
|
||||||
|
golang.org/x/text v0.39.0 h1:UbZz4pLOvn600D6Oh6GGEI6VAmndrEBLv8/6BEXzyus=
|
||||||
|
golang.org/x/text v0.39.0/go.mod h1:3UwRclnC2g0TU9x8PZiyfOajCd1zaUNHF9cvqcQZ+ZM=
|
||||||
|
gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4=
|
||||||
|
gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E=
|
||||||
|
google.golang.org/genproto/googleapis/api v0.0.0-20260414002931-afd174a4e478 h1:yQugLulqltosq0B/f8l4w9VryjV+N/5gcW0jQ3N8Qec=
|
||||||
|
google.golang.org/genproto/googleapis/api v0.0.0-20260414002931-afd174a4e478/go.mod h1:C6ADNqOxbgdUUeRTU+LCHDPB9ttAMCTff6auwCVa4uc=
|
||||||
|
google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478 h1:RmoJA1ujG+/lRGNfUnOMfhCy5EipVMyvUE+KNbPbTlw=
|
||||||
|
google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8=
|
||||||
|
google.golang.org/grpc v1.82.1 h1:NnAxzGRA0677vCa4BUkOAnO5+FfQqVl9iUXeD0IqcGE=
|
||||||
|
google.golang.org/grpc v1.82.1/go.mod h1:yzTZ1TB1Z3SG+LIYaI+WiE8D5+PZ3ArnrSp8zF3+/ZA=
|
||||||
|
google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
|
||||||
|
google.golang.org/protobuf v1.36.11/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=
|
||||||
341
apps/api/http.go
Normal file
341
apps/api/http.go
Normal file
|
|
@ -0,0 +1,341 @@
|
||||||
|
package main
|
||||||
|
|
||||||
|
import (
|
||||||
|
"crypto/rand"
|
||||||
|
"encoding/hex"
|
||||||
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"io"
|
||||||
|
"log/slog"
|
||||||
|
"math"
|
||||||
|
mathrand "math/rand/v2"
|
||||||
|
"net/http"
|
||||||
|
"strconv"
|
||||||
|
"strings"
|
||||||
|
"sync/atomic"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/go-chi/chi/v5"
|
||||||
|
"github.com/jackc/pgx/v5"
|
||||||
|
"github.com/prometheus/client_golang/prometheus"
|
||||||
|
"github.com/prometheus/client_golang/prometheus/promhttp"
|
||||||
|
"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
|
||||||
|
"go.opentelemetry.io/otel/trace"
|
||||||
|
)
|
||||||
|
|
||||||
|
type api struct {
|
||||||
|
store store
|
||||||
|
metrics *metrics
|
||||||
|
logger *slog.Logger
|
||||||
|
version string
|
||||||
|
chaosRate float64
|
||||||
|
migrated *atomic.Bool
|
||||||
|
prometheus prometheus.Gatherer
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *api) routes() http.Handler {
|
||||||
|
r := chi.NewRouter()
|
||||||
|
r.Use(a.versionHeader)
|
||||||
|
r.Use(a.observeRequest)
|
||||||
|
r.Get("/healthz", func(w http.ResponseWriter, _ *http.Request) {
|
||||||
|
writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
|
||||||
|
})
|
||||||
|
r.Get("/readyz", a.ready)
|
||||||
|
r.Handle("/metrics", promhttp.HandlerFor(a.prometheus, promhttp.HandlerOpts{}))
|
||||||
|
r.Route("/api/v1", func(r chi.Router) {
|
||||||
|
r.Use(a.chaos)
|
||||||
|
r.Get("/buoys", a.listBuoys)
|
||||||
|
r.Post("/buoys", a.createBuoy)
|
||||||
|
r.Get("/buoys/{id}", a.getBuoy)
|
||||||
|
r.Delete("/buoys/{id}", a.deleteBuoy)
|
||||||
|
r.Get("/readings", a.listReadings)
|
||||||
|
r.Post("/readings", a.createReading)
|
||||||
|
r.Get("/readings/aggregate", a.aggregateReadings)
|
||||||
|
})
|
||||||
|
return otelhttp.NewHandler(r, "http.request")
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *api) versionHeader(next http.Handler) http.Handler {
|
||||||
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
w.Header().Set("X-Nereus-Version", a.version)
|
||||||
|
next.ServeHTTP(w, r)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
type responseRecorder struct {
|
||||||
|
http.ResponseWriter
|
||||||
|
status int
|
||||||
|
}
|
||||||
|
|
||||||
|
func (w *responseRecorder) WriteHeader(status int) {
|
||||||
|
w.status = status
|
||||||
|
w.ResponseWriter.WriteHeader(status)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *api) observeRequest(next http.Handler) http.Handler {
|
||||||
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
started := time.Now()
|
||||||
|
recorder := &responseRecorder{ResponseWriter: w, status: http.StatusOK}
|
||||||
|
next.ServeHTTP(recorder, r)
|
||||||
|
path := chi.RouteContext(r.Context()).RoutePattern()
|
||||||
|
if path == "" {
|
||||||
|
path = "unmatched"
|
||||||
|
}
|
||||||
|
status := strconv.Itoa(recorder.status)
|
||||||
|
a.metrics.requests.WithLabelValues(r.Method, path, status, a.version).Inc()
|
||||||
|
a.metrics.duration.WithLabelValues(r.Method, path, a.version).Observe(time.Since(started).Seconds())
|
||||||
|
span := trace.SpanFromContext(r.Context()).SpanContext()
|
||||||
|
a.logger.InfoContext(r.Context(), "request", "method", r.Method, "path", path, "status", recorder.status, "duration_ms", time.Since(started).Milliseconds(), "trace_id", span.TraceID().String())
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *api) chaos(next http.Handler) http.Handler {
|
||||||
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
if a.chaosRate > 0 && mathrand.Float64() < a.chaosRate {
|
||||||
|
writeError(w, http.StatusInternalServerError, "injected failure")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
next.ServeHTTP(w, r)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *api) ready(w http.ResponseWriter, r *http.Request) {
|
||||||
|
ctx, cancel := contextWithTimeout(r, 2*time.Second)
|
||||||
|
defer cancel()
|
||||||
|
if !a.migrated.Load() || a.store.Ping(ctx) != nil {
|
||||||
|
writeError(w, http.StatusServiceUnavailable, "database unavailable")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
writeJSON(w, http.StatusOK, map[string]string{"status": "ready"})
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *api) listBuoys(w http.ResponseWriter, r *http.Request) {
|
||||||
|
items, err := a.store.ListBuoys(r.Context())
|
||||||
|
if err != nil {
|
||||||
|
a.internalError(w, r, err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if items == nil {
|
||||||
|
items = []buoy{}
|
||||||
|
}
|
||||||
|
a.metrics.active.Set(float64(len(items)))
|
||||||
|
writeJSON(w, http.StatusOK, items)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *api) createBuoy(w http.ResponseWriter, r *http.Request) {
|
||||||
|
var input buoy
|
||||||
|
if !decodeJSON(w, r, &input) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
input.Name = strings.TrimSpace(input.Name)
|
||||||
|
if input.Name == "" || len(input.Name) > 120 || input.Latitude < -90 || input.Latitude > 90 || input.Longitude < -180 || input.Longitude > 180 {
|
||||||
|
writeError(w, http.StatusBadRequest, "invalid buoy")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
id, err := newUUID()
|
||||||
|
if err != nil {
|
||||||
|
a.internalError(w, r, err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
input.ID = id
|
||||||
|
created, err := a.store.CreateBuoy(r.Context(), input)
|
||||||
|
if err != nil {
|
||||||
|
a.internalError(w, r, err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
a.metrics.active.Inc()
|
||||||
|
writeJSON(w, http.StatusCreated, created)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *api) getBuoy(w http.ResponseWriter, r *http.Request) {
|
||||||
|
id := chi.URLParam(r, "id")
|
||||||
|
if !validUUID(id) {
|
||||||
|
writeError(w, http.StatusBadRequest, "invalid buoy ID")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
item, err := a.store.GetBuoy(r.Context(), id)
|
||||||
|
if errors.Is(err, pgx.ErrNoRows) {
|
||||||
|
writeError(w, http.StatusNotFound, "buoy not found")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
a.internalError(w, r, err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
writeJSON(w, http.StatusOK, item)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *api) deleteBuoy(w http.ResponseWriter, r *http.Request) {
|
||||||
|
id := chi.URLParam(r, "id")
|
||||||
|
if !validUUID(id) {
|
||||||
|
writeError(w, http.StatusBadRequest, "invalid buoy ID")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
err := a.store.DeleteBuoy(r.Context(), id)
|
||||||
|
if errors.Is(err, pgx.ErrNoRows) {
|
||||||
|
writeError(w, http.StatusNotFound, "buoy not found")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
a.internalError(w, r, err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
a.metrics.active.Dec()
|
||||||
|
w.WriteHeader(http.StatusNoContent)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *api) listReadings(w http.ResponseWriter, r *http.Request) {
|
||||||
|
query := r.URL.Query()
|
||||||
|
from, ok := optionalTime(w, query.Get("from"))
|
||||||
|
if !ok {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
to, ok := optionalTime(w, query.Get("to"))
|
||||||
|
if !ok {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if from != nil && to != nil && from.After(*to) {
|
||||||
|
writeError(w, http.StatusBadRequest, "from must not be after to")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if id := query.Get("buoy_id"); id != "" && !validUUID(id) {
|
||||||
|
writeError(w, http.StatusBadRequest, "invalid buoy ID")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
limit, ok := boundedInt(query.Get("limit"), 100, 1, 1000)
|
||||||
|
if !ok {
|
||||||
|
writeError(w, http.StatusBadRequest, "invalid limit")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
offset, ok := boundedInt(query.Get("offset"), 0, 0, math.MaxInt)
|
||||||
|
if !ok {
|
||||||
|
writeError(w, http.StatusBadRequest, "invalid offset")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
items, err := a.store.ListReadings(r.Context(), query.Get("buoy_id"), from, to, limit, offset)
|
||||||
|
if err != nil {
|
||||||
|
a.internalError(w, r, err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if items == nil {
|
||||||
|
items = []reading{}
|
||||||
|
}
|
||||||
|
writeJSON(w, http.StatusOK, items)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *api) createReading(w http.ResponseWriter, r *http.Request) {
|
||||||
|
var input reading
|
||||||
|
if !decodeJSON(w, r, &input) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if !validUUID(input.BuoyID) || input.WaveHeight < 0 || input.Salinity != nil && *input.Salinity < 0 {
|
||||||
|
writeError(w, http.StatusBadRequest, "invalid reading")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
id, err := newUUID()
|
||||||
|
if err != nil {
|
||||||
|
a.internalError(w, r, err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
input.ID = id
|
||||||
|
created, err := a.store.CreateReading(r.Context(), input)
|
||||||
|
if err != nil {
|
||||||
|
a.internalError(w, r, err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
a.metrics.ingested.Inc()
|
||||||
|
writeJSON(w, http.StatusCreated, created)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *api) aggregateReadings(w http.ResponseWriter, r *http.Request) {
|
||||||
|
window, err := time.ParseDuration(r.URL.Query().Get("window"))
|
||||||
|
if err != nil || window < time.Minute || window > 24*time.Hour {
|
||||||
|
writeError(w, http.StatusBadRequest, "window must be between 1m and 24h")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
items, err := a.store.AggregateReadings(r.Context(), window)
|
||||||
|
if err != nil {
|
||||||
|
a.internalError(w, r, err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if items == nil {
|
||||||
|
items = []aggregate{}
|
||||||
|
}
|
||||||
|
writeJSON(w, http.StatusOK, items)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *api) internalError(w http.ResponseWriter, r *http.Request, err error) {
|
||||||
|
a.logger.ErrorContext(r.Context(), "request failed", "error", err)
|
||||||
|
writeError(w, http.StatusInternalServerError, "internal error")
|
||||||
|
}
|
||||||
|
func writeError(w http.ResponseWriter, status int, message string) {
|
||||||
|
writeJSON(w, status, map[string]string{"error": message})
|
||||||
|
}
|
||||||
|
func writeJSON(w http.ResponseWriter, status int, value any) {
|
||||||
|
w.Header().Set("Content-Type", "application/json")
|
||||||
|
w.WriteHeader(status)
|
||||||
|
_ = json.NewEncoder(w).Encode(value)
|
||||||
|
}
|
||||||
|
|
||||||
|
func decodeJSON(w http.ResponseWriter, r *http.Request, dst any) bool {
|
||||||
|
decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20))
|
||||||
|
decoder.DisallowUnknownFields()
|
||||||
|
if err := decoder.Decode(dst); err != nil {
|
||||||
|
writeError(w, http.StatusBadRequest, "invalid JSON")
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) {
|
||||||
|
writeError(w, http.StatusBadRequest, "request body must contain one JSON object")
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
|
||||||
|
func optionalTime(w http.ResponseWriter, value string) (*time.Time, bool) {
|
||||||
|
if value == "" {
|
||||||
|
return nil, true
|
||||||
|
}
|
||||||
|
parsed, err := time.Parse(time.RFC3339, value)
|
||||||
|
if err != nil {
|
||||||
|
writeError(w, http.StatusBadRequest, "invalid timestamp")
|
||||||
|
return nil, false
|
||||||
|
}
|
||||||
|
return &parsed, true
|
||||||
|
}
|
||||||
|
func boundedInt(value string, fallback, min, max int) (int, bool) {
|
||||||
|
if value == "" {
|
||||||
|
return fallback, true
|
||||||
|
}
|
||||||
|
parsed, err := strconv.Atoi(value)
|
||||||
|
return parsed, err == nil && parsed >= min && parsed <= max
|
||||||
|
}
|
||||||
|
|
||||||
|
func validUUID(value string) bool {
|
||||||
|
if len(value) != 36 || value[8] != '-' || value[13] != '-' || value[18] != '-' || value[23] != '-' {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
compact := strings.ReplaceAll(value, "-", "")
|
||||||
|
_, err := hex.DecodeString(compact)
|
||||||
|
return err == nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func newUUID() (string, error) {
|
||||||
|
var b [16]byte
|
||||||
|
if _, err := rand.Read(b[:]); err != nil {
|
||||||
|
return "", fmt.Errorf("generate UUID: %w", err)
|
||||||
|
}
|
||||||
|
b[6] = (b[6] & 0x0f) | 0x40
|
||||||
|
b[8] = (b[8] & 0x3f) | 0x80
|
||||||
|
encoded := make([]byte, 36)
|
||||||
|
hex.Encode(encoded[0:8], b[0:4])
|
||||||
|
encoded[8] = '-'
|
||||||
|
hex.Encode(encoded[9:13], b[4:6])
|
||||||
|
encoded[13] = '-'
|
||||||
|
hex.Encode(encoded[14:18], b[6:8])
|
||||||
|
encoded[18] = '-'
|
||||||
|
hex.Encode(encoded[19:23], b[8:10])
|
||||||
|
encoded[23] = '-'
|
||||||
|
hex.Encode(encoded[24:36], b[10:16])
|
||||||
|
return string(encoded), nil
|
||||||
|
}
|
||||||
256
apps/api/http_test.go
Normal file
256
apps/api/http_test.go
Normal file
|
|
@ -0,0 +1,256 @@
|
||||||
|
package main
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"errors"
|
||||||
|
"io"
|
||||||
|
"log/slog"
|
||||||
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
|
"strings"
|
||||||
|
"sync/atomic"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/jackc/pgx/v5"
|
||||||
|
"github.com/prometheus/client_golang/prometheus"
|
||||||
|
)
|
||||||
|
|
||||||
|
type fakeStore struct{ pingErr error }
|
||||||
|
|
||||||
|
const (
|
||||||
|
testBuoyID = "00000000-0000-4000-8000-000000000001"
|
||||||
|
missingBuoyID = "00000000-0000-4000-8000-000000000002"
|
||||||
|
)
|
||||||
|
|
||||||
|
func (f *fakeStore) Ping(context.Context) error { return f.pingErr }
|
||||||
|
func (f *fakeStore) Migrate(context.Context) error { return nil }
|
||||||
|
func (f *fakeStore) Close() {}
|
||||||
|
func (f *fakeStore) ListBuoys(context.Context) ([]buoy, error) {
|
||||||
|
return []buoy{{ID: testBuoyID, Name: "Atlantic"}}, nil
|
||||||
|
}
|
||||||
|
func (f *fakeStore) CreateBuoy(_ context.Context, b buoy) (buoy, error) {
|
||||||
|
b.CreatedAt = time.Unix(1, 0)
|
||||||
|
return b, nil
|
||||||
|
}
|
||||||
|
func (f *fakeStore) GetBuoy(_ context.Context, id string) (buoy, error) {
|
||||||
|
if id == missingBuoyID {
|
||||||
|
return buoy{}, pgx.ErrNoRows
|
||||||
|
}
|
||||||
|
return buoy{ID: id}, nil
|
||||||
|
}
|
||||||
|
func (f *fakeStore) DeleteBuoy(_ context.Context, id string) error {
|
||||||
|
if id == missingBuoyID {
|
||||||
|
return pgx.ErrNoRows
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
func (f *fakeStore) ListReadings(context.Context, string, *time.Time, *time.Time, int, int) ([]reading, error) {
|
||||||
|
return []reading{}, nil
|
||||||
|
}
|
||||||
|
func (f *fakeStore) CreateReading(_ context.Context, r reading) (reading, error) {
|
||||||
|
r.RecordedAt = time.Unix(1, 0)
|
||||||
|
return r, nil
|
||||||
|
}
|
||||||
|
func (f *fakeStore) AggregateReadings(context.Context, time.Duration) ([]aggregate, error) {
|
||||||
|
return []aggregate{}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
type failingStore struct {
|
||||||
|
fakeStore
|
||||||
|
operation string
|
||||||
|
}
|
||||||
|
|
||||||
|
var errDatabaseUnavailable = errors.New("database unavailable")
|
||||||
|
|
||||||
|
func (f *failingStore) ListBuoys(context.Context) ([]buoy, error) {
|
||||||
|
if f.operation == "list buoys" {
|
||||||
|
return nil, errDatabaseUnavailable
|
||||||
|
}
|
||||||
|
return f.fakeStore.ListBuoys(context.Background())
|
||||||
|
}
|
||||||
|
func (f *failingStore) CreateBuoy(ctx context.Context, b buoy) (buoy, error) {
|
||||||
|
if f.operation == "create buoy" {
|
||||||
|
return buoy{}, errDatabaseUnavailable
|
||||||
|
}
|
||||||
|
return f.fakeStore.CreateBuoy(ctx, b)
|
||||||
|
}
|
||||||
|
func (f *failingStore) GetBuoy(ctx context.Context, id string) (buoy, error) {
|
||||||
|
if f.operation == "get buoy" {
|
||||||
|
return buoy{}, errDatabaseUnavailable
|
||||||
|
}
|
||||||
|
return f.fakeStore.GetBuoy(ctx, id)
|
||||||
|
}
|
||||||
|
func (f *failingStore) DeleteBuoy(ctx context.Context, id string) error {
|
||||||
|
if f.operation == "delete buoy" {
|
||||||
|
return errDatabaseUnavailable
|
||||||
|
}
|
||||||
|
return f.fakeStore.DeleteBuoy(ctx, id)
|
||||||
|
}
|
||||||
|
func (f *failingStore) ListReadings(ctx context.Context, buoyID string, from, to *time.Time, limit, offset int) ([]reading, error) {
|
||||||
|
if f.operation == "list readings" {
|
||||||
|
return nil, errDatabaseUnavailable
|
||||||
|
}
|
||||||
|
return f.fakeStore.ListReadings(ctx, buoyID, from, to, limit, offset)
|
||||||
|
}
|
||||||
|
func (f *failingStore) CreateReading(ctx context.Context, r reading) (reading, error) {
|
||||||
|
if f.operation == "create reading" {
|
||||||
|
return reading{}, errDatabaseUnavailable
|
||||||
|
}
|
||||||
|
return f.fakeStore.CreateReading(ctx, r)
|
||||||
|
}
|
||||||
|
func (f *failingStore) AggregateReadings(ctx context.Context, window time.Duration) ([]aggregate, error) {
|
||||||
|
if f.operation == "aggregate readings" {
|
||||||
|
return nil, errDatabaseUnavailable
|
||||||
|
}
|
||||||
|
return f.fakeStore.AggregateReadings(ctx, window)
|
||||||
|
}
|
||||||
|
|
||||||
|
func testAPI(s store, chaos float64) http.Handler {
|
||||||
|
handler, _ := testAPIWithMetrics(s, chaos)
|
||||||
|
return handler
|
||||||
|
}
|
||||||
|
|
||||||
|
func testAPIWithMetrics(s store, chaos float64) (http.Handler, *metrics) {
|
||||||
|
reg := prometheus.NewRegistry()
|
||||||
|
ready := &atomic.Bool{}
|
||||||
|
ready.Store(true)
|
||||||
|
m := newMetrics(reg)
|
||||||
|
a := &api{store: s, metrics: m, logger: slog.New(slog.NewTextHandler(io.Discard, nil)), version: "test", chaosRate: chaos, migrated: ready, prometheus: reg}
|
||||||
|
return a.routes(), m
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestHandlers(t *testing.T) {
|
||||||
|
tests := []struct {
|
||||||
|
name, method, path, body string
|
||||||
|
want int
|
||||||
|
}{
|
||||||
|
{"health", http.MethodGet, "/healthz", "", http.StatusOK},
|
||||||
|
{"ready", http.MethodGet, "/readyz", "", http.StatusOK},
|
||||||
|
{"metrics", http.MethodGet, "/metrics", "", http.StatusOK},
|
||||||
|
{"list buoys", http.MethodGet, "/api/v1/buoys", "", http.StatusOK},
|
||||||
|
{"create buoy", http.MethodPost, "/api/v1/buoys", `{"name":"North","latitude":42,"longitude":-8}`, http.StatusCreated},
|
||||||
|
{"invalid buoy", http.MethodPost, "/api/v1/buoys", `{"name":"","latitude":42,"longitude":-8}`, http.StatusBadRequest},
|
||||||
|
{"get buoy", http.MethodGet, "/api/v1/buoys/" + testBuoyID, "", http.StatusOK},
|
||||||
|
{"missing buoy", http.MethodGet, "/api/v1/buoys/" + missingBuoyID, "", http.StatusNotFound},
|
||||||
|
{"invalid buoy ID", http.MethodGet, "/api/v1/buoys/not-a-uuid", "", http.StatusBadRequest},
|
||||||
|
{"delete buoy", http.MethodDelete, "/api/v1/buoys/" + testBuoyID, "", http.StatusNoContent},
|
||||||
|
{"list readings", http.MethodGet, "/api/v1/readings?limit=10", "", http.StatusOK},
|
||||||
|
{"bad timestamp", http.MethodGet, "/api/v1/readings?from=yesterday", "", http.StatusBadRequest},
|
||||||
|
{"create reading", http.MethodPost, "/api/v1/readings", `{"buoy_id":"` + testBuoyID + `","water_temp":14,"wave_height":2}`, http.StatusCreated},
|
||||||
|
{"aggregate", http.MethodGet, "/api/v1/readings/aggregate?window=1h", "", http.StatusOK},
|
||||||
|
{"bad aggregate", http.MethodGet, "/api/v1/readings/aggregate?window=bad", "", http.StatusBadRequest},
|
||||||
|
}
|
||||||
|
handler := testAPI(&fakeStore{}, 0)
|
||||||
|
for _, tt := range tests {
|
||||||
|
t.Run(tt.name, func(t *testing.T) {
|
||||||
|
req := httptest.NewRequest(tt.method, tt.path, strings.NewReader(tt.body))
|
||||||
|
recorder := httptest.NewRecorder()
|
||||||
|
handler.ServeHTTP(recorder, req)
|
||||||
|
if recorder.Code != tt.want {
|
||||||
|
t.Fatalf("status = %d, want %d; body=%s", recorder.Code, tt.want, recorder.Body.String())
|
||||||
|
}
|
||||||
|
if got := recorder.Header().Get("X-Nereus-Version"); got != "test" {
|
||||||
|
t.Errorf("X-Nereus-Version = %q, want test", got)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestReadinessFailure(t *testing.T) {
|
||||||
|
recorder := httptest.NewRecorder()
|
||||||
|
testAPI(&fakeStore{pingErr: context.DeadlineExceeded}, 0).ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/readyz", nil))
|
||||||
|
if recorder.Code != http.StatusServiceUnavailable {
|
||||||
|
t.Fatalf("status = %d, want %d", recorder.Code, http.StatusServiceUnavailable)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestChaosMiddleware(t *testing.T) {
|
||||||
|
recorder := httptest.NewRecorder()
|
||||||
|
testAPI(&fakeStore{}, 1).ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/api/v1/buoys", nil))
|
||||||
|
if recorder.Code != http.StatusInternalServerError {
|
||||||
|
t.Fatalf("status = %d, want %d", recorder.Code, http.StatusInternalServerError)
|
||||||
|
}
|
||||||
|
recorder = httptest.NewRecorder()
|
||||||
|
testAPI(&fakeStore{}, 1).ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/healthz", nil))
|
||||||
|
if recorder.Code != http.StatusOK {
|
||||||
|
t.Fatalf("health status = %d, want %d", recorder.Code, http.StatusOK)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestDataEndpointsReturnInternalErrorOnDatabaseFailure(t *testing.T) {
|
||||||
|
tests := []struct {
|
||||||
|
name, method, path, body string
|
||||||
|
}{
|
||||||
|
{"list buoys", http.MethodGet, "/api/v1/buoys", ""},
|
||||||
|
{"create buoy", http.MethodPost, "/api/v1/buoys", `{"name":"North","latitude":42,"longitude":-8}`},
|
||||||
|
{"get buoy", http.MethodGet, "/api/v1/buoys/" + testBuoyID, ""},
|
||||||
|
{"delete buoy", http.MethodDelete, "/api/v1/buoys/" + testBuoyID, ""},
|
||||||
|
{"list readings", http.MethodGet, "/api/v1/readings", ""},
|
||||||
|
{"create reading", http.MethodPost, "/api/v1/readings", `{"buoy_id":"` + testBuoyID + `","water_temp":14,"wave_height":2}`},
|
||||||
|
{"aggregate readings", http.MethodGet, "/api/v1/readings/aggregate?window=1h", ""},
|
||||||
|
}
|
||||||
|
for _, tt := range tests {
|
||||||
|
t.Run(tt.name, func(t *testing.T) {
|
||||||
|
recorder := httptest.NewRecorder()
|
||||||
|
testAPI(&failingStore{operation: tt.name}, 0).ServeHTTP(recorder, httptest.NewRequest(tt.method, tt.path, strings.NewReader(tt.body)))
|
||||||
|
if recorder.Code != http.StatusInternalServerError {
|
||||||
|
t.Fatalf("status = %d, want %d; body=%s", recorder.Code, http.StatusInternalServerError, recorder.Body.String())
|
||||||
|
}
|
||||||
|
if recorder.Body.String() != "{\"error\":\"internal error\"}\n" {
|
||||||
|
t.Fatalf("body = %q", recorder.Body.String())
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestJSONBodyValidation(t *testing.T) {
|
||||||
|
tests := []struct {
|
||||||
|
name, body string
|
||||||
|
}{
|
||||||
|
{"malformed", `{"name":`},
|
||||||
|
{"multiple values", `{"name":"North","latitude":42,"longitude":-8} {}`},
|
||||||
|
{"oversized", `{"name":"` + strings.Repeat("x", 1<<20) + `","latitude":42,"longitude":-8}`},
|
||||||
|
}
|
||||||
|
for _, tt := range tests {
|
||||||
|
t.Run(tt.name, func(t *testing.T) {
|
||||||
|
recorder := httptest.NewRecorder()
|
||||||
|
testAPI(&fakeStore{}, 0).ServeHTTP(recorder, httptest.NewRequest(http.MethodPost, "/api/v1/buoys", strings.NewReader(tt.body)))
|
||||||
|
if recorder.Code != http.StatusBadRequest {
|
||||||
|
t.Fatalf("status = %d, want %d", recorder.Code, http.StatusBadRequest)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestMetricNamesAndRouteTemplateLabels(t *testing.T) {
|
||||||
|
handler, metrics := testAPIWithMetrics(&fakeStore{}, 0)
|
||||||
|
metrics.db.WithLabelValues("test").Observe(0)
|
||||||
|
|
||||||
|
recorder := httptest.NewRecorder()
|
||||||
|
handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/api/v1/buoys/"+testBuoyID, nil))
|
||||||
|
recorder = httptest.NewRecorder()
|
||||||
|
handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/metrics", nil))
|
||||||
|
body := recorder.Body.String()
|
||||||
|
|
||||||
|
for _, name := range []string{
|
||||||
|
"nereus_http_requests_total",
|
||||||
|
"nereus_http_request_duration_seconds",
|
||||||
|
"nereus_db_query_duration_seconds",
|
||||||
|
"nereus_readings_ingested_total",
|
||||||
|
"nereus_buoys_active",
|
||||||
|
} {
|
||||||
|
if !strings.Contains(body, "# HELP "+name+" ") {
|
||||||
|
t.Errorf("metrics output does not contain %q", name)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if !strings.Contains(body, `nereus_http_requests_total{method="GET",path="/api/v1/buoys/{id}",status="200",version="test"} 1`) {
|
||||||
|
t.Error("request counter does not contain the exact labels and buoy route template")
|
||||||
|
}
|
||||||
|
if !strings.Contains(body, `nereus_http_request_duration_seconds_count{method="GET",path="/api/v1/buoys/{id}",version="test"} 1`) {
|
||||||
|
t.Error("request histogram does not contain the exact labels and buoy route template")
|
||||||
|
}
|
||||||
|
if strings.Contains(body, `path="/api/v1/buoys/`+testBuoyID+`"`) {
|
||||||
|
t.Error("metrics output contains a resolved resource path")
|
||||||
|
}
|
||||||
|
}
|
||||||
31
apps/api/integration_test.go
Normal file
31
apps/api/integration_test.go
Normal file
|
|
@ -0,0 +1,31 @@
|
||||||
|
package main
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"os"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/prometheus/client_golang/prometheus"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestPostgresIntegration(t *testing.T) {
|
||||||
|
if testing.Short() {
|
||||||
|
t.Skip("integration test disabled in short mode")
|
||||||
|
}
|
||||||
|
dsn, configured := os.LookupEnv("DATABASE_URL")
|
||||||
|
if !configured {
|
||||||
|
t.Skip("DATABASE_URL is not configured")
|
||||||
|
}
|
||||||
|
m := newMetrics(prometheus.NewRegistry())
|
||||||
|
db, err := newPostgresStore(context.Background(), dsn, m)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("new store: %v", err)
|
||||||
|
}
|
||||||
|
defer db.Close()
|
||||||
|
if err := db.Migrate(context.Background()); err != nil {
|
||||||
|
t.Fatalf("migrate: %v", err)
|
||||||
|
}
|
||||||
|
if err := db.Ping(context.Background()); err != nil {
|
||||||
|
t.Fatalf("ping: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
165
apps/api/main.go
Normal file
165
apps/api/main.go
Normal file
|
|
@ -0,0 +1,165 @@
|
||||||
|
package main
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"log/slog"
|
||||||
|
"net/http"
|
||||||
|
"os"
|
||||||
|
"os/signal"
|
||||||
|
"strconv"
|
||||||
|
"sync/atomic"
|
||||||
|
"syscall"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/prometheus/client_golang/prometheus"
|
||||||
|
"go.opentelemetry.io/otel"
|
||||||
|
"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
|
||||||
|
"go.opentelemetry.io/otel/sdk/resource"
|
||||||
|
sdktrace "go.opentelemetry.io/otel/sdk/trace"
|
||||||
|
semconv "go.opentelemetry.io/otel/semconv/v1.34.0"
|
||||||
|
)
|
||||||
|
|
||||||
|
var appVersion = "dev"
|
||||||
|
|
||||||
|
type config struct {
|
||||||
|
port string
|
||||||
|
database string
|
||||||
|
chaosRate float64
|
||||||
|
logLevel slog.Level
|
||||||
|
traceReady bool
|
||||||
|
}
|
||||||
|
|
||||||
|
func main() {
|
||||||
|
if err := run(); err != nil {
|
||||||
|
slog.Error("startup failed", "error", err)
|
||||||
|
os.Exit(1)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func run() error {
|
||||||
|
cfg, err := loadConfig()
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: cfg.logLevel}))
|
||||||
|
slog.SetDefault(logger)
|
||||||
|
|
||||||
|
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
|
||||||
|
defer stop()
|
||||||
|
shutdownTrace, err := configureTracing(ctx, cfg.traceReady)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
defer func() {
|
||||||
|
if err := shutdownTrace(context.Background()); err != nil {
|
||||||
|
logger.Error("trace shutdown failed", "error", err)
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
|
registry := prometheus.NewRegistry()
|
||||||
|
m := newMetrics(registry)
|
||||||
|
db, err := newPostgresStore(ctx, cfg.database, m)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
defer db.Close()
|
||||||
|
|
||||||
|
var migrated atomic.Bool
|
||||||
|
go migrateUntilReady(ctx, db, &migrated, logger)
|
||||||
|
application := &api{store: db, metrics: m, logger: logger, version: appVersion, chaosRate: cfg.chaosRate, migrated: &migrated, prometheus: registry}
|
||||||
|
server := &http.Server{
|
||||||
|
Addr: ":" + cfg.port,
|
||||||
|
Handler: application.routes(),
|
||||||
|
ReadHeaderTimeout: 5 * time.Second,
|
||||||
|
WriteTimeout: 30 * time.Second,
|
||||||
|
IdleTimeout: 60 * time.Second,
|
||||||
|
MaxHeaderBytes: 1 << 20,
|
||||||
|
}
|
||||||
|
errCh := make(chan error, 1)
|
||||||
|
go func() { errCh <- server.ListenAndServe() }()
|
||||||
|
logger.Info("server started", "port", cfg.port, "version", appVersion)
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||||
|
defer cancel()
|
||||||
|
if err := server.Shutdown(shutdownCtx); err != nil {
|
||||||
|
return fmt.Errorf("shutdown server: %w", err)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
case err := <-errCh:
|
||||||
|
if errors.Is(err, http.ErrServerClosed) {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return fmt.Errorf("serve HTTP: %w", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func loadConfig() (config, error) {
|
||||||
|
cfg := config{port: envOr("PORT", "8080"), database: os.Getenv("DATABASE_URL"), traceReady: os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT") != ""}
|
||||||
|
if cfg.database == "" {
|
||||||
|
return config{}, errors.New("DATABASE_URL is required")
|
||||||
|
}
|
||||||
|
chaos, err := strconv.ParseFloat(envOr("CHAOS_ERROR_RATE", "0.0"), 64)
|
||||||
|
if err != nil || chaos < 0 || chaos > 1 {
|
||||||
|
return config{}, errors.New("CHAOS_ERROR_RATE must be between 0.0 and 1.0")
|
||||||
|
}
|
||||||
|
cfg.chaosRate = chaos
|
||||||
|
switch envOr("LOG_LEVEL", "info") {
|
||||||
|
case "debug":
|
||||||
|
cfg.logLevel = slog.LevelDebug
|
||||||
|
case "info":
|
||||||
|
cfg.logLevel = slog.LevelInfo
|
||||||
|
case "warn":
|
||||||
|
cfg.logLevel = slog.LevelWarn
|
||||||
|
case "error":
|
||||||
|
cfg.logLevel = slog.LevelError
|
||||||
|
default:
|
||||||
|
return config{}, errors.New("LOG_LEVEL must be debug, info, warn, or error")
|
||||||
|
}
|
||||||
|
return cfg, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func envOr(name, fallback string) string {
|
||||||
|
if value := os.Getenv(name); value != "" {
|
||||||
|
return value
|
||||||
|
}
|
||||||
|
return fallback
|
||||||
|
}
|
||||||
|
|
||||||
|
func configureTracing(ctx context.Context, enabled bool) (func(context.Context) error, error) {
|
||||||
|
if !enabled {
|
||||||
|
return func(context.Context) error { return nil }, nil
|
||||||
|
}
|
||||||
|
exporter, err := otlptracegrpc.New(ctx)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("create trace exporter: %w", err)
|
||||||
|
}
|
||||||
|
res, err := resource.New(ctx, resource.WithAttributes(semconv.ServiceName("nereus-api")))
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("create trace resource: %w", err)
|
||||||
|
}
|
||||||
|
provider := sdktrace.NewTracerProvider(sdktrace.WithBatcher(exporter), sdktrace.WithResource(res))
|
||||||
|
otel.SetTracerProvider(provider)
|
||||||
|
return provider.Shutdown, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func migrateUntilReady(ctx context.Context, db store, ready *atomic.Bool, logger *slog.Logger) {
|
||||||
|
for {
|
||||||
|
attempt, cancel := context.WithTimeout(ctx, 5*time.Second)
|
||||||
|
err := db.Migrate(attempt)
|
||||||
|
cancel()
|
||||||
|
if err == nil {
|
||||||
|
ready.Store(true)
|
||||||
|
logger.Info("database migrations applied")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
logger.Warn("database unavailable; migration will retry", "error", err)
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return
|
||||||
|
case <-time.After(5 * time.Second):
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
23
apps/api/metrics.go
Normal file
23
apps/api/metrics.go
Normal file
|
|
@ -0,0 +1,23 @@
|
||||||
|
package main
|
||||||
|
|
||||||
|
import "github.com/prometheus/client_golang/prometheus"
|
||||||
|
|
||||||
|
type metrics struct {
|
||||||
|
requests *prometheus.CounterVec
|
||||||
|
duration *prometheus.HistogramVec
|
||||||
|
db *prometheus.HistogramVec
|
||||||
|
ingested prometheus.Counter
|
||||||
|
active prometheus.Gauge
|
||||||
|
}
|
||||||
|
|
||||||
|
func newMetrics(reg prometheus.Registerer) *metrics {
|
||||||
|
m := &metrics{
|
||||||
|
requests: prometheus.NewCounterVec(prometheus.CounterOpts{Name: "nereus_http_requests_total", Help: "HTTP requests processed."}, []string{"method", "path", "status", "version"}),
|
||||||
|
duration: prometheus.NewHistogramVec(prometheus.HistogramOpts{Name: "nereus_http_request_duration_seconds", Help: "HTTP request duration."}, []string{"method", "path", "version"}),
|
||||||
|
db: prometheus.NewHistogramVec(prometheus.HistogramOpts{Name: "nereus_db_query_duration_seconds", Help: "Database query duration."}, []string{"operation"}),
|
||||||
|
ingested: prometheus.NewCounter(prometheus.CounterOpts{Name: "nereus_readings_ingested_total", Help: "Readings successfully ingested."}),
|
||||||
|
active: prometheus.NewGauge(prometheus.GaugeOpts{Name: "nereus_buoys_active", Help: "Current number of buoys."}),
|
||||||
|
}
|
||||||
|
reg.MustRegister(m.requests, m.duration, m.db, m.ingested, m.active)
|
||||||
|
return m
|
||||||
|
}
|
||||||
25
apps/api/migrations/001_init.sql
Normal file
25
apps/api/migrations/001_init.sql
Normal file
|
|
@ -0,0 +1,25 @@
|
||||||
|
BEGIN;
|
||||||
|
|
||||||
|
-- Replicas start together during a rollout, so serialize schema creation.
|
||||||
|
SELECT pg_advisory_xact_lock(hashtext('nereus_schema_migration'));
|
||||||
|
|
||||||
|
CREATE TABLE IF NOT EXISTS buoys (
|
||||||
|
id UUID PRIMARY KEY,
|
||||||
|
name TEXT NOT NULL,
|
||||||
|
latitude DOUBLE PRECISION NOT NULL,
|
||||||
|
longitude DOUBLE PRECISION NOT NULL,
|
||||||
|
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE TABLE IF NOT EXISTS readings (
|
||||||
|
id UUID PRIMARY KEY,
|
||||||
|
buoy_id UUID NOT NULL REFERENCES buoys(id) ON DELETE CASCADE,
|
||||||
|
water_temp DOUBLE PRECISION NOT NULL,
|
||||||
|
wave_height DOUBLE PRECISION NOT NULL,
|
||||||
|
salinity DOUBLE PRECISION,
|
||||||
|
recorded_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE INDEX IF NOT EXISTS idx_readings_buoy_time ON readings (buoy_id, recorded_at DESC);
|
||||||
|
|
||||||
|
COMMIT;
|
||||||
28
apps/api/model.go
Normal file
28
apps/api/model.go
Normal file
|
|
@ -0,0 +1,28 @@
|
||||||
|
package main
|
||||||
|
|
||||||
|
import "time"
|
||||||
|
|
||||||
|
type buoy struct {
|
||||||
|
ID string `db:"id" json:"id"`
|
||||||
|
Name string `db:"name" json:"name"`
|
||||||
|
Latitude float64 `db:"latitude" json:"latitude"`
|
||||||
|
Longitude float64 `db:"longitude" json:"longitude"`
|
||||||
|
CreatedAt time.Time `db:"created_at" json:"created_at"`
|
||||||
|
}
|
||||||
|
|
||||||
|
type reading struct {
|
||||||
|
ID string `db:"id" json:"id"`
|
||||||
|
BuoyID string `db:"buoy_id" json:"buoy_id"`
|
||||||
|
WaterTemp float64 `db:"water_temp" json:"water_temp"`
|
||||||
|
WaveHeight float64 `db:"wave_height" json:"wave_height"`
|
||||||
|
Salinity *float64 `db:"salinity" json:"salinity,omitempty"`
|
||||||
|
RecordedAt time.Time `db:"recorded_at" json:"recorded_at"`
|
||||||
|
}
|
||||||
|
|
||||||
|
type aggregate struct {
|
||||||
|
Bucket time.Time `db:"bucket" json:"bucket"`
|
||||||
|
AverageTemp float64 `db:"average_water_temp" json:"average_water_temp"`
|
||||||
|
AverageWave float64 `db:"average_wave_height" json:"average_wave_height"`
|
||||||
|
AverageSaline *float64 `db:"average_salinity" json:"average_salinity"`
|
||||||
|
Count int64 `db:"count" json:"count"`
|
||||||
|
}
|
||||||
159
apps/api/store.go
Normal file
159
apps/api/store.go
Normal file
|
|
@ -0,0 +1,159 @@
|
||||||
|
package main
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
_ "embed"
|
||||||
|
"fmt"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/jackc/pgx/v5"
|
||||||
|
"github.com/jackc/pgx/v5/pgxpool"
|
||||||
|
"go.opentelemetry.io/otel"
|
||||||
|
)
|
||||||
|
|
||||||
|
//go:embed migrations/001_init.sql
|
||||||
|
var migrationSQL string
|
||||||
|
|
||||||
|
type store interface {
|
||||||
|
Ping(context.Context) error
|
||||||
|
Migrate(context.Context) error
|
||||||
|
ListBuoys(context.Context) ([]buoy, error)
|
||||||
|
CreateBuoy(context.Context, buoy) (buoy, error)
|
||||||
|
GetBuoy(context.Context, string) (buoy, error)
|
||||||
|
DeleteBuoy(context.Context, string) error
|
||||||
|
ListReadings(context.Context, string, *time.Time, *time.Time, int, int) ([]reading, error)
|
||||||
|
CreateReading(context.Context, reading) (reading, error)
|
||||||
|
AggregateReadings(context.Context, time.Duration) ([]aggregate, error)
|
||||||
|
Close()
|
||||||
|
}
|
||||||
|
|
||||||
|
type postgresStore struct {
|
||||||
|
pool *pgxpool.Pool
|
||||||
|
metrics *metrics
|
||||||
|
}
|
||||||
|
|
||||||
|
func newPostgresStore(ctx context.Context, dsn string, m *metrics) (*postgresStore, error) {
|
||||||
|
pool, err := pgxpool.New(ctx, dsn)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("create database pool: %w", err)
|
||||||
|
}
|
||||||
|
return &postgresStore{pool: pool, metrics: m}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *postgresStore) observe(ctx context.Context, operation string) (context.Context, func()) {
|
||||||
|
ctx, span := otel.Tracer("nereus-api/database").Start(ctx, operation)
|
||||||
|
started := time.Now()
|
||||||
|
return ctx, func() {
|
||||||
|
s.metrics.db.WithLabelValues(operation).Observe(time.Since(started).Seconds())
|
||||||
|
span.End()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *postgresStore) Ping(ctx context.Context) error { return s.pool.Ping(ctx) }
|
||||||
|
func (s *postgresStore) Close() { s.pool.Close() }
|
||||||
|
|
||||||
|
func (s *postgresStore) Migrate(ctx context.Context) error {
|
||||||
|
ctx, done := s.observe(ctx, "migrate")
|
||||||
|
defer done()
|
||||||
|
if _, err := s.pool.Exec(ctx, migrationSQL); err != nil {
|
||||||
|
return fmt.Errorf("apply migrations: %w", err)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *postgresStore) ListBuoys(ctx context.Context) ([]buoy, error) {
|
||||||
|
ctx, done := s.observe(ctx, "list_buoys")
|
||||||
|
defer done()
|
||||||
|
rows, err := s.pool.Query(ctx, `SELECT id, name, latitude, longitude, created_at FROM buoys ORDER BY created_at`)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("query buoys: %w", err)
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
items, err := pgx.CollectRows(rows, pgx.RowToStructByName[buoy])
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("collect buoys: %w", err)
|
||||||
|
}
|
||||||
|
return items, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *postgresStore) CreateBuoy(ctx context.Context, b buoy) (buoy, error) {
|
||||||
|
ctx, done := s.observe(ctx, "create_buoy")
|
||||||
|
defer done()
|
||||||
|
err := s.pool.QueryRow(ctx, `INSERT INTO buoys (id,name,latitude,longitude) VALUES ($1,$2,$3,$4) RETURNING created_at`, b.ID, b.Name, b.Latitude, b.Longitude).Scan(&b.CreatedAt)
|
||||||
|
if err != nil {
|
||||||
|
return buoy{}, fmt.Errorf("insert buoy: %w", err)
|
||||||
|
}
|
||||||
|
return b, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *postgresStore) GetBuoy(ctx context.Context, id string) (buoy, error) {
|
||||||
|
ctx, done := s.observe(ctx, "get_buoy")
|
||||||
|
defer done()
|
||||||
|
var b buoy
|
||||||
|
err := s.pool.QueryRow(ctx, `SELECT id, name, latitude, longitude, created_at FROM buoys WHERE id=$1`, id).Scan(&b.ID, &b.Name, &b.Latitude, &b.Longitude, &b.CreatedAt)
|
||||||
|
if err != nil {
|
||||||
|
return buoy{}, fmt.Errorf("select buoy: %w", err)
|
||||||
|
}
|
||||||
|
return b, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *postgresStore) DeleteBuoy(ctx context.Context, id string) error {
|
||||||
|
ctx, done := s.observe(ctx, "delete_buoy")
|
||||||
|
defer done()
|
||||||
|
tag, err := s.pool.Exec(ctx, `DELETE FROM buoys WHERE id=$1`, id)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("delete buoy: %w", err)
|
||||||
|
}
|
||||||
|
if tag.RowsAffected() == 0 {
|
||||||
|
return pgx.ErrNoRows
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *postgresStore) ListReadings(ctx context.Context, buoyID string, from, to *time.Time, limit, offset int) ([]reading, error) {
|
||||||
|
ctx, done := s.observe(ctx, "list_readings")
|
||||||
|
defer done()
|
||||||
|
rows, err := s.pool.Query(ctx, `SELECT id, buoy_id, water_temp, wave_height, salinity, recorded_at FROM readings WHERE ($1::text='' OR buoy_id=NULLIF($1,'')::uuid) AND ($2::timestamptz IS NULL OR recorded_at >= $2) AND ($3::timestamptz IS NULL OR recorded_at <= $3) ORDER BY recorded_at DESC LIMIT $4 OFFSET $5`, buoyID, from, to, limit, offset)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("query readings: %w", err)
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
items, err := pgx.CollectRows(rows, pgx.RowToStructByName[reading])
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("collect readings: %w", err)
|
||||||
|
}
|
||||||
|
return items, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *postgresStore) CreateReading(ctx context.Context, r reading) (reading, error) {
|
||||||
|
ctx, done := s.observe(ctx, "create_reading")
|
||||||
|
defer done()
|
||||||
|
err := s.pool.QueryRow(ctx, `INSERT INTO readings (id,buoy_id,water_temp,wave_height,salinity,recorded_at) VALUES ($1,$2,$3,$4,$5,COALESCE($6,now())) RETURNING recorded_at`, r.ID, r.BuoyID, r.WaterTemp, r.WaveHeight, r.Salinity, nullableTime(r.RecordedAt)).Scan(&r.RecordedAt)
|
||||||
|
if err != nil {
|
||||||
|
return reading{}, fmt.Errorf("insert reading: %w", err)
|
||||||
|
}
|
||||||
|
return r, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func nullableTime(value time.Time) any {
|
||||||
|
if value.IsZero() {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return value
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *postgresStore) AggregateReadings(ctx context.Context, window time.Duration) ([]aggregate, error) {
|
||||||
|
ctx, done := s.observe(ctx, "aggregate_readings")
|
||||||
|
defer done()
|
||||||
|
seconds := int64(window.Seconds())
|
||||||
|
rows, err := s.pool.Query(ctx, `SELECT to_timestamp(floor(extract(epoch FROM recorded_at)/$1)*$1) AS bucket, avg(water_temp) AS average_water_temp, avg(wave_height) AS average_wave_height, avg(salinity) AS average_salinity, count(*) AS count FROM readings GROUP BY bucket ORDER BY bucket DESC`, seconds)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("aggregate readings: %w", err)
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
items, err := pgx.CollectRows(rows, pgx.RowToStructByName[aggregate])
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("collect aggregates: %w", err)
|
||||||
|
}
|
||||||
|
return items, nil
|
||||||
|
}
|
||||||
Loading…
Add table
Reference in a new issue