Go Microservices Patterns
gRPC Service Implementation
Protocol Definition
// proto/service.proto
syntax = "proto3";
package myservice.v1;
option go_package = "github.com/user/project/proto";
service MyService {
rpc Create(CreateRequest) returns (CreateResponse);
rpc Get(GetRequest) returns (GetResponse);
rpc List(ListRequest) returns (stream ListResponse);
}
message CreateRequest {
string name = 1;
string description = 2;
}
message CreateResponse {
string id = 1;
int64 created_at = 2;
}
message GetRequest {
string id = 1;
}
message GetResponse {
string id = 1;
string name = 2;
string description = 3;
}
message ListRequest {
int32 page_size = 1;
string page_token = 2;
}
message ListResponse {
repeated Item items = 1;
string next_page_token = 2;
}
Server Implementation
package server
import (
"context"
"log/slog"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
type server struct {
pb.UnimplementedMyServiceServer
svc Service
log *slog.Logger
}
func NewServer(svc Service, log *slog.Logger) *grpc.Server {
s := grpc.NewServer(
grpc.ChainUnaryInterceptor(
loggingInterceptor(log),
recoveryInterceptor(),
validationInterceptor(),
),
)
pb.RegisterMyServiceServer(s, &server{
svc: svc,
log: log,
})
return s
}
func (s *server) Create(ctx context.Context, req *pb.CreateRequest) (*pb.CreateResponse, error) {
if err := validateCreateRequest(req); err != nil {
return nil, status.Error(codes.InvalidArgument, err.Error())
}
result, err := s.svc.Create(ctx, CreateRequest{
Name: req.Name,
Description: req.Description,
})
if err != nil {
return nil, handleError(err)
}
return &pb.CreateResponse{
Id: result.ID,
CreatedAt: result.CreatedAt,
}, nil
}
func (s *server) Get(ctx context.Context, req *pb.GetRequest) (*pb.GetResponse, error) {
result, err := s.svc.Get(ctx, req.Id)
if err != nil {
if errors.Is(err, ErrNotFound) {
return nil, status.Error(codes.NotFound, "item not found")
}
return nil, status.Error(codes.Internal, err.Error())
}
return toProtoResponse(result), nil
}
func (s *server) List(req *pb.ListRequest, stream pb.MyService_ListServer) error {
for {
resp, err := s.svc.List(stream.Context(), ListRequest{
PageSize: req.PageSize,
PageToken: req.PageToken,
})
if err != nil {
return err
}
for _, item := range resp.Items {
if err := stream.Send(&pb.ListResponse{Items: item}); err != nil {
return err
}
}
if resp.NextPageToken == "" {
return nil
}
req.PageToken = resp.NextPageToken
}
}
Unary Interceptors
func loggingInterceptor(log *slog.Logger) grpc.UnaryServerInterceptor {
return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
start := time.Now()
resp, err := handler(ctx, req)
duration := time.Since(start)
log.Info("gRPC request",
"method", info.FullMethod,
"duration", duration,
"error", err,
)
return resp, err
}
}
func recoveryInterceptor() grpc.UnaryServerInterceptor {
return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
defer func() {
if r := recover(); r != nil {
err = status.Error(codes.Internal, "internal server error")
slog.Error("panic recovered", "panic", r)
}
}()
return handler(ctx, req)
}
}
REST API with Middleware
Chi Router Pattern
package api
import (
"github.com/go-chi/chi/v5"
"github.com/go-chi/chi/v5/middleware"
"github.com/go-chi/cors"
)
type Server struct {
router *chi.Mux
svc Service
log *slog.Logger
}
func NewServer(svc Service, log *slog.Logger) *Server {
r := chi.NewRouter()
// Middleware
r.Use(middleware.RequestID)
r.Use(middleware.RealIP)
r.Use(middleware.Logger)
r.Use(middleware.Recoverer)
r.Use(middleware.Timeout(60 * time.Second))
r.Use(middleware.SetHeader("Content-Type", "application/json"))
// CORS
r.Use(cors.Handler(cors.Options{
AllowedOrigins: []string{"https://*", "http://*"},
AllowedMethods: []string{"GET", "POST", "PUT", "DELETE", "OPTIONS"},
AllowedHeaders: []string{"Accept", "Authorization", "Content-Type"},
ExposedHeaders: []string{"Link"},
AllowCredentials: true,
MaxAge: 300,
}))
s := &Server{router: r, svc: svc, log: log}
// Routes
r.Route("/api/v1", func(r chi.Router) {
r.Post("/items", s.createItem)
r.Get("/items/{id}", s.getItem)
r.Get("/items", s.listItems)
r.Put("/items/{id}", s.updateItem)
r.Delete("/items/{id}", s.deleteItem)
})
return s
}
func (s *Server) createItem(w http.ResponseWriter, r *http.Request) {
var req CreateRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
http.Error(w, "invalid request", http.StatusBadRequest)
return
}
result, err := s.svc.Create(r.Context(), req)
if err != nil {
handleError(w, err)
return
}
w.WriteHeader(http.StatusCreated)
json.NewEncoder(w).Encode(result)
}
func handleError(w http.ResponseWriter, err error) {
switch {
case errors.Is(err, ErrNotFound):
http.Error(w, "not found", http.StatusNotFound)
case errors.Is(err, ErrValidation):
http.Error(w, err.Error(), http.StatusBadRequest)
default:
http.Error(w, "internal error", http.StatusInternalServerError)
}
}
Service Discovery Integration
Consul Registration
package discovery
import (
"github.com/hashicorp/consul/api"
)
type Registrar struct {
client *api.Client
id string
}
func NewRegistrar(consulAddr, serviceName, serviceID string) (*Registrar, error) {
config := api.DefaultConfig()
config.Address = consulAddr
client, err := api.NewClient(config)
if err != nil {
return nil, err
}
return &Registrar{
client: client,
id: serviceID,
}, nil
}
func (r *Registrar) Register(addr string, port int) error {
registration := &api.AgentServiceRegistration{
ID: r.id,
Name: "my-service",
Port: port,
Address: addr,
Check: &api.AgentServiceCheck{
GRPC: fmt.Sprintf("%s:%d", addr, port),
Interval: "10s",
Timeout: "3s",
},
Tags: []string{"go", "microservice"},
}
return r.client.Agent().ServiceRegister(registration)
}
func (r *Registrar) Deregister() error {
return r.client.Agent().ServiceDeregister(r.id)
}
Circuit Breaker Pattern
package circuitbreaker
import (
"github.com/sony/gobreaker"
)
type CircuitBreaker struct {
breaker *gobreaker.CircuitBreaker
}
func NewCircuitBreaker() *CircuitBreaker {
settings := gobreaker.Settings{
Name: "my-service",
MaxRequests: 100,
Interval: 10 * time.Second,
Timeout: 30 * time.Second,
ReadyToTrip: func(counts gobreaker.Counts) bool {
return counts.ConsecutiveFailures > 5
},
OnStateChange: func(name string, from, to gobreaker.State) {
slog.Info("circuit breaker state changed",
"name", name,
"from", from,
"to", to,
)
},
}
return &CircuitBreaker{
breaker: gobreaker.NewCircuitBreaker(settings),
}
}
func (cb *CircuitBreaker) Execute(fn func() (interface{}, error)) (interface{}, error) {
return cb.breaker.Execute(fn)
}
// Usage in service
func (s *Service) CallExternal(ctx context.Context, req Request) (Response, error) {
result, err := s.cb.Execute(func() (interface{}, error) {
return s.externalAPI.Call(ctx, req)
})
if err != nil {
return Response{}, fmt.Errorf("circuit breaker: %w", err)
}
return result.(Response), nil
}
Health Checks
package health
import (
"google.golang.org/grpc/health"
"google.golang.org/grpc/health/grpc_health_v1"
)
type Checker struct {
healthServer *health.Server
db *sql.DB
redis *redis.Client
}
func NewChecker(db *sql.DB, redis *redis.Client) *Checker {
hc := health.NewServer()
return &Checker{
healthServer: hc,
db: db,
redis: redis,
}
}
func (c *Checker) Start() {
c.healthServer.SetServingStatus("", grpc_health_v1.HealthCheckResponse_SERVING)
// Start background checks
go c.checkDatabase()
go c.checkRedis()
}
func (c *Checker) checkDatabase() {
ticker := time.NewTicker(30 * time.Second)
defer ticker.Stop()
for range ticker.C {
if err := c.db.Ping(); err != nil {
c.healthServer.SetServingStatus("db", grpc_health_v1.HealthCheckResponse_NOT_SERVING)
} else {
c.healthServer.SetServingStatus("db", grpc_health_v1.HealthCheckResponse_SERVING)
}
}
}
func (c *Checker) checkRedis() {
ticker := time.NewTicker(30 * time.Second)
defer ticker.Stop()
for range ticker.C {
if err := c.redis.Ping(context.Background()).Err(); err != nil {
c.healthServer.SetServingStatus("redis", grpc_health_v1.HealthCheckResponse_NOT_SERVING)
} else {
c.healthServer.SetServingStatus("redis", grpc_health_v1.HealthCheckResponse_SERVING)
}
}
}
Graceful Shutdown
package server
import (
"context"
"net"
"net/http"
"time"
"google.golang.org/grpc"
)
type Server struct {
httpServer *http.Server
grpcServer *grpc.Server
log *slog.Logger
}
func NewServer(httpAddr, grpcAddr string, handlers http.Handler) *Server {
return &Server{
httpServer: &http.Server{
Addr: httpAddr,
Handler: handlers,
ReadTimeout: 15 * time.Second,
WriteTimeout: 15 * time.Second,
IdleTimeout: 60 * time.Second,
MaxHeaderBytes: 1 << 20, // 1MB
},
grpcServer: grpc.NewServer(),
log: slog.Default(),
}
}
func (s *Server) Start() error {
// Start HTTP
go func() {
s.log.Info("starting HTTP server", "addr", s.httpServer.Addr)
if err := s.httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
s.log.Error("HTTP server error", "error", err)
}
}()
// Start gRPC
lis, err := net.Listen("tcp", s.grpcServerAddr)
if err != nil {
return err
}
go func() {
s.log.Info("starting gRPC server", "addr", s.grpcServerAddr)
if err := s.grpcServer.Serve(lis); err != nil {
s.log.Error("gRPC server error", "error", err)
}
}()
return nil
}
func (s *Server) Shutdown(ctx context.Context) error {
var wg sync.WaitGroup
errs := make(chan error, 2)
// Shutdown HTTP
wg.Add(1)
go func() {
defer wg.Done()
errs <- s.httpServer.Shutdown(ctx)
}()
// Shutdown gRPC
wg.Add(1)
go func() {
defer wg.Done()
s.grpcServer.GracefulStop()
errs <- nil
}()
wg.Wait()
close(errs)
for err := range errs {
if err != nil {
return err
}
}
return nil
}
// Main with signal handling
func main() {
srv := NewServer(":8080", ":9090", handlers)
if err := srv.Start(); err != nil {
log.Fatal(err)
}
// Wait for interrupt signal
sigChan := make(chan os.Signal, 1)
signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)
<-sigChan
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
if err := srv.Shutdown(ctx); err != nil {
log.Fatal(err)
}
log.Println("server shutdown complete")
}
Configuration Management
Environment-based Configuration
package config
import (
"github.com/kelseyhightower/envconfig"
)
type Config struct {
// Server
HTTPPort int `env:"HTTP_PORT" envDefault:"8080"`
GRPCPort int `env:"GRPC_PORT" envDefault:"9090"`
// Database
DatabaseURL string `env:"DATABASE_URL" envDefault:"postgres://localhost:5432/mydb"`
DBPoolSize int `env:"DB_POOL_SIZE" envDefault:"10"`
// Redis
RedisURL string `env:"REDIS_URL" envDefault:"redis://localhost:6379"`
// Service Discovery
ConsulAddr string `env:"CONSUL_ADDR" envDefault:"localhost:8500"`
// Observability
LogLevel string `env:"LOG_LEVEL" envDefault:"info"`
MetricsPort int `env:"METRICS_PORT" envDefault:"9091"`
TracingAddr string `env:"TRACING_ADDR" envDefault:"localhost:4318"`
ServiceName string `env:"SERVICE_NAME" envDefault:"my-service"`
ServiceVersion string `env:"SERVICE_VERSION" envDefault:"1.0.0"`
}
func Load() (*Config, error) {
var cfg Config
if err := envconfig.Process("", &cfg); err != nil {
return nil, err
}
return &cfg, nil
}
Observability Setup
Structured Logging
package logging
import (
"log/slog"
"os"
)
func NewLogger(level string, service, version string) *slog.Logger {
var logLevel slog.Level
switch level {
case "debug":
logLevel = slog.LevelDebug
case "info":
logLevel = slog.LevelInfo
case "warn":
logLevel = slog.LevelWarn
case "error":
logLevel = slog.LevelError
default:
logLevel = slog.LevelInfo
}
opts := &slog.HandlerOptions{
Level: logLevel,
}
handler := slog.NewJSONHandler(os.Stdout, opts)
logger := slog.New(handler)
return logger.With(
"service", service,
"version", version,
)
}
Prometheus Metrics
package metrics
import (
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
)
var (
requestsTotal = promauto.NewCounterVec(
prometheus.CounterOpts{
Name: "my_service_requests_total",
Help: "Total number of requests",
},
[]string{"method", "endpoint", "status"},
)
requestDuration = promauto.NewHistogramVec(
prometheus.HistogramOpts{
Name: "my_service_request_duration_seconds",
Help: "Request duration in seconds",
Buckets: prometheus.DefBuckets,
},
[]string{"method", "endpoint"},
)
activeConnections = promauto.NewGauge(
prometheus.GaugeOpts{
Name: "my_service_active_connections",
Help: "Number of active connections",
},
)
)
func RecordRequest(method, endpoint, status string, duration float64) {
requestsTotal.WithLabelValues(method, endpoint, status).Inc()
requestDuration.WithLabelValues(method, endpoint).Observe(duration)
}
func IncrementConnections() {
activeConnections.Inc()
}
func DecrementConnections() {
activeConnections.Dec()
}
Distributed Tracing with OpenTelemetry
package tracing
import (
"context"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
"go.opentelemetry.io/otel/sdk/resource"
sdktrace "go.opentelemetry.io/otel/sdk/trace"
semconv "go.opentelemetry.io/otel/semconv/v1.4.0"
)
func InitTracer(ctx context.Context, addr, service, version string) (func(context.Context) error, error) {
res, err := resource.New(ctx,
resource.WithAttributes(
semconv.ServiceNameKey.String(service),
semconv.ServiceVersionKey.String(version),
),
)
if err != nil {
return nil, err
}
exporter, err := otlptracegrpc.New(ctx,
otlptracegrpc.WithInsecure(),
otlptracegrpc.WithEndpoint(addr),
)
if err != nil {
return nil, err
}
tp := sdktrace.NewTracerProvider(
sdktrace.WithBatcher(exporter),
sdktrace.WithResource(res),
sdktrace.WithSampler(sdktrace.TraceIDRatioBased(1.0)),
)
otel.SetTracerProvider(tp)
return tp.Shutdown, nil
}
Middleware Chain for Observability
package middleware
import (
"github.com/prometheus/client_golang/prometheus/promhttp"
)
func ObservabilityMiddleware(service, version string) func(http.Handler) http.Handler {
return func(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
start := time.Now()
// Wrap writer to capture status
ww := &responseWriter{ResponseWriter: w}
next.ServeHTTP(ww, r)
// Record metrics
duration := time.Since(start).Seconds()
metrics.RecordRequest(
r.Method,
r.URL.Path,
fmt.Sprintf("%d", ww.status),
duration,
)
// Log request
slog.Info("request",
"method", r.Method,
"path", r.URL.Path,
"status", ww.status,
"duration", duration,
"user_agent", r.UserAgent(),
)
})
}
}
type responseWriter struct {
http.ResponseWriter
status int
}
func (w *responseWriter) WriteHeader(status int) {
w.status = status
w.ResponseWriter.WriteHeader(status)
}
// Metrics endpoint
func MetricsHandler() http.Handler {
return promhttp.Handler()
}
Docker Compose for Development
version: '3.8'
services:
app:
build: .
ports:
- "8080:8080"
- "9090:9090"
- "9091:9091"
environment:
- DATABASE_URL=postgres://postgres:password@db:5432/mydb
- REDIS_URL=redis://redis:6379
- CONSUL_ADDR=consul:8500
- TRACING_ADDR=jaeger:4317
depends_on:
- db
- redis
- consul
- jaeger
db:
image: postgres:16
environment:
- POSTGRES_PASSWORD=password
- POSTGRES_DB=mydb
volumes:
- postgres_data:/var/lib/postgresql/data
redis:
image: redis:7
volumes:
- redis_data:/data
consul:
image: consul:latest
ports:
- "8500:8500"
command: consul agent -server -ui -bootstrap-expect=1 -client=0.0.0.0
jaeger:
image: jaegertracing/all-in-one:latest
ports:
- "5775:5775/udp"
- "6831:6831/udp"
- "6832:6832/udp"
- "5778:5778"
- "16686:16686"
- "14268:14268"
- "14250:14250"
- "9411:9411"
volumes:
postgres_data:
redis_data:
Best Practices Summary
- Always implement health checks - For readiness/liveness probes
- Use circuit breakers - Prevent cascade failures
- Register with service discovery - Enable dynamic routing
- Implement graceful shutdown - Handle SIGTERM properly
- Add observability - Logging, metrics, tracing from day one
- Use structured logging - JSON format with contextual fields
- Set reasonable timeouts - Prevent resource leaks
- Implement backpressure - Rate limiting and throttling
- Secure inter-service communication - mTLS where needed
- Design for failure - Assume dependencies will fail
andvl1