FEAT: Added Redis Caching
This commit is contained in:
FahdShalhoub
@@ -23,10 +23,12 @@ catalogue search — see below) and the Swagger UI. Migration `0001_init` still
|
||||
creates nothing (it predates the domain and exists only because
|
||||
`internal/db/migrate.go` embeds `migrations/*.sql`, which will not compile
|
||||
against an empty directory); `0002_create_videos_table` is where the schema
|
||||
actually starts, and `0003`–`0005` add what the search needs. It now has an `internal/services` (SQS only) and an
|
||||
actually starts, and `0003`–`0005` add what the search needs. It now has an `internal/services` (SQS and Redis) and an
|
||||
`internal/consumers`, and a task role scoped to that one queue — receive,
|
||||
delete, get-attributes, and nothing else. It cannot publish back onto the
|
||||
topic: it is a subscriber, not a participant.
|
||||
topic: it is a subscriber, not a participant. The Redis client needs nothing
|
||||
from that role: ElastiCache is reached over the Redis protocol on the private
|
||||
network, so the security group is what grants access, not IAM.
|
||||
|
||||
## Commands
|
||||
|
||||
@@ -49,7 +51,7 @@ never had a v1.
|
||||
|
||||
```bash
|
||||
cd discovery
|
||||
go run . # serves on :8080 (requires DB_* plus AWS_REGION/CATALOGUE_EVENTS_QUEUE_URL — no S3/MediaConvert)
|
||||
go run . # serves on :8080 (requires DB_* plus AWS_REGION/CATALOGUE_EVENTS_QUEUE_URL/REDIS_ADDR — no S3/MediaConvert)
|
||||
go run . migrate # applies pending DB migrations, then exits (no HTTP server)
|
||||
go build ./...
|
||||
go vet ./...
|
||||
@@ -58,7 +60,8 @@ go vet ./...
|
||||
Its OpenAPI spec is generated into `discovery/docs` by the same `swag init`
|
||||
invocation as `cms`, run from `discovery/`. In `docker-compose.yml` it is a
|
||||
service of its own on `127.0.0.1:8080`, wired to the `discovery` database and
|
||||
role LocalStack provisions, plus the AWS_* vars its consumer needs — and, like
|
||||
role LocalStack provisions, plus the AWS_* vars its consumer needs and
|
||||
`REDIS_ADDR` pointing at the `redis` container — and, like
|
||||
`cms`, started in server mode only, so a freshly created local database needs
|
||||
`docker exec discovery ./discovery migrate` once. Until that runs, the consumer
|
||||
logs `relation "videos" does not exist` per announcement and leaves them on the
|
||||
@@ -124,6 +127,41 @@ browse path (no title term) is a true index seek — that is what
|
||||
`videos_recent_idx` is for, and it takes a 200k-row browse from ~15 ms to
|
||||
~0.03 ms.
|
||||
|
||||
#### Search cache (Redis / ElastiCache)
|
||||
|
||||
Because of that caveat, the search reads through a Redis cache before it
|
||||
touches Postgres — `handlers.cachedSearch`/`cacheSearch` around the repository
|
||||
call, storing the `api.SearchResults` a page serialises to. `REDIS_ADDR` points
|
||||
at the ElastiCache node `infrastructure/main.go` provisions (a `redis`
|
||||
container under compose). Locally it takes a repeated ranked search from
|
||||
~6.8 ms to ~0.5 ms.
|
||||
|
||||
- **The key is `discovery:search:v1:` + a canonical JSON encoding of
|
||||
`{t: title, c: categories, l: limit, p: cursor}`.** The cursor *is* the page
|
||||
identity — keyset paging has no page number to key on. `limit` is in the key
|
||||
because the same title/categories/cursor at a different limit is a different
|
||||
set of rows. JSON rather than pasted-together separators so a category name
|
||||
containing `,` or `|` cannot be mistaken for a boundary. The title is
|
||||
lowercased (`tsQueryFor` folds case anyway) but the **categories are not**:
|
||||
`categories && $2` compares them verbatim, so `News` and `news` really are
|
||||
different searches. Categories are sorted and deduped — `&&` means "any of
|
||||
these", so order never changed the answer — and never nil, so `"categories":
|
||||
[]` and an omitted member share an entry.
|
||||
- **TTL is 60 s and nothing invalidates on write.** The consumer writes rows
|
||||
continuously and would have to know which cached pages a new title belongs
|
||||
on — for a ranked search, every page it outranks. Expiry is cheaper and
|
||||
bounds staleness to roughly the announcement's own delivery lag.
|
||||
- **Only 200s are cached**, so a malformed cursor still reaches the repository
|
||||
and is still a 400.
|
||||
- **Every cache failure is a miss.** A Get/Set error is logged and the search
|
||||
is answered from Postgres; the service also *starts* with an unreachable
|
||||
cache, warning rather than panicking as the DB and SQS assertions do — a slow
|
||||
read side beats no read side. Each call is bounded by
|
||||
`services.cacheOperationTimeout` (100 ms, retries off), because an unbounded
|
||||
cache miss on a dead node costs the dial timeout *plus* the query it was
|
||||
avoiding. A total cache outage adds ~200 ms per search (a failed read and a
|
||||
failed write), not seconds.
|
||||
|
||||
`cms`, `discovery` and `infrastructure` hold no test files of their own. The
|
||||
only tests in the repo are the black-box BDD scenarios in `tests/` — see below.
|
||||
|
||||
@@ -353,6 +391,18 @@ on boot if any required var is empty.
|
||||
backups — intentionally minimal). Each app (`cms`, `discovery`) gets its
|
||||
own login role and same-named database via the `postgresql` provider
|
||||
(`newServiceDatabase`), so services never share DB credentials.
|
||||
- **ElastiCache**: one `cache.t4g.micro` Redis node (`search-cache`, engine
|
||||
7.1, single-AZ, no replica, no snapshots) that `discovery` answers repeated
|
||||
catalogue searches from. Its contents are derivable from Postgres by
|
||||
definition, so there is nothing to back up. Its security group admits 6379
|
||||
from `ecs-service-sg` **only** — narrower than the database's, which also
|
||||
admits the deployer's IP for the `postgresql` provider; there is nothing to
|
||||
administer here from a laptop. The endpoint is the single node's address
|
||||
(`CacheNodes[0]`, not `ConfigurationEndpoint` — that is a Memcached thing),
|
||||
exported as `searchCacheAddress` and injected as `REDIS_ADDR`. Skipped under
|
||||
LocalStack, where docker-compose runs a plain `redis:7-alpine` container
|
||||
instead: ElastiCache picks its own endpoint, and compose needs a literal
|
||||
`REDIS_ADDR` before anything is provisioned.
|
||||
- **ECS Fargate**: one cluster (`app-cluster`), one ALB *per service* (each
|
||||
gets its own DNS name rather than sharing a load balancer on different
|
||||
ports). `deployFargateService(...)` is the shared helper building a
|
||||
@@ -402,7 +452,9 @@ on boot if any required var is empty.
|
||||
- `discovery-task-role` — the `discovery` container's own AWS identity, and
|
||||
its only one: receive/delete/get-attributes on
|
||||
`discovery-catalogue-events`. No S3, no MediaConvert, and no `sns:Publish`
|
||||
— it consumes the catalogue, it does not add to it.
|
||||
— it consumes the catalogue, it does not add to it. The search cache is
|
||||
absent from it on purpose: ElastiCache is reached over the Redis protocol
|
||||
inside the VPC, so `search-cache-sg` is the grant, not IAM.
|
||||
- `mediaconvert-service-role` — trusted by `mediaconvert.amazonaws.com`,
|
||||
not by ECS; the role MediaConvert itself assumes (passed as
|
||||
`CreateJobInput.Role`) to read `raw-uploads-bucket` and write
|
||||
|
||||
+6
-2
@@ -3,8 +3,11 @@ module thamanyah/discovery
|
||||
go 1.25.12
|
||||
|
||||
require (
|
||||
github.com/aws/aws-sdk-go-v2/config v1.33.1
|
||||
github.com/aws/aws-sdk-go-v2/service/sqs v1.48.1
|
||||
github.com/golang-migrate/migrate/v4 v4.19.1
|
||||
github.com/lib/pq v1.12.3
|
||||
github.com/redis/go-redis/v9 v9.22.0
|
||||
github.com/swaggo/http-swagger/v2 v2.0.2
|
||||
github.com/swaggo/swag v1.16.6
|
||||
)
|
||||
@@ -12,7 +15,6 @@ require (
|
||||
require (
|
||||
github.com/KyleBanks/depth v1.2.1 // indirect
|
||||
github.com/aws/aws-sdk-go-v2 v1.45.1 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/config v1.33.1 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/credentials v1.20.1 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.19.1 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/internal/configsources v1.5.1 // indirect
|
||||
@@ -21,11 +23,11 @@ require (
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.19 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.14.1 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/signin v1.7.1 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/sqs v1.48.1 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/sso v1.35.1 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/ssooidc v1.40.1 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/sts v1.47.1 // indirect
|
||||
github.com/aws/smithy-go v1.28.1 // indirect
|
||||
github.com/cespare/xxhash/v2 v2.3.0 // indirect
|
||||
github.com/go-openapi/jsonpointer v0.19.5 // indirect
|
||||
github.com/go-openapi/jsonreference v0.20.0 // indirect
|
||||
github.com/go-openapi/spec v0.20.6 // indirect
|
||||
@@ -33,8 +35,10 @@ require (
|
||||
github.com/josharian/intern v1.0.0 // indirect
|
||||
github.com/mailru/easyjson v0.7.6 // indirect
|
||||
github.com/swaggo/files/v2 v2.0.0 // indirect
|
||||
go.uber.org/atomic v1.11.0 // indirect
|
||||
golang.org/x/mod v0.29.0 // indirect
|
||||
golang.org/x/sync v0.18.0 // indirect
|
||||
golang.org/x/sys v0.38.0 // indirect
|
||||
golang.org/x/tools v0.38.0 // indirect
|
||||
gopkg.in/yaml.v2 v2.4.0 // indirect
|
||||
)
|
||||
|
||||
@@ -34,6 +34,12 @@ github.com/aws/aws-sdk-go-v2/service/sts v1.47.1 h1:Sv2xPnRHlThSUtVujYuUBPI/Il8s
|
||||
github.com/aws/aws-sdk-go-v2/service/sts v1.47.1/go.mod h1:mKo/CzaCz8qytGW70NG4vIIGAx1HXTlb5lHNkC5k3lk=
|
||||
github.com/aws/smithy-go v1.28.1 h1:R/nXH00c8qcfCzQVELtRw+eLQWtzv+VAIEFJ1/xxXlQ=
|
||||
github.com/aws/smithy-go v1.28.1/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc=
|
||||
github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
|
||||
github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
|
||||
github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
|
||||
github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
|
||||
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/containerd/errdefs v1.0.0 h1:tg5yIfIlQIrxYtu9ajqY42W3lpS19XqdxRQeEwYG8PI=
|
||||
github.com/containerd/errdefs v1.0.0/go.mod h1:+YBYIdtsnF4Iw6nWZhJcqGSg/dwvV7tyJ/kCkyJ2k+M=
|
||||
github.com/containerd/errdefs/pkg v0.3.0 h1:9IKJ06FvyNlexW690DXuQNx2KA2cUJXx151Xdx3ZPPE=
|
||||
@@ -77,6 +83,8 @@ github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
|
||||
github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
|
||||
github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY=
|
||||
github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y=
|
||||
github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE=
|
||||
github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0=
|
||||
github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
|
||||
github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
|
||||
github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
|
||||
@@ -105,6 +113,8 @@ github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINE
|
||||
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
|
||||
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/redis/go-redis/v9 v9.22.0 h1:laDvpYXTJtZLloinw1fA5Kqd6HAEH2XKxOkG/PDq2F0=
|
||||
github.com/redis/go-redis/v9 v9.22.0/go.mod h1:y2g0Wj8rQvuK0ELM+oxSudcLtC09JScs98I/X9gRWY4=
|
||||
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.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
|
||||
@@ -116,6 +126,8 @@ github.com/swaggo/http-swagger/v2 v2.0.2 h1:FKCdLsl+sFCx60KFsyM0rDarwiUSZ8DqbfSy
|
||||
github.com/swaggo/http-swagger/v2 v2.0.2/go.mod h1:r7/GBkAWIfK6E/OLnE8fXnviHiDeAHmgIyooa4xm3AQ=
|
||||
github.com/swaggo/swag v1.16.6 h1:qBNcx53ZaX+M5dxVyTrgQ0PJ/ACK+NzhwcbieTt+9yI=
|
||||
github.com/swaggo/swag v1.16.6/go.mod h1:ngP2etMK5a0P3QBizic5MEwpRmluJZPHjXcMoj4Xesg=
|
||||
github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs=
|
||||
github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s=
|
||||
go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA=
|
||||
go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A=
|
||||
go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0 h1:F7Jx+6hwnZ41NSFTO5q4LYDtJRXBf2PD0rNBkeB/lus=
|
||||
@@ -126,6 +138,8 @@ go.opentelemetry.io/otel/metric v1.37.0 h1:mvwbQS5m0tbmqML4NqK+e3aDiO02vsf/Wgbsd
|
||||
go.opentelemetry.io/otel/metric v1.37.0/go.mod h1:04wGrZurHYKOc+RKeye86GwKiTb9FKm1WHtO+4EVr2E=
|
||||
go.opentelemetry.io/otel/trace v1.37.0 h1:HLdcFNbRQBE2imdSEgm/kwqmQj1Or1l/7bW6mxVK7z4=
|
||||
go.opentelemetry.io/otel/trace v1.37.0/go.mod h1:TlgrlQ+PtQO5XFerSPUYG0JSgGyryXewPGyayAWSBS0=
|
||||
go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
|
||||
go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
|
||||
golang.org/x/mod v0.29.0 h1:HV8lRxZC4l2cr3Zq1LvtOsi/ThTgWnUk/y64QSs8GwA=
|
||||
golang.org/x/mod v0.29.0/go.mod h1:NyhrlYXJ2H4eJiRy/WDBO6HMqZQ6q9nk4JzS3NuCK+w=
|
||||
golang.org/x/sync v0.18.0 h1:kr88TuHDroi+UVf+0hZnirlk8o8T+4MrK6mr60WkH/I=
|
||||
|
||||
@@ -1,12 +1,17 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"log"
|
||||
"net/http"
|
||||
"slices"
|
||||
"strings"
|
||||
"thamanyah/discovery/internal/api"
|
||||
"thamanyah/discovery/internal/db/repositories"
|
||||
"thamanyah/discovery/internal/services"
|
||||
"time"
|
||||
)
|
||||
|
||||
// GetVideo serves the catalogue's copy of one video.
|
||||
@@ -64,8 +69,112 @@ const (
|
||||
// cursor still reaches the rest — because a client that wants everything
|
||||
// is doing something legitimate, just not in one request.
|
||||
maxSearchPageSize = 100
|
||||
|
||||
// searchCacheTTL is how long a page of results stays servable from the
|
||||
// cache before it has to be built again.
|
||||
//
|
||||
// Nothing invalidates an entry when a video is announced: the consumer
|
||||
// writes rows continuously and would have to know which of the cached
|
||||
// pages a new title belongs on, which for a ranked search is every page it
|
||||
// outranks. Expiry is the cheaper answer, and it bounds the staleness — a
|
||||
// video is findable within a minute of being catalogued, which is the same
|
||||
// order as the announcement's own delivery lag.
|
||||
searchCacheTTL = time.Minute
|
||||
)
|
||||
|
||||
// searchCacheKey is what one page of search results is stored under: the title
|
||||
// searched for, the categories it was narrowed to, and the page within that
|
||||
// result set.
|
||||
//
|
||||
// The page is named by the cursor, because with keyset paging that is what a
|
||||
// page *is* — there is no page number to key on, and the cursor identifies the
|
||||
// same rows whenever it is presented. The empty cursor is the first page.
|
||||
//
|
||||
// The limit is in the key too, even though a reader would not call it part of
|
||||
// "which page". It has to be: the same title, categories and cursor with a
|
||||
// different limit is a different set of rows, and sharing an entry between
|
||||
// them would hand a client asking for 50 a page of 20.
|
||||
//
|
||||
// The parts are JSON-encoded rather than pasted together with separators so
|
||||
// that no category name can be mistaken for the boundary between two of them —
|
||||
// {"c":["a,b"]} and {"c":["a","b"]} stay distinguishable, which "a,b" and
|
||||
// "a,b" would not. The key stays readable in redis-cli either way.
|
||||
func searchCacheKey(title string, categories []string, limit int, cursor string) string {
|
||||
// The title is lowercased because the search itself is: tsQueryFor folds
|
||||
// case before it builds the tsquery, so two spellings that differ only in
|
||||
// case are the same search and should be the same entry. The categories
|
||||
// are not — they are compared to the stored array verbatim, so "News" and
|
||||
// "news" really do ask different questions.
|
||||
//
|
||||
// Sorting the categories is safe for the same reason the SQL uses &&:
|
||||
// they mean "any of these", so their order never changed the answer. Two
|
||||
// readers naming the same categories in a different order now share one
|
||||
// entry instead of building the same page twice.
|
||||
//
|
||||
// Never nil, so that a client sending "categories": [] and one omitting
|
||||
// the member entirely — the same search, since both mean "every category"
|
||||
// — land on one entry rather than on `null` and `[]`.
|
||||
sorted := append([]string{}, categories...)
|
||||
slices.Sort(sorted)
|
||||
sorted = slices.Compact(sorted)
|
||||
|
||||
parts := struct {
|
||||
Title string `json:"t"`
|
||||
Categories []string `json:"c"`
|
||||
Limit int `json:"l"`
|
||||
Cursor string `json:"p"`
|
||||
}{
|
||||
Title: strings.ToLower(strings.TrimSpace(title)),
|
||||
Categories: sorted,
|
||||
Limit: limit,
|
||||
Cursor: cursor,
|
||||
}
|
||||
|
||||
// Cannot fail: every field is a string, a string slice or an int.
|
||||
encoded, _ := json.Marshal(parts)
|
||||
return "discovery:search:v1:" + string(encoded)
|
||||
}
|
||||
|
||||
// cachedSearch returns the page held under key, or false when there is none to
|
||||
// serve.
|
||||
//
|
||||
// Every failure reads as "no cached page": a cache that is unreachable, slow
|
||||
// or holding something unreadable must cost a reader nothing more than the
|
||||
// database query they would have paid for anyway. Only faults worth acting on
|
||||
// are logged — a miss is not one.
|
||||
func cachedSearch(ctx context.Context, key string) (api.SearchResults, bool) {
|
||||
encoded, err := services.CacheClient.Get(ctx, key)
|
||||
if errors.Is(err, services.ErrCacheMiss) {
|
||||
return api.SearchResults{}, false
|
||||
}
|
||||
if err != nil {
|
||||
log.Printf("Could Not Read The Search Cache, Falling Back To The Database: key=%q: %s", key, err)
|
||||
return api.SearchResults{}, false
|
||||
}
|
||||
|
||||
var results api.SearchResults
|
||||
if err := json.Unmarshal(encoded, &results); err != nil {
|
||||
// Only reachable if something else wrote this key, or the shape
|
||||
// changed without the v1 in the prefix changing with it.
|
||||
log.Printf("Could Not Read A Cached Search Page, Falling Back To The Database: key=%q: %s", key, err)
|
||||
return api.SearchResults{}, false
|
||||
}
|
||||
|
||||
return results, true
|
||||
}
|
||||
|
||||
func cacheSearch(ctx context.Context, key string, results api.SearchResults) {
|
||||
encoded, err := json.Marshal(results)
|
||||
if err != nil {
|
||||
log.Printf("Could Not Encode A Search Page For The Cache: key=%q: %s", key, err)
|
||||
return
|
||||
}
|
||||
|
||||
if err := services.CacheClient.Set(ctx, key, encoded, searchCacheTTL); err != nil {
|
||||
log.Printf("Could Not Write The Search Cache: key=%q: %s", key, err)
|
||||
}
|
||||
}
|
||||
|
||||
// SearchVideos serves the catalogue's lexical search.
|
||||
//
|
||||
// It is served on QUERY rather than GET or POST: a search is safe and
|
||||
@@ -94,6 +203,19 @@ func SearchVideos(w http.ResponseWriter, r *http.Request) {
|
||||
limit = maxSearchPageSize
|
||||
}
|
||||
|
||||
// Read through the cache before touching Postgres. A ranked search is the
|
||||
// expensive request this service serves — ts_rank is computed per matching
|
||||
// row, so deep pages scan rather than seek — and readers ask for the same
|
||||
// few things over and over, which is exactly the shape a cache pays for.
|
||||
//
|
||||
// Only successful pages are ever stored, so a malformed cursor still
|
||||
// reaches the repository and is still rejected below.
|
||||
cacheKey := searchCacheKey(request.Title, request.Categories, limit, request.Cursor)
|
||||
if cached, hit := cachedSearch(r.Context(), cacheKey); hit {
|
||||
writeJSON(w, http.StatusOK, cached)
|
||||
return
|
||||
}
|
||||
|
||||
found, next, err := repositories.VideoRepo.SearchVideos(r.Context(), request.Title, request.Categories, limit, request.Cursor)
|
||||
if errors.Is(err, repositories.ErrInvalidCursor) {
|
||||
writeProblem(w, http.StatusBadRequest, "Malformed Search Cursor",
|
||||
@@ -119,5 +241,8 @@ func SearchVideos(w http.ResponseWriter, r *http.Request) {
|
||||
})
|
||||
}
|
||||
|
||||
writeJSON(w, http.StatusOK, api.SearchResults{Videos: videos, NextCursor: next})
|
||||
results := api.SearchResults{Videos: videos, NextCursor: next}
|
||||
cacheSearch(r.Context(), cacheKey, results)
|
||||
|
||||
writeJSON(w, http.StatusOK, results)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,58 @@
|
||||
package services
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"time"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
var CacheClient Cache
|
||||
|
||||
var ErrCacheMiss = errors.New("cache: miss")
|
||||
|
||||
type Cache interface {
|
||||
Get(ctx context.Context, key string) ([]byte, error)
|
||||
Set(ctx context.Context, key string, value []byte, ttl time.Duration) error
|
||||
}
|
||||
|
||||
type RedisConcrete struct {
|
||||
Redis *redis.Client
|
||||
}
|
||||
|
||||
const cacheOperationTimeout = 100 * time.Millisecond
|
||||
|
||||
func (svc RedisConcrete) Get(ctx context.Context, key string) ([]byte, error) {
|
||||
ctx, cancel := context.WithTimeout(ctx, cacheOperationTimeout)
|
||||
defer cancel()
|
||||
|
||||
value, err := svc.Redis.Get(ctx, key).Bytes()
|
||||
if errors.Is(err, redis.Nil) {
|
||||
return nil, ErrCacheMiss
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return value, nil
|
||||
}
|
||||
|
||||
func (svc RedisConcrete) Set(ctx context.Context, key string, value []byte, ttl time.Duration) error {
|
||||
ctx, cancel := context.WithTimeout(ctx, cacheOperationTimeout)
|
||||
defer cancel()
|
||||
|
||||
return svc.Redis.Set(ctx, key, value, ttl).Err()
|
||||
}
|
||||
|
||||
func (svc RedisConcrete) Close() error {
|
||||
return svc.Redis.Close()
|
||||
}
|
||||
|
||||
func (svc RedisConcrete) AssertSuccessfulConnection(ctx context.Context) {
|
||||
if err := svc.Redis.Ping(ctx).Err(); err != nil {
|
||||
log.Printf("WARNING: %s", fmt.Errorf("redis: cannot reach cache at %q, searches will be served from the database: %w",
|
||||
svc.Redis.Options().Addr, err))
|
||||
}
|
||||
}
|
||||
@@ -11,10 +11,12 @@ import (
|
||||
"thamanyah/discovery/internal/db/repositories"
|
||||
"thamanyah/discovery/internal/handlers"
|
||||
"thamanyah/discovery/internal/services"
|
||||
"time"
|
||||
|
||||
awsconfig "github.com/aws/aws-sdk-go-v2/config"
|
||||
"github.com/aws/aws-sdk-go-v2/service/sqs"
|
||||
_ "github.com/lib/pq"
|
||||
"github.com/redis/go-redis/v9"
|
||||
httpSwagger "github.com/swaggo/http-swagger/v2"
|
||||
|
||||
_ "thamanyah/discovery/docs"
|
||||
@@ -65,6 +67,38 @@ func runServer() {
|
||||
|
||||
services.SQSClient = concreteSQSClient
|
||||
|
||||
// The ElastiCache Redis node the catalogue search is answered from. Not an
|
||||
// AWS API call — the address resolves inside the VPC and the security
|
||||
// group is what grants access — so it needs nothing from the task role and
|
||||
// nothing from awsConfig.
|
||||
//
|
||||
// The address is required because a missing one is a deployment mistake,
|
||||
// not a choice; whether the node actually answers is a different question,
|
||||
// and one the service deliberately survives getting "no" to.
|
||||
redisAddress := os.Getenv("REDIS_ADDR")
|
||||
if redisAddress == "" {
|
||||
panic(fmt.Errorf("missing required env var: REDIS_ADDR"))
|
||||
}
|
||||
|
||||
concreteCacheClient := &services.RedisConcrete{
|
||||
Redis: redis.NewClient(&redis.Options{
|
||||
Addr: redisAddress,
|
||||
// Per-call deadlines on top of the budget RedisConcrete already
|
||||
// applies, so a connection that hangs rather than refusing cannot
|
||||
// tie up a goroutine past the request that opened it. Retries are
|
||||
// off for the same reason the budget is small: a second attempt at
|
||||
// a cache costs more than the query it is saving.
|
||||
DialTimeout: 100 * time.Millisecond,
|
||||
ReadTimeout: 100 * time.Millisecond,
|
||||
WriteTimeout: 100 * time.Millisecond,
|
||||
MaxRetries: -1,
|
||||
}),
|
||||
}
|
||||
defer func() { _ = concreteCacheClient.Close() }()
|
||||
concreteCacheClient.AssertSuccessfulConnection(context.Background())
|
||||
|
||||
services.CacheClient = concreteCacheClient
|
||||
|
||||
concreteDBClient := db.CreateDBConnection(requireDBConnectionString())
|
||||
defer db.CloseConnection(concreteDBClient)
|
||||
db.AssertSuccessfulConnection(context.Background(), concreteDBClient)
|
||||
|
||||
@@ -126,6 +126,23 @@ services:
|
||||
- action: rebuild
|
||||
path: ./cms
|
||||
|
||||
# Stands in for the ElastiCache node infrastructure/main.go provisions on the
|
||||
# AWS stacks. Not provisioned through LocalStack like the buckets and the
|
||||
# database are: ElastiCache hands out an endpoint on a port it chooses, and
|
||||
# the literal REDIS_ADDR below has to be known before anything is created.
|
||||
# Nothing here is worth persisting — every entry is derivable from Postgres,
|
||||
# which is the whole point of a cache — so there is no volume.
|
||||
redis:
|
||||
container_name: "${REDIS_DOCKER_NAME:-redis}"
|
||||
image: redis:7-alpine
|
||||
ports:
|
||||
- "127.0.0.1:6379:6379"
|
||||
healthcheck:
|
||||
test: ["CMD", "redis-cli", "ping"]
|
||||
interval: 5s
|
||||
timeout: 3s
|
||||
retries: 10
|
||||
|
||||
discovery:
|
||||
container_name: "${DISCOVERY_DOCKER_NAME:-discovery}"
|
||||
build:
|
||||
@@ -143,6 +160,10 @@ services:
|
||||
- AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY:-test}
|
||||
- AWS_ENDPOINT_URL=${AWS_ENDPOINT_URL:-http://localhost.localstack.cloud:4566}
|
||||
- CATALOGUE_EVENTS_QUEUE_URL=${CATALOGUE_EVENTS_QUEUE_URL:-http://localhost.localstack.cloud:4566/000000000000/discovery-catalogue-events}
|
||||
# The catalogue search's cache — the redis service below, standing in for
|
||||
# ElastiCache. main.go panics without an address but only warns if the
|
||||
# node does not answer: a cold cache means slower searches, not none.
|
||||
- REDIS_ADDR=${REDIS_ADDR:-redis:6379}
|
||||
# Postgres: its own role and database, provisioned alongside the cms ones
|
||||
# by newServiceDatabase, so the two services share no credentials.
|
||||
- DB_HOST=${DISCOVERY_DB_HOST:-localhost.localstack.cloud}
|
||||
@@ -154,6 +175,8 @@ services:
|
||||
depends_on:
|
||||
infra:
|
||||
condition: service_completed_successfully
|
||||
redis:
|
||||
condition: service_healthy
|
||||
develop:
|
||||
watch:
|
||||
# Same as cms: a compiled binary, so every change means a new image.
|
||||
|
||||
@@ -13,6 +13,7 @@ import (
|
||||
"github.com/pulumi/pulumi-aws/sdk/v7/go/aws/ec2"
|
||||
"github.com/pulumi/pulumi-aws/sdk/v7/go/aws/ecr"
|
||||
"github.com/pulumi/pulumi-aws/sdk/v7/go/aws/ecs"
|
||||
"github.com/pulumi/pulumi-aws/sdk/v7/go/aws/elasticache"
|
||||
"github.com/pulumi/pulumi-aws/sdk/v7/go/aws/iam"
|
||||
"github.com/pulumi/pulumi-aws/sdk/v7/go/aws/lb"
|
||||
"github.com/pulumi/pulumi-aws/sdk/v7/go/aws/rds"
|
||||
@@ -384,6 +385,7 @@ func main() {
|
||||
cmsRepo, discoveryRepo *ecr.Repository
|
||||
cmsService, discoveryService *ecs.Service
|
||||
cmsAlb, discoveryAlb *lb.LoadBalancer
|
||||
searchCacheAddress pulumi.StringOutput
|
||||
)
|
||||
|
||||
// Bucket names are left to Pulumi's auto-naming in AWS, but pinned under
|
||||
@@ -771,6 +773,67 @@ func main() {
|
||||
}
|
||||
}
|
||||
|
||||
if !localstack {
|
||||
cacheSubnetGroup, err := elasticache.NewSubnetGroup(ctx, "search-cache-subnet-group", &elasticache.SubnetGroupArgs{
|
||||
SubnetIds: pulumi.ToStringArray(subnetIDs),
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
cacheSecurityGroup, err := ec2.NewSecurityGroup(ctx, "search-cache-sg", &ec2.SecurityGroupArgs{
|
||||
Description: pulumi.String("Allow the ECS tasks to reach the search cache"),
|
||||
VpcId: pulumi.String(vpcID),
|
||||
Ingress: ec2.SecurityGroupIngressArray{
|
||||
&ec2.SecurityGroupIngressArgs{
|
||||
Protocol: pulumi.String("tcp"),
|
||||
FromPort: pulumi.Int(6379),
|
||||
ToPort: pulumi.Int(6379),
|
||||
SecurityGroups: pulumi.StringArray{serviceSecurityGroup.ID()},
|
||||
Description: pulumi.String("Redis, from the service tasks only"),
|
||||
},
|
||||
},
|
||||
Egress: ec2.SecurityGroupEgressArray{
|
||||
&ec2.SecurityGroupEgressArgs{
|
||||
Protocol: pulumi.String("-1"),
|
||||
FromPort: pulumi.Int(0),
|
||||
ToPort: pulumi.Int(0),
|
||||
CidrBlocks: pulumi.ToStringArray([]string{"0.0.0.0/0"}),
|
||||
},
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
searchCache, err := elasticache.NewCluster(ctx, "search-cache", &elasticache.ClusterArgs{
|
||||
Engine: pulumi.String("redis"),
|
||||
EngineVersion: pulumi.String("7.1"),
|
||||
NodeType: pulumi.String("cache.t4g.micro"),
|
||||
NumCacheNodes: pulumi.Int(1),
|
||||
ParameterGroupName: pulumi.String("default.redis7"),
|
||||
Port: pulumi.Int(6379),
|
||||
SubnetGroupName: cacheSubnetGroup.Name,
|
||||
SecurityGroupIds: pulumi.StringArray{cacheSecurityGroup.ID()},
|
||||
ApplyImmediately: pulumi.Bool(true),
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// A single-node Redis cluster has no configuration endpoint — that
|
||||
// is a Memcached thing — so the address is the one cache node's.
|
||||
searchCacheAddress = pulumi.All(searchCache.CacheNodes, searchCache.Port).ApplyT(
|
||||
func(args []any) (string, error) {
|
||||
nodes := args[0].([]elasticache.ClusterCacheNode)
|
||||
if len(nodes) == 0 || nodes[0].Address == nil {
|
||||
return "", fmt.Errorf("search cache reported no node address")
|
||||
}
|
||||
return fmt.Sprintf("%s:%d", *nodes[0].Address, args[1].(int)), nil
|
||||
},
|
||||
).(pulumi.StringOutput)
|
||||
}
|
||||
|
||||
// Private bucket for raw video uploads (pre-transcode). Kept separate
|
||||
// from encoded-bucket, which is fronted by CloudFront/OAC for public
|
||||
// delivery of finished renditions — raw source video must not be
|
||||
@@ -1359,6 +1422,10 @@ func main() {
|
||||
discoveryExtraEnv := []envVar{
|
||||
{Name: "AWS_REGION", Value: pulumi.String(awsRegion).ToStringOutput()},
|
||||
{Name: "CATALOGUE_EVENTS_QUEUE_URL", Value: catalogueQueue.Url},
|
||||
// The search cache. Reached over the Redis protocol on the
|
||||
// private network, so unlike the queue above it needs no
|
||||
// matching grant on discovery-task-role.
|
||||
{Name: "REDIS_ADDR", Value: searchCacheAddress},
|
||||
}
|
||||
|
||||
discoveryRepo, discoveryService, discoveryAlb, err = deployFargateService(ctx, "discovery", 8080,
|
||||
@@ -1452,6 +1519,7 @@ func main() {
|
||||
ctx.Export("ecsClusterArn", cluster.Arn)
|
||||
ctx.Export("cmsServiceArn", cmsService.Arn)
|
||||
ctx.Export("discoveryServiceArn", discoveryService.Arn)
|
||||
ctx.Export("searchCacheAddress", searchCacheAddress)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user