Project Structure and Architecture — Go Package Layout for a Distributed Log
Before writing any code, set up the Go project and understand how the packages are organized. This page walks through the directory layout, what each package is responsible for, and the layered architecture that ties everything together.
The project is split into independent modules along team/deployment boundaries: everything that runs on a broker node lives under broker/, the wire format and transport that broker and clients both share live under api/, and the producer and consumer are each self-contained packages with their own client library and CLI.
Step 1: Initialize the Go module
Create the project directory and initialize a Go module:
mkdir distributed-log && cd distributed-log
go mod init github.com/mohitkumar/mlog
Step 2: Create the project layout
Use these commands to create the directory structure:
mkdir -p api/protocol/pb api/protocol/proto api/transport
mkdir -p broker/{cluster,cmd,config,consumer,log,rpc,segment,topic}
mkdir -p broker/cluster/{discovery,raft} broker/cmd/{server,topic}
mkdir -p producer/{client,cmd} consumer/{client,cmd}
mkdir -p client scripts infra tests
Each package owns one layer or concern of the system:
distributed-log/
├── api/
│ ├── protocol/ # Wire format: frames, message types, protobuf codec
│ │ ├── types.go # Go-native request/response structs (used by all callers)
│ │ ├── frame.go # Length-prefixed frame header + MessageType enum
│ │ ├── codec.go # Encode/Decode: Go structs ⇄ protobuf ⇄ frame bytes
│ │ ├── convert.go # LogEntry/TopicInfo ⇄ pb.* conversion helpers
│ │ ├── metadata.go # Raft-replicated metadata event encode/decode
│ │ ├── replication_batch.go # Raw [offset][len][value] batch format for replica fetch
│ │ ├── error.go # RPCError, error codes, ShouldReconnect
│ │ ├── proto/mlog.proto # Protobuf source
│ │ └── pb/mlog.pb.go # Generated protobuf types
│ └── transport/ # TCP transport (server and client)
│ └── transport.go
├── broker/
│ ├── cmd/
│ │ ├── server/ # Broker daemon binary
│ │ │ ├── main.go
│ │ │ └── helper.go # Wires cluster → topic manager → RPC server → discovery
│ │ └── topic/main.go # Topic management CLI binary
│ ├── config/ # Configuration structs
│ │ └── config.go
│ ├── segment/ # Segment and index (byte encoding, mmap)
│ │ ├── segment.go
│ │ ├── index.go
│ │ ├── errors.go
│ │ └── test_util.go
│ ├── log/ # Log API (multiple segments) + LogManager (LEO/HW)
│ │ ├── log.go
│ │ ├── log_manager.go
│ │ └── errors.go
│ ├── rpc/ # Server-side RPC handlers (produce, consume, admin)
│ │ ├── server.go
│ │ ├── producer.go
│ │ ├── consumer.go
│ │ ├── leader.go
│ │ └── error.go
│ ├── cluster/ # Cluster metadata + consensus + membership
│ │ ├── cluster.go # Cluster: the one object topic manager talks to
│ │ ├── cluster_metadata.go # ClusterMetadataStore: Raft-applied topic/replica state
│ │ ├── discovery/ # Serf cluster membership
│ │ │ └── discovery.go
│ │ └── raft/ # Raft plumbing
│ │ ├── node.go # RaftNode: setup, join/leave, voter config
│ │ ├── fsm.go # Raft FSM shim → MetadataStore
│ │ ├── logstore.go # Raft LogStore backed by broker/log
│ │ ├── metadata.go # Metadata event encode/decode helpers
│ │ └── errors.go
│ ├── topic/ # Topic management, replication, ISR
│ │ ├── topic.go # TopicManager: per-node topic state
│ │ ├── topic_coordinator.go # TopicCoordinator interface (decouples topic from cluster)
│ │ ├── placement.go # Replica placement policy
│ │ ├── producer.go # HandleProduce/HandleProduceBatch
│ │ ├── replication.go # Pull-based replication threads
│ │ └── errors.go
│ └── consumer/ # Server-side consumer offset management
│ ├── consumer.go
│ └── error.go
├── client/ # Shared: admin client + reconnect/retry toolkit
│ ├── rpc.go # RemoteClient: CreateTopic/DeleteTopic/FindTopicLeader/...
│ ├── bootstrap.go # TryAddrs: dial bootstrap addresses in turn
│ ├── leader.go # ResolveTopicLeader, ReconnectBackoff
│ └── reconnect.go # ShouldReconnect
│ └── retry.go # RetryTopicNotReady
├── producer/
│ ├── client/
│ │ ├── producer.go # ProducerClient: one connection, Produce/ProduceBatch
│ │ └── client.go # Client: topic-aware, self-healing, reconnecting producer
│ └── cmd/main.go # Producer CLI binary
├── consumer/
│ ├── client/
│ │ ├── consumer.go # ConsumerClient: one connection, Fetch/Commit/FetchOffset
│ │ └── client.go # Client: topic-aware, self-healing, reconnecting consumer
│ └── cmd/main.go # Consumer CLI binary
├── tests/ # Integration and end-to-end tests
├── scripts/ # Helper scripts
│ ├── start-local-cluster.sh
│ └── stop-local-cluster.sh
├── infra/ # Docker files
│ ├── Dockerfile
│ └── docker-compose.yml
├── go.mod
├── go.sum
└── Makefile
Package responsibilities
Each package has a single, clear responsibility:
| Package | Layer | Responsibility |
|---|---|---|
| broker/segment | Storage | One segment: append-only .log file + sparse .idx index. Byte-level encoding of records. Memory-mapped index with binary search. |
| broker/log | Storage | Manages multiple segments. Append to active segment, roll when full, find segment for reads. LogManager adds LEO and high watermark tracking. |
| api/protocol | Network | All message types (Produce, Fetch, CreateTopic, etc.), frame format (length-prefixed), protobuf codec, replication batch encoding, error codes. |
| api/transport | Network | TCP server (accept connections, dispatch to handlers) and TCP client (dial, send request, read response). Connection management and keepalive. |
| broker/rpc | Network | Server-side RPC handlers. Maps message types to handler functions. Implements produce, fetch, topic CRUD, and leader discovery. |
| broker/cluster/raft | Consensus | Raft node lifecycle (setup, join, leave). FSM for metadata events. Adapts broker/log as a Raft LogStore. |
| broker/cluster/discovery | Cluster | Serf-based cluster membership. Gossip, join/leave events, member list with tags (RPC/Raft addresses). |
| broker/cluster | Cluster | Cluster: owns the Raft node and the metadata store; the single object broker/topic talks to for cluster-wide state. |
| broker/topic | Application | TopicManager: tracks topics, leaders, replicas on this node. Handles produce at the application level. Runs replication and reconciliation threads. Computes ISR and high watermark. |
| broker/consumer | Application | Server-side consumer offset tracking. Persists committed offsets to a local log. Recovers offsets on restart. |
| client | Client | Shared toolkit: RemoteClient (admin/discovery RPCs), bootstrap/leader-resolution, and the reconnect/retry helpers producer/client and consumer/client both build on. |
| producer/client | Client | Producer-side client library: a low-level ProducerClient (one connection) and a self-healing Client (leader discovery, reconnection) used by the producer CLI. |
| consumer/client | Client | Consumer-side client library: a low-level ConsumerClient (one connection, also reused internally by broker-to-broker replication) and a self-healing Client used by the consumer CLI. |
| broker/config | Infra | Configuration structs for node, Raft, Serf, and replication settings. |
| broker/cmd, producer/cmd, consumer/cmd | CLI | Cobra-based CLI binaries: server, topic management, producer, consumer. |
Note there's no central errs/coordinator package anymore — each package (segment, log, topic, cluster/raft, consumer) owns its own errors.go, and Raft/cluster concerns live entirely under broker/cluster.
Three-layer architecture
The system is organized into three layers. Each layer depends only on the layers below it:
┌─────────────────────────────────────────────────────────┐
│ Layer 3: Cluster │
│ │
│ broker/cluster/discovery broker/topic broker/rpc │
│ client producer/client consumer/client cmd/ │
│ │
│ Serf membership, topic management, replication, │
│ RPC handlers, client libraries, CLI tools │
├─────────────────────────────────────────────────────────┤
│ Layer 2: Consensus │
│ │
│ broker/cluster broker/cluster/raft │
│ │
│ Raft leader election, metadata replication, │
│ FSM for applying committed events │
├─────────────────────────────────────────────────────────┤
│ Layer 1: Storage + Network │
│ │
│ broker/segment broker/log api/protocol api/transport │
│ │
│ On-disk segments, indexes, log management, │
│ wire format, TCP transport │
└─────────────────────────────────────────────────────────┘
Layer 1 — Storage + Network: Pure local operations. Segments read and write bytes to disk. The protocol defines message formats and serializes them as protobuf. The transport sends and receives bytes over TCP. No cluster awareness.
Layer 2 — Consensus: Raft coordinates metadata across the cluster. broker/cluster/raft uses Layer 1 (the log as a Raft LogStore, transport for Raft's own RPCs), and broker/cluster wraps it together with the applied metadata store to give Layer 3 a consistent view of cluster state.
Layer 3 — Cluster: Everything that makes the system distributed. Topics are managed through Raft via the TopicCoordinator interface (implemented by broker/cluster.Cluster in production, and by an in-memory fake in tests). Replication threads pull data between nodes, reusing the same consumer/client.ConsumerClient the consumer CLI uses. Serf discovers peers. RPC handlers serve client requests. Producer and consumer client libraries find leaders and reconnect on failure.
Step 3: Add key dependencies
Add the core dependencies to your go.mod:
# Raft consensus
go get github.com/hashicorp/raft
go get github.com/hashicorp/raft-boltdb
# Serf cluster membership
go get github.com/hashicorp/serf
# Memory-mapped files (for the sparse index)
go get github.com/tysonmote/gommap
# Protobuf wire encoding
go get google.golang.org/protobuf
# CLI framework
go get github.com/spf13/cobra
go get github.com/spf13/viper
# Structured logging
go get go.uber.org/zap
| Dependency | Purpose |
|---|---|
hashicorp/raft |
Raft consensus: leader election, log replication, FSM interface |
hashicorp/raft-boltdb |
BoltDB-backed stable store for Raft's term and vote state |
hashicorp/serf |
Gossip-based cluster membership and failure detection |
tysonmote/gommap |
Memory-mapped file I/O for the sparse index |
google.golang.org/protobuf |
Protobuf marshal/unmarshal for every wire message payload |
spf13/cobra |
CLI command framework for server, producer, consumer, topic binaries |
spf13/viper |
Configuration management (flags, env vars) |
go.uber.org/zap |
Structured, leveled logging |
There's no gRPC dependency: protobuf here is used purely as the payload serialization format inside a custom length-prefixed TCP frame, not as a full RPC framework. api/protocol/proto/mlog.proto is the source of truth; api/protocol/pb/mlog.pb.go is generated from it and only ever touched by api/protocol/codec.go and the Raft snapshot/log-store code in broker/cluster. Everything else — handler signatures, client APIs — works with the plain Go structs in api/protocol/types.go.
Data flow through the layers
Here is how a produce request flows through the system:
Producer CLI (producer/cmd)
│
▼
Client (producer/client/client.go)
│ Resolves leader via client.ResolveTopicLeader
│ Delegates to ProducerClient.Produce
▼
ProducerClient (producer/client/producer.go)
│ Encodes ProduceRequest as protobuf, sends over TCP
▼
Transport (api/transport/transport.go)
│ Decodes frame, routes to handler
▼
RPC Server (broker/rpc/producer.go)
│ Validates request, checks TopicManager.IsLeader
▼
TopicManager (broker/topic/topic.go, broker/topic/producer.go)
│ Calls LogManager.Append()
│ If AckAll: waits for replicas to catch up
▼
LogManager (broker/log/log_manager.go)
│ Tracks LEO, delegates to Log
▼
Log (broker/log/log.go)
│ Appends to active segment
│ Rolls segment if full
▼
Segment (broker/segment/segment.go)
│ Encodes record: [Offset][Len][Value]
│ Writes to .log file
│ Updates sparse index
▼
Disk (.log and .idx files)
And for replication:
Follower node
│
Replication thread (broker/topic/replication.go)
│ runReplicateLoop wakes every second
│ For each (topic, leader) this node replicates:
▼
ConsumerClient → Leader node (consumer/client/consumer.go)
│ FetchBatch(topic, startOffset) with SetReplicaNodeID set
▼
Leader's RPC Server (broker/rpc/consumer.go)
│ Reads using ReadUncommitted (up to LEO)
│ Records follower's LEO for ISR computation
▼
Follower applies records to local log
│ TopicManager.ApplyRecordBatch()
▼
Leader computes ISR, advances HW
A separate runReconcileLoop (every 50ms) opens or closes each node's local logs to match whatever topic/replica assignment Raft has replicated — this is what makes a newly created topic (or a newly assigned replica) show up on a follower without a restart.
Build order
Build the system bottom-up in this order:
- Segment + Index (
broker/segment) — byte-level record encoding, sparse index with mmap, append and read. - Log + LogManager (
broker/log) — multiple segments, rotation, LEO and high watermark tracking. - Protocol (
api/protocol) — frame format, message types, protobuf codec, replication batch format. - Transport (
api/transport) — TCP server and client, connection management. - RPC (
broker/rpc) — handler registration, produce/consume/admin handlers. - Discovery (
broker/cluster/discovery) — Serf setup, join/leave event handling. - Cluster + Raft (
broker/cluster,broker/cluster/raft) — Raft setup, FSM, LogStore adapter, metadata events. - Topic (
broker/topic) — TopicManager, topic lifecycle, replication and reconciliation threads, ISR/HW logic. - Clients (
client,producer/client,consumer/client) — admin, producer, and consumer clients with leader discovery and reconnection. - CLI (
broker/cmd,producer/cmd,consumer/cmd) — server, producer, consumer, topic management commands.
Each step builds on the previous one. You can test each layer independently before moving to the next.
Next steps
The next page builds the low-level log API: record encoding, segments, the sparse index, and the Log/LogManager types that wrap it all together.