Compare commits

...

8 Commits

17 changed files with 402 additions and 69 deletions
+1 -1
View File
@@ -56,7 +56,7 @@ jobs:
apt-get install ca-certificates make -y
update-ca-certificates
go mod tidy
get -u -v ./...
go get -u -v ./...
go mod tidy -v
- name: Run unit tests
+15
View File
@@ -28,6 +28,7 @@ This project is in active use by [telegram-bot.app](https://telegram-bot.app), a
- [Cache operations](#cache-operations)
- [General](#general)
- [Metrics which matter](#metrics-which-matter)
- [Tracing](#tracing)
- [Healthcheck](#healthcheck)
- [Monitoring endpoint](#monitoring-endpoint)
@@ -144,6 +145,8 @@ You can still use the non-prefixed environment variables in the spirit of the ba
| `HASURA_EVENT_CLEANER` | Enable the hasura event cleaner | `false` |
| `HASURA_EVENT_CLEANER_OLDER_THAN` | The interval for the hasura event cleaner (in days) | `1` |
| `HASURA_EVENT_METADATA_DB` | URL to the hasura metadata database | `postgresql://localhost:5432/hasura` |
| `ENABLE_TRACE` | Enables tracing | `false` |
| `TRACER_ENDPOINT` | Tracing endpoint | `localhost:4317` |
### Speed
@@ -293,6 +296,18 @@ With the `PURGE_METRICS_ON_CRAWL` enabled, the `graphql_proxy_requests_failed`,
If you prefer more control over the metrics purging - you can enable `PURGE_METRICS_ON_TIMER` environment variable and set the interval in seconds. This will allow you to purge the metrics on a regular basis, for example every 90 seconds. It could be better solution if you have multiple crawlers checking the metrics endpoints and you want to avoid the situation when metrics are purged by for example healthcheck.
#### Tracing
Tracing can be enabled by setting `ENABLE_TRACE` to `true` and providing compatible with OTEL `TRACER_ENDPOINT` value ( default is `localhost:4317` ). From that moment you can include `X-Trace-Span` with content being json encoded in your requests to the proxy endpoint.
The value of X-Trace-Span should be in following format:
```json
{
"traceparent": "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01",
}
```
#### Healthcheck
If you'd like the `/healthz` endpoint to perform actual check for the connectivity to the graphql endpoint - set the `HEALTHCHECK_GRAPHQL_URL` environment variable to the exact URL of the graphql endpoint. The query executed will be `query { __typename }` and if the response is not `200 OK` - the healthcheck will fail. Remember that the endpoint is a full URL which you'd like to check, so it should include the protocol, host and path - for example `http://localhost:8080/v1/graphql` and it's NOT the same as value of `HOST_GRAPHQL` environment variable which should provide only the host, without path, ending with slash.
+1 -1
View File
@@ -15,10 +15,10 @@ type CacheConfig struct {
Logger *libpack_logger.Logger
Client CacheClient
Redis struct {
Enable bool `json:"enable"`
URL string `json:"url"`
Password string `json:"password"`
DB int `json:"db"`
Enable bool `json:"enable"`
}
TTL int `json:"ttl"`
}
+3 -3
View File
@@ -11,10 +11,10 @@ import (
)
type RedisConfig struct {
client *redis.Client
ctx context.Context
prefix string
client *redis.Client
builderPool *sync.Pool
prefix string
}
func (c *RedisConfig) prependKeyName(key string) string {
@@ -29,8 +29,8 @@ func (c *RedisConfig) prependKeyName(key string) string {
type RedisClientConfig struct {
RedisServer string
RedisPassword string
RedisDB int
Prefix string
RedisDB int
}
func New(redisClientConfig *RedisClientConfig) *RedisConfig {
+4
View File
@@ -74,4 +74,8 @@ func cleanEvents() {
})
}
}
cfg.Logger.Info(&libpack_logger.LogMessage{
Message: "Old events cleaned up",
Pairs: nil,
})
}
+15 -2
View File
@@ -19,25 +19,32 @@ require (
github.com/redis/go-redis/v9 v9.5.3
github.com/stretchr/testify v1.9.0
github.com/valyala/fasthttp v1.55.0
go.opentelemetry.io/otel v1.27.0
go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.27.0
go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.27.0
go.opentelemetry.io/otel/sdk v1.27.0
go.opentelemetry.io/otel/trace v1.27.0
)
require (
github.com/alicebob/gopher-json v0.0.0-20200520072559-a9ecdc9d1d3a // indirect
github.com/andybalholm/brotli v1.1.0 // indirect
github.com/cenkalti/backoff/v4 v4.3.0 // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/davecgh/go-spew v1.1.1 // indirect
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
github.com/go-logr/logr v1.4.2 // indirect
github.com/go-logr/stdr v1.2.2 // indirect
github.com/gookit/color v1.5.4 // indirect
github.com/grpc-ecosystem/grpc-gateway/v2 v2.20.0 // indirect
github.com/jackc/pgpassfile v1.0.0 // indirect
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
github.com/klauspost/compress v1.17.9 // indirect
github.com/kr/pretty v0.3.1 // indirect
github.com/mattn/go-colorable v0.1.13 // indirect
github.com/mattn/go-isatty v0.0.20 // indirect
github.com/mattn/go-runewidth v0.0.15 // indirect
github.com/pmezard/go-difflib v1.0.0 // indirect
github.com/rivo/uniseg v0.4.7 // indirect
github.com/rogpeppe/go-internal v1.11.0 // indirect
github.com/rs/zerolog v1.33.0 // indirect
github.com/valyala/bytebufferpool v1.0.0 // indirect
github.com/valyala/fastrand v1.1.0 // indirect
@@ -45,6 +52,8 @@ require (
github.com/valyala/tcplisten v1.0.0 // indirect
github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e // indirect
github.com/yuin/gopher-lua v1.1.1 // indirect
go.opentelemetry.io/otel/metric v1.27.0 // indirect
go.opentelemetry.io/proto/otlp v1.3.1 // indirect
golang.org/x/crypto v0.24.0 // indirect
golang.org/x/exp v0.0.0-20240506185415-9bf2ced13842 // indirect
golang.org/x/net v0.26.0 // indirect
@@ -52,5 +61,9 @@ require (
golang.org/x/sys v0.21.0 // indirect
golang.org/x/term v0.21.0 // indirect
golang.org/x/text v0.16.0 // indirect
google.golang.org/genproto/googleapis/api v0.0.0-20240617180043-68d350f18fd4 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20240617180043-68d350f18fd4 // indirect
google.golang.org/grpc v1.64.0 // indirect
google.golang.org/protobuf v1.34.2 // indirect
gopkg.in/yaml.v3 v3.0.1 // indirect
)
+37 -5
View File
@@ -12,15 +12,21 @@ 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/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8=
github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
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/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
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/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY=
github.com/go-logr/logr v1.4.2/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/goccy/go-json v0.10.3 h1:KZ5WoDbxAIgm2HNbYckL0se1fHD6rz5j4ywS6ebzDqA=
github.com/goccy/go-json v0.10.3/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M=
github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
@@ -28,6 +34,8 @@ github.com/gofiber/fiber/v2 v2.52.4 h1:P+T+4iK7VaqUsq2PALYEfBBo6bJZ4q3FP8cZ84Egg
github.com/gofiber/fiber/v2 v2.52.4/go.mod h1:KEOE+cXMhXG0zHc9d8+E38hoX+ZN7bhOtgeF2oT6jrQ=
github.com/gofrs/flock v0.8.1 h1:+gYjHKf32LDeiEEFhQaotPbLuUXjY5ZqxKgXy7n59aw=
github.com/gofrs/flock v0.8.1/go.mod h1:F1TvTiK9OcQqauNUHlbJvyl9Qa1QvF/gOUDKA14jxHU=
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/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/gookit/color v1.5.4 h1:FZmqs7XOyGgCAxmWyPslpiok1k05wmY3SJTytgvYFs0=
@@ -36,6 +44,8 @@ github.com/gookit/goutil v0.6.15 h1:mMQ0ElojNZoyPD0eVROk5QXJPh2uKR4g06slgPDF5Jo=
github.com/gookit/goutil v0.6.15/go.mod h1:qdKdYEHQdEtyH+4fNdQNZfJHhI0jUZzHxQVAV3DaMDY=
github.com/graphql-go/graphql v0.8.1 h1:p7/Ou/WpmulocJeEx7wjQy611rtXGQaAcXGqanuMMgc=
github.com/graphql-go/graphql v0.8.1/go.mod h1:nKiHzRM0qopJEwCITUuIsxk9PlVlwIiiI8pnJEhordQ=
github.com/grpc-ecosystem/grpc-gateway/v2 v2.20.0 h1:bkypFPDjIYGfCYD5mRBvpqxfYX1YCS1PXdKYWi8FsN0=
github.com/grpc-ecosystem/grpc-gateway/v2 v2.20.0/go.mod h1:P+Lt/0by1T8bfcF3z737NnSbmxQAppXMRziHUxPOC8k=
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=
@@ -64,7 +74,6 @@ github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWE
github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
github.com/mattn/go-runewidth v0.0.15 h1:UNAjwbU9l54TA3KzvqLGxwWjHmMgBUVhBiTjelZgg3U=
github.com/mattn/go-runewidth v0.0.15/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w=
github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA=
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
@@ -73,9 +82,8 @@ github.com/redis/go-redis/v9 v9.5.3/go.mod h1:hdY0cQFCN4fnSYT6TkisLufl/4W5UIXyv0
github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
github.com/rivo/uniseg v0.4.7 h1:WUdvkW8uEhrYfLC4ZzdpI2ztxP1I582+49Oc5Mq64VQ=
github.com/rivo/uniseg v0.4.7/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88=
github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
github.com/rogpeppe/go-internal v1.11.0 h1:cWPaGQEPrBb5/AsnsZesgZZ9yb1OQ+GOISoDNXVBh4M=
github.com/rogpeppe/go-internal v1.11.0/go.mod h1:ddIwULY96R17DhadqLgMfk9H9tvdUzkipdSkR5nkCZA=
github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8=
github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4=
github.com/rs/xid v1.5.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg=
github.com/rs/zerolog v1.33.0 h1:1cU2KZkvPxNyfgEmhHAz/1A9Bz+llsdYzklWFzgp0r8=
github.com/rs/zerolog v1.33.0/go.mod h1:/7mN4D5sKwJLZQ2b/znpjC3/GQWY/xaDXUM0kKWRHss=
@@ -98,6 +106,22 @@ github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e h1:JVG44RsyaB9T2KIHavM
github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e/go.mod h1:RbqR21r5mrJuqunuUZ/Dhy/avygyECGrLceyNeo4LiM=
github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M=
github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw=
go.opentelemetry.io/otel v1.27.0 h1:9BZoF3yMK/O1AafMiQTVu0YDj5Ea4hPhxCs7sGva+cg=
go.opentelemetry.io/otel v1.27.0/go.mod h1:DMpAK8fzYRzs+bi3rS5REupisuqTheUlSZJ1WnZaPAQ=
go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.27.0 h1:R9DE4kQ4k+YtfLI2ULwX82VtNQ2J8yZmA7ZIF/D+7Mc=
go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.27.0/go.mod h1:OQFyQVrDlbe+R7xrEyDr/2Wr67Ol0hRUgsfA+V5A95s=
go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.27.0 h1:qFffATk0X+HD+f1Z8lswGiOQYKHRlzfmdJm0wEaVrFA=
go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.27.0/go.mod h1:MOiCmryaYtc+V0Ei+Tx9o5S1ZjA7kzLucuVuyzBZloQ=
go.opentelemetry.io/otel/metric v1.27.0 h1:hvj3vdEKyeCi4YaYfNjv2NUje8FqKqUY8IlF0FxV/ik=
go.opentelemetry.io/otel/metric v1.27.0/go.mod h1:mVFgmRlhljgBiuk/MP/oKylr4hs85GZAylncepAX/ak=
go.opentelemetry.io/otel/sdk v1.27.0 h1:mlk+/Y1gLPLn84U4tI8d3GNJmGT/eXe3ZuOXN9kTWmI=
go.opentelemetry.io/otel/sdk v1.27.0/go.mod h1:Ha9vbLwJE6W86YstIywK2xFfPjbWlCuwPtMkKdz/Y4A=
go.opentelemetry.io/otel/trace v1.27.0 h1:IqYb813p7cmbHk0a5y6pD5JPakbVfftRXABGt5/Rscw=
go.opentelemetry.io/otel/trace v1.27.0/go.mod h1:6RiD1hkAprV4/q+yd2ln1HG9GoPx39SuvvstaLBl+l4=
go.opentelemetry.io/proto/otlp v1.3.1 h1:TrMUixzpM0yuc/znrFTP9MMRh8trP93mkCiDVeXrui0=
go.opentelemetry.io/proto/otlp v1.3.1/go.mod h1:0X1WI4de4ZsLrrJNLAQbFeLCm3T7yBkR0XqQ7niQU+8=
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
golang.org/x/crypto v0.24.0 h1:mnl8DM0o513X8fdIkmyFE/5hTYxbwYOjDS/+rK6qpRI=
golang.org/x/crypto v0.24.0/go.mod h1:Z1PMYSOR5nyMcyAVAIQSKCDwalqy85Aqn1x3Ws4L5DM=
golang.org/x/exp v0.0.0-20240506185415-9bf2ced13842 h1:vr/HnozRka3pE4EsMEg1lgkXJkTFJCVUX+S/ZT6wYzM=
@@ -115,6 +139,14 @@ golang.org/x/term v0.21.0 h1:WVXCp+/EBEHOj53Rvu+7KiT/iElMrO8ACK16SMZ3jaA=
golang.org/x/term v0.21.0/go.mod h1:ooXLefLobQVslOqselCNF4SxFAaoS6KujMbsGzSDmX0=
golang.org/x/text v0.16.0 h1:a94ExnEXNtEwYLGJSIUxnWoxoRz/ZcCsV63ROupILh4=
golang.org/x/text v0.16.0/go.mod h1:GhwF1Be+LQoKShO3cGOHzqOgRrGaYc9AvblQOmPVHnI=
google.golang.org/genproto/googleapis/api v0.0.0-20240617180043-68d350f18fd4 h1:MuYw1wJzT+ZkybKfaOXKp5hJiZDn2iHaXRw0mRYdHSc=
google.golang.org/genproto/googleapis/api v0.0.0-20240617180043-68d350f18fd4/go.mod h1:px9SlOOZBg1wM1zdnr8jEL4CNGUBZ+ZKYtNPApNQc4c=
google.golang.org/genproto/googleapis/rpc v0.0.0-20240617180043-68d350f18fd4 h1:Di6ANFilr+S60a4S61ZM00vLdw0IrQOSMS2/6mrnOU0=
google.golang.org/genproto/googleapis/rpc v0.0.0-20240617180043-68d350f18fd4/go.mod h1:Ue6ibwXGpU+dqIcODieyLOcgj7z8+IcskoNIgZxtrFY=
google.golang.org/grpc v1.64.0 h1:KH3VH9y/MgNQg1dE7b3XfVK0GsPSIzJwdF617gUSbvY=
google.golang.org/grpc v1.64.0/go.mod h1:oxjF8E3FBnjp+/gVFYdWacaLDx9na1aqy9oovLpxQYg=
google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg=
google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw=
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=
+3 -3
View File
@@ -42,16 +42,16 @@ const (
var defaultOutput = os.Stdout
type Logger struct {
output io.Writer
format string
minLogLevel int
showCaller bool
output io.Writer
}
type LogMessage struct {
Message string
Pairs map[string]any
output io.Writer
Pairs map[string]any
Message string
}
func (m *LogMessage) String() string {
+3 -3
View File
@@ -59,13 +59,13 @@ func (suite *LoggerTestSuite) Test_LogsLevelsPrint() {
logger := New().SetOutput(output)
tests := []struct {
pairs map[string]any
name string
method string
message string
loggerMinLevel int
messageLogLevel int
message string
pairs map[string]any
wantOutput bool // Whether we expect output to be written
wantOutput bool
}{
{
name: "Log: Debug, Level: Debug - no pairs",
+16
View File
@@ -12,6 +12,7 @@ import (
libpack_cache "github.com/lukaszraczylo/graphql-monitoring-proxy/cache"
libpack_config "github.com/lukaszraczylo/graphql-monitoring-proxy/config"
libpack_logging "github.com/lukaszraczylo/graphql-monitoring-proxy/logging"
libpack_trace "github.com/lukaszraczylo/graphql-monitoring-proxy/tracing"
)
var cfg *config
@@ -88,6 +89,8 @@ func parseConfig() {
c.HasuraEventCleaner.Enable = getDetailsFromEnv("HASURA_EVENT_CLEANER", false)
c.HasuraEventCleaner.ClearOlderThan = getDetailsFromEnv("HASURA_EVENT_CLEANER_OLDER_THAN", 1)
c.HasuraEventCleaner.EventMetadataDb = getDetailsFromEnv("HASURA_EVENT_METADATA_DB", "")
c.Trace.Enable = getDetailsFromEnv("ENABLE_TRACE", false)
c.Trace.TraceEndpoint = getDetailsFromEnv("TRACER_ENDPOINT", "localhost:4317")
cfg = &c
if cfg.Cache.CacheEnable || cfg.Cache.CacheRedisEnable {
@@ -103,8 +106,21 @@ func parseConfig() {
}
libpack_cache.EnableCache(cacheConfig)
}
loadRatelimitConfig()
once.Do(func() {
if cfg.Trace.Enable {
var err error
cfg.Trace.Client, err = libpack_trace.NewClient(cfg.Logger, cfg.Trace.TraceEndpoint)
if err != nil {
cfg.Logger.Error(&libpack_logging.LogMessage{
Message: "Failed to start tracer",
Pairs: map[string]interface{}{
"error": err,
},
})
}
}
go enableApi()
go enableHasuraEventCleaner()
})
+1 -1
View File
@@ -14,8 +14,8 @@ import (
// Cache for sorted label keys to avoid repeated sorting
var sortedLabelKeysCache = struct {
sync.RWMutex
m map[string][]string
sync.RWMutex
}{m: make(map[string][]string)}
func (ms *MetricsSetup) get_metrics_name(name string, labels map[string]string) string {
+1 -1
View File
@@ -167,9 +167,9 @@ func TestEnsureDefaultLabels(t *testing.T) {
libpack_config.PKG_NAME = "example_microservice"
tests := []struct {
name string
inputLabels map[string]string
expectedLabels map[string]string
name string
}{
{
name: "Nil labels",
+30 -1
View File
@@ -1,15 +1,18 @@
package main
import (
"context"
"crypto/tls"
"fmt"
"time"
"github.com/avast/retry-go/v4"
"github.com/goccy/go-json"
fiber "github.com/gofiber/fiber/v2"
"github.com/gofiber/fiber/v2/middleware/proxy"
libpack_logger "github.com/lukaszraczylo/graphql-monitoring-proxy/logging"
libpack_monitoring "github.com/lukaszraczylo/graphql-monitoring-proxy/monitoring"
libpack_trace "github.com/lukaszraczylo/graphql-monitoring-proxy/tracing"
"github.com/valyala/fasthttp"
)
@@ -29,7 +32,7 @@ func createFasthttpClient(timeout int) *fasthttp.Client {
}
}
func proxyTheRequest(c *fiber.Ctx, currentEndpoint string) error {
func proxyTheRequest(c *fiber.Ctx, currentEndpoint string, ctx context.Context) error {
if !checkAllowedURLs(c) {
cfg.Logger.Error(&libpack_logger.LogMessage{
Message: "Request blocked",
@@ -119,9 +122,35 @@ func proxyTheRequest(c *fiber.Ctx, currentEndpoint string) error {
if ifNotInTest() {
cfg.Monitoring.Increment(libpack_monitoring.MetricsFailed, nil)
}
cfg.Logger.Error(&libpack_logger.LogMessage{
Message: "Received non-200 response from the GraphQL server",
Pairs: map[string]interface{}{
"status_code": c.Response().StatusCode(),
},
})
return fmt.Errorf("Received non-200 response from the GraphQL server: %d", c.Response().StatusCode())
}
c.Response().Header.Del(fiber.HeaderServer)
if cfg.Trace.Enable {
tracingContext := libpack_trace.TraceContextInject(ctx)
if tracingContext == nil {
cfg.Logger.Error(&libpack_logger.LogMessage{
Message: "Can't inject empty tracing context",
})
return nil
}
traceJsonEncoded, err := json.Marshal(tracingContext)
if err != nil {
cfg.Logger.Error(&libpack_logger.LogMessage{
Message: "Can't convert tracing context to JSON",
Pairs: map[string]interface{}{
"error": err.Error(),
},
})
return err
}
c.Response().Header.Set("X-Trace-Span", string(traceJsonEncoded))
}
return nil
}
+60 -48
View File
@@ -1,6 +1,7 @@
package main
import (
"context"
"fmt"
"strconv"
"time"
@@ -14,6 +15,7 @@ import (
libpack_config "github.com/lukaszraczylo/graphql-monitoring-proxy/config"
libpack_logger "github.com/lukaszraczylo/graphql-monitoring-proxy/logging"
libpack_monitoring "github.com/lukaszraczylo/graphql-monitoring-proxy/monitoring"
libpack_trace "github.com/lukaszraczylo/graphql-monitoring-proxy/tracing"
)
// StartHTTPProxy starts the HTTP and points it to the GraphQL server.
@@ -75,6 +77,26 @@ func checkAllowedURLs(c *fiber.Ctx) bool {
return ok
}
func extractTraceHeaders(c *fiber.Ctx) (found bool, traceHeaders map[string]string) {
if !cfg.Trace.Enable {
return
}
headers := c.Request().Header
traceHeader := headers.Peek("X-Trace-Span")
if traceHeader != nil {
traceHeaders = make(map[string]string)
if err := json.Unmarshal(traceHeader, &traceHeaders); err != nil {
cfg.Logger.Error(&libpack_logger.LogMessage{
Message: "Error unmarshalling tracer header",
Pairs: map[string]interface{}{"error": err},
})
return
}
found = true
}
return
}
func healthCheck(c *fiber.Ctx) error {
if len(cfg.Server.HealthcheckGraphQL) > 0 {
cfg.Logger.Debug(&libpack_logger.LogMessage{
@@ -105,63 +127,63 @@ func processGraphQLRequest(c *fiber.Ctx) error {
startTime := time.Now()
// Initialize variables with default values
extractedUserID := "-"
extractedRoleName := "-"
var queryCacheHash string
extractedUserID, extractedRoleName := "-", "-"
authorization := c.Request().Header.Peek("Authorization")
// Pre-fetch headers and trace header processing
headers := c.Request().Header
authorization := headers.Peek("Authorization")
ctx := context.Background()
traceHeaderFound, traceHeader := extractTraceHeaders(c)
if traceHeaderFound {
ctx = libpack_trace.TraceContextExtract(ctx, traceHeader)
_, span := libpack_trace.ContinueSpanFromContext(ctx, "GraphQLRequest")
defer span.End()
}
// JWT and role extraction with pre-check
if authorization != nil && (len(cfg.Client.JWTUserClaimPath) > 0 || len(cfg.Client.JWTRoleClaimPath) > 0) {
extractedUserID, extractedRoleName = extractClaimsFromJWTHeader(string(authorization))
}
// Check for banned users early
if checkIfUserIsBanned(c, extractedUserID) {
c.Status(403).SendString("User is banned")
return nil
return c.Status(403).SendString("User is banned")
}
// Role extraction from header
if len(cfg.Client.RoleFromHeader) > 0 {
extractedRoleName = string(c.Request().Header.Peek(cfg.Client.RoleFromHeader))
extractedRoleName = string(headers.Peek(cfg.Client.RoleFromHeader))
if extractedRoleName == "" {
extractedRoleName = "-"
}
}
// Implementing rate limiting if enabled
if cfg.Client.RoleRateLimit {
// Rate limiting check
if cfg.Client.RoleRateLimit && !rateLimitedRequest(extractedUserID, extractedRoleName) {
cfg.Logger.Debug(&libpack_logger.LogMessage{
Message: "Rate limiting enabled",
Pairs: map[string]interface{}{"user_id": extractedUserID, "role_name": extractedRoleName},
})
if !rateLimitedRequest(extractedUserID, extractedRoleName) {
c.Status(429).SendString("Rate limit exceeded, try again later")
return nil
}
return c.Status(429).SendString("Rate limit exceeded, try again later")
}
// Parsing GraphQL query
parsedResult := parseGraphQLQuery(c)
if parsedResult.shouldBlock {
c.Status(403).SendString("Request blocked")
return nil
return c.Status(403).SendString("Request blocked")
}
if parsedResult.shouldIgnore {
cfg.Logger.Debug(&libpack_logger.LogMessage{
Message: "Request passed as-is - probably not a GraphQL",
Pairs: nil,
})
return proxyTheRequest(c, parsedResult.activeEndpoint)
return proxyTheRequest(c, parsedResult.activeEndpoint, ctx)
}
calculatedQueryHash := libpack_cache.CalculateHash(c)
if parsedResult.cacheTime > 0 {
cfg.Logger.Debug(&libpack_logger.LogMessage{
Message: "Cache time set via query",
Pairs: map[string]interface{}{"cacheTime": parsedResult.cacheTime},
})
} else {
// If not set via query, try setting via header
cacheQuery := c.Request().Header.Peek("X-Cache-Graphql-Query")
// Cache handling logic
queryCacheHash := libpack_cache.CalculateHash(c)
if parsedResult.cacheTime == 0 {
cacheQuery := headers.Peek("X-Cache-Graphql-Query")
if cacheQuery != nil {
parsedResult.cacheTime, _ = strconv.Atoi(string(cacheQuery))
cfg.Logger.Debug(&libpack_logger.LogMessage{
@@ -173,39 +195,34 @@ func processGraphQLRequest(c *fiber.Ctx) error {
}
}
wasCached := false
if parsedResult.cacheRefresh {
cfg.Logger.Debug(&libpack_logger.LogMessage{
Message: "Cache refresh requested via query",
Pairs: map[string]interface{}{"user_id": extractedUserID, "request_uuid": c.Locals("request_uuid")},
})
libpack_cache.CacheDelete(calculatedQueryHash)
libpack_cache.CacheDelete(queryCacheHash)
}
// Handling Cache Logic
wasCached := false
if parsedResult.cacheRequest || cfg.Cache.CacheEnable || cfg.Cache.CacheRedisEnable {
cfg.Logger.Debug(&libpack_logger.LogMessage{
Message: "Cache enabled",
Pairs: map[string]interface{}{"via_query": parsedResult.cacheRequest, "via_env": cfg.Cache.CacheEnable},
})
queryCacheHash = calculatedQueryHash
if cachedResponse := libpack_cache.CacheLookup(queryCacheHash); cachedResponse != nil {
cfg.Monitoring.Increment(libpack_monitoring.MetricsCacheHit, nil)
cfg.Logger.Debug(&libpack_logger.LogMessage{
Message: "Cache hit",
Pairs: map[string]interface{}{"hash": queryCacheHash, "user_id": extractedUserID, "request_uuid": c.Locals("request_uuid")},
})
c.Request().Header.Add("X-Cache-Hit", "true")
err := c.Send(cachedResponse)
if err != nil {
headers.Add("X-Cache-Hit", "true")
if err := c.Send(cachedResponse); err != nil {
cfg.Logger.Error(&libpack_logger.LogMessage{
Message: "Can't send the cached response",
Pairs: map[string]interface{}{"error": err.Error()},
})
cfg.Monitoring.Increment(libpack_monitoring.MetricsFailed, nil)
c.Status(500).SendString("Can't send the cached response - try again later")
return c.Status(500).SendString("Can't send the cached response - try again later")
}
wasCached = true
} else {
@@ -214,32 +231,27 @@ func processGraphQLRequest(c *fiber.Ctx) error {
Message: "Cache miss",
Pairs: map[string]interface{}{"hash": queryCacheHash, "user_id": extractedUserID, "request_uuid": c.Locals("request_uuid")},
})
proxyAndCacheTheRequest(c, queryCacheHash, parsedResult.cacheTime, parsedResult.activeEndpoint)
proxyAndCacheTheRequest(c, queryCacheHash, parsedResult.cacheTime, parsedResult.activeEndpoint, ctx)
}
} else {
err := proxyTheRequest(c, parsedResult.activeEndpoint)
if err != nil {
if err := proxyTheRequest(c, parsedResult.activeEndpoint, ctx); err != nil {
cfg.Logger.Error(&libpack_logger.LogMessage{
Message: "Can't proxy the request",
Pairs: map[string]interface{}{"error": err.Error()},
})
cfg.Monitoring.Increment(libpack_monitoring.MetricsFailed, nil)
c.Status(500).SendString("Can't proxy the request - try again later")
return nil
return c.Status(500).SendString("Can't proxy the request - try again later")
}
}
timeTaken := time.Since(startTime)
// Logging & Monitoring
logAndMonitorRequest(c, extractedUserID, parsedResult.operationType, parsedResult.operationName, wasCached, timeTaken, startTime)
return nil
}
// Additional helper function to avoid code repetition
func proxyAndCacheTheRequest(c *fiber.Ctx, queryCacheHash string, cacheTime int, currentEndpoint string) {
err := proxyTheRequest(c, currentEndpoint)
func proxyAndCacheTheRequest(c *fiber.Ctx, queryCacheHash string, cacheTime int, currentEndpoint string, ctx context.Context) {
err := proxyTheRequest(c, currentEndpoint, ctx)
if err != nil {
cfg.Logger.Error(&libpack_logger.LogMessage{
Message: "Can't proxy the request",
+5
View File
@@ -9,6 +9,11 @@ import (
// config is a struct that holds the configuration of the application.
type config struct {
Trace struct {
Client func()
TraceEndpoint string
Enable bool
}
Logger *libpack_logging.Logger
LogLevel string
Monitoring *libpack_monitoring.MetricsSetup
+98
View File
@@ -0,0 +1,98 @@
package libpack_trace
import (
"context"
"fmt"
"time"
libpack_config "github.com/lukaszraczylo/graphql-monitoring-proxy/config"
libpack_logging "github.com/lukaszraczylo/graphql-monitoring-proxy/logging"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/exporters/otlp/otlptrace"
"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
"go.opentelemetry.io/otel/propagation"
"go.opentelemetry.io/otel/sdk/resource"
"go.opentelemetry.io/otel/sdk/trace"
oteltrace "go.opentelemetry.io/otel/trace"
semconv "go.opentelemetry.io/otel/semconv/v1.4.0"
)
func NewClient(log *libpack_logging.Logger, otelGRPCCollector string, attr ...attribute.KeyValue) (func(), error) {
attr = append(attr, semconv.ServiceNameKey.String(libpack_config.PKG_NAME))
fmt.Printf("Starting OpenTelemetry tracer: otlp, configured with endpoint: %s\n", otelGRPCCollector)
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
client := otlptracegrpc.NewClient(
otlptracegrpc.WithInsecure(),
otlptracegrpc.WithEndpoint(otelGRPCCollector),
)
exporter, err := otlptrace.New(ctx, client)
if err != nil {
log.Error(&libpack_logging.LogMessage{
Message: "Failed to create exporter",
Pairs: map[string]interface{}{"error": err},
})
return nil, err
}
tp := trace.NewTracerProvider(
trace.WithSampler(trace.AlwaysSample()),
trace.WithBatcher(exporter, trace.WithMaxExportBatchSize(1), trace.WithBatchTimeout(30*time.Second)),
trace.WithResource(resource.NewWithAttributes(semconv.SchemaURL, attr...)),
)
otel.SetTracerProvider(tp)
otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(propagation.TraceContext{}, propagation.Baggage{}))
shutdownFunc := func() {
shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second)
defer shutdownCancel()
log.Info(&libpack_logging.LogMessage{
Message: "Shutting down tracer",
Pairs: nil,
})
if err := tp.Shutdown(shutdownCtx); err != nil {
log.Warning(&libpack_logging.LogMessage{
Message: "Failed to shutdown tracer provider",
Pairs: map[string]interface{}{"error": err},
})
}
}
return shutdownFunc, nil
}
func TraceContextInject(ctx context.Context) map[string]string {
carrier := propagation.MapCarrier{}
propagator := otel.GetTextMapPropagator()
propagator.Inject(ctx, carrier)
return map[string]string(carrier)
}
func TraceContextExtract(ctx context.Context, traceContext map[string]string) context.Context {
carrier := propagation.MapCarrier(traceContext)
propagator := otel.GetTextMapPropagator()
return propagator.Extract(ctx, carrier)
}
func StartSpanFromContext(ctx context.Context, operationName string) (context.Context, oteltrace.Span) {
tr := otel.GetTracerProvider().Tracer("")
return tr.Start(ctx, operationName, oteltrace.WithSpanKind(oteltrace.SpanKindServer))
}
func ContinueSpanFromContext(ctx context.Context, operationName string) (context.Context, oteltrace.Span) {
tr := otel.GetTracerProvider().Tracer("")
options := []oteltrace.SpanStartOption{
oteltrace.WithSpanKind(oteltrace.SpanKindInternal),
oteltrace.WithAttributes(attribute.String("cont", "true")),
}
return tr.Start(ctx, operationName, options...)
}
func AddAttributesToSpan(span oteltrace.Span, attributes ...attribute.KeyValue) {
span.SetAttributes(attributes...)
}
+109
View File
@@ -0,0 +1,109 @@
package libpack_trace
import (
"testing"
libpack_logging "github.com/lukaszraczylo/graphql-monitoring-proxy/logging"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/suite"
)
type TraceTestSuite struct {
suite.Suite
logger *libpack_logging.Logger
}
func (suite *TraceTestSuite) SetupTest() {
suite.logger = libpack_logging.New()
}
func (suite *TraceTestSuite) TearDownTest() {
// Any cleanup logic can be added here
}
func TestTraceTestSuite(t *testing.T) {
suite.Run(t, new(TraceTestSuite))
}
func (suite *TraceTestSuite) Test_NewClient() {
shutdownFunc, err := NewClient(suite.logger, "localhost:4317")
assert.NoError(suite.T(), err)
assert.NotNil(suite.T(), shutdownFunc)
shutdownFunc()
}
// func (suite *TraceTestSuite) Test_TraceContextInjectExtract() {
// ctx := context.Background()
// traceContext := TraceContextInject(ctx)
// assert.NotEmpty(suite.T(), traceContext)
// extractedCtx := TraceContextExtract(ctx, traceContext)
// assert.NotNil(suite.T(), extractedCtx)
// }
// func (suite *TraceTestSuite) Test_StartSpanFromContext() {
// ctx := context.Background()
// ctx, span := StartSpanFromContext(ctx, "operation")
// assert.NotNil(suite.T(), ctx)
// assert.NotNil(suite.T(), span)
// span.End()
// }
// func (suite *TraceTestSuite) Test_ContinueSpanFromContext() {
// ctx := context.Background()
// ctx, span := ContinueSpanFromContext(ctx, "operation")
// assert.NotNil(suite.T(), ctx)
// assert.NotNil(suite.T(), span)
// span.End()
// }
// func (suite *TraceTestSuite) Test_AddAttributesToSpan() {
// ctx := context.Background()
// _, span := StartSpanFromContext(ctx, "operation")
// attributes := []attribute.KeyValue{
// attribute.String("key1", "value1"),
// attribute.Int("key2", 2),
// }
// AddAttributesToSpan(span, attributes...)
// span.End()
// // Create an in-memory span exporter
// exporter := tracetest.NewSpanRecorder()
// tracerProvider := trace.NewTracerProvider(trace.WithSpanProcessor(exporter))
// otel.SetTracerProvider(tracerProvider)
// // Verify the span attributes
// spans := exporter.Ended()
// assert.Len(suite.T(), spans, 1)
// exportedSpan := spans[0]
// for _, attr := range attributes {
// assert.Contains(suite.T(), exportedSpan.Attributes(), attr)
// }
// }
// func (suite *TraceTestSuite) Test_Shutdown() {
// shutdownFunc, err := NewClient(suite.logger, "localhost:4317")
// assert.NoError(suite.T(), err)
// assert.NotNil(suite.T(), shutdownFunc)
// shutdownFunc()
// logOutput := captureStdOut(func() { suite.logger.Info(&libpack_logging.LogMessage{Message: "Shutting down tracer"}) })
// assert.Contains(suite.T(), logOutput, "Shutting down tracer")
// }
// // Helper function to capture standard output for testing logs
// func captureStdOut(f func()) string {
// originalStdout := os.Stdout
// r, w, _ := os.Pipe()
// os.Stdout = w
// f()
// w.Close()
// var buf bytes.Buffer
// buf.ReadFrom(r)
// os.Stdout = originalStdout
// return buf.String()
// }