I’ve been thinking a lot about how to build systems that not only scale but also remain understandable when things inevitably go wrong. In my experience, the combination of Go’s simplicity, NATS JetStream’s reliability, and OpenTelemetry’s observability creates a powerful foundation for production-ready event-driven microservices. Let me show you how these pieces fit together in a real system.

Why choose this stack? Go’s concurrency model and performance characteristics make it ideal for handling high volumes of events. NATS JetStream provides durable, at-least-once message delivery without the operational overhead of traditional message brokers. OpenTelemetry gives us the visibility we need to understand our distributed system’s behavior.

Setting up our infrastructure starts with a simple Docker Compose file. This gives us NATS with JetStream enabled, Jaeger for tracing, and Prometheus for metrics—everything we need for local development and testing.

version: '3.8'
services:
  nats:
    image: nats:2.10-alpine
    ports:
      - "4222:4222"
      - "8222:8222"
    command: 
      - "--jetstream"
      - "--store_dir=/data"
    volumes:
      - nats_data:/data

  jaeger:
    image: jaegertracing/all-in-one:1.50
    ports:
      - "16686:16686"

volumes:
  nats_data:

Our event schema design is crucial. We need events that are self-describing and include necessary metadata for tracing and correlation. Have you considered how your events will evolve over time?

type BaseEvent struct {
    ID          string            `json:"id"`
    Type        string            `json:"type"`
    Version     string            `json:"version"`
    Timestamp   time.Time         `json:"timestamp"`
    Source      string            `json:"source"`
    TraceID     string            `json:"traceId"`
}

type OrderCreated struct {
    BaseEvent
    Data OrderData `json:"data"`
}

func NewOrderCreatedEvent(orderID string) *OrderCreated {
    return &OrderCreated{
        BaseEvent: BaseEvent{
            ID:        uuid.New().String(),
            Type:      "order.created",
            Version:   "v1",
            Timestamp: time.Now().UTC(),
            Source:    "order-service",
        },
        Data: OrderData{OrderID: orderID},
    }
}

Connecting to NATS JetStream requires proper configuration for production use. We need to handle connection failures, implement proper cleanup, and ensure our publishers and consumers are resilient.

func ConnectJetStream(url string) (nats.JetStreamContext, error) {
    nc, err := nats.Connect(url, 
        nats.MaxReconnects(5),
        nats.ReconnectWait(2*time.Second),
        nats.DisconnectErrHandler(func(nc *nats.Conn, err error) {
            log.Printf("Disconnected from NATS: %v", err)
        }),
    )
    if err != nil {
        return nil, err
    }

    js, err := nc.JetStream()
    if err != nil {
        return nil, err
    }

    return js, nil
}

Publishing events isn’t just about sending messages—it’s about ensuring they’re delivered reliably. How do you handle scenarios where the message broker is temporarily unavailable?

func (p *Publisher) PublishOrderCreated(ctx context.Context, event *OrderCreated) error {
    span := trace.SpanFromContext(ctx)
    event.TraceID = span.SpanContext().TraceID().String()

    data, err := json.Marshal(event)
    if err != nil {
        return fmt.Errorf("marshaling event: %w", err)
    }

    _, err = p.js.Publish("orders.created", data, 
        nats.MsgId(event.ID),
        nats.ExpectStream("ORDERS"),
    )
    if err != nil {
        span.RecordError(err)
        return fmt.Errorf("publishing event: %w", err)
    }

    return nil
}

On the consumer side, we need to handle message processing with proper error handling and observability. What happens when your consumer crashes mid-processing?

func (c *Consumer) ProcessOrders() {
    c.js.Subscribe("orders.created", func(m *nats.Msg) {
        ctx := context.Background()
        ctx, span := tracer.Start(ctx, "process_order")
        defer span.End()

        var event OrderCreated
        if err := json.Unmarshal(m.Data, &event); err != nil {
            span.RecordError(err)
            log.Printf("Failed to unmarshal event: %v", err)
            m.Nak()
            return
        }

        if err := c.processOrder(ctx, event); err != nil {
            span.RecordError(err)
            log.Printf("Failed to process order: %v", err)
            m.Nak()
            return
        }

        m.Ack()
    }, nats.ManualAck())
}

OpenTelemetry integration gives us the visibility we need across our distributed system. The real power comes from correlating traces across service boundaries.

func InitTracing(serviceName string) (*sdktrace.TracerProvider, error) {
    exporter, err := jaeger.New(jaeger.WithCollectorEndpoint(
        jaeger.WithEndpoint("http://jaeger:14268/api/traces"),
    ))
    if err != nil {
        return nil, err
    }

    tp := sdktrace.NewTracerProvider(
        sdktrace.WithBatcher(exporter),
        sdktrace.WithResource(resource.NewWithAttributes(
            semconv.SchemaURL,
            semconv.ServiceName(serviceName),
        )),
    )

    otel.SetTracerProvider(tp)
    return tp, nil
}

Error handling and retries are where many event-driven systems fall short. We need strategies for handling transient failures without losing messages.

func (c *Consumer) processOrderWithRetry(ctx context.Context, event OrderCreated) error {
    retryPolicy := backoff.NewExponentialBackOff()
    retryPolicy.MaxElapsedTime = 5 * time.Minute

    operation := func() error {
        return c.processOrder(ctx, event)
    }

    if err := backoff.Retry(operation, retryPolicy); err != nil {
        return fmt.Errorf("failed after retries: %w", err)
    }

    return nil
}

Testing event-driven systems requires a different approach. We need to verify not just individual components, but the entire flow of events through our system.

func TestOrderProcessing(t *testing.T) {
    js := setupTestJetStream(t)
    defer cleanupTest(t)

    publisher := NewPublisher(js)
    consumer := NewConsumer(js)

    event := NewOrderCreatedEvent("test-order")
    if err := publisher.PublishOrderCreated(context.Background(), event); err != nil {
        t.Fatalf("Failed to publish event: %v", err)
    }

    // Wait for processing
    time.Sleep(100 * time.Millisecond)

    // Verify side effects
    if !orderWasProcessed("test-order") {
        t.Error("Order was not processed correctly")
    }
}

Deployment considerations include proper health checks, resource limits, and monitoring configuration. How do you know your event processors are keeping up with the load?

# Kubernetes deployment snippet
livenessProbe:
  httpGet:
    path: /health
    port: 8080
  initialDelaySeconds: 30
  periodSeconds: 10

resources:
  limits:
    memory: "256Mi"
    cpu: "500m"

Monitoring production systems means watching both infrastructure metrics and business-level indicators. Are orders being processed within acceptable latency bounds?

func MonitorOrderProcessing() {
    ordersProcessed := promauto.NewCounter(prometheus.CounterOpts{
        Name: "orders_processed_total",
        Help: "Total number of orders processed",
    })

    processingTime := promauto.NewHistogram(prometheus.HistogramOpts{
        Name:    "order_processing_seconds",
        Help:    "Time taken to process an order",
        Buckets: prometheus.DefBuckets,
    })
}

Building production-ready event-driven systems is challenging but incredibly rewarding when done right. The patterns I’ve shared here come from real-world experience building and operating these systems at scale.

I’d love to hear about your experiences with event-driven architectures. What challenges have you faced? Share your thoughts in the comments below, and if you found this useful, please consider liking and sharing with others who might benefit from these patterns.