From 4679a95d877fae0d9f6b76f3832c200fb855149e Mon Sep 17 00:00:00 2001 From: Martin Joehren Date: Wed, 1 Jul 2026 17:01:57 +0200 Subject: [PATCH] feat: athena history integration --- go.mod | 28 +++++++++ go.sum | 66 +++++++++++++++++++++ lib/history/athena.go | 123 +++++++++++++++++++++++++++++++++++++++ lib/history/lib.go | 19 +++++- lib/history/lib_test.go | 48 +++++++++++++++ lib/history/render.go | 18 +++++- lib/history/singleton.go | 40 ++++++++++++- 7 files changed, 338 insertions(+), 4 deletions(-) diff --git a/go.mod b/go.mod index 32eb8d2..3fbef2b 100644 --- a/go.mod +++ b/go.mod @@ -30,6 +30,7 @@ require ( github.com/prometheus/client_golang v1.23.2 github.com/robfig/cron/v3 v3.0.1 github.com/rubenv/sql-migrate v1.8.1 + github.com/speee/go-athena v1.0.5 github.com/stretchr/testify v1.11.1 go.opentelemetry.io/contrib/instrumentation/runtime v0.69.0 go.opentelemetry.io/otel v1.44.0 @@ -51,6 +52,25 @@ require ( github.com/Azure/azure-sdk-for-go/sdk/internal v1.12.0 // indirect github.com/AzureAD/microsoft-authentication-library-for-go v1.7.2 // indirect github.com/Masterminds/semver/v3 v3.5.0 // indirect + github.com/aws/aws-sdk-go-v2 v1.26.1 // indirect + github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.1 // indirect + github.com/aws/aws-sdk-go-v2/config v1.27.10 // indirect + github.com/aws/aws-sdk-go-v2/credentials v1.17.10 // indirect + github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.1 // indirect + github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.5 // indirect + github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.5 // indirect + github.com/aws/aws-sdk-go-v2/internal/ini v1.8.0 // indirect + github.com/aws/aws-sdk-go-v2/internal/v4a v1.3.3 // indirect + github.com/aws/aws-sdk-go-v2/service/athena v1.40.2 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.11.2 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.3.5 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.11.7 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.17.3 // indirect + github.com/aws/aws-sdk-go-v2/service/s3 v1.51.4 // indirect + github.com/aws/aws-sdk-go-v2/service/sso v1.20.4 // indirect + github.com/aws/aws-sdk-go-v2/service/ssooidc v1.23.4 // indirect + github.com/aws/aws-sdk-go-v2/service/sts v1.28.6 // indirect + github.com/aws/smithy-go v1.20.2 // indirect 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 @@ -83,9 +103,11 @@ require ( github.com/google/gnostic-models v0.7.1 // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 // indirect github.com/hashicorp/errwrap v1.1.0 // indirect + github.com/hashicorp/go-uuid v1.0.2 // 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/jcmturner/gofork v1.0.0 // indirect github.com/json-iterator/go v1.1.12 // indirect github.com/kardianos/osext v0.0.0-20190222173326-2bc1f35cddc0 // indirect github.com/kylelemons/godebug v1.1.0 // indirect @@ -101,10 +123,12 @@ require ( github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect + github.com/prestodb/presto-go-client v0.0.0-20201204133205-8958eb37e584 // indirect github.com/prometheus/client_model v0.6.2 // indirect github.com/prometheus/common v0.68.1 // indirect github.com/prometheus/otlptranslator v1.0.0 // indirect github.com/prometheus/procfs v0.20.1 // indirect + github.com/satori/go.uuid v1.2.0 // indirect github.com/spf13/pflag v1.0.10 // indirect github.com/stretchr/objx v0.5.3 // indirect github.com/twmb/murmur3 v1.1.8 // indirect @@ -129,6 +153,10 @@ require ( google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af // indirect gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect gopkg.in/inf.v0 v0.9.1 // indirect + gopkg.in/jcmturner/aescts.v1 v1.0.1 // indirect + gopkg.in/jcmturner/dnsutils.v1 v1.0.1 // indirect + gopkg.in/jcmturner/gokrb5.v6 v6.1.1 // indirect + gopkg.in/jcmturner/rpc.v1 v1.1.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect k8s.io/klog/v2 v2.140.0 // indirect k8s.io/kube-openapi v0.0.0-20260603220949-865597e52e25 // indirect diff --git a/go.sum b/go.sum index c6d360d..af0faf3 100644 --- a/go.sum +++ b/go.sum @@ -16,6 +16,44 @@ github.com/Masterminds/semver/v3 v3.5.0 h1:kQceYJfbupGfZOKZQg0kou0DgAKhzDg2NZPAw github.com/Masterminds/semver/v3 v3.5.0/go.mod h1:4V+yj/TJE1HU9XfppCwVMZq3I84lprf4nC11bSS5beM= github.com/Unleash/unleash-go-sdk/v5 v5.1.0 h1:W+HHQklU5/H9kjYTn/T4TKvDHE0BxnZ0+MyTk06RdYw= github.com/Unleash/unleash-go-sdk/v5 v5.1.0/go.mod h1:1u8BfdyjlkV5j43la61n9A9ul4E+YQC2kKQotz8z7BE= +github.com/aws/aws-sdk-go-v2 v1.26.1 h1:5554eUqIYVWpU0YmeeYZ0wU64H2VLBs8TlhRB2L+EkA= +github.com/aws/aws-sdk-go-v2 v1.26.1/go.mod h1:ffIFB97e2yNsv4aTSGkqtHnppsIJzw7G7BReUZ3jCXM= +github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.1 h1:gTK2uhtAPtFcdRRJilZPx8uJLL2J85xK11nKtWL0wfU= +github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.1/go.mod h1:sxpLb+nZk7tIfCWChfd+h4QwHNUR57d8hA1cleTkjJo= +github.com/aws/aws-sdk-go-v2/config v1.27.10 h1:PS+65jThT0T/snC5WjyfHHyUgG+eBoupSDV+f838cro= +github.com/aws/aws-sdk-go-v2/config v1.27.10/go.mod h1:BePM7Vo4OBpHreKRUMuDXX+/+JWP38FLkzl5m27/Jjs= +github.com/aws/aws-sdk-go-v2/credentials v1.17.10 h1:qDZ3EA2lv1KangvQB6y258OssCHD0xvaGiEDkG4X/10= +github.com/aws/aws-sdk-go-v2/credentials v1.17.10/go.mod h1:6t3sucOaYDwDssHQa0ojH1RpmVmF5/jArkye1b2FKMI= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.1 h1:FVJ0r5XTHSmIHJV6KuDmdYhEpvlHpiSd38RQWhut5J4= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.1/go.mod h1:zusuAeqezXzAB24LGuzuekqMAEgWkVYukBec3kr3jUg= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.5 h1:aw39xVGeRWlWx9EzGVnhOR4yOjQDHPQ6o6NmBlscyQg= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.5/go.mod h1:FSaRudD0dXiMPK2UjknVwwTYyZMRsHv3TtkabsZih5I= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.5 h1:PG1F3OD1szkuQPzDw3CIQsRIrtTlUC3lP84taWzHlq0= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.5/go.mod h1:jU1li6RFryMz+so64PpKtudI+QzbKoIEivqdf6LNpOc= +github.com/aws/aws-sdk-go-v2/internal/ini v1.8.0 h1:hT8rVHwugYE2lEfdFE0QWVo81lF7jMrYJVDWI+f+VxU= +github.com/aws/aws-sdk-go-v2/internal/ini v1.8.0/go.mod h1:8tu/lYfQfFe6IGnaOdrpVgEL2IrrDOf6/m9RQum4NkY= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.3.3 h1:mDnFOE2sVkyphMWtTH+stv0eW3k0OTx94K63xpxHty4= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.3.3/go.mod h1:V8MuRVcCRt5h1S+Fwu8KbC7l/gBGo3yBAyUbJM2IJOk= +github.com/aws/aws-sdk-go-v2/service/athena v1.40.2 h1:8J9He28jsrIWPcVfdCxltJj9haEsSN2SB4keoRQtQKE= +github.com/aws/aws-sdk-go-v2/service/athena v1.40.2/go.mod h1:77mOTf8npeWpoOpyF7Wj/75Pk1v5su7sBYqgGw46Emg= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.11.2 h1:Ji0DY1xUsUr3I8cHps0G+XM3WWU16lP6yG8qu1GAZAs= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.11.2/go.mod h1:5CsjAbs3NlGQyZNFACh+zztPDI7fU6eW9QsxjfnuBKg= +github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.3.5 h1:mbWNpfRUTT6bnacmvOTKXZjR/HycibdWzNpfbrbLDIs= +github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.3.5/go.mod h1:FCOPWGjsshkkICJIn9hq9xr6dLKtyaWpuUojiN3W1/8= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.11.7 h1:ogRAwT1/gxJBcSWDMZlgyFUM962F51A5CRhDLbxLdmo= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.11.7/go.mod h1:YCsIZhXfRPLFFCl5xxY+1T9RKzOKjCut+28JSX2DnAk= +github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.17.3 h1:4t+QEX7BsXz98W8W1lNvMAG+NX8qHz2CjLBxQKku40g= +github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.17.3/go.mod h1:oFcjjUq5Hm09N9rpxTdeMeLeQcxS7mIkBkL8qUKng+A= +github.com/aws/aws-sdk-go-v2/service/s3 v1.51.4 h1:lW5xUzOPGAMY7HPuNF4FdyBwRc3UJ/e8KsapbesVeNU= +github.com/aws/aws-sdk-go-v2/service/s3 v1.51.4/go.mod h1:MGTaf3x/+z7ZGugCGvepnx2DS6+caCYYqKhzVoLNYPk= +github.com/aws/aws-sdk-go-v2/service/sso v1.20.4 h1:WzFol5Cd+yDxPAdnzTA5LmpHYSWinhmSj4rQChV0ee8= +github.com/aws/aws-sdk-go-v2/service/sso v1.20.4/go.mod h1:qGzynb/msuZIE8I75DVRCUXw3o3ZyBmUvMwQ2t/BrGM= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.23.4 h1:Jux+gDDyi1Lruk+KHF91tK2KCuY61kzoCpvtvJJBtOE= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.23.4/go.mod h1:mUYPBhaF2lGiukDEjJX2BLRRKTmoUSitGDUgM4tRxak= +github.com/aws/aws-sdk-go-v2/service/sts v1.28.6 h1:cwIxeBttqPN3qkaAjcEcsh8NYr8n2HZPkcKgPAi1phU= +github.com/aws/aws-sdk-go-v2/service/sts v1.28.6/go.mod h1:FZf1/nKNEkHdGGJP/cI2MoIMquumuRK6ol3QQJNDxmw= +github.com/aws/smithy-go v1.20.2 h1:tbp628ireGtzcHDDmLT/6ADHidqnwgF57XOXZe6tp4Q= +github.com/aws/smithy-go v1.20.2/go.mod h1:krry+ya/rV9RDcV/Q16kpu6ypI4K2czasz0NC3qS14E= github.com/benbjohnson/clock v1.3.5 h1:VvXlSJBzZpA/zum6Sj74hxwYI2DIxRWuNIoXAzHZz5o= github.com/benbjohnson/clock v1.3.5/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA= github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= @@ -124,6 +162,8 @@ github.com/hashicorp/errwrap v1.1.0 h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= github.com/hashicorp/go-multierror v1.1.1 h1:H5DkEtf6CXdFp0N0Em5UCwQpXMWke8IA0+lD48awMYo= github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9/fmwbPZ6JB6eMoM= +github.com/hashicorp/go-uuid v1.0.2 h1:cfejS+Tpcp13yd5nYHWDI6qVCny6wyX2Mt5SGur2IGE= +github.com/hashicorp/go-uuid v1.0.2/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro= github.com/hashicorp/golang-lru v1.0.2 h1:dV3g9Z/unq5DpblPpw+Oqcv4dU/1omnb4Ok8iPY6p1c= github.com/hashicorp/golang-lru v1.0.2/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4= github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= @@ -134,6 +174,8 @@ 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/jcmturner/gofork v1.0.0 h1:J7uCkflzTEhUZ64xqKnkDxq3kzc96ajM1Gli5ktUem8= +github.com/jcmturner/gofork v1.0.0/go.mod h1:MK8+TM0La+2rjBD4jE12Kj1pCCxK7d2LK/UM3ncEo0o= github.com/jessevdk/go-flags v1.6.1 h1:Cvu5U8UGrLay1rZfv/zP7iLpSHGUZ/Ou68T0iX1bBK4= github.com/jessevdk/go-flags v1.6.1/go.mod h1:Mk8T1hIAWpOiJiHa9rJASDK2UGWji0EuPGBnNLMooyc= github.com/jmoiron/sqlx v1.4.0 h1:1PLqN7S1UYp5t4SrVVnt4nUVNemrDAtxlulVe+Qgm3o= @@ -207,6 +249,8 @@ github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRI github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/poy/onpar v1.1.2 h1:QaNrNiZx0+Nar5dLgTVp5mXkyoVFIbepjyEoGSnhbAY= github.com/poy/onpar v1.1.2/go.mod h1:6X8FLNoxyr9kkmnlqpK6LSoiOtrO6MICtWwEuWkLjzg= +github.com/prestodb/presto-go-client v0.0.0-20201204133205-8958eb37e584 h1:utxLhK2pYK3qDug5474YP2bbm2dbAPXA+dDrTvrkxeU= +github.com/prestodb/presto-go-client v0.0.0-20201204133205-8958eb37e584/go.mod h1:cwaFkElLIrI4vTXo5A1oDobUBFad0aVtZiZvfxJyX6I= 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= @@ -223,6 +267,10 @@ github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0t github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= github.com/rubenv/sql-migrate v1.8.1 h1:EPNwCvjAowHI3TnZ+4fQu3a915OpnQoPAjTXCGOy2U0= github.com/rubenv/sql-migrate v1.8.1/go.mod h1:BTIKBORjzyxZDS6dzoiw6eAFYJ1iNlGAtjn4LGeVjS8= +github.com/satori/go.uuid v1.2.0 h1:0uYX9dsZ2yD7q2RtLRtPSdGDWzjeM3TbMJP9utgA0ww= +github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0= +github.com/speee/go-athena v1.0.5 h1:zGIEPtMoQ4uFuB1NMozkGb71kcvDSqActnbGtdF4Bts= +github.com/speee/go-athena v1.0.5/go.mod h1:+udzNpwA9Sv/Y+5CmBQ7KafRFlzd1GNGgg7Nak6kWqQ= github.com/spf13/pflag v1.0.10 h1:4EBh2KAYBwaONj6b2Ye1GiHfwjqyROoF4RwYO+vPwFk= github.com/spf13/pflag v1.0.10/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= @@ -230,6 +278,7 @@ github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSS github.com/stretchr/objx v0.5.3 h1:jmXUvGomnU1o3W/V5h2VEradbpJDwGrzugQQvL0POH4= github.com/stretchr/objx v0.5.3/go.mod h1:rDQraq+vQZU7Fde9LOZLr8Tax6zZvy4kuNKF+QYS+U0= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA= github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.7.5/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= @@ -271,21 +320,27 @@ go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ= go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ= go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc= go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20200221231518-2aa609cf4a9d/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= golang.org/x/crypto v0.53.0 h1:QZ4Muo8THX6CizN2vPPd5fBGHyogrdK9fG4wLPFUsto= golang.org/x/crypto v0.53.0/go.mod h1:DNLU434OwVakk9PzuwV8w62mAJpRJL3vsgcfp4Qnsio= golang.org/x/exp v0.0.0-20260611194520-c48552f49976 h1:X8Hz2ImujgbmetVuW+w2YkyZChE3cBpZi2P158rTG9M= golang.org/x/exp v0.0.0-20260611194520-c48552f49976/go.mod h1:vnf4pv9iKZXY58sQE1L86zmNWJ4159e1RkcWiLCkeEY= +golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= golang.org/x/net v0.56.0 h1:Rw8j/hFzGvJUZwNBXnAtf5sVDVt+65SK2C7IxCxZt5o= golang.org/x/net v0.56.0/go.mod h1:D3Ku6r+V6JROoZK144D2XfMHFcMq/0zSfLelVTCFKec= golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs= golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q= 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.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.46.0 h1:noSf2Fq6F8DBgS+LysIkx7rIExoNHJsxOAtPp4rthXw= golang.org/x/sys v0.46.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/term v0.44.0 h1:0rLvDRCtNj0gZkyIXhCyOb2OAzEhLVqc4B+hrsBhrmc= golang.org/x/term v0.44.0/go.mod h1:7ze4MdzUzLXpSAoFP1H0bOI9aXDqveSvatT5vKcFh2Y= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.38.0 h1:sXmwo9DwP3OK9EZ7PqAdaooSGozfl/3a6/xJcbzPRhE= golang.org/x/text v0.38.0/go.mod h1:YXZt3QhHUKYT53r2lLKFIVi6Ao1jdzrTR/KQ09qyxF4= golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= @@ -307,6 +362,17 @@ gopkg.in/evanphx/json-patch.v4 v4.13.0 h1:czT3CmqEaQ1aanPc5SdlgQrrEIb8w/wwCvWWnf gopkg.in/evanphx/json-patch.v4 v4.13.0/go.mod h1:p8EYWUEYMpynmqDbY58zCKCFZw8pRWMG4EsWvDvM72M= gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc= gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw= +gopkg.in/jcmturner/aescts.v1 v1.0.1 h1:cVVZBK2b1zY26haWB4vbBiZrfFQnfbTVrE3xZq6hrEw= +gopkg.in/jcmturner/aescts.v1 v1.0.1/go.mod h1:nsR8qBOg+OucoIW+WMhB3GspUQXq9XorLnQb9XtvcOo= +gopkg.in/jcmturner/dnsutils.v1 v1.0.1 h1:cIuC1OLRGZrld+16ZJvvZxVJeKPsvd5eUIvxfoN5hSM= +gopkg.in/jcmturner/dnsutils.v1 v1.0.1/go.mod h1:m3v+5svpVOhtFAP/wSz+yzh4Mc0Fg7eRhxkJMWSIz9Q= +gopkg.in/jcmturner/goidentity.v3 v3.0.0 h1:1duIyWiTaYvVx3YX2CYtpJbUFd7/UuPYCfgXtQ3VTbI= +gopkg.in/jcmturner/goidentity.v3 v3.0.0/go.mod h1:oG2kH0IvSYNIu80dVAyu/yoefjq1mNfM5bm88whjWx4= +gopkg.in/jcmturner/gokrb5.v6 v6.1.1 h1:n0KFjpbuM5pFMN38/Ay+Br3l91netGSVqHPHEXeWUqk= +gopkg.in/jcmturner/gokrb5.v6 v6.1.1/go.mod h1:NFjHNLrHQiruory+EmqDXCGv6CrjkeYeA+bR9mIfNFk= +gopkg.in/jcmturner/rpc.v1 v1.1.0 h1:QHIUxTX1ISuAv9dD2wJ9HWQVuWDX/Zc0PfeC2tjc4rU= +gopkg.in/jcmturner/rpc.v1 v1.1.0/go.mod h1:YIdkC4XfD6GXbzje11McwsDuOlZQSb9W4vfLvuNnlv8= +gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= 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/lib/history/athena.go b/lib/history/athena.go index 52f6f83..e6dc4f9 100755 --- a/lib/history/athena.go +++ b/lib/history/athena.go @@ -9,9 +9,132 @@ import ( "strings" "time" + "github.com/flachnetz/startup/v2/lib/clock" + "github.com/flachnetz/startup/v2/lib/ql" "github.com/flachnetz/startup/v2/startup_base" + sl "github.com/flachnetz/startup/v2/startup_logging" + + // registers the "athena" database/sql driver used by AthenaQuery. + _ "github.com/speee/go-athena" +) + +// Default thresholds applied by RecordsAt when AthenaConfig leaves them zero. +const ( + defaultLookupThreshold = 24 * time.Hour + defaultLookbackMargin = 30 * time.Minute ) +// AthenaConfig enables the Athena read fallback on a Service (see WithAthena). +type AthenaConfig struct { + // required Athena configuration + Database string + Table string + WorkGroup string + OutputLocation string + + // optional AWS region + Region string + + // LookupThreshold selects Athena over the local table once the tracked + // object is older than this. Defaults to 24h when zero. + LookupThreshold time.Duration + + // LookbackMargin is subtracted from the object creation time to form the + // Athena query's MinTimestamp, bounding the scanned partitions. Defaults + // to 30m when zero. + LookbackMargin time.Duration +} + +// RecordsAt returns the records for groupId. With Athena configured (WithAthena) +// it bridges the local table and long-term Athena storage, because a cleanup job +// deletes local rows after a retention window while Athena keeps them forever: +// +// - createdTime older than AthenaConfig.LookupThreshold: read from Athena, +// falling back to the local table if Athena fails. +// - createdTime within the threshold: read the local table only. +// - createdTime zero (age unknown): read the local table first and, only when +// it returns nothing, fall back to Athena — so records aged out of the local +// table are still found. +// +// Any Athena failure is logged and never fails the read. +func (h *Service) RecordsAt(ctx ql.TxContext, groupId GroupId, createdTime time.Time) ([]Record, error) { + cfg := h.athena + if cfg == nil { + return h.Records(ctx, groupId) + } + + threshold := cfg.LookupThreshold + if threshold <= 0 { + threshold = defaultLookupThreshold + } + + knownOld := !createdTime.IsZero() && clock.GlobalClock.Now().Sub(createdTime) > threshold + if knownOld { + if records, ok := h.recordsFromAthena(ctx, cfg, groupId, createdTime); ok { + return records, nil + } + return h.Records(ctx, groupId) + } + + records, err := h.Records(ctx, groupId) + if err != nil { + return nil, err + } + + // ponytail: unknown-age group with an empty local table always hits Athena + // (aged out, or never existed). Fine for the rare history-page render; pass + // createdTime when the caller knows it to skip this probe. + if len(records) == 0 && createdTime.IsZero() { + if athenaRecords, ok := h.recordsFromAthena(ctx, cfg, groupId, time.Time{}); ok { + return athenaRecords, nil + } + } + + return records, nil +} + +// recordsFromAthena runs the Athena query for groupId. A non-zero createdTime +// bounds the scan to createdTime-LookbackMargin; a zero createdTime means no +// lower bound (full scan). ok is false when Athena is misconfigured or errored, +// signalling the caller to fall back to the local table. +func (h *Service) recordsFromAthena(ctx ql.TxContext, cfg *AthenaConfig, groupId GroupId, createdTime time.Time) (_ []Record, ok bool) { + loc, err := url.Parse(cfg.OutputLocation) + if err != nil { + sl.LoggerOf(ctx).WarnContext(ctx, "invalid athena output location, using local history", sl.Error(err)) + return nil, false + } + + var minTimestamp *time.Time + if !createdTime.IsZero() { + margin := cfg.LookbackMargin + if margin <= 0 { + margin = defaultLookbackMargin + } + minTimestamp = new(createdTime.Add(-margin)) + } + + query := AthenaQuery{ + GroupId: groupId, + Database: cfg.Database, + Table: cfg.Table, + WorkGroup: cfg.WorkGroup, + OutputLocation: loc, + Region: cfg.Region, + MinTimestamp: minTimestamp, + } + + queryCtx, cancel := context.WithTimeout(ctx, 15*time.Second) + defer cancel() + + records, err := query.Records(queryCtx) + if err != nil { + sl.LoggerOf(ctx).WarnContext(ctx, "failed to load history from athena, using local history", sl.Error(err)) + return nil, false + } + + return records, true +} + type AthenaQuery struct { GroupId GroupId diff --git a/lib/history/lib.go b/lib/history/lib.go index 7944a8e..d6dd3fe 100755 --- a/lib/history/lib.go +++ b/lib/history/lib.go @@ -79,11 +79,22 @@ type Service struct { txStarter ql.TxStarter table pgx.Identifier eventSending *EventSending + athena *AthenaConfig // record to send out async queue chan RecordToSend } +// Option customizes a Service created with New. +type Option func(*Service) + +// WithAthena enables the Athena fallback for reads: RecordsAt loads records +// from Athena instead of the local table when the tracked object is older than +// AthenaConfig.LookupThreshold. See RecordsAt. +func WithAthena(cfg AthenaConfig) Option { + return func(s *Service) { s.athena = &cfg } +} + // New creates a new history.Service instance to trace events. By default the service writes // records to the history table given by table. // @@ -97,7 +108,7 @@ type Service struct { // // You can specify parameter table as nil to not write to the history table. If you specify an // EventSending config, history entries are then only sent out as events. -func New(txStarter ql.TxStarter, table pgx.Identifier, eventSending *EventSending) *Service { +func New(txStarter ql.TxStarter, table pgx.Identifier, eventSending *EventSending, opts ...Option) *Service { if eventSending != nil { if eventSending.EventSender == nil || eventSending.EventCreator == nil { panic(errors.New("history: EventSending requires both EventSender and EventCreator")) @@ -115,11 +126,15 @@ func New(txStarter ql.TxStarter, table pgx.Identifier, eventSending *EventSendin } } } - return &Service{ + s := &Service{ txStarter: txStarter, table: table, eventSending: eventSending, } + for _, opt := range opts { + opt(s) + } + return s } // Track records the given item under groupId. Depending on the Service configuration the diff --git a/lib/history/lib_test.go b/lib/history/lib_test.go index 85499ff..bbafe9c 100755 --- a/lib/history/lib_test.go +++ b/lib/history/lib_test.go @@ -2,6 +2,7 @@ package history import ( "context" + "encoding/json" "io" "testing" "time" @@ -201,3 +202,50 @@ func TestTrackAsyncFlushesQueuedRecords(t *testing.T) { return nil }) } + +// TestRecordsAtRoutesToLocal covers the RecordsAt paths that resolve from the +// local table without touching Athena: no Athena config, a zero createdTime with +// data still present locally, and a createdTime newer than the lookup threshold. +// The Athena-hitting paths (known-old, and empty-local fallback) are not +// exercised here because they require a live AWS Athena endpoint. +func TestRecordsAtRoutesToLocal(t *testing.T) { + db := testx.NewConnection(t, "history_migrations") + + testx.MustTransactErr(t, db, func(ctx ql.TxContext) error { + return CreateTable(ctx, "history") + }) + + testx.MustTransact(t, db, func(ctx ql.TxContext) { + New(db, pgx.Identifier{"history"}, nil). + Track(ctx, GroupId("group-1"), item{Value: "hello"}) + }) + + athenaCfg := AthenaConfig{Database: "db", Table: "t", WorkGroup: "wg", OutputLocation: "s3://bucket/out/"} + + cases := []struct { + name string + opts []Option + createdTime time.Time + }{ + {"no athena config", nil, time.Now().Add(-72 * time.Hour)}, + {"zero created time", []Option{WithAthena(athenaCfg)}, time.Time{}}, + {"newer than threshold", []Option{WithAthena(athenaCfg)}, time.Now()}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + service := New(db, pgx.Identifier{"history"}, nil, tc.opts...) + testx.MustTransactErr(t, db, func(ctx ql.TxContext) error { + records, err := service.RecordsAt(ctx, GroupId("group-1"), tc.createdTime) + require.NoError(t, err) + require.Len(t, records, 1) + require.Equal(t, "hello", func() string { + var i item + _ = json.Unmarshal(records[0].Payload, &i) + return i.Value + }()) + return nil + }) + }) + } +} diff --git a/lib/history/render.go b/lib/history/render.go index 630e8ca..9a7f062 100644 --- a/lib/history/render.go +++ b/lib/history/render.go @@ -88,11 +88,27 @@ func (h *Service) RenderPage(ctx context.Context, w io.Writer, groupId GroupId, return h.RenderPageSummary(ctx, w, groupId, title, nil) } +// RenderPageAt is RenderPage with an Athena fallback: records are loaded via +// RecordsAt using createdTime to decide between the local table and Athena. +func (h *Service) RenderPageAt(ctx context.Context, w io.Writer, groupId GroupId, title string, createdTime time.Time) error { + return h.renderPage(ctx, w, groupId, title, nil, createdTime) +} + // RenderPageSummary is RenderPage with an extra current-state summary rendered // above the ledger. func (h *Service) RenderPageSummary(ctx context.Context, w io.Writer, groupId GroupId, title string, summary []SummaryItem) error { + // zero time: RecordsAt always reads the local table. + return h.renderPage(ctx, w, groupId, title, summary, time.Time{}) +} + +// RenderPageSummaryAt is RenderPageSummary with the Athena fallback (see RenderPageAt). +func (h *Service) RenderPageSummaryAt(ctx context.Context, w io.Writer, groupId GroupId, title string, summary []SummaryItem, createdTime time.Time) error { + return h.renderPage(ctx, w, groupId, title, summary, createdTime) +} + +func (h *Service) renderPage(ctx context.Context, w io.Writer, groupId GroupId, title string, summary []SummaryItem, createdTime time.Time) error { records, err := ql.InNewTransactionWithResult(ctx, h.txStarter, func(ctx ql.TxContext) ([]Record, error) { - return h.Records(ctx, groupId) + return h.RecordsAt(ctx, groupId, createdTime) }) if err != nil { return fmt.Errorf("load records: %w", err) diff --git a/lib/history/singleton.go b/lib/history/singleton.go index d2ebceb..33055b2 100644 --- a/lib/history/singleton.go +++ b/lib/history/singleton.go @@ -6,6 +6,7 @@ import ( "fmt" "io" "log/slog" + "time" "github.com/flachnetz/startup/v2/lib/events" "github.com/flachnetz/startup/v2/lib/ql" @@ -24,6 +25,9 @@ type Options struct { HistoryTable string // EventCreator builds the event that is sent out for every tracked record. EventCreator EventCreator + // Athena, when set, enables the Athena read fallback for old records + // (see WithAthena and Service.RecordsAt). + Athena *AthenaConfig } // instance holds the global history Service initialized by InitializeGlobal. @@ -40,13 +44,18 @@ func InitializeGlobal(ctx context.Context, opts Options) error { return fmt.Errorf("create history table: %w", err) } + var histOpts []Option + if opts.Athena != nil { + histOpts = append(histOpts, WithAthena(*opts.Athena)) + } + instance = New(opts.DB, pgx.Identifier{opts.HistoryTable}, &EventSending{ EventSender: events.Sender, EventCreator: opts.EventCreator, ServiceId: opts.ServiceId, ServiceVersion: startup_base.BuildVersion, WriteToOutbox: true, - }) + }, histOpts...) // start the background task that flushes records tracked outside of a // transaction. Cancel ctx to stop it. @@ -77,6 +86,16 @@ func RenderPage[T ~string](ctx context.Context, w io.Writer, groupId T, title st return instance.RenderPage(ctx, w, GroupId(groupId), title) } +// RenderPageAt is RenderPage with the Athena fallback: createdTime decides +// whether records are read from the local table or from Athena. +func RenderPageAt[T ~string](ctx context.Context, w io.Writer, groupId T, title string, createdTime time.Time) error { + if instance == nil { + return errors.New("history: global instance not initialized") + } + + return instance.RenderPageAt(ctx, w, GroupId(groupId), title, createdTime) +} + // RenderPageSummary is RenderPage with a current-state summary above the ledger. func RenderPageSummary[T ~string](ctx context.Context, w io.Writer, groupId T, title string, summary []SummaryItem) error { if instance == nil { @@ -85,3 +104,22 @@ func RenderPageSummary[T ~string](ctx context.Context, w io.Writer, groupId T, t return instance.RenderPageSummary(ctx, w, GroupId(groupId), title, summary) } + +// RenderPageSummaryAt is RenderPageSummary with the Athena fallback (see RenderPageAt). +func RenderPageSummaryAt[T ~string](ctx context.Context, w io.Writer, groupId T, title string, summary []SummaryItem, createdTime time.Time) error { + if instance == nil { + return errors.New("history: global instance not initialized") + } + + return instance.RenderPageSummaryAt(ctx, w, GroupId(groupId), title, summary, createdTime) +} + +// RecordsAt uses the global history singleton to load records for groupId with +// the Athena fallback (see Service.RecordsAt). +func RecordsAt[T ~string](ctx ql.TxContext, groupId T, createdTime time.Time) ([]Record, error) { + if instance == nil { + return nil, errors.New("history: global instance not initialized") + } + + return instance.RecordsAt(ctx, GroupId(groupId), createdTime) +}