From 6c1329c2a9d29c46efb667dc0c8d6b207b9183d5 Mon Sep 17 00:00:00 2001 From: Fi3w0 Date: Mon, 24 Aug 2026 23:17:08 +0200 Subject: [PATCH] feat(api): serve buoy telemetry --- apps/api/context.go | 11 + apps/api/go.mod | 45 ++++ apps/api/go.sum | 109 ++++++++++ apps/api/http.go | 341 +++++++++++++++++++++++++++++++ apps/api/http_test.go | 256 +++++++++++++++++++++++ apps/api/integration_test.go | 31 +++ apps/api/main.go | 165 +++++++++++++++ apps/api/metrics.go | 23 +++ apps/api/migrations/001_init.sql | 25 +++ apps/api/model.go | 28 +++ apps/api/store.go | 159 ++++++++++++++ 11 files changed, 1193 insertions(+) create mode 100644 apps/api/context.go create mode 100644 apps/api/go.mod create mode 100644 apps/api/go.sum create mode 100644 apps/api/http.go create mode 100644 apps/api/http_test.go create mode 100644 apps/api/integration_test.go create mode 100644 apps/api/main.go create mode 100644 apps/api/metrics.go create mode 100644 apps/api/migrations/001_init.sql create mode 100644 apps/api/model.go create mode 100644 apps/api/store.go diff --git a/apps/api/context.go b/apps/api/context.go new file mode 100644 index 0000000..6100bea --- /dev/null +++ b/apps/api/context.go @@ -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) +} diff --git a/apps/api/go.mod b/apps/api/go.mod new file mode 100644 index 0000000..585715d --- /dev/null +++ b/apps/api/go.mod @@ -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 +) diff --git a/apps/api/go.sum b/apps/api/go.sum new file mode 100644 index 0000000..6fff21f --- /dev/null +++ b/apps/api/go.sum @@ -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= diff --git a/apps/api/http.go b/apps/api/http.go new file mode 100644 index 0000000..3c669c8 --- /dev/null +++ b/apps/api/http.go @@ -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 +} diff --git a/apps/api/http_test.go b/apps/api/http_test.go new file mode 100644 index 0000000..f6ff372 --- /dev/null +++ b/apps/api/http_test.go @@ -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") + } +} diff --git a/apps/api/integration_test.go b/apps/api/integration_test.go new file mode 100644 index 0000000..c24741e --- /dev/null +++ b/apps/api/integration_test.go @@ -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) + } +} diff --git a/apps/api/main.go b/apps/api/main.go new file mode 100644 index 0000000..b8b6aed --- /dev/null +++ b/apps/api/main.go @@ -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): + } + } +} diff --git a/apps/api/metrics.go b/apps/api/metrics.go new file mode 100644 index 0000000..5abe32c --- /dev/null +++ b/apps/api/metrics.go @@ -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 +} diff --git a/apps/api/migrations/001_init.sql b/apps/api/migrations/001_init.sql new file mode 100644 index 0000000..7945465 --- /dev/null +++ b/apps/api/migrations/001_init.sql @@ -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; diff --git a/apps/api/model.go b/apps/api/model.go new file mode 100644 index 0000000..23a4c76 --- /dev/null +++ b/apps/api/model.go @@ -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"` +} diff --git a/apps/api/store.go b/apps/api/store.go new file mode 100644 index 0000000..5dd3f1d --- /dev/null +++ b/apps/api/store.go @@ -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 +}