From d2dbd1a1bb295eb044381af39149509616241597 Mon Sep 17 00:00:00 2001 From: Ret2Hell Date: Thu, 19 Mar 2026 03:32:05 +0100 Subject: [PATCH 1/8] add a basic ci --- .github/workflows/ci.yml | 41 ++++++++++++++++++++++++++++++++++++++++ .golangci.yml | 19 +++++++++++++++++++ go.mod | 2 +- mise.toml | 2 ++ 4 files changed, 63 insertions(+), 1 deletion(-) create mode 100644 .github/workflows/ci.yml create mode 100644 .golangci.yml diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..3fe0928 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,41 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + branches: [main] + +jobs: + test: + name: Test + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4.3.1 + + - uses: actions/setup-go@40f1582b2485089dde7abd97c1529aa768e1baff # v5.6.0 + with: + go-version-file: go.mod + cache: true + + - name: go vet + run: go vet ./... + + - name: go test + run: go test ./... + + lint: + name: Lint + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4.3.1 + + - uses: actions/setup-go@40f1582b2485089dde7abd97c1529aa768e1baff # v5.6.0 + with: + go-version-file: go.mod + cache: true + + - name: golangci-lint + uses: golangci/golangci-lint-action@1e7e51e771db61008b38414a730f564565cf7c20 # v9.2.0 + with: + version: v2.11.3 diff --git a/.golangci.yml b/.golangci.yml new file mode 100644 index 0000000..7679fd5 --- /dev/null +++ b/.golangci.yml @@ -0,0 +1,19 @@ +version: "2" + +linters: + default: none + enable: + - errcheck + - govet + - staticcheck + settings: + errcheck: + # Report errors from type assertions (e.g. v, ok := x.(T) when ok is ignored) + check-type-assertions: true + # Report when errors are assigned to the blank identifier + check-blank: true + +issues: + # Don't limit the number of issues per linter + max-issues-per-linter: 0 + max-same-issues: 0 diff --git a/go.mod b/go.mod index 0944d0a..9e8c1d2 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module github.com/Tanilytics/Processing -go 1.25.0 +go 1.26.0 require ( github.com/andybalholm/brotli v1.2.0 // indirect diff --git a/mise.toml b/mise.toml index 047ff9e..49a7489 100644 --- a/mise.toml +++ b/mise.toml @@ -1,10 +1,12 @@ [tools] go = "1.26" air = "latest" +golangci-lint = "2.11.3" [env] CGO_ENABLED = "0" GOCACHE = "{{env.HOME}}/.cache/go-build" +_.file = { path = ".env" } [tasks.dev] run = "air" From eabd09622af75ee45593d599548c3453709b2f8f Mon Sep 17 00:00:00 2001 From: Ret2Hell Date: Fri, 20 Mar 2026 19:12:35 +0100 Subject: [PATCH 2/8] basic setup --- .air.toml | 2 +- .env.example | 9 +++- Dockerfile | 16 +++--- cmd/processor/main.go | 44 +++++++++++++++ go.mod | 40 ++++++++------ go.sum | 90 ++++++++++++++++++++----------- internal/config/config.go | 38 +++++++++++++ internal/observability/logger.go | 27 ++++++++++ internal/observability/tracing.go | 33 ++++++++++++ main.go | 29 ---------- mise.toml | 4 +- 11 files changed, 245 insertions(+), 87 deletions(-) create mode 100644 cmd/processor/main.go create mode 100644 internal/config/config.go create mode 100644 internal/observability/logger.go create mode 100644 internal/observability/tracing.go delete mode 100644 main.go diff --git a/.air.toml b/.air.toml index 725c0b8..d519222 100644 --- a/.air.toml +++ b/.air.toml @@ -8,7 +8,7 @@ tmp_dir = "tmp" [build] args_bin = [] bin = "./tmp/main" - cmd = "go build -o ./tmp/main ." + cmd = "go build -o ./tmp/main ./cmd/processor" delay = 1000 entrypoint = ["./tmp/main"] exclude_dir = ["assets", "tmp", "vendor", "testdata", ".git"] diff --git a/.env.example b/.env.example index 2fc80e3..707dc49 100644 --- a/.env.example +++ b/.env.example @@ -1 +1,8 @@ -PORT=3000 +REDPANDA_BROKERS=localhost:19092 +REDPANDA_SASL_USER=superuser +REDPANDA_SASL_PASSWORD=12345678 +# SCRAM-SHA-256 (default) or SCRAM-SHA-512 +REDPANDA_SASL_MECHANISM=SCRAM-SHA-256 +REDIS_URL=redis://localhost:6379 +LOG_LEVEL=info +OTEL_EXPORTER_OTLP_ENDPOINT=localhost:4318 diff --git a/Dockerfile b/Dockerfile index 75695d0..ae43f9f 100644 --- a/Dockerfile +++ b/Dockerfile @@ -5,20 +5,21 @@ WORKDIR /app COPY go.mod go.sum ./ RUN go mod download -COPY main.go . +COPY cmd/ ./cmd/ +COPY internal/ ./internal/ RUN CGO_ENABLED=0 GOOS=linux go build \ - -ldflags="-s -w" \ - -o app \ - . + -ldflags="-s -w" \ + -o bin/processor \ + ./cmd/processor -FROM alpine:latest +FROM alpine:3.21.6 RUN adduser -D -g '' appuser WORKDIR /app -COPY --from=builder /app/processing . +COPY --from=builder /app/bin/processor/ /bin/processor RUN chown -R appuser:appuser /app @@ -26,4 +27,5 @@ USER appuser EXPOSE 3000 -CMD ["./app"] + +CMD ["/bin/processor"] diff --git a/cmd/processor/main.go b/cmd/processor/main.go new file mode 100644 index 0000000..6471d46 --- /dev/null +++ b/cmd/processor/main.go @@ -0,0 +1,44 @@ +package main + +import ( + "context" + "os" + "os/signal" + "syscall" + + "github.com/Tanilytics/processing/internal/config" + "github.com/Tanilytics/processing/internal/observability" +) + +func main() { + // 1. Load configuration + cfg := config.LoadProcessorConfig() + + // 2. Init structured logger + logger := observability.NewLogger("processor-service", cfg.LogLevel) + logger.Info("starting processor service", + "brokers", cfg.RedpandaBrokers, + "redis_url", cfg.RedisURL, + ) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + // 3. Init OTel Tracer + shutdownTracer, err := observability.InitTracer(ctx, "processor-service", cfg.OTelEndpoint) + if err != nil { + logger.Error("failed to initialize tracer", "error", err) + } else { + defer shutdownTracer() + } + + // 4. Wait for shutdown signal + sigCh := make(chan os.Signal, 1) + signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT) + select { + case sig := <-sigCh: + logger.Info("received signal, shutting down", "signal", sig) + case <-ctx.Done(): + logger.Info("context cancelled, shutting down") + } +} diff --git a/go.mod b/go.mod index 9e8c1d2..a0b7d8c 100644 --- a/go.mod +++ b/go.mod @@ -1,22 +1,30 @@ -module github.com/Tanilytics/Processing +module github.com/Tanilytics/processing go 1.26.0 require ( - github.com/andybalholm/brotli v1.2.0 // indirect - github.com/gofiber/fiber/v3 v3.0.0 // indirect - github.com/gofiber/schema v1.6.0 // indirect - github.com/gofiber/utils/v2 v2.0.0 // indirect + go.opentelemetry.io/otel v1.42.0 + go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.42.0 + go.opentelemetry.io/otel/sdk v1.42.0 +) + +require ( + github.com/cenkalti/backoff/v5 v5.0.3 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/go-logr/logr v1.4.3 // indirect + github.com/go-logr/stdr v1.2.2 // indirect github.com/google/uuid v1.6.0 // indirect - github.com/klauspost/compress v1.18.3 // indirect - github.com/mattn/go-colorable v0.1.14 // indirect - github.com/mattn/go-isatty v0.0.20 // indirect - github.com/philhofer/fwd v1.2.0 // indirect - github.com/tinylib/msgp v1.6.3 // indirect - github.com/valyala/bytebufferpool v1.0.0 // indirect - github.com/valyala/fasthttp v1.69.0 // indirect - golang.org/x/crypto v0.47.0 // indirect - golang.org/x/net v0.49.0 // indirect - golang.org/x/sys v0.40.0 // indirect - golang.org/x/text v0.33.0 // indirect + github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 // indirect + go.opentelemetry.io/auto/sdk v1.2.1 // indirect + go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.42.0 // indirect + go.opentelemetry.io/otel/metric v1.42.0 // indirect + go.opentelemetry.io/otel/trace v1.42.0 // indirect + go.opentelemetry.io/proto/otlp v1.9.0 // indirect + golang.org/x/net v0.51.0 // indirect + golang.org/x/sys v0.41.0 // indirect + golang.org/x/text v0.34.0 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20260209200024-4cfbd4190f57 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260209200024-4cfbd4190f57 // indirect + google.golang.org/grpc v1.79.2 // indirect + google.golang.org/protobuf v1.36.11 // indirect ) diff --git a/go.sum b/go.sum index f7ca47f..7b77052 100644 --- a/go.sum +++ b/go.sum @@ -1,33 +1,61 @@ -github.com/andybalholm/brotli v1.2.0 h1:ukwgCxwYrmACq68yiUqwIWnGY0cTPox/M94sVwToPjQ= -github.com/andybalholm/brotli v1.2.0/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY= -github.com/gofiber/fiber/v3 v3.0.0 h1:GPeCG8X60L42wLKrzgeewDHBr6pE6veAvwaXsqD3Xjk= -github.com/gofiber/fiber/v3 v3.0.0/go.mod h1:kVZiO/AwyT5Pq6PgC8qRCJ+j/BHrMy5jNw1O9yH38aY= -github.com/gofiber/schema v1.6.0 h1:rAgVDFwhndtC+hgV7Vu5ItQCn7eC2mBA4Eu1/ZTiEYY= -github.com/gofiber/schema v1.6.0/go.mod h1:WNZWpQx8LlPSK7ZaX0OqOh+nQo/eW2OevsXs1VZfs/s= -github.com/gofiber/utils/v2 v2.0.0 h1:SCC3rpsEDWupFSHtc0RKxg/BKgV0s1qKfZg9Jv6D0sM= -github.com/gofiber/utils/v2 v2.0.0/go.mod h1:xF9v89FfmbrYqI/bQUGN7gR8ZtXot2jxnZvmAUtiavE= +github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM= +github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= +github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= +github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/klauspost/compress v1.18.3 h1:9PJRvfbmTabkOX8moIpXPbMMbYN60bWImDDU7L+/6zw= -github.com/klauspost/compress v1.18.3/go.mod h1:R0h/fSBs8DE4ENlcrlib3PsXS61voFxhIs2DeRhCvJ4= -github.com/mattn/go-colorable v0.1.14 h1:9A9LHSqF/7dyVVX6g0U9cwm9pG3kP9gSzcuIPHPsaIE= -github.com/mattn/go-colorable v0.1.14/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stgPZH1UqBm1s8= -github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= -github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= -github.com/philhofer/fwd v1.2.0 h1:e6DnBTl7vGY+Gz322/ASL4Gyp1FspeMvx1RNDoToZuM= -github.com/philhofer/fwd v1.2.0/go.mod h1:RqIHx9QI14HlwKwm98g9Re5prTQ6LdeRQn+gXJFxsJM= -github.com/tinylib/msgp v1.6.3 h1:bCSxiTz386UTgyT1i0MSCvdbWjVW+8sG3PjkGsZQt4s= -github.com/tinylib/msgp v1.6.3/go.mod h1:RSp0LW9oSxFut3KzESt5Voq4GVWyS+PSulT77roAqEA= -github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw= -github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc= -github.com/valyala/fasthttp v1.69.0 h1:fNLLESD2SooWeh2cidsuFtOcrEi4uB4m1mPrkJMZyVI= -github.com/valyala/fasthttp v1.69.0/go.mod h1:4wA4PfAraPlAsJ5jMSqCE2ug5tqUPwKXxVj8oNECGcw= -golang.org/x/crypto v0.47.0 h1:V6e3FRj+n4dbpw86FJ8Fv7XVOql7TEwpHapKoMJ/GO8= -golang.org/x/crypto v0.47.0/go.mod h1:ff3Y9VzzKbwSSEzWqJsJVBnWmRwRSHt/6Op5n9bQc4A= -golang.org/x/net v0.49.0 h1:eeHFmOGUTtaaPSGNmjBKpbng9MulQsJURQUAfUwY++o= -golang.org/x/net v0.49.0/go.mod h1:/ysNB2EvaqvesRkuLAyjI1ycPZlQHM3q01F02UY/MV8= -golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.40.0 h1:DBZZqJ2Rkml6QMQsZywtnjnnGvHza6BTfYFWY9kjEWQ= -golang.org/x/sys v0.40.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= -golang.org/x/text v0.33.0 h1:B3njUFyqtHDUI5jMn1YIr5B0IE2U0qck04r6d4KPAxE= -golang.org/x/text v0.33.0/go.mod h1:LuMebE6+rBincTi9+xWTY8TztLzKHc/9C1uBCG27+q8= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 h1:HWRh5R2+9EifMyIHV7ZV+MIZqgz+PMpZ14Jynv3O2Zs= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0/go.mod h1:JfhWUomR1baixubs02l85lZYYOm7LV6om4ceouMv45c= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= +go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +go.opentelemetry.io/otel v1.42.0 h1:lSQGzTgVR3+sgJDAU/7/ZMjN9Z+vUip7leaqBKy4sho= +go.opentelemetry.io/otel v1.42.0/go.mod h1:lJNsdRMxCUIWuMlVJWzecSMuNjE7dOYyWlqOXWkdqCc= +go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.42.0 h1:THuZiwpQZuHPul65w4WcwEnkX2QIuMT+UFoOrygtoJw= +go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.42.0/go.mod h1:J2pvYM5NGHofZ2/Ru6zw/TNWnEQp5crgyDeSrYpXkAw= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.42.0 h1:uLXP+3mghfMf7XmV4PkGfFhFKuNWoCvvx5wP/wOXo0o= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.42.0/go.mod h1:v0Tj04armyT59mnURNUJf7RCKcKzq+lgJs6QSjHjaTc= +go.opentelemetry.io/otel/metric v1.42.0 h1:2jXG+3oZLNXEPfNmnpxKDeZsFI5o4J+nz6xUlaFdF/4= +go.opentelemetry.io/otel/metric v1.42.0/go.mod h1:RlUN/7vTU7Ao/diDkEpQpnz3/92J9ko05BIwxYa2SSI= +go.opentelemetry.io/otel/sdk v1.42.0 h1:LyC8+jqk6UJwdrI/8VydAq/hvkFKNHZVIWuslJXYsDo= +go.opentelemetry.io/otel/sdk v1.42.0/go.mod h1:rGHCAxd9DAph0joO4W6OPwxjNTYWghRWmkHuGbayMts= +go.opentelemetry.io/otel/sdk/metric v1.42.0 h1:D/1QR46Clz6ajyZ3G8SgNlTJKBdGp84q9RKCAZ3YGuA= +go.opentelemetry.io/otel/sdk/metric v1.42.0/go.mod h1:Ua6AAlDKdZ7tdvaQKfSmnFTdHx37+J4ba8MwVCYM5hc= +go.opentelemetry.io/otel/trace v1.42.0 h1:OUCgIPt+mzOnaUTpOQcBiM/PLQ/Op7oq6g4LenLmOYY= +go.opentelemetry.io/otel/trace v1.42.0/go.mod h1:f3K9S+IFqnumBkKhRJMeaZeNk9epyhnCmQh/EysQCdc= +go.opentelemetry.io/proto/otlp v1.9.0 h1:l706jCMITVouPOqEnii2fIAuO3IVGBRPV5ICjceRb/A= +go.opentelemetry.io/proto/otlp v1.9.0/go.mod h1:xE+Cx5E/eEHw+ISFkwPLwCZefwVjY+pqKg1qcK03+/4= +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/net v0.51.0 h1:94R/GTO7mt3/4wIKpcR5gkGmRLOuE/2hNGeWq/GBIFo= +golang.org/x/net v0.51.0/go.mod h1:aamm+2QF5ogm02fjy5Bb7CQ0WMt1/WVM7FtyaTLlA9Y= +golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k= +golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= +golang.org/x/text v0.34.0 h1:oL/Qq0Kdaqxa1KbNeMKwQq0reLCCaFtqu2eNuSeNHbk= +golang.org/x/text v0.34.0/go.mod h1:homfLqTYRFyVYemLBFl5GgL/DWEiH5wcsQ5gSh1yziA= +gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk= +gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E= +google.golang.org/genproto/googleapis/api v0.0.0-20260209200024-4cfbd4190f57 h1:JLQynH/LBHfCTSbDWl+py8C+Rg/k1OVH3xfcaiANuF0= +google.golang.org/genproto/googleapis/api v0.0.0-20260209200024-4cfbd4190f57/go.mod h1:kSJwQxqmFXeo79zOmbrALdflXQeAYcUbgS7PbpMknCY= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260209200024-4cfbd4190f57 h1:mWPCjDEyshlQYzBpMNHaEof6UX1PmHcaUODUywQ0uac= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260209200024-4cfbd4190f57/go.mod h1:j9x/tPzZkyxcgEFkiKEEGxfvyumM01BEtsW8xzOahRQ= +google.golang.org/grpc v1.79.2 h1:fRMD94s2tITpyJGtBBn7MkMseNpOZU8ZxgC3MMBaXRU= +google.golang.org/grpc v1.79.2/go.mod h1:KmT0Kjez+0dde/v2j9vzwoAScgEPx/Bw1CYChhHLrHQ= +google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= +google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +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/internal/config/config.go b/internal/config/config.go new file mode 100644 index 0000000..8f4dd71 --- /dev/null +++ b/internal/config/config.go @@ -0,0 +1,38 @@ +package config + +import ( + "os" + "strings" +) + +type ProcessorConfig struct { + RedpandaBrokers []string + RedpandaSASLUser string + RedpandaSASLPassword string + // RedpandaSASLMechanism is the SASL mechanism to use. + // Supported values: "SCRAM-SHA-256" (default), "SCRAM-SHA-512". + // Leave empty to disable SASL. + RedpandaSASLMechanism string + RedisURL string + LogLevel string + OTelEndpoint string +} + +func LoadProcessorConfig() ProcessorConfig { + return ProcessorConfig{ + RedpandaBrokers: strings.Split(getEnv("REDPANDA_BROKERS", "localhost:19092"), ","), + RedpandaSASLUser: getEnv("REDPANDA_SASL_USER", ""), + RedpandaSASLPassword: getEnv("REDPANDA_SASL_PASSWORD", ""), + RedpandaSASLMechanism: getEnv("REDPANDA_SASL_MECHANISM", "SCRAM-SHA-256"), + RedisURL: getEnv("REDIS_URL", "redis://localhost:6379"), + LogLevel: getEnv("LOG_LEVEL", "info"), + OTelEndpoint: getEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "localhost:4318"), + } +} + +func getEnv(key, fallback string) string { + if v := os.Getenv(key); v != "" { + return v + } + return fallback +} diff --git a/internal/observability/logger.go b/internal/observability/logger.go new file mode 100644 index 0000000..5c2d818 --- /dev/null +++ b/internal/observability/logger.go @@ -0,0 +1,27 @@ +package observability + +import ( + "log/slog" + "os" +) + +func NewLogger(service, level string) *slog.Logger { + var lvl slog.Level + switch level { + case "debug": + lvl = slog.LevelDebug + case "warn": + lvl = slog.LevelWarn + case "error": + lvl = slog.LevelError + default: + lvl = slog.LevelInfo + } + + handler := slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{ + Level: lvl, + }) + return slog.New(handler).With( + "service", service, + ) +} diff --git a/internal/observability/tracing.go b/internal/observability/tracing.go new file mode 100644 index 0000000..1406a0a --- /dev/null +++ b/internal/observability/tracing.go @@ -0,0 +1,33 @@ +package observability + +import ( + "context" + + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp" + "go.opentelemetry.io/otel/sdk/resource" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + semconv "go.opentelemetry.io/otel/semconv/v1.24.0" +) + +func InitTracer(ctx context.Context, serviceName, otlpEndpoint string) (func(), error) { + exporter, err := otlptracehttp.New(ctx, + otlptracehttp.WithEndpoint(otlpEndpoint), + otlptracehttp.WithInsecure(), + ) + if err != nil { + return nil, err + } + + tp := sdktrace.NewTracerProvider( + sdktrace.WithBatcher(exporter), + sdktrace.WithResource(resource.NewWithAttributes( + semconv.SchemaURL, + semconv.ServiceNameKey.String(serviceName), + )), + sdktrace.WithSampler(sdktrace.AlwaysSample()), // sample everything in dev + ) + otel.SetTracerProvider(tp) + + return func() { _ = tp.Shutdown(ctx) }, nil // nolint: errcheck +} diff --git a/main.go b/main.go deleted file mode 100644 index 8bd6b74..0000000 --- a/main.go +++ /dev/null @@ -1,29 +0,0 @@ -package main - -import ( - "fmt" - "os" - - "github.com/gofiber/fiber/v3" -) - -func getEnv(key, defaultValue string) string { - if value := os.Getenv(key); value != "" { - return value - } - return defaultValue -} - -func main() { - app := fiber.New(fiber.Config{ - Immutable: true, - }) - - app.Get("/healthz", func(c fiber.Ctx) error { - return c.SendString("OK") - }) - - port := getEnv("PORT", "3000") - fmt.Printf("Server starting on port %s...\n", port) - app.Listen(":" + port) -} diff --git a/mise.toml b/mise.toml index 49a7489..fc229eb 100644 --- a/mise.toml +++ b/mise.toml @@ -13,10 +13,10 @@ run = "air" description = "Hot reload Fiber API" [tasks.build] -run = "go build -ldflags='-s -w' -o bin/app" +run = "go build -ldflags='-s -w' -o bin/processor ./cmd/processor/" [tasks.start] -run = "./bin/app" +run = "./bin/processor" [tasks.test] run = "go test ./..." From 519f01a3b8a3c6e49e6fa256047783d03f639c4c Mon Sep 17 00:00:00 2001 From: Ret2Hell Date: Fri, 20 Mar 2026 20:13:49 +0100 Subject: [PATCH 3/8] added healthcheck --- .env.example | 1 + Dockerfile | 1 + cmd/processor/main.go | 43 +++++++++++++++++++++++++++++++- internal/config/config.go | 2 ++ internal/server/health.go | 52 +++++++++++++++++++++++++++++++++++++++ internal/server/server.go | 14 +++++++++++ 6 files changed, 112 insertions(+), 1 deletion(-) create mode 100644 internal/server/health.go create mode 100644 internal/server/server.go diff --git a/.env.example b/.env.example index 707dc49..04b954c 100644 --- a/.env.example +++ b/.env.example @@ -1,3 +1,4 @@ +PROCESSING_PORT=:3000 REDPANDA_BROKERS=localhost:19092 REDPANDA_SASL_USER=superuser REDPANDA_SASL_PASSWORD=12345678 diff --git a/Dockerfile b/Dockerfile index ae43f9f..223d23c 100644 --- a/Dockerfile +++ b/Dockerfile @@ -27,5 +27,6 @@ USER appuser EXPOSE 3000 +HEALTHCHECK --interval=10s --timeout=3s CMD wget -qO- http://127.0.0.1:3000/healthz || exit 1 CMD ["/bin/processor"] diff --git a/cmd/processor/main.go b/cmd/processor/main.go index 6471d46..bb34f13 100644 --- a/cmd/processor/main.go +++ b/cmd/processor/main.go @@ -2,12 +2,17 @@ package main import ( "context" + "errors" + "log/slog" + "net/http" "os" "os/signal" "syscall" + "time" "github.com/Tanilytics/processing/internal/config" "github.com/Tanilytics/processing/internal/observability" + "github.com/Tanilytics/processing/internal/server" ) func main() { @@ -17,6 +22,7 @@ func main() { // 2. Init structured logger logger := observability.NewLogger("processor-service", cfg.LogLevel) logger.Info("starting processor service", + "port", cfg.Port, "brokers", cfg.RedpandaBrokers, "redis_url", cfg.RedisURL, ) @@ -32,7 +38,23 @@ func main() { defer shutdownTracer() } - // 4. Wait for shutdown signal + // 4. Create HTTP server + app := server.NewServer(cfg.Port) + + // 5. Mark service as ready + server.SetReady() + logger.Info("service is ready") + + // 6. Start HTTP server in a goroutine + go func() { + logger.Info("http server listening", "addr", cfg.Port) + if err := app.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + logger.Error("http server error", "error", err) + cancel() + } + }() + + // 7. Wait for shutdown signal sigCh := make(chan os.Signal, 1) signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT) select { @@ -41,4 +63,23 @@ func main() { case <-ctx.Done(): logger.Info("context cancelled, shutting down") } + + // 8. Graceful shutdown + gracefulShutdown(app, logger) +} + +func gracefulShutdown( + app interface{ Shutdown(context.Context) error }, + logger *slog.Logger, +) { + server.SetNotReady() + + shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer shutdownCancel() + + if err := app.Shutdown(shutdownCtx); err != nil { + logger.Error("http server shutdown error", "error", err) + } + + logger.Info("shutdown complete") } diff --git a/internal/config/config.go b/internal/config/config.go index 8f4dd71..00fd9dc 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -6,6 +6,7 @@ import ( ) type ProcessorConfig struct { + Port string RedpandaBrokers []string RedpandaSASLUser string RedpandaSASLPassword string @@ -20,6 +21,7 @@ type ProcessorConfig struct { func LoadProcessorConfig() ProcessorConfig { return ProcessorConfig{ + Port: getEnv("PROCESSING_PORT", ":3000"), RedpandaBrokers: strings.Split(getEnv("REDPANDA_BROKERS", "localhost:19092"), ","), RedpandaSASLUser: getEnv("REDPANDA_SASL_USER", ""), RedpandaSASLPassword: getEnv("REDPANDA_SASL_PASSWORD", ""), diff --git a/internal/server/health.go b/internal/server/health.go new file mode 100644 index 0000000..04989ff --- /dev/null +++ b/internal/server/health.go @@ -0,0 +1,52 @@ +package server + +import ( + "encoding/json" + "net/http" + "sync/atomic" +) + +var ready atomic.Bool + +type statusResponse struct { + Status string `json:"status"` +} + +// SetReady marks the service as ready to serve traffic. +func SetReady() { + ready.Store(true) +} + +// SetNotReady marks the service as not ready to serve traffic. +func SetNotReady() { + ready.Store(false) +} + +// Healthz is a liveness probe. Returns 200 OK if the process is running. +func Healthz(w http.ResponseWriter, _ *http.Request) { + writeJSON(w, http.StatusOK, statusResponse{Status: "ok"}) +} + +// Readyz is a readiness probe. Returns 200 OK only when service is ready. +func Readyz(w http.ResponseWriter, _ *http.Request) { + if !ready.Load() { + writeJSON(w, http.StatusServiceUnavailable, statusResponse{Status: "not ready"}) + return + } + + writeJSON(w, http.StatusOK, statusResponse{Status: "ready"}) +} + +func writeJSON(w http.ResponseWriter, status int, payload statusResponse) { + body, err := json.Marshal(payload) + if err != nil { + http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError) + return + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + if _, err := w.Write(body); err != nil { + return + } +} diff --git a/internal/server/server.go b/internal/server/server.go new file mode 100644 index 0000000..0446f35 --- /dev/null +++ b/internal/server/server.go @@ -0,0 +1,14 @@ +package server + +import "net/http" + +func NewServer(addr string) *http.Server { + mux := http.NewServeMux() + mux.HandleFunc("/healthz", Healthz) + mux.HandleFunc("/readyz", Readyz) + + return &http.Server{ + Addr: addr, + Handler: mux, + } +} From 68235f142c1fe4f036a701ac94e5ad36b10cef21 Mon Sep 17 00:00:00 2001 From: Ret2Hell Date: Fri, 20 Mar 2026 20:20:20 +0100 Subject: [PATCH 4/8] added consumer group --- .env.example | 1 + internal/config/config.go | 2 ++ 2 files changed, 3 insertions(+) diff --git a/.env.example b/.env.example index 04b954c..9f2b838 100644 --- a/.env.example +++ b/.env.example @@ -7,3 +7,4 @@ REDPANDA_SASL_MECHANISM=SCRAM-SHA-256 REDIS_URL=redis://localhost:6379 LOG_LEVEL=info OTEL_EXPORTER_OTLP_ENDPOINT=localhost:4318 +CONSUMER_GROUP=event-processor diff --git a/internal/config/config.go b/internal/config/config.go index 00fd9dc..860816b 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -17,6 +17,7 @@ type ProcessorConfig struct { RedisURL string LogLevel string OTelEndpoint string + ConsumerGroup string } func LoadProcessorConfig() ProcessorConfig { @@ -29,6 +30,7 @@ func LoadProcessorConfig() ProcessorConfig { RedisURL: getEnv("REDIS_URL", "redis://localhost:6379"), LogLevel: getEnv("LOG_LEVEL", "info"), OTelEndpoint: getEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "localhost:4318"), + ConsumerGroup: getEnv("CONSUMER_GROUP", "event-processor"), } } From 7b082ca46e87af089b70dd9494cf0afcb883846d Mon Sep 17 00:00:00 2001 From: Ret2Hell Date: Fri, 20 Mar 2026 21:08:21 +0100 Subject: [PATCH 5/8] no fallback for env vars changed the way I'm handling env vars as default values were causing confusion --- internal/config/config.go | 30 +++++++++++++++++------------- 1 file changed, 17 insertions(+), 13 deletions(-) diff --git a/internal/config/config.go b/internal/config/config.go index 860816b..a3e9375 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -1,6 +1,7 @@ package config import ( + "log" "os" "strings" ) @@ -22,21 +23,24 @@ type ProcessorConfig struct { func LoadProcessorConfig() ProcessorConfig { return ProcessorConfig{ - Port: getEnv("PROCESSING_PORT", ":3000"), - RedpandaBrokers: strings.Split(getEnv("REDPANDA_BROKERS", "localhost:19092"), ","), - RedpandaSASLUser: getEnv("REDPANDA_SASL_USER", ""), - RedpandaSASLPassword: getEnv("REDPANDA_SASL_PASSWORD", ""), - RedpandaSASLMechanism: getEnv("REDPANDA_SASL_MECHANISM", "SCRAM-SHA-256"), - RedisURL: getEnv("REDIS_URL", "redis://localhost:6379"), - LogLevel: getEnv("LOG_LEVEL", "info"), - OTelEndpoint: getEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "localhost:4318"), - ConsumerGroup: getEnv("CONSUMER_GROUP", "event-processor"), + Port: getEnv("PROCESSING_PORT"), + RedpandaBrokers: strings.Split(getEnv("REDPANDA_BROKERS"), ","), + RedpandaSASLUser: getEnv("REDPANDA_SASL_USER"), + RedpandaSASLPassword: getEnv("REDPANDA_SASL_PASSWORD"), + RedpandaSASLMechanism: getEnv("REDPANDA_SASL_MECHANISM"), + RedisURL: getEnv("REDIS_URL"), + LogLevel: getEnv("LOG_LEVEL"), + OTelEndpoint: getEnv("OTEL_EXPORTER_OTLP_ENDPOINT"), + ConsumerGroup: getEnv("CONSUMER_GROUP"), } } -func getEnv(key, fallback string) string { - if v := os.Getenv(key); v != "" { - return v +func getEnv(key string) string { + v, ok := os.LookupEnv(key) + if !ok || v == "" { + log.Printf("missing required environment variable: %s", key) + os.Exit(1) } - return fallback + + return v } From 7bb063c8d413a11c7efefdc70c05b44553c197cb Mon Sep 17 00:00:00 2001 From: Ret2Hell Date: Fri, 20 Mar 2026 23:59:30 +0100 Subject: [PATCH 6/8] implemented the consumer --- cmd/processor/main.go | 46 +++++++++++--- go.mod | 7 ++- go.sum | 10 ++++ internal/consumer/consumer.go | 95 +++++++++++++++++++++++++++++ internal/models/event.go | 109 ++++++++++++++++++++++++++++++++++ 5 files changed, 258 insertions(+), 9 deletions(-) create mode 100644 internal/consumer/consumer.go create mode 100644 internal/models/event.go diff --git a/cmd/processor/main.go b/cmd/processor/main.go index bb34f13..a455cb5 100644 --- a/cmd/processor/main.go +++ b/cmd/processor/main.go @@ -7,10 +7,12 @@ import ( "net/http" "os" "os/signal" + "sync" "syscall" "time" "github.com/Tanilytics/processing/internal/config" + "github.com/Tanilytics/processing/internal/consumer" "github.com/Tanilytics/processing/internal/observability" "github.com/Tanilytics/processing/internal/server" ) @@ -41,11 +43,31 @@ func main() { // 4. Create HTTP server app := server.NewServer(cfg.Port) - // 5. Mark service as ready - server.SetReady() - logger.Info("service is ready") + // 5. Initialize event consumer + eventConsumer, err := consumer.NewEventConsumer( + cfg.RedpandaBrokers, + cfg.ConsumerGroup, + cfg.RedpandaSASLUser, + cfg.RedpandaSASLPassword, + cfg.RedpandaSASLMechanism, + logger, + ) + if err != nil { + logger.Error("failed to initialize consumer", "error", err) + return + } + + var consumerWG sync.WaitGroup + + // 6. Start event consumer in a goroutine + consumerWG.Go(func() { + if err := eventConsumer.Run(ctx); err != nil && !errors.Is(err, context.Canceled) { + logger.Error("consumer stopped with error", "error", err) + cancel() + } + }) - // 6. Start HTTP server in a goroutine + // 7. Start HTTP server in a goroutine go func() { logger.Info("http server listening", "addr", cfg.Port) if err := app.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { @@ -54,9 +76,14 @@ func main() { } }() - // 7. Wait for shutdown signal + // 8. Mark service as ready + server.SetReady() + logger.Info("service is ready") + + // 9. Wait for shutdown signal sigCh := make(chan os.Signal, 1) signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT) + defer signal.Stop(sigCh) select { case sig := <-sigCh: logger.Info("received signal, shutting down", "signal", sig) @@ -64,8 +91,13 @@ func main() { logger.Info("context cancelled, shutting down") } - // 8. Graceful shutdown + // 10. Graceful shutdown + cancel() gracefulShutdown(app, logger) + eventConsumer.Close() + consumerWG.Wait() + + logger.Info("shutdown complete") } func gracefulShutdown( @@ -80,6 +112,4 @@ func gracefulShutdown( if err := app.Shutdown(shutdownCtx); err != nil { logger.Error("http server shutdown error", "error", err) } - - logger.Info("shutdown complete") } diff --git a/go.mod b/go.mod index a0b7d8c..d5ef2d3 100644 --- a/go.mod +++ b/go.mod @@ -3,6 +3,8 @@ module github.com/Tanilytics/processing go 1.26.0 require ( + github.com/google/uuid v1.6.0 + github.com/twmb/franz-go v1.20.7 go.opentelemetry.io/otel v1.42.0 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.42.0 go.opentelemetry.io/otel/sdk v1.42.0 @@ -13,13 +15,16 @@ require ( github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/go-logr/logr v1.4.3 // indirect github.com/go-logr/stdr v1.2.2 // indirect - github.com/google/uuid v1.6.0 // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 // indirect + github.com/klauspost/compress v1.18.4 // indirect + github.com/pierrec/lz4/v4 v4.1.25 // indirect + github.com/twmb/franz-go/pkg/kmsg v1.12.0 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.42.0 // indirect go.opentelemetry.io/otel/metric v1.42.0 // indirect go.opentelemetry.io/otel/trace v1.42.0 // indirect go.opentelemetry.io/proto/otlp v1.9.0 // indirect + golang.org/x/crypto v0.48.0 // indirect golang.org/x/net v0.51.0 // indirect golang.org/x/sys v0.41.0 // indirect golang.org/x/text v0.34.0 // indirect diff --git a/go.sum b/go.sum index 7b77052..ec0c458 100644 --- a/go.sum +++ b/go.sum @@ -17,10 +17,18 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 h1:HWRh5R2+9EifMyIHV7ZV+MIZqgz+PMpZ14Jynv3O2Zs= github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0/go.mod h1:JfhWUomR1baixubs02l85lZYYOm7LV6om4ceouMv45c= +github.com/klauspost/compress v1.18.4 h1:RPhnKRAQ4Fh8zU2FY/6ZFDwTVTxgJ/EMydqSTzE9a2c= +github.com/klauspost/compress v1.18.4/go.mod h1:R0h/fSBs8DE4ENlcrlib3PsXS61voFxhIs2DeRhCvJ4= +github.com/pierrec/lz4/v4 v4.1.25 h1:kocOqRffaIbU5djlIBr7Wh+cx82C0vtFb0fOurZHqD0= +github.com/pierrec/lz4/v4 v4.1.25/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/twmb/franz-go v1.20.7 h1:P4MGSXJjjAPP3NRGPCks/Lrq+j+twWMVl1qYCVgNmWY= +github.com/twmb/franz-go v1.20.7/go.mod h1:0bRX9HZVaoueqFWhPZNi2ODnJL7DNa6mK0HeCrC2bNU= +github.com/twmb/franz-go/pkg/kmsg v1.12.0 h1:CbatD7ers1KzDNgJqPbKOq0Bz/WLBdsTH75wgzeVaPc= +github.com/twmb/franz-go/pkg/kmsg v1.12.0/go.mod h1:+DPt4NC8RmI6hqb8G09+3giKObE6uD2Eya6CfqBpeJY= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= go.opentelemetry.io/otel v1.42.0 h1:lSQGzTgVR3+sgJDAU/7/ZMjN9Z+vUip7leaqBKy4sho= @@ -41,6 +49,8 @@ go.opentelemetry.io/proto/otlp v1.9.0 h1:l706jCMITVouPOqEnii2fIAuO3IVGBRPV5ICjce go.opentelemetry.io/proto/otlp v1.9.0/go.mod h1:xE+Cx5E/eEHw+ISFkwPLwCZefwVjY+pqKg1qcK03+/4= 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.48.0 h1:/VRzVqiRSggnhY7gNRxPauEQ5Drw9haKdM0jqfcCFts= +golang.org/x/crypto v0.48.0/go.mod h1:r0kV5h3qnFPlQnBSrULhlsRfryS2pmewsg+XfMgkVos= golang.org/x/net v0.51.0 h1:94R/GTO7mt3/4wIKpcR5gkGmRLOuE/2hNGeWq/GBIFo= golang.org/x/net v0.51.0/go.mod h1:aamm+2QF5ogm02fjy5Bb7CQ0WMt1/WVM7FtyaTLlA9Y= golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k= diff --git a/internal/consumer/consumer.go b/internal/consumer/consumer.go new file mode 100644 index 0000000..340d3b0 --- /dev/null +++ b/internal/consumer/consumer.go @@ -0,0 +1,95 @@ +package consumer + +import ( + "context" + "encoding/json" + "log/slog" + + "github.com/Tanilytics/processing/internal/models" + "github.com/twmb/franz-go/pkg/kgo" + "github.com/twmb/franz-go/pkg/sasl/scram" +) + +type EventConsumer struct { + client *kgo.Client + logger *slog.Logger +} + +func NewEventConsumer(brokers []string, groupID, saslUser, saslPassword, saslMechanism string, logger *slog.Logger) (*EventConsumer, error) { + opts := []kgo.Opt{ + kgo.SeedBrokers(brokers...), + kgo.ConsumerGroup(groupID), + kgo.ConsumeTopics("raw-events"), + kgo.DisableAutoCommit(), // manual commit after processing + kgo.FetchMinBytes(1), + kgo.FetchMaxBytes(50 * 1024 * 1024), // 50MB max fetch + kgo.ConsumeResetOffset(kgo.NewOffset().AtStart()), // start from beginning on new group + } + + if saslUser != "" { + auth := scram.Auth{ + User: saslUser, + Pass: saslPassword, + } + switch saslMechanism { + case "SCRAM-SHA-512": + opts = append(opts, kgo.SASL(auth.AsSha512Mechanism())) + default: + // "SCRAM-SHA-256" and any unrecognized value fall back to SHA-256, + // which is the Redpanda default. + opts = append(opts, kgo.SASL(auth.AsSha256Mechanism())) + } + } + + client, err := kgo.NewClient(opts...) + if err != nil { + return nil, err + } + return &EventConsumer{client: client, logger: logger}, nil +} + +func (c *EventConsumer) Run(ctx context.Context) error { + for { + fetches := c.client.PollFetches(ctx) + if ctx.Err() != nil { + return ctx.Err() + } + if errs := fetches.Errors(); len(errs) > 0 { + for _, e := range errs { + c.logger.Error("fetch error", + "topic", e.Topic, + "partition", e.Partition, + "err", e.Err, + ) + } + } + + var events []*models.InternalEvent + fetches.EachRecord(func(record *kgo.Record) { + var event models.InternalEvent + if err := json.Unmarshal(record.Value, &event); err != nil { + c.logger.Error("unmarshal event", + "err", err, + "partition", record.Partition, + "offset", record.Offset, + ) + return // skip malformed records (will be DLQ'd when I feel like doing it) + } + events = append(events, &event) + }) + + if len(events) > 0 { + // Process Events + c.logger.Info("Process Events") + } + + // Commit offsets after successful processing + if err := c.client.CommitUncommittedOffsets(ctx); err != nil { + c.logger.Error("offset commit failed", "err", err) + } + } +} + +func (c *EventConsumer) Close() { + c.client.Close() +} diff --git a/internal/models/event.go b/internal/models/event.go new file mode 100644 index 0000000..e37f880 --- /dev/null +++ b/internal/models/event.go @@ -0,0 +1,109 @@ +package models + +import ( + "encoding/json" + "time" + + "github.com/google/uuid" +) + +// EventType enumerates all valid event types. +type EventType string + +const ( + EventPageView EventType = "page_view" + EventPageLeave EventType = "page_leave" + EventClick EventType = "click" + EventScroll EventType = "scroll" + EventMediaPlay EventType = "media_play" + EventMediaPause EventType = "media_pause" + EventMediaSeek EventType = "media_seek" + EventMediaProgress EventType = "media_progress" + EventMediaBuffer EventType = "media_buffer" + EventMediaComplete EventType = "media_complete" + EventCustom EventType = "custom" +) + +// ValidEventTypes is used for validation. +var ValidEventTypes = map[EventType]bool{ + EventPageView: true, EventPageLeave: true, + EventClick: true, EventScroll: true, + EventMediaPlay: true, EventMediaPause: true, + EventMediaSeek: true, EventMediaProgress: true, + EventMediaBuffer: true, EventMediaComplete: true, + EventCustom: true, +} + +// SessionContext holds device/browser context from the SDK. +type SessionContext struct { + ScreenWidth int `json:"screen_width"` + ScreenHeight int `json:"screen_height"` + Language string `json:"language"` + Timezone string `json:"timezone"` +} + +// EventBatch is the HTTP inbound payload from the SDK. +type EventBatch struct { + SiteID string `json:"site_id"` + VisitorID string `json:"visitor_id"` + SessionContext SessionContext `json:"session_context"` + Events []RawEvent `json:"events"` +} + +// RawEvent is a single event as received from the SDK. +type RawEvent struct { + EventID string `json:"event_id"` + EventType EventType `json:"event_type"` + Timestamp int64 `json:"timestamp"` // Unix milliseconds + URL string `json:"url"` + Referrer string `json:"referrer,omitempty"` + UTMSource string `json:"utm_source,omitempty"` + UTMMedium string `json:"utm_medium,omitempty"` + UTMCampaign string `json:"utm_campaign,omitempty"` + Properties json.RawMessage `json:"properties,omitempty"` +} + +// InternalEvent is the enriched event produced to Redpanda. +// Contains server-side metadata not present in RawEvent. +type InternalEvent struct { + EventID string `json:"event_id"` + SiteID string `json:"site_id"` + VisitorID string `json:"visitor_id"` + EventType EventType `json:"event_type"` + Timestamp int64 `json:"timestamp"` + URL string `json:"url"` + Referrer string `json:"referrer,omitempty"` + UTMSource string `json:"utm_source,omitempty"` + UTMMedium string `json:"utm_medium,omitempty"` + UTMCampaign string `json:"utm_campaign,omitempty"` + SessionContext SessionContext `json:"session_context"` + Properties json.RawMessage `json:"properties,omitempty"` + ServerTimestamp int64 `json:"server_timestamp"` + IP string `json:"ip"` + UserAgent string `json:"user_agent"` +} + +// ProcessedEvent is the final event written to ClickHouse. +// IP and UserAgent are gone —> replaced by anonymized/parsed fields. +type ProcessedEvent struct { + EventID uuid.UUID + SiteID string + VisitorID string + SessionID string + EventType EventType + Timestamp time.Time + URL string + Referrer string + UTMSource string + UTMMedium string + UTMCampaign string + Country string + Region string + DeviceType string // desktop, mobile, tablet + Browser string + OS string + ScreenWidth uint16 + Properties json.RawMessage + IPHash string + ConsentGiven bool +} From f20f040de5fce91ec2addf096c98abcaf76bfbd8 Mon Sep 17 00:00:00 2001 From: Ret2Hell Date: Sat, 21 Mar 2026 01:40:10 +0100 Subject: [PATCH 7/8] consumer tests --- go.mod | 4 + go.sum | 9 ++ internal/consumer/consumer_test.go | 197 +++++++++++++++++++++++++++++ 3 files changed, 210 insertions(+) create mode 100644 internal/consumer/consumer_test.go diff --git a/go.mod b/go.mod index d5ef2d3..cd5aa03 100644 --- a/go.mod +++ b/go.mod @@ -4,6 +4,7 @@ go 1.26.0 require ( github.com/google/uuid v1.6.0 + github.com/stretchr/testify v1.11.1 github.com/twmb/franz-go v1.20.7 go.opentelemetry.io/otel v1.42.0 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.42.0 @@ -13,11 +14,13 @@ require ( require ( github.com/cenkalti/backoff/v5 v5.0.3 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/davecgh/go-spew v1.1.1 // indirect github.com/go-logr/logr v1.4.3 // indirect github.com/go-logr/stdr v1.2.2 // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 // indirect github.com/klauspost/compress v1.18.4 // indirect github.com/pierrec/lz4/v4 v4.1.25 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect github.com/twmb/franz-go/pkg/kmsg v1.12.0 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.42.0 // indirect @@ -32,4 +35,5 @@ require ( google.golang.org/genproto/googleapis/rpc v0.0.0-20260209200024-4cfbd4190f57 // indirect google.golang.org/grpc v1.79.2 // indirect google.golang.org/protobuf v1.36.11 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/go.sum b/go.sum index ec0c458..2ab3c17 100644 --- a/go.sum +++ b/go.sum @@ -19,10 +19,16 @@ github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 h1:HWRh5R2+9EifMyIHV7ZV+MIZqgz github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0/go.mod h1:JfhWUomR1baixubs02l85lZYYOm7LV6om4ceouMv45c= github.com/klauspost/compress v1.18.4 h1:RPhnKRAQ4Fh8zU2FY/6ZFDwTVTxgJ/EMydqSTzE9a2c= github.com/klauspost/compress v1.18.4/go.mod h1:R0h/fSBs8DE4ENlcrlib3PsXS61voFxhIs2DeRhCvJ4= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/pierrec/lz4/v4 v4.1.25 h1:kocOqRffaIbU5djlIBr7Wh+cx82C0vtFb0fOurZHqD0= github.com/pierrec/lz4/v4 v4.1.25/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= +github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/twmb/franz-go v1.20.7 h1:P4MGSXJjjAPP3NRGPCks/Lrq+j+twWMVl1qYCVgNmWY= @@ -67,5 +73,8 @@ google.golang.org/grpc v1.79.2 h1:fRMD94s2tITpyJGtBBn7MkMseNpOZU8ZxgC3MMBaXRU= google.golang.org/grpc v1.79.2/go.mod h1:KmT0Kjez+0dde/v2j9vzwoAScgEPx/Bw1CYChhHLrHQ= google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/internal/consumer/consumer_test.go b/internal/consumer/consumer_test.go new file mode 100644 index 0000000..fbdf514 --- /dev/null +++ b/internal/consumer/consumer_test.go @@ -0,0 +1,197 @@ +package consumer + +import ( + "context" + "encoding/json" + "log/slog" + "testing" + + "github.com/Tanilytics/processing/internal/models" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestNewEventConsumerWithoutSASL(t *testing.T) { + logger := slog.Default() + brokers := []string{"localhost:9092"} + + consumer, err := NewEventConsumer(brokers, "test-group", "", "", "", logger) + + require.NoError(t, err) + assert.NotNil(t, consumer) + assert.NotNil(t, consumer.client) + assert.NotNil(t, consumer.logger) +} + +func TestNewEventConsumerWithSASLSHA256(t *testing.T) { + logger := slog.Default() + brokers := []string{"localhost:9092"} + + consumer, err := NewEventConsumer(brokers, "test-group", "user", "pass", "SCRAM-SHA-256", logger) + + require.NoError(t, err) + assert.NotNil(t, consumer) + assert.NotNil(t, consumer.client) +} + +func TestNewEventConsumerWithSASLSHA512(t *testing.T) { + logger := slog.Default() + brokers := []string{"localhost:9092"} + + consumer, err := NewEventConsumer(brokers, "test-group", "user", "pass", "SCRAM-SHA-512", logger) + + require.NoError(t, err) + assert.NotNil(t, consumer) + assert.NotNil(t, consumer.client) +} + +func TestNewEventConsumerWithEmptyMechanismFallsBackToSHA256(t *testing.T) { + logger := slog.Default() + brokers := []string{"localhost:9092"} + + consumer, err := NewEventConsumer(brokers, "test-group", "user", "pass", "", logger) + + require.NoError(t, err) + assert.NotNil(t, consumer) +} + +func TestNewEventConsumerWithUnknownMechanismFallsBackToSHA256(t *testing.T) { + logger := slog.Default() + brokers := []string{"localhost:9092"} + + consumer, err := NewEventConsumer(brokers, "test-group", "user", "pass", "UNKNOWN-MECH", logger) + + require.NoError(t, err) + assert.NotNil(t, consumer) +} + +func TestNewEventConsumerWithMultipleBrokers(t *testing.T) { + logger := slog.Default() + brokers := []string{"broker1:9092", "broker2:9092", "broker3:9092"} + + consumer, err := NewEventConsumer(brokers, "test-group", "", "", "", logger) + + require.NoError(t, err) + assert.NotNil(t, consumer) +} + +func TestEventConsumerClose(t *testing.T) { + logger := slog.Default() + brokers := []string{"localhost:9092"} + + consumer, err := NewEventConsumer(brokers, "test-group", "", "", "", logger) + require.NoError(t, err) + + assert.NotPanics(t, func() { + consumer.Close() + }) +} + +func TestInternalEventUnmarshal(t *testing.T) { + eventJSON := `{ + "event_id": "evt-123", + "site_id": "site-456", + "visitor_id": "vis-789", + "event_type": "page_view", + "timestamp": 1700000000000, + "url": "https://example.com/page", + "referrer": "https://google.com", + "utm_source": "google", + "utm_medium": "organic", + "utm_campaign": "search", + "session_context": { + "screen_width": 1920, + "screen_height": 1080, + "language": "en-US", + "timezone": "America/New_York" + }, + "properties": {"key": "value"}, + "server_timestamp": 1700000001000, + "ip": "192.168.1.1", + "user_agent": "Mozilla/5.0" + }` + + var event models.InternalEvent + err := json.Unmarshal([]byte(eventJSON), &event) + + require.NoError(t, err) + assert.Equal(t, "evt-123", event.EventID) + assert.Equal(t, "site-456", event.SiteID) + assert.Equal(t, "vis-789", event.VisitorID) + assert.Equal(t, models.EventPageView, event.EventType) + assert.Equal(t, int64(1700000000000), event.Timestamp) + assert.Equal(t, "https://example.com/page", event.URL) + assert.Equal(t, "https://google.com", event.Referrer) + assert.Equal(t, "google", event.UTMSource) + assert.Equal(t, "organic", event.UTMMedium) + assert.Equal(t, "search", event.UTMCampaign) + assert.Equal(t, 1920, event.SessionContext.ScreenWidth) + assert.Equal(t, 1080, event.SessionContext.ScreenHeight) + assert.Equal(t, "en-US", event.SessionContext.Language) + assert.Equal(t, "America/New_York", event.SessionContext.Timezone) + assert.Equal(t, int64(1700000001000), event.ServerTimestamp) + assert.Equal(t, "192.168.1.1", event.IP) + assert.Equal(t, "Mozilla/5.0", event.UserAgent) +} + +func TestInternalEventUnmarshalMinimalEvent(t *testing.T) { + eventJSON := `{ + "event_id": "evt-min", + "site_id": "site-min", + "visitor_id": "vis-min", + "event_type": "click", + "timestamp": 1700000000000, + "url": "https://example.com" + }` + + var event models.InternalEvent + err := json.Unmarshal([]byte(eventJSON), &event) + + require.NoError(t, err) + assert.Equal(t, "evt-min", event.EventID) + assert.Equal(t, models.EventClick, event.EventType) + assert.Empty(t, event.Referrer) + assert.Empty(t, event.UTMSource) + assert.Empty(t, event.IP) +} + +func TestInternalEventUnmarshalInvalidJSON(t *testing.T) { + invalidJSON := `{invalid json}` + + var event models.InternalEvent + err := json.Unmarshal([]byte(invalidJSON), &event) + + assert.Error(t, err) +} + +func TestInternalEventUnmarshalInvalidEventType(t *testing.T) { + eventJSON := `{ + "event_id": "evt-123", + "site_id": "site-456", + "visitor_id": "vis-789", + "event_type": "invalid_type", + "timestamp": 1700000000000, + "url": "https://example.com" + }` + + var event models.InternalEvent + err := json.Unmarshal([]byte(eventJSON), &event) + + require.NoError(t, err) + assert.Equal(t, models.EventType("invalid_type"), event.EventType) +} + +func TestRunContextCancellation(t *testing.T) { + logger := slog.Default() + brokers := []string{"localhost:9092"} + + consumer, err := NewEventConsumer(brokers, "test-group", "", "", "", logger) + require.NoError(t, err) + defer consumer.Close() + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + err = consumer.Run(ctx) + assert.ErrorIs(t, err, context.Canceled) +} From 3d858672a2fe9aaa1c56eb867121ef37ceaf7d5a Mon Sep 17 00:00:00 2001 From: Ret2Hell Date: Sat, 21 Mar 2026 01:46:05 +0100 Subject: [PATCH 8/8] fix issues --- internal/consumer/consumer_test.go | 35 +++++++++++++++++------------- 1 file changed, 20 insertions(+), 15 deletions(-) diff --git a/internal/consumer/consumer_test.go b/internal/consumer/consumer_test.go index fbdf514..20834de 100644 --- a/internal/consumer/consumer_test.go +++ b/internal/consumer/consumer_test.go @@ -11,11 +11,16 @@ import ( "github.com/stretchr/testify/require" ) +const ( + testBroker = "localhost:9092" + testGroupID = "test-group" +) + func TestNewEventConsumerWithoutSASL(t *testing.T) { logger := slog.Default() - brokers := []string{"localhost:9092"} + brokers := []string{testBroker} - consumer, err := NewEventConsumer(brokers, "test-group", "", "", "", logger) + consumer, err := NewEventConsumer(brokers, testGroupID, "", "", "", logger) require.NoError(t, err) assert.NotNil(t, consumer) @@ -25,9 +30,9 @@ func TestNewEventConsumerWithoutSASL(t *testing.T) { func TestNewEventConsumerWithSASLSHA256(t *testing.T) { logger := slog.Default() - brokers := []string{"localhost:9092"} + brokers := []string{testBroker} - consumer, err := NewEventConsumer(brokers, "test-group", "user", "pass", "SCRAM-SHA-256", logger) + consumer, err := NewEventConsumer(brokers, testGroupID, "user", "pass", "SCRAM-SHA-256", logger) require.NoError(t, err) assert.NotNil(t, consumer) @@ -36,9 +41,9 @@ func TestNewEventConsumerWithSASLSHA256(t *testing.T) { func TestNewEventConsumerWithSASLSHA512(t *testing.T) { logger := slog.Default() - brokers := []string{"localhost:9092"} + brokers := []string{testBroker} - consumer, err := NewEventConsumer(brokers, "test-group", "user", "pass", "SCRAM-SHA-512", logger) + consumer, err := NewEventConsumer(brokers, testGroupID, "user", "pass", "SCRAM-SHA-512", logger) require.NoError(t, err) assert.NotNil(t, consumer) @@ -47,9 +52,9 @@ func TestNewEventConsumerWithSASLSHA512(t *testing.T) { func TestNewEventConsumerWithEmptyMechanismFallsBackToSHA256(t *testing.T) { logger := slog.Default() - brokers := []string{"localhost:9092"} + brokers := []string{testBroker} - consumer, err := NewEventConsumer(brokers, "test-group", "user", "pass", "", logger) + consumer, err := NewEventConsumer(brokers, testGroupID, "user", "pass", "", logger) require.NoError(t, err) assert.NotNil(t, consumer) @@ -57,9 +62,9 @@ func TestNewEventConsumerWithEmptyMechanismFallsBackToSHA256(t *testing.T) { func TestNewEventConsumerWithUnknownMechanismFallsBackToSHA256(t *testing.T) { logger := slog.Default() - brokers := []string{"localhost:9092"} + brokers := []string{testBroker} - consumer, err := NewEventConsumer(brokers, "test-group", "user", "pass", "UNKNOWN-MECH", logger) + consumer, err := NewEventConsumer(brokers, testGroupID, "user", "pass", "UNKNOWN-MECH", logger) require.NoError(t, err) assert.NotNil(t, consumer) @@ -69,7 +74,7 @@ func TestNewEventConsumerWithMultipleBrokers(t *testing.T) { logger := slog.Default() brokers := []string{"broker1:9092", "broker2:9092", "broker3:9092"} - consumer, err := NewEventConsumer(brokers, "test-group", "", "", "", logger) + consumer, err := NewEventConsumer(brokers, testGroupID, "", "", "", logger) require.NoError(t, err) assert.NotNil(t, consumer) @@ -77,9 +82,9 @@ func TestNewEventConsumerWithMultipleBrokers(t *testing.T) { func TestEventConsumerClose(t *testing.T) { logger := slog.Default() - brokers := []string{"localhost:9092"} + brokers := []string{testBroker} - consumer, err := NewEventConsumer(brokers, "test-group", "", "", "", logger) + consumer, err := NewEventConsumer(brokers, testGroupID, "", "", "", logger) require.NoError(t, err) assert.NotPanics(t, func() { @@ -183,9 +188,9 @@ func TestInternalEventUnmarshalInvalidEventType(t *testing.T) { func TestRunContextCancellation(t *testing.T) { logger := slog.Default() - brokers := []string{"localhost:9092"} + brokers := []string{testBroker} - consumer, err := NewEventConsumer(brokers, "test-group", "", "", "", logger) + consumer, err := NewEventConsumer(brokers, testGroupID, "", "", "", logger) require.NoError(t, err) defer consumer.Close()