diff --git a/README.md b/README.md index 0c87d12..69a0b59 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,8 @@ Shared Go module for services on the platform. The `runtime` package implements the runtime contract once, so a service's `main` is: load the -runtime, register readiness checks, serve. +runtime, register readiness checks, serve. The `natsclient` package is the +NATS client every service shares. ```go ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) @@ -67,6 +68,79 @@ curl -i localhost:8081/readyz # 503 for two seconds, then 200 curl localhost:9464/metrics ``` +## natsclient + +The `natsclient` package is the one NATS client every Go service uses: it +connects with the projected token and the service's identity, serves +request/reply endpoints and makes requests with deadlines, tracing and +metrics. Events and commands (JetStream) come from the service's AsyncAPI +contract, not from this package. + +```go +nc, err := natsclient.Connect(ctx, natsclient.FromRuntime(rt.Config), rt.Logger) +if err != nil { ... } +rt.Ready("nats", nc.Connected()) + +svc, err := nc.Service("orders", version) // $SRV discovery for free +err = svc.AddEndpoint("get", func(ctx context.Context, req natsclient.Request) error { + return req.Respond(data) // orders.svc.get, queue group "orders" +}) + +ctx, cancel := context.WithTimeout(ctx, 2*time.Second) // a deadline is required +reply, err := nc.Request(ctx, natsclient.EndpointSubject("orders", "get"), body, nil) +``` + +Subjects follow ADR 0006: `.svc.` for synchronous calls on +core NATS, `.evt.` and `.cmd.` on JetStream. +Endpoints join a queue group named after the service, so replicas share the +load; `WithQueueGroup` picks another group for an endpoint that must see every +request. The connection re-reads the token on every (re)connect, reconnects +forever, is named after the service, and logs every connection event: the +hourly credential expiry at Info, a closed connection at Error. + +Replies arrive on a private inbox, `_INBOX__` from `SERVICE_IDENTITY` +with dots turned into `_`, which the auth callout grants to that identity +alone. Until the grant is issued (ENG-447) an identity requests with +`WithSharedInbox()`, which keeps nats.go's `_INBOX` prefix. + +`Request` tells failures apart, so a caller can act on them: + +| Error | Meaning | +|---|---| +| `ErrNoDeadline` | the context has no deadline; a request without one would hang on a lost reply | +| `ErrNoResponders` | nobody serves the subject: the target is down or not deployed (wraps `nats.ErrNoResponders`) | +| `ErrTimeout` | the deadline passed with no reply: the target is slow (wraps `context.DeadlineExceeded`) | +| `*ServiceError` | the endpoint answered with an error response; a handler returns one to pick the code, any other error is 500 | + +Every request is counted in `nats_requests_total{service,endpoint,outcome}` +(`ok`, `no_responders`, `timeout`, `error`) and timed in +`nats_request_duration_seconds` with the same labels. The caller's trace +goes along as a W3C `traceparent` header, and the handler's context carries +it, so logs and spans on both sides share one trace. + +### Conventions for Rust and .NET + +The other languages follow the same conventions with their own clients; +nothing is wrapped. + +- Connect as the user `SERVICE_IDENTITY` with the projected token at + `NATS_TOKEN_FILE` as the password, read on every connect attempt, never + once at start (async-nats: `ConnectOptions::auth_callback`; NATS.Net: + `NatsAuthOpts.AuthCredCallback`), reconnect without limit, and name the + connection after the service. +- Set the private inbox prefix `_INBOX__` (async-nats: + `ConnectOptions::custom_inbox_prefix`; NATS.Net: `NatsOpts.InboxPrefix`). +- Subjects are `.svc.`, `.evt.` and + `.cmd.`; endpoints join the queue group named after the + service. +- Every request has a deadline, and "no responders" is handled apart from a + timeout (async-nats: `RequestErrorKind::NoResponders` vs `TimedOut`; + NATS.Net: `NatsNoRespondersException` vs `NatsNoReplyException`). +- The W3C `traceparent` header carries the trace; the handler continues it in + a server span. +- Metrics are `nats_requests_total{service,endpoint,outcome}` and + `nats_request_duration_seconds` with the same labels. + ## Working here ``` diff --git a/go.mod b/go.mod index 6fdf0a2..aa4b86c 100644 --- a/go.mod +++ b/go.mod @@ -4,12 +4,15 @@ go 1.26.0 require ( github.com/caarlos0/env/v11 v11.4.1 + github.com/nats-io/nats-server/v2 v2.15.1 + github.com/nats-io/nats.go v1.54.0 github.com/prometheus/client_golang v1.23.2 github.com/stretchr/testify v1.11.1 go.opentelemetry.io/contrib/instrumentation/runtime v0.68.0 go.opentelemetry.io/otel v1.44.0 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.44.0 go.opentelemetry.io/otel/exporters/prometheus v0.66.0 + go.opentelemetry.io/otel/metric v1.44.0 go.opentelemetry.io/otel/sdk v1.44.0 go.opentelemetry.io/otel/sdk/metric v1.44.0 go.opentelemetry.io/otel/trace v1.44.0 @@ -18,15 +21,22 @@ require ( ) require ( + github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op // indirect github.com/beorn7/perks v1.0.1 // indirect github.com/cenkalti/backoff/v5 v5.0.3 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect 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/google/go-tpm v0.9.8 // indirect github.com/google/uuid v1.6.0 // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 // indirect + github.com/klauspost/compress v1.20.1 // indirect + github.com/minio/highwayhash v1.0.4 // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect + github.com/nats-io/jwt/v2 v2.8.2 // indirect + github.com/nats-io/nkeys v0.4.16 // indirect + github.com/nats-io/nuid v1.0.1 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect github.com/prometheus/client_model v0.6.2 // indirect github.com/prometheus/common v0.67.5 // indirect @@ -34,11 +44,12 @@ require ( github.com/prometheus/procfs v0.20.1 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0 // indirect - go.opentelemetry.io/otel/metric v1.44.0 // indirect go.yaml.in/yaml/v2 v2.4.4 // indirect - golang.org/x/net v0.55.0 // indirect - golang.org/x/sys v0.45.0 // indirect - golang.org/x/text v0.37.0 // indirect + golang.org/x/crypto v0.57.0 // indirect + golang.org/x/net v0.58.0 // indirect + golang.org/x/sys v0.48.0 // indirect + golang.org/x/text v0.42.0 // indirect + golang.org/x/time v0.16.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect google.golang.org/protobuf v1.36.11 // indirect diff --git a/go.sum b/go.sum index da002b1..2f0dbc4 100644 --- a/go.sum +++ b/go.sum @@ -1,3 +1,5 @@ +github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op h1:1BOWQJweNyvZMlpAHXGLiZQn9S+QXGcz3xh94lC0w6E= +github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= github.com/caarlos0/env/v11 v11.4.1 h1:fYwH0sWEsBSMPG7t4e/PEfTFzrWrpjyygXyUnWiSwEw= @@ -17,20 +19,34 @@ 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/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= +github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY= 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.29.0 h1:5VipnvEpbqr2gA2VbM+nYVbkIF28c5ZQfqCBQ5g2xfk= github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0/go.mod h1:Hyl3n6Twe1hvtd9XUXDec4pTvgMSEixRuQKPTMH2bNs= -github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo= -github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= +github.com/klauspost/compress v1.20.1 h1:T7kKElXUMXrUJ2E9QhQhxFtcK5rPyLdsGZvdbLMPdiQ= +github.com/klauspost/compress v1.20.1/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= +github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4= +github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= +github.com/nats-io/jwt/v2 v2.8.2 h1:XXRgB60MSTnqsRwejQurVDs/hcv2dkt+86GjI+I/bMc= +github.com/nats-io/jwt/v2 v2.8.2/go.mod h1:Ag/56sq9OblL4JgdYufDd16Egb17Kr/8WwwuO/forVc= +github.com/nats-io/nats-server/v2 v2.15.1 h1:fK3AeYFIji1Smng60jB0hVoAPZezA/UzsaI3BqXFNEI= +github.com/nats-io/nats-server/v2 v2.15.1/go.mod h1:yeuYRtKVKIFtmMz2/YPk+6hvmNBnk+WwTUzErBdH6IE= +github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA= +github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k= +github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg= +github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs= +github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/prometheus/client_golang v1.23.2 h1:Je96obch5RDVy3FDMndoUsjAhG5Edi49h0RJWRi/o0o= @@ -75,12 +91,17 @@ go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ= go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ= -golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8= -golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww= -golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY= -golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= -golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc= -golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38= +golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= +golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= +golang.org/x/net v0.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To= +golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU= +golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= +golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI= +golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E= +golang.org/x/time v0.16.0 h1:vMb6ptszcQMkcwiRTAuNNU50gom6++Q/6gY2hDM6VDE= +golang.org/x/time v0.16.0/go.mod h1:rVKOqvZeKvrDKTQiAHJ7wmwP0RzleSphoEA9RcdLA0s= gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa h1:Kjn0N0tCrDgiAFW+lGO4JZ3ck44CehvJQMAwj9QF0G8= diff --git a/natsclient/connect.go b/natsclient/connect.go new file mode 100644 index 0000000..7028f49 --- /dev/null +++ b/natsclient/connect.go @@ -0,0 +1,266 @@ +package natsclient + +import ( + "context" + "errors" + "fmt" + "log/slog" + "os" + "strings" + "time" + + "github.com/nats-io/nats.go" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/metric" + "go.opentelemetry.io/otel/propagation" + "go.opentelemetry.io/otel/trace" + + "github.com/MetsaApp/platform-kit/runtime" +) + +// scope names the OTel instruments and spans this package emits. +const scope = "github.com/MetsaApp/platform-kit/natsclient" + +// pingInterval is how often the client pings the server; with nats.go's +// default of two missed pings, a wedged connection is noticed in under a +// minute rather than the default six. +const pingInterval = 20 * time.Second + +// Config is what Connect needs. FromRuntime fills it from the runtime +// contract; a test fills it by hand. +type Config struct { + // URL is the server to dial (NATS_URL). + URL string + // TokenFile holds the projected ServiceAccount token (NATS_TOKEN_FILE). + // It rotates hourly and is read on every (re)connect. Empty connects + // without credentials, for a dev server only. + TokenFile string + // Identity is / (SERVICE_IDENTITY), the NATS + // user and the owner of the private reply inbox. + Identity string + // ServiceName names the connection on the server and the service in the + // request metrics (OTEL_SERVICE_NAME). + ServiceName string +} + +// FromRuntime maps the runtime contract's names onto a Config. +func FromRuntime(cfg runtime.Config) Config { + return Config{ + URL: cfg.NATSURL, + TokenFile: cfg.NATSTokenFile, + Identity: cfg.ServiceIdentity, + ServiceName: cfg.ServiceName, + } +} + +// Option tunes Connect. +type Option func(*options) + +type options struct { + sharedInbox bool + natsOpts []nats.Option +} + +// WithSharedInbox keeps nats.go's default _INBOX prefix instead of the +// identity's private one. Today's auth callout grants only _INBOX.>, so an +// identity whose grant does not yet include its private prefix (ENG-447) +// requests with this option; without it the reply subscription is refused +// and every request times out. +func WithSharedInbox() Option { + return func(o *options) { o.sharedInbox = true } +} + +// WithNATSOptions appends raw nats.go options, applied after this package's +// own, for the rare setting (TLS, a test hook) the Config does not cover. +func WithNATSOptions(opts ...nats.Option) Option { + return func(o *options) { o.natsOpts = append(o.natsOpts, opts...) } +} + +// InboxPrefix is the identity's private reply inbox prefix, +// "_INBOX__", with dots inside the names turned into "_" so the +// prefix is one token. The auth callout grants the identity +// ".>" and nobody else, which is what keeps one identity's replies +// out of another's reach on the shared account. +func InboxPrefix(identity string) (string, error) { + ns, sa, ok := strings.Cut(identity, "/") + if !ok || ns == "" || sa == "" { + return "", fmt.Errorf("natsclient: identity %q: want /", identity) + } + safe := func(s string) string { return strings.ReplaceAll(s, ".", "_") } + return "_INBOX_" + safe(ns) + "_" + safe(sa), nil +} + +// Client is one process's connection to NATS and the instruments its +// requests record into. +type Client struct { + nc *nats.Conn + log *slog.Logger + service string + tracer trace.Tracer + propagator propagation.TextMapPropagator + requests metric.Int64Counter + latency metric.Float64Histogram +} + +// Connect dials NATS as cfg.Identity, authenticating with the projected +// token at cfg.TokenFile, and returns once the connection is set up, not +// once it is established: an unreachable server is retried in the +// background rather than failing the caller. +// +// One connection serves a process for its lifetime. It reconnects forever +// with nats.go's paced, jittered sweep. The token is re-read on every +// (re)connect: both it and the credential the auth callout mints rotate +// hourly, and because a stale read is possible at the moment of rotation, +// repeated auth errors do not close the connection the way nats.go +// otherwise would. Every connection event is logged: the hourly expiry at +// Info, a disconnect or reconnect at Info, a closed connection at Error, +// since that one is permanent and the process should exit. A nil log +// discards. +// +// The reply inbox is the identity's private prefix (see InboxPrefix) unless +// WithSharedInbox says otherwise. +func Connect(ctx context.Context, cfg Config, log *slog.Logger, opts ...Option) (*Client, error) { + if cfg.URL == "" { + return nil, errors.New("natsclient: URL is empty (NATS_URL)") + } + if log == nil { + log = slog.New(slog.DiscardHandler) + } + var o options + for _, opt := range opts { + opt(&o) + } + natsOpts := connectOptions(cfg, log) + inbox := "_INBOX" + if !o.sharedInbox { + prefix, err := InboxPrefix(cfg.Identity) + if err != nil { + return nil, fmt.Errorf("%w (set SERVICE_IDENTITY or use WithSharedInbox)", err) + } + inbox = prefix + natsOpts = append(natsOpts, nats.CustomInboxPrefix(prefix)) + } + + c := &Client{log: log, service: cfg.ServiceName, tracer: otel.Tracer(scope), propagator: propagation.TraceContext{}} + if err := c.initInstruments(); err != nil { + return nil, err + } + nc, err := nats.Connect(cfg.URL, append(natsOpts, o.natsOpts...)...) + if err != nil { + return nil, fmt.Errorf("natsclient: connect: %w", err) + } + c.nc = nc + log.InfoContext(ctx, "nats connecting", "url", cfg.URL, "identity", cfg.Identity, "inbox", inbox) + return c, nil +} + +// connectOptions is Connect's nats.go configuration. It is built outside +// Connect so the callbacks, which run for the life of the connection, do not +// close over Connect's context. +func connectOptions(cfg Config, log *slog.Logger) []nats.Option { + opts := []nats.Option{ + nats.Name(cfg.ServiceName), + nats.MaxReconnects(-1), + nats.RetryOnFailedConnect(true), + nats.IgnoreAuthErrorAbort(), + nats.PingInterval(pingInterval), + nats.ErrorHandler(asyncErrorLogger(log)), + nats.DisconnectErrHandler(func(nc *nats.Conn, err error) { + log.Info("nats disconnected, reconnecting", "error", err, "server", nc.ConnectedUrl()) + }), + nats.ReconnectHandler(func(nc *nats.Conn) { + log.Info("nats reconnected", "server", nc.ConnectedUrl()) + }), + nats.ClosedHandler(func(nc *nats.Conn) { + log.Error("nats connection closed for good", "error", nc.LastError()) + }), + } + if cfg.TokenFile != "" { + opts = append(opts, nats.UserInfoHandler(func() (string, string) { + tok, err := os.ReadFile(cfg.TokenFile) + if err != nil { + // Empty fails this attempt; nats.go keeps retrying and the + // projected token will reappear. + return cfg.Identity, "" + } + return cfg.Identity, strings.TrimSpace(string(tok)) + })) + } + return opts +} + +// asyncErrorLogger replaces nats.go's default handler, which wrote a raw +// "nats: authentication expired on connection [N]" line to stderr that no +// log parser could read. The hourly expiry is by design, so it is Info; any +// other async error (slow consumer, permission violation) stays a Warn. +func asyncErrorLogger(log *slog.Logger) nats.ErrHandler { + return func(nc *nats.Conn, sub *nats.Subscription, err error) { + attrs := []any{"error", err} + if nc != nil { + attrs = append(attrs, "server", nc.ConnectedUrl()) + } + if sub != nil { + attrs = append(attrs, "subject", sub.Subject) + } + if errors.Is(err, nats.ErrAuthExpired) { + log.Info("nats credential expired, reconnecting", attrs...) + return + } + log.Warn("nats async error", attrs...) + } +} + +func (c *Client) initInstruments() error { + meter := otel.Meter(scope) + var err error + c.requests, err = meter.Int64Counter("nats_requests_total", + metric.WithDescription("Requests sent, by subject and outcome.")) + if err != nil { + return fmt.Errorf("natsclient: request counter: %w", err) + } + c.latency, err = meter.Float64Histogram("nats_request_duration_seconds", + metric.WithDescription("Request round trip, by subject and outcome."), metric.WithUnit("s")) + if err != nil { + return fmt.Errorf("natsclient: request histogram: %w", err) + } + return nil +} + +// Conn is the underlying connection, for JetStream and anything else this +// package does not wrap. +func (c *Client) Conn() *nats.Conn { return c.nc } + +// Connected is a readiness check: an error while the client has no server. +// The hourly credential expiry reconnects within a probe period, so only an +// outage that lasts fails the probe's threshold. +func (c *Client) Connected() func(context.Context) error { + return func(context.Context) error { + if !c.nc.IsConnected() { + return fmt.Errorf("natsclient: connection is %s", c.nc.Status()) + } + return nil + } +} + +// Drain shuts the connection down gracefully: no new messages are +// delivered, in-flight replies and publishes are flushed, and only then is +// the connection closed. It returns once the connection has closed or ctx is +// done, whichever comes first, so a SIGTERM handler waits for the work drain +// was meant to save rather than exiting on Drain's return. +func (c *Client) Drain(ctx context.Context) error { + closed := make(chan struct{}) + c.nc.SetClosedHandler(func(*nats.Conn) { close(closed) }) + if c.nc.IsClosed() { + return nil + } + if err := c.nc.Drain(); err != nil { + return fmt.Errorf("natsclient: drain: %w", err) + } + select { + case <-closed: + return nil + case <-ctx.Done(): + c.nc.Close() + return fmt.Errorf("natsclient: drain: %w", ctx.Err()) + } +} diff --git a/natsclient/doc.go b/natsclient/doc.go new file mode 100644 index 0000000..0ac2a8f --- /dev/null +++ b/natsclient/doc.go @@ -0,0 +1,21 @@ +// Package natsclient is the one NATS client every Go service on the platform +// uses: +// +// - Connect dials with the projected token and the service's identity, +// re-reads the token on every (re)connect, reconnects forever, logs +// every connection event and gives the identity a private reply inbox +// (_INBOX__) so no other identity can read its replies; +// - Request is a synchronous call with a deadline, W3C trace context in the +// headers, and distinct errors for "nobody serves this subject" +// (ErrNoResponders) and "the deadline passed" (ErrTimeout), counted and +// timed as nats_requests_total and nats_request_duration_seconds; +// - Service registers request/reply endpoints on .svc. +// through nats.go's micro package, in a queue group named after the +// service by default, with the caller's trace continued in a server span. +// +// The subject shapes follow infrastructure ADR 0006: .svc. +// for synchronous calls on core NATS, .evt. for JetStream +// facts and .cmd. for JetStream work. This package covers +// the synchronous half; events and commands are generated from the service's +// AsyncAPI contract. +package natsclient diff --git a/natsclient/natsclient_test.go b/natsclient/natsclient_test.go new file mode 100644 index 0000000..934acae --- /dev/null +++ b/natsclient/natsclient_test.go @@ -0,0 +1,168 @@ +package natsclient_test + +import ( + "bytes" + "context" + "log/slog" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + natsserver "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/MetsaApp/platform-kit/natsclient" +) + +const ( + serverStart = 5 * time.Second + requestTimeout = 2 * time.Second + identity = "metsa-test/api" +) + +// startServer runs an embedded core NATS server. With users it enforces +// their publish and subscribe permissions, the way the auth callout's +// grants do in the cluster. +func startServer(t *testing.T, users ...*natsserver.User) *natsserver.Server { + t.Helper() + srv, err := natsserver.NewServer(&natsserver.Options{Port: -1, NoLog: true, NoSigs: true, Users: users}) + require.NoError(t, err) + srv.Start() + require.True(t, srv.ReadyForConnections(serverStart), "nats server not ready") + t.Cleanup(srv.Shutdown) + return srv +} + +// connect dials srv as id with a token file holding token, which is how a +// pod authenticates: the user is the identity, the password the projected +// token. An empty token connects without credentials. +func connect( + t *testing.T, srv *natsserver.Server, id, token string, log *slog.Logger, opts ...natsclient.Option, +) *natsclient.Client { + t.Helper() + cfg := natsclient.Config{URL: srv.ClientURL(), Identity: id, ServiceName: "test"} + if token != "" { + cfg.TokenFile = filepath.Join(t.TempDir(), "token") + require.NoError(t, os.WriteFile(cfg.TokenFile, []byte(token+"\n"), 0o600)) + } + c, err := natsclient.Connect(t.Context(), cfg, log, opts...) + require.NoError(t, err) + t.Cleanup(c.Conn().Close) + require.Eventually(t, c.Conn().IsConnected, serverStart, 10*time.Millisecond) + return c +} + +// echo is an endpoint handler that answers with the request data. +func echo(_ context.Context, req natsclient.Request) error { + return req.Respond(req.Data()) +} + +// serve registers a service with one echo endpoint and returns its subject. +func serve(t *testing.T, c *natsclient.Client, service, endpoint string, opts ...natsclient.EndpointOption) string { + t.Helper() + svc, err := c.Service(service, "1.0.0") + require.NoError(t, err) + t.Cleanup(func() { _ = svc.Stop() }) + require.NoError(t, svc.AddEndpoint(endpoint, echo, opts...)) + return natsclient.EndpointSubject(service, endpoint) +} + +func request(t *testing.T, c *natsclient.Client, subject string, data string) (*nats.Msg, error) { + t.Helper() + ctx, cancel := context.WithTimeout(t.Context(), requestTimeout) + defer cancel() + return c.Request(ctx, subject, []byte(data), nil) +} + +// logBuffer collects log lines from nats.go's callback goroutines. +type logBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (b *logBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.Write(p) +} + +func (b *logBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.String() +} + +func TestInboxPrefix(t *testing.T) { + for _, tc := range []struct{ identity, prefix string }{ + {"metsa-production/api", "_INBOX_metsa-production_api"}, + {"a.b/c.d", "_INBOX_a_b_c_d"}, + } { + prefix, err := natsclient.InboxPrefix(tc.identity) + require.NoError(t, err) + assert.Equal(t, tc.prefix, prefix) + } + for _, bad := range []string{"", "no-slash", "/sa", "ns/"} { + _, err := natsclient.InboxPrefix(bad) + assert.Error(t, err, bad) + } +} + +func TestConnectNeedsIdentityForPrivateInbox(t *testing.T) { + srv := startServer(t) + _, err := natsclient.Connect(t.Context(), natsclient.Config{URL: srv.ClientURL()}, nil) + require.ErrorContains(t, err, "WithSharedInbox") + + c := connect(t, srv, "", "", nil, natsclient.WithSharedInbox()) + assert.True(t, strings.HasPrefix(c.Conn().NewRespInbox(), "_INBOX.")) +} + +// The private inbox: each identity's grant allows its own prefix and nothing +// else, so a request's reply is readable by its sender only, and the +// responder, which publishes replies, cannot subscribe to any inbox. +func TestPrivateInboxIsolation(t *testing.T) { + const idA, idB, idSvc = "metsa-test/a", "metsa-test/b", "metsa-test/orders" + subject := natsclient.EndpointSubject("orders", "get") + inboxA, _ := natsclient.InboxPrefix(idA) + inboxB, _ := natsclient.InboxPrefix(idB) + caller := func(name, inbox string) *natsserver.User { + return &natsserver.User{Username: name, Password: "tok-" + name, Permissions: &natsserver.Permissions{ + Publish: &natsserver.SubjectPermission{Allow: []string{"orders.svc.>", inbox + ".>"}}, + Subscribe: &natsserver.SubjectPermission{Allow: []string{inbox + ".>"}}, + }} + } + srv := startServer(t, caller(idA, inboxA), caller(idB, inboxB), &natsserver.User{ + Username: idSvc, Password: "tok-" + idSvc, Permissions: &natsserver.Permissions{ + Publish: &natsserver.SubjectPermission{Allow: []string{inboxA + ".>", inboxB + ".>", "$SRV.>"}}, + Subscribe: &natsserver.SubjectPermission{Allow: []string{"orders.svc.>", "$SRV.>"}}, + }, + }) + serve(t, connect(t, srv, idSvc, "tok-"+idSvc, nil), "orders", "get") + + var logs logBuffer + a := connect(t, srv, idA, "tok-"+idA, nil) + b := connect(t, srv, idB, "tok-"+idB, slog.New(slog.NewTextHandler(&logs, nil))) + assert.True(t, strings.HasPrefix(a.Conn().NewRespInbox(), inboxA+".")) + + // B tries to read A's replies: the server refuses the subscription. + eavesdrop, err := b.Conn().SubscribeSync(inboxA + ".>") + require.NoError(t, err) + require.Eventually(t, func() bool { + return strings.Contains(logs.String(), "permissions violation") + }, serverStart, 10*time.Millisecond) + + reply, err := request(t, a, subject, "hello") + require.NoError(t, err) + assert.Equal(t, "hello", string(reply.Data)) + _, err = eavesdrop.NextMsg(100 * time.Millisecond) + require.ErrorIs(t, err, nats.ErrTimeout, "B must not see A's reply") + + // B's own requests still work on B's own inbox. + reply, err = request(t, b, subject, "hi") + require.NoError(t, err) + assert.Equal(t, "hi", string(reply.Data)) +} diff --git a/natsclient/request.go b/natsclient/request.go new file mode 100644 index 0000000..20ac7c1 --- /dev/null +++ b/natsclient/request.go @@ -0,0 +1,120 @@ +package natsclient + +import ( + "context" + "errors" + "fmt" + "maps" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/micro" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/metric" + "go.opentelemetry.io/otel/propagation" + "go.opentelemetry.io/otel/trace" +) + +// The request outcomes, as recorded in nats_requests_total{outcome}. +var ( + // ErrNoDeadline is returned for a context without a deadline: a request + // with no bound would hang a handler forever on a lost reply. + ErrNoDeadline = errors.New("natsclient: request context needs a deadline") + // ErrNoResponders means nobody is subscribed to the subject: the target + // service is down or not deployed, which is a different problem from a + // slow one. It wraps nats.ErrNoResponders. + ErrNoResponders = errors.New("natsclient: no responders") + // ErrTimeout means the deadline passed with no reply: the target is + // there but slow, or the reply was lost. It wraps + // [context.DeadlineExceeded]. + ErrTimeout = errors.New("natsclient: request timed out") +) + +// ServiceError is an endpoint's error response, the micro protocol's +// Nats-Service-Error-Code and Nats-Service-Error headers. A handler returns +// one to choose the code the caller sees; any other error is reported as +// code 500 with the error's text. +type ServiceError struct { + Code string + Description string +} + +func (e *ServiceError) Error() string { + return "natsclient: service error " + e.Code + ": " + e.Description +} + +// Outcome values, one per error the caller can tell apart. +const ( + outcomeOK = "ok" + outcomeNoResponders = "no_responders" + outcomeTimeout = "timeout" + outcomeError = "error" +) + +// Request sends data to subject and waits for the reply until ctx's +// deadline, which it must have. The caller's trace context goes along as a +// W3C traceparent header, so the handler's span continues the caller's +// trace. hdr may be nil; it is not modified. +// +// The error says what went wrong in a way the caller can act on: +// ErrNoResponders when nobody serves the subject, ErrTimeout when the +// deadline passed, a *ServiceError when the endpoint answered with an error +// response, and anything else wrapped as is. Each request is counted in +// nats_requests_total{service,endpoint,outcome} and timed in +// nats_request_duration_seconds. +func (c *Client) Request(ctx context.Context, subject string, data []byte, hdr nats.Header) (*nats.Msg, error) { + if _, ok := ctx.Deadline(); !ok { + return nil, fmt.Errorf("%w: %s", ErrNoDeadline, subject) + } + ctx, span := c.tracer.Start(ctx, subject, + trace.WithSpanKind(trace.SpanKindClient), + trace.WithAttributes(attribute.String("messaging.system", "nats"), + attribute.String("messaging.destination.name", subject))) + defer span.End() + + msg := &nats.Msg{Subject: subject, Data: data, Header: maps.Clone(hdr)} + if msg.Header == nil { + msg.Header = nats.Header{} + } + c.propagator.Inject(ctx, propagation.HeaderCarrier(msg.Header)) + + start := time.Now() + reply, err := c.nc.RequestMsgWithContext(ctx, msg) + outcome, err := classify(subject, reply, err) + elapsed := time.Since(start) + + attrs := metric.WithAttributes( + attribute.String("service", c.service), + attribute.String("endpoint", subject), + attribute.String("outcome", outcome), + ) + c.requests.Add(ctx, 1, attrs) + c.latency.Record(ctx, elapsed.Seconds(), attrs) + span.SetAttributes(attribute.String("nats.outcome", outcome)) + if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, outcome) + return nil, err + } + return reply, nil +} + +// classify sorts a request's result into its outcome and the error the +// caller gets, including an error response from the endpoint, which nats.go +// delivers as a plain reply with the micro error headers set. +func classify(subject string, reply *nats.Msg, err error) (string, error) { + switch { + case err == nil: + if code := reply.Header.Get(micro.ErrorCodeHeader); code != "" { + return outcomeError, &ServiceError{Code: code, Description: reply.Header.Get(micro.ErrorHeader)} + } + return outcomeOK, nil + case errors.Is(err, nats.ErrNoResponders): + return outcomeNoResponders, fmt.Errorf("%w: %s: %w", ErrNoResponders, subject, err) + case errors.Is(err, context.DeadlineExceeded): + return outcomeTimeout, fmt.Errorf("%w: %s: %w", ErrTimeout, subject, err) + default: + return outcomeError, fmt.Errorf("natsclient: request %s: %w", subject, err) + } +} diff --git a/natsclient/request_test.go b/natsclient/request_test.go new file mode 100644 index 0000000..7a58f0c --- /dev/null +++ b/natsclient/request_test.go @@ -0,0 +1,138 @@ +package natsclient_test + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" + + "github.com/MetsaApp/platform-kit/natsclient" +) + +func TestRequestNeedsDeadline(t *testing.T) { + c := connect(t, startServer(t), identity, "", nil) + _, err := c.Request(t.Context(), "orders.svc.get", nil, nil) + require.ErrorIs(t, err, natsclient.ErrNoDeadline) +} + +// A slow endpoint and an absent one fail differently: the first is a +// timeout the caller may retry, the second a deployment problem. +func TestRequestTimeoutAndNoResponders(t *testing.T) { + srv := startServer(t) + c := connect(t, srv, identity, "", nil) + svc, err := c.Service("slow", "1.0.0") + require.NoError(t, err) + t.Cleanup(func() { _ = svc.Stop() }) + // The handler's context is not cancelled by the caller's deadline (NATS + // does not carry one), so the slow handler simply outlasts it. + require.NoError(t, svc.AddEndpoint("get", func(_ context.Context, req natsclient.Request) error { + time.Sleep(requestTimeout) + return req.Respond(nil) + })) + + ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond) + defer cancel() + _, timeoutErr := c.Request(ctx, natsclient.EndpointSubject("slow", "get"), nil, nil) + require.ErrorIs(t, timeoutErr, natsclient.ErrTimeout) + require.ErrorIs(t, timeoutErr, context.DeadlineExceeded) + require.NotErrorIs(t, timeoutErr, natsclient.ErrNoResponders) + + _, noneErr := request(t, c, natsclient.EndpointSubject("nobody", "get"), "") + require.ErrorIs(t, noneErr, natsclient.ErrNoResponders) + require.NotErrorIs(t, noneErr, natsclient.ErrTimeout) + require.NotErrorIs(t, noneErr, context.DeadlineExceeded) +} + +// A handler's error reaches the caller as a ServiceError with the code the +// handler chose, or 500 when it did not. +func TestRequestServiceError(t *testing.T) { + c := connect(t, startServer(t), identity, "", nil) + svc, err := c.Service("orders", "1.0.0") + require.NoError(t, err) + t.Cleanup(func() { _ = svc.Stop() }) + require.NoError(t, svc.AddEndpoint("missing", func(context.Context, natsclient.Request) error { + return &natsclient.ServiceError{Code: "404", Description: "no such order"} + })) + require.NoError(t, svc.AddEndpoint("broken", func(context.Context, natsclient.Request) error { + return errors.New("boom") + })) + + _, err = request(t, c, natsclient.EndpointSubject("orders", "missing"), "") + serviceErr, ok := errors.AsType[*natsclient.ServiceError](err) + require.True(t, ok, err) + assert.Equal(t, &natsclient.ServiceError{Code: "404", Description: "no such order"}, serviceErr) + + _, err = request(t, c, natsclient.EndpointSubject("orders", "broken"), "") + serviceErr, ok = errors.AsType[*natsclient.ServiceError](err) + require.True(t, ok, err) + assert.Equal(t, &natsclient.ServiceError{Code: "500", Description: "boom"}, serviceErr) +} + +// Every request lands in nats_requests_total and +// nats_request_duration_seconds with the service, the subject and its +// outcome, which is what the dashboards and alerts key on. +func TestRequestMetrics(t *testing.T) { + reader := sdkmetric.NewManualReader() + otel.SetMeterProvider(sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader))) + + srv := startServer(t) + c := connect(t, srv, identity, "", nil) + subject := serve(t, c, "orders", "get") + _, err := request(t, c, subject, "x") + require.NoError(t, err) + _, err = request(t, c, subject, "y") + require.NoError(t, err) + _, err = request(t, c, "nobody.svc.get", "") + require.ErrorIs(t, err, natsclient.ErrNoResponders) + + var rm metricdata.ResourceMetrics + require.NoError(t, reader.Collect(t.Context(), &rm)) + okAttrs := attribute.NewSet( + attribute.String("service", "test"), attribute.String("endpoint", subject), attribute.String("outcome", "ok")) + noneAttrs := attribute.NewSet( + attribute.String("service", "test"), attribute.String("endpoint", "nobody.svc.get"), + attribute.String("outcome", "no_responders")) + + counter := findMetric(t, rm, "nats_requests_total") + sum, ok := counter.Data.(metricdata.Sum[int64]) + require.True(t, ok, "counter data is %T", counter.Data) + counts := map[attribute.Set]int64{} + for _, dp := range sum.DataPoints { + counts[dp.Attributes] = dp.Value + } + assert.Equal(t, map[attribute.Set]int64{okAttrs: 2, noneAttrs: 1}, counts) + + histogram := findMetric(t, rm, "nats_request_duration_seconds") + assert.Equal(t, "s", histogram.Unit) + hist, ok := histogram.Data.(metricdata.Histogram[float64]) + require.True(t, ok, "histogram data is %T", histogram.Data) + samples := map[attribute.Set]uint64{} + for _, dp := range hist.DataPoints { + samples[dp.Attributes] = dp.Count + assert.Positive(t, dp.Sum) + } + assert.Equal(t, map[attribute.Set]uint64{okAttrs: 2, noneAttrs: 1}, samples) +} + +func findMetric(t *testing.T, rm metricdata.ResourceMetrics, name string) metricdata.Metrics { + t.Helper() + for _, sm := range rm.ScopeMetrics { + if sm.Scope.Name != "github.com/MetsaApp/platform-kit/natsclient" { + continue + } + for _, m := range sm.Metrics { + if m.Name == name { + return m + } + } + } + t.Fatalf("metric %s not collected", name) + return metricdata.Metrics{} +} diff --git a/natsclient/service.go b/natsclient/service.go new file mode 100644 index 0000000..a0bf671 --- /dev/null +++ b/natsclient/service.go @@ -0,0 +1,135 @@ +package natsclient + +import ( + "context" + "errors" + "fmt" + "log/slog" + + "github.com/nats-io/nats.go/micro" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/propagation" + "go.opentelemetry.io/otel/trace" +) + +// serviceErrorCode is the code of an error response from a handler that +// did not choose one with a *ServiceError. +const serviceErrorCode = "500" + +// EndpointSubject is the subject of a synchronous endpoint per ADR 0006: +// ".svc.". Callers build their request subjects with +// it, so the convention lives in one place. +func EndpointSubject(service, endpoint string) string { + return service + ".svc." + endpoint +} + +// Request is one request to an endpoint: its data and headers, and the +// reply methods. A handler answers with Respond or RespondJSON, or returns +// an error; it does not do both. +type Request = micro.Request + +// Handler serves one request. ctx carries the caller's trace, so spans and +// *Context log calls inside join it. A returned error becomes an error +// response: a *ServiceError chooses the code, anything else is code 500 +// with the error's text, which the caller sees as a *ServiceError. +type Handler func(ctx context.Context, req Request) error + +// EndpointOption tunes one endpoint. +type EndpointOption func(*endpointOptions) + +type endpointOptions struct { + queueGroup string +} + +// WithQueueGroup overrides the endpoint's queue group, which is the +// service's name by default so every replica shares the load. A different +// group receives every request too, for an endpoint that must see all of +// them. +func WithQueueGroup(name string) EndpointOption { + return func(o *endpointOptions) { o.queueGroup = name } +} + +// Service is a set of request/reply endpoints under one service name. It is +// a nats.go micro service, so it also answers the $SRV.PING, $SRV.INFO and +// $SRV.STATS discovery subjects, which the nats CLI's "micro" commands read. +type Service struct { + micro micro.Service + name string + client *Client + log *slog.Logger +} + +// Service registers a micro service named name at version (SemVer) on the +// connection. Endpoints are added with AddEndpoint; Stop drains them. +func (c *Client) Service(name, version string) (*Service, error) { + log := c.log.With("nats_service", name) + svc, err := micro.AddService(c.nc, micro.Config{ + Name: name, + Version: version, + QueueGroup: name, + ErrorHandler: func(_ micro.Service, natsErr *micro.NATSError) { + log.Warn("nats service error", "subject", natsErr.Subject, "error", natsErr.Description) + }, + }) + if err != nil { + return nil, fmt.Errorf("natsclient: add service %s: %w", name, err) + } + return &Service{micro: svc, name: name, client: c, log: log}, nil +} + +// AddEndpoint serves EndpointSubject(service, endpoint) with handler, in +// the service's queue group unless WithQueueGroup says otherwise. Requests +// to one endpoint are handled one at a time, in order, as nats.go delivers +// them; a slow handler is throttled by its own subscription, not the +// service's others. +func (s *Service) AddEndpoint(endpoint string, handler Handler, opts ...EndpointOption) error { + o := endpointOptions{queueGroup: s.name} + for _, opt := range opts { + opt(&o) + } + subject := EndpointSubject(s.name, endpoint) + err := s.micro.AddEndpoint(endpoint, micro.HandlerFunc(s.handle(subject, handler)), + micro.WithEndpointSubject(subject), micro.WithEndpointQueueGroup(o.queueGroup)) + if err != nil { + return fmt.Errorf("natsclient: add endpoint %s: %w", subject, err) + } + return nil +} + +// Stop drains the endpoint subscriptions and answers no more requests. +func (s *Service) Stop() error { + if err := s.micro.Stop(); err != nil { + return fmt.Errorf("natsclient: stop service %s: %w", s.name, err) + } + return nil +} + +// handle wraps handler for micro: it continues the caller's trace from the +// request headers in a server span and turns a returned error into an +// error response. +func (s *Service) handle(subject string, handler Handler) func(micro.Request) { + return func(req micro.Request) { + ctx := s.client.propagator.Extract(context.Background(), propagation.HeaderCarrier(req.Headers())) + ctx, span := s.client.tracer.Start(ctx, subject, + trace.WithSpanKind(trace.SpanKindServer), + trace.WithAttributes(attribute.String("messaging.system", "nats"), + attribute.String("messaging.destination.name", subject))) + defer span.End() + + err := handler(ctx, req) + if err == nil { + return + } + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) + code, description := serviceErrorCode, err.Error() + if serviceErr, ok := errors.AsType[*ServiceError](err); ok { + code, description = serviceErr.Code, serviceErr.Description + } + s.log.WarnContext(ctx, "endpoint failed", "subject", subject, "code", code, "error", err) + if respondErr := req.Error(code, description, nil); respondErr != nil { + s.log.WarnContext(ctx, "error response not sent", "subject", subject, "error", respondErr) + } + } +} diff --git a/natsclient/service_test.go b/natsclient/service_test.go new file mode 100644 index 0000000..2a5ae3a --- /dev/null +++ b/natsclient/service_test.go @@ -0,0 +1,96 @@ +package natsclient_test + +import ( + "context" + "sync/atomic" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + "go.opentelemetry.io/otel/trace" + + "github.com/MetsaApp/platform-kit/natsclient" +) + +// counting is an endpoint handler that counts its requests before echoing. +func counting(n *atomic.Int64) natsclient.Handler { + return func(_ context.Context, req natsclient.Request) error { + n.Add(1) + return req.Respond(req.Data()) + } +} + +// Two instances of one service are two replicas: they share one queue +// group, so a burst is split between them. An endpoint in another group is +// a separate subscriber and sees every request. +func TestQueueGroupSharing(t *testing.T) { + const burst = 40 + srv := startServer(t) + c := connect(t, srv, identity, "", nil) + var replicaA, replicaB, observer atomic.Int64 + for _, n := range []*atomic.Int64{&replicaA, &replicaB} { + svc, err := c.Service("orders", "1.0.0") + require.NoError(t, err) + t.Cleanup(func() { _ = svc.Stop() }) + require.NoError(t, svc.AddEndpoint("get", counting(n))) + } + audit, err := c.Service("orders", "1.0.0") + require.NoError(t, err) + t.Cleanup(func() { _ = audit.Stop() }) + require.NoError(t, audit.AddEndpoint("get", counting(&observer), natsclient.WithQueueGroup("audit"))) + + subject := natsclient.EndpointSubject("orders", "get") + for range burst { + _, reqErr := request(t, c, subject, "x") + require.NoError(t, reqErr) + } + assert.Equal(t, int64(burst), replicaA.Load()+replicaB.Load(), "the replicas split the burst") + assert.Positive(t, replicaA.Load()) + assert.Positive(t, replicaB.Load()) + assert.Equal(t, int64(burst), observer.Load(), "another group sees every request") +} + +// The caller's trace continues in the handler: the request carries a W3C +// traceparent, the handler's context holds the same trace id, and the +// exported spans form caller -> client -> server. +func TestTracePropagation(t *testing.T) { + exporter := tracetest.NewInMemoryExporter() + tp := sdktrace.NewTracerProvider(sdktrace.WithSyncer(exporter)) + otel.SetTracerProvider(tp) + t.Cleanup(func() { _ = tp.Shutdown(context.Background()) }) + + c := connect(t, startServer(t), identity, "", nil) + svc, err := c.Service("orders", "1.0.0") + require.NoError(t, err) + t.Cleanup(func() { _ = svc.Stop() }) + seen := make(chan trace.SpanContext, 1) + require.NoError(t, svc.AddEndpoint("get", func(ctx context.Context, req natsclient.Request) error { + seen <- trace.SpanContextFromContext(ctx) + return req.Respond(nil) + })) + + ctx, caller := tp.Tracer("test").Start(t.Context(), "caller") + ctx, cancel := context.WithTimeout(ctx, requestTimeout) + defer cancel() + _, err = c.Request(ctx, natsclient.EndpointSubject("orders", "get"), nil, nil) + require.NoError(t, err) + caller.End() + + handlerSC := <-seen + assert.Equal(t, caller.SpanContext().TraceID(), handlerSC.TraceID()) + assert.NotEqual(t, caller.SpanContext().SpanID(), handlerSC.SpanID()) + + kinds := map[trace.SpanKind]tracetest.SpanStub{} + for _, s := range exporter.GetSpans() { + kinds[s.SpanKind] = s + } + client, server := kinds[trace.SpanKindClient], kinds[trace.SpanKindServer] + assert.Equal(t, "orders.svc.get", client.Name) + assert.Equal(t, "orders.svc.get", server.Name) + assert.Equal(t, caller.SpanContext().SpanID(), client.Parent.SpanID()) + assert.Equal(t, client.SpanContext.SpanID(), server.Parent.SpanID()) + assert.Equal(t, server.SpanContext.SpanID(), handlerSC.SpanID()) +}