diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..1b59ba1 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,21 @@ +# Build Stage +FROM golang:1.25-alpine AS builder +WORKDIR /app + +# Cache dependencies +COPY go.mod go.sum ./ +RUN go mod download + +# Copy the rest of the code and build +COPY . . +RUN go build -o /fs-node + +# Run Stage +FROM alpine:latest +WORKDIR /app + +# Copy the compiled binary from the builder stage +COPY --from=builder /fs-node /app/fs-node + +# Start the server +CMD ["/app/fs-node"] \ No newline at end of file diff --git a/Readme.md b/Readme.md index a38f701..617af55 100644 --- a/Readme.md +++ b/Readme.md @@ -1,39 +1,48 @@ -# Distributed File Server (DFS) +# 🚀 Go Distributed File Server (DFS) -A work-in-progress **distributed file storage system** built from first principles in Go. -This project focuses on understanding **networking, systems design, and distributed systems internals** by building each layer manually instead of relying on frameworks. +A decentralized, peer-to-peer, content-addressable file storage network built natively in Go. ---- +This project aims to bypass the scalability bottlenecks of traditional Distributed File Systems (like the HDFS NameNode memory limit) by utilizing a decentralized Hash Ring for metadata-free peer routing, and a deeply nested Content-Addressable Storage (CAS) engine. -## 🎯 Project Goals +## 🧱 Architecture & Features -- Learn distributed systems by **building**, not just reading -- Implement a fault-tolerant, scalable file storage system -- Gain deep understanding of: - - TCP networking - - Message framing & protocols - - Data replication - - Node coordination +### 1. Peer-to-Peer Networking ---- +- **Custom TCP Transport Layer:** Server-side `Listen`, `Accept`, and active outbound `Dial` capabilities. +- **Message Framing:** Custom `LengthPrefixDecoder` to handle raw TCP byte streams reliably. +- **Wire Protocol:** Structured binary message encoding (`MessagePayload`) using Go's `encoding/gob`, supporting `PUT`, `GET`, and `DELETE` commands. + +### 2. Decentralized Routing & Replication + +- **Consistent Hash Ring:** Uses SHA-1 and binary tree search (`sort.Search`) to map files to nodes mathematically, completely eliminating the need for a central metadata database. +- **Dynamic Peer Discovery:** Nodes dynamically join the ring upon TCP handshake. +- **Automated Replication:** Files uploaded to any node are automatically broadcast to all connected peers, with local CAS lookups to prevent infinite replication loops. + +### 3. Local Storage Engine (CAS) -## 🧱 Current Status +- **Content Addressable Storage:** Files are stored based on the SHA-1 hash of their key, ensuring deduplication. +- **Optimized Disk I/O:** Hashes are split into deeply nested directory structures (e.g., `a1/b2/c3/...`) to prevent OS-level directory limitations during massive file ingestion. +- **Deep Clean:** The `DELETE` command not only removes the file but recursively prunes empty parent directories. -### ✅ Implemented -- Custom **TCP transport layer** -- Server-side `Listen` and `Accept` loop -- Incoming connection handling -- Manual testing using **Telnet** -- Clean separation of transport logic +### 4. Cluster Orchestration -### 🔜 In Progress / Planned -- Message framing (length-prefixed protocol) -- Wire protocol (PUT / GET / DELETE) -- Local file storage engine -- Peer discovery & membership -- Data replication & fault tolerance +- **Containerized Testing:** Fully containerized using a multi-stage `Dockerfile` (Alpine Linux) and `docker-compose.yml`. +- Spins up an isolated virtual network with one seed node and multiple dynamic peers to simulate real-world cluster topologies. --- -## 🏗️ Architecture (Early Stage) +## 🛠️ How to Run the Cluster + +### Prerequisites + +- [Docker Desktop](https://www.docker.com/products/docker-desktop/) installed and running. +- Go 1.25+ (if running the test client locally). + +### Booting the Network + +You can spin up a 3-node cluster with a single command. Open your terminal in the project root and run: +`bash +docker-compose up --build +` +_Node 1 will act as the seed node. Nodes 2 and 3 will automatically boot, read their Environment Variables, and dial Node 1 to join the Hash Ring._ diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..9c9ed0f --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,48 @@ +version: '3.8' + +services: + node1: + build: . + container_name: fs-node1 + environment: + - LISTEN_ADDR=:3000 + - STORAGE_ROOT=/data + - BOOTSTRAP_NODES= + ports: + - "3000:3000" # Expose to host machine on 3000 + volumes: + - node1_data:/data + + node2: + build: . + container_name: fs-node2 + environment: + - LISTEN_ADDR=:3000 + - STORAGE_ROOT=/data + - BOOTSTRAP_NODES=node1:3000 # Dial Node 1 + ports: + - "4000:3000" # Expose to host machine on 4000 + volumes: + - node2_data:/data + depends_on: + - node1 + + node3: + build: . + container_name: fs-node3 + environment: + - LISTEN_ADDR=:3000 + - STORAGE_ROOT=/data + - BOOTSTRAP_NODES=node1:3000 # Dial Node 1 + ports: + - "5000:3000" # Expose to host machine on 5000 + volumes: + - node3_data:/data + depends_on: + - node1 + +volumes: + # Persistent virtual hard drives for each node + node1_data: + node2_data: + node3_data: \ No newline at end of file diff --git a/main.go b/main.go index f7ea618..c275e3e 100644 --- a/main.go +++ b/main.go @@ -2,33 +2,58 @@ package main import ( "log" + "os" + "strings" "github.com/girish/storage/p2p" "github.com/girish/storage/store" ) -func main() { +func makeServer(listenAddr string, storageroot string, bootstrapNodes []string) *FileServer { tcpOpts := p2p.TCPTransportOps{ - ListenAdder: ":3000", + ListenAdder: listenAddr, Handshakefunc: p2p.NOPHandshakeFunc, Decode: p2p.LengthPrefixDecoder{}, } tr := p2p.NewTCPTransport(tcpOpts) storeOpts := store.StoreOpts{ - Root: "my_network_data", + Root: storageroot, PathTransformFunc: store.CASPathTransfromFunc, } localStore := store.NewStore(storeOpts) serverOpts := FileServerOpts{ - StorageRoot: "my_network_data", // This is where files will be saved later - Transport: tr, - Store: localStore, + ListenAddr: listenAddr, + StorageRoot: storageroot, + Transport: tr, + Store: localStore, + BootStrapNoeds: bootstrapNodes, + } + return NewFileServer(serverOpts) +} + +func main() { + // 1. Read the network port from Docker (default to :3000) + listenAddr := os.Getenv("LISTEN_ADDR") + if listenAddr == "" { + listenAddr = ":3000" + } + + // 2. Read the storage folder from Docker + storageRoot := os.Getenv("STORAGE_ROOT") + if storageRoot == "" { + storageRoot = "network_data" } - server := NewFileServer(serverOpts) - if err := server.Start(); err != nil { - log.Fatal(err) + // 3. Read the comma-separated list of peers to dial on startup + bootstrapNodesStr := os.Getenv("BOOTSTRAP_NODES") + var bootstrapNodes []string + if bootstrapNodesStr != "" { + bootstrapNodes = strings.Split(bootstrapNodesStr, ",") } + + // 4. Spin up the single node + server := makeServer(listenAddr, storageRoot, bootstrapNodes) + log.Fatal(server.Start()) } diff --git a/p2p/TCP_transport.go b/p2p/TCP_transport.go index 4de1435..4bbc2aa 100644 --- a/p2p/TCP_transport.go +++ b/p2p/TCP_transport.go @@ -61,6 +61,21 @@ func (t *TCPTransport) ListenAndAccept() error { return nil } +// Dial implement the Transport interface. It connects to a remote node +// and treats that connection exactly like an incomming one +func (t *TCPTransport) Dial(addr string) error { + conn, err := net.Dial("tcp", addr) + if err != nil { + return err + } + + fmt.Printf("Successfully dialed and connected to peer: %s\n", addr) + + go t.handleConn(conn) + + return nil +} + func (t *TCPTransport) startAcceptLoop() { for { conn, err := t.listener.Accept() @@ -105,8 +120,9 @@ func (t *TCPTransport) handleConn(conn net.Conn) { } t.rpcCh <- RPC{ - From: conn.RemoteAddr(), + From: conn.RemoteAddr(), Payload: msg.Payload, + Peer: peer, } } @@ -131,3 +147,10 @@ func (p *TCPPeer) Send(b []byte) error { func (t *TCPTransport) Consume() <-chan RPC { return t.rpcCh } + +func (t *TCPTransport) Close() error { + if t.listener != nil { + return t.listener.Close() + } + return nil +} diff --git a/p2p/hashring.go b/p2p/hashring.go new file mode 100644 index 0000000..1c0a3a8 --- /dev/null +++ b/p2p/hashring.go @@ -0,0 +1,80 @@ +package p2p + +import ( + "crypto/sha1" + "encoding/binary" + "sort" + "sync" +) + +type HashRing struct { + mu sync.RWMutex + nodes []uint32 + nodeMap map[uint32]string +} + +func NewHashring() *HashRing { + return &HashRing{ + nodes: []uint32{}, + nodeMap: make(map[uint32]string), + } +} + +func (h *HashRing) hash(key string) uint32 { + hasher := sha1.New() + hasher.Write([]byte(key)) + sum := hasher.Sum(nil) + + return binary.BigEndian.Uint32(sum[:4]) +} + +func (h *HashRing) AddNode(addr string) { + h.mu.Lock() + defer h.mu.Unlock() + + hash := h.hash(addr) + if _, existed := h.nodeMap[hash]; !existed { + h.nodes = append(h.nodes, hash) + + sort.Slice(h.nodes, func(i, j int) bool { + return h.nodes[i] < h.nodes[j] + }) + h.nodeMap[hash] = addr + } +} + +func (h *HashRing) RemoveNode(addr string) { + h.mu.Lock() + defer h.mu.Unlock() + hash := h.hash(addr) + if _, exists := h.nodeMap[hash]; exists { + delete(h.nodeMap, hash) + + var newNode []uint32 + for _, n := range h.nodes { + if n != hash { + newNode = append(newNode, n) + } + } + h.nodes = newNode + } +} + +func (h *HashRing) GetNode(key string) string { + h.mu.RLock() + defer h.mu.RUnlock() + + if len(h.nodes) == 0 { + return "" + } + hash := h.hash(key) + + idx := sort.Search(len(h.nodes), func(i int) bool { + return h.nodes[i] >= hash + }) + + if idx == len(h.nodes) { + idx = 0 + } + return h.nodeMap[h.nodes[idx]] +} diff --git a/p2p/hashring_test.go b/p2p/hashring_test.go new file mode 100644 index 0000000..0c2e9bb --- /dev/null +++ b/p2p/hashring_test.go @@ -0,0 +1,56 @@ +package p2p + +import ( + "testing" +) + +func TestHashRing_AddAndGet(t *testing.T) { + ring := NewHashring() + + nodes := []string{"192.168.1.1:3000", "192.168.1.2:3000", "192.168.1.3:3000"} + for _, n := range nodes { + ring.AddNode(n) + } + + if len(ring.nodes) != 3 { + t.Fatalf("expected 3 nodes, got %d", len(ring.nodes)) + } + + // Verify that a specific key consistently maps to the exact same node + key := "my_test_file.txt" + expectedNode := ring.GetNode(key) + + if expectedNode == "" { + t.Fatal("expected a valid node address, got empty string") + } + + // Do it 100 times to ensure the mutexes and mapping remain stable + for i := 0; i < 100; i++ { + node := ring.GetNode(key) + if node != expectedNode { + t.Fatalf("consistent hash failed: expected %s, got %s", expectedNode, node) + } + } +} + +func TestHashRing_RemoveNode(t *testing.T) { + ring := NewHashring() + ring.AddNode("nodeA") + ring.AddNode("nodeB") + + key := "test_file" + firstOwner := ring.GetNode(key) + + // Remove the node that owns the key + ring.RemoveNode(firstOwner) + + if len(ring.nodes) != 1 { + t.Fatalf("expected 1 node, got %d", len(ring.nodes)) + } + + // The key should now smoothly map to the only remaining node + secondOwner := ring.GetNode(key) + if secondOwner == firstOwner { + t.Fatalf("expected owner to change after removal") + } +} \ No newline at end of file diff --git a/p2p/message.go b/p2p/message.go index 3525356..b407841 100644 --- a/p2p/message.go +++ b/p2p/message.go @@ -5,6 +5,7 @@ import "net" type RPC struct { From net.Addr Payload []byte + Peer Peer } type Message struct { diff --git a/p2p/transport.go b/p2p/transport.go index 09c4df3..ac469db 100644 --- a/p2p/transport.go +++ b/p2p/transport.go @@ -5,8 +5,10 @@ type Peer interface { Send([]byte) error } -//Transport is anything that handles the communication +//Transport is anything that handles the communication between nodes in the network type Transport interface { ListenAndAccept() error Consume() <-chan RPC + Close() error + Dial(string) error } \ No newline at end of file diff --git a/server.go b/server.go index bc96c03..233b351 100644 --- a/server.go +++ b/server.go @@ -4,33 +4,46 @@ import ( "bytes" "encoding/gob" "fmt" + "io" + "sync" "github.com/girish/storage/p2p" "github.com/girish/storage/store" ) type FileServerOpts struct { - StorageRoot string //StorageRoot is the folder on hard drive where this node will save files - Transport p2p.Transport - Store *store.Store + ListenAddr string + StorageRoot string //StorageRoot is the folder on hard drive where this node will save files + Transport p2p.Transport + Store *store.Store + BootStrapNoeds []string } // DataMessage is wire protocol payload! -type DataMessage struct { - Key string - Data []byte +type MessagePayload struct { + Command string + Key string + Data []byte } type FileServer struct { FileServerOpts - quitCh chan struct{} + peerLock sync.Mutex //Thread sefty for concurrent connectons + peers map[string]p2p.Peer //Maps addresses to active network connection + Ring *p2p.HashRing //The decentralized routing table + quitCh chan struct{} } func NewFileServer(opts FileServerOpts) *FileServer { - return &FileServer{ + fs := &FileServer{ FileServerOpts: opts, + peers: make(map[string]p2p.Peer), + Ring: p2p.NewHashring(), quitCh: make(chan struct{}), } + fs.Ring.AddNode(opts.ListenAddr) + + return fs } // Start boots up the transport layer and begins processing messages @@ -41,10 +54,29 @@ func (s *FileServer) Start() error { return err } + // Bootstrap the network! Connect to all known peers. + s.bootstrapNetwork() + s.loop() return nil } +// BootStrapNetwork iterates through all provideed nodes and attempts to connect +func (s *FileServer) bootstrapNetwork() { + for _, addr := range s.BootStrapNoeds { + if len(addr) == 0 { + continue + } + fmt.Printf("Attempting to connect with bootstrap node at %s...\n", addr) + // running this in goroutine so one slow connection dosen't block the others + go func(peerAddr string) { + if err := s.Transport.Dial(peerAddr); err != nil { + fmt.Printf("Failed to dial bootstrap node %s: %s\n", peerAddr, err) + } + }(addr) + } +} + func (s *FileServer) loop() { for { select { @@ -58,19 +90,108 @@ func (s *FileServer) loop() { } func (s *FileServer) handleMessage(rpc p2p.RPC) { - //1 Decode the binary network payload back into Structured DataMessage - var msg DataMessage + s.registerPeer(rpc.Peer, rpc.From.String()) + + var msg MessagePayload if err := gob.NewDecoder(bytes.NewReader(rpc.Payload)).Decode(&msg); err != nil { fmt.Printf("Failed to decode network payload: %s\n", err) return } - fmt.Printf("FileServer received command to store file: '%s' (%d bytes)\n", msg.Key, len(msg.Data)) - //2 Write the file to disk using CAS store engine + switch msg.Command { + case "PUT": + s.handlePutCommand(rpc.From.String(), msg) + case "GET": + s.handleGetCommand(rpc) + case "DELETE": + s.handleDeleteCommand(rpc.From.String(), msg) + default: + fmt.Printf("Unknown command recived: %s\n", msg.Command) + } +} + +func (s *FileServer) handlePutCommand(from string, msg MessagePayload) { + fmt.Printf("Receiving 'PUT' command for '%s' from %s\n", msg.Key, from) + err := s.Store.WriteStream(msg.Key, bytes.NewReader(msg.Data)) if err != nil { - fmt.Printf("Error Storing file to disk: %s\n", err) + fmt.Printf("Error storing file: %s\n", err) + return + } + fmt.Println("File successfully saved to disk!") + + err = s.broadcast(msg) + if err != nil { + fmt.Printf("Error broadcasting file: %s\n", err) + } +} + +func (s *FileServer) handleGetCommand(rpc p2p.RPC) { + var msg MessagePayload + gob.NewDecoder(bytes.NewReader(rpc.Payload)).Decode(&msg) + + fmt.Printf("Receiving 'GET' command for '%s' from %s\n", msg.Key, rpc.From) + + //1 Open the file stream + r, err := s.Store.ReadStream(msg.Key) + if err != nil { + fmt.Printf("Error reading file from disk: %s\n", err) + return + } + defer r.Close() + + //2 Read file into memory + fileBytes, err := io.ReadAll(r) + if err != nil { + fmt.Printf("Error reading file bytes: %s\n", err) + return + } + + //3 Send it back to the client + if err := rpc.Peer.Send(fileBytes); err != nil { + fmt.Printf("Error sending file to peer: %s\n", err) + } + fmt.Printf("Successfully sent file '%s' back to client!\n", msg.Key) +} + +// handleDeleteCommand handles request to remove files from the network +func (s *FileServer) handleDeleteCommand(from string, msg MessagePayload) { + fmt.Printf("Receiving 'DELETE' command for '%s' from %s\n", msg.Key, from) + + err := s.Store.Delete(msg.Key) + if err != nil { + fmt.Printf("Error deleting file from disk: %s\n", err) return } - fmt.Println("File successfully saved from the network!") + fmt.Printf("File '%s' and its directories successfully deleted!\n", msg.Key) +} + +func (s *FileServer) registerPeer(peer p2p.Peer, addr string) { + s.peerLock.Lock() + defer s.peerLock.Unlock() + + if _, exists := s.peers[addr]; !exists { + s.peers[addr] = peer + s.Ring.AddNode(addr) + fmt.Printf("New peer Added %s\n", addr) + } +} + +func (s *FileServer) broadcast(msg MessagePayload) error { + s.peerLock.Lock() + defer s.peerLock.Unlock() + + payloadBuff := new(bytes.Buffer) + if err := gob.NewEncoder(payloadBuff).Encode(msg); err != nil { + return err + } + payloadBytes := payloadBuff.Bytes() + + for addr, peer := range s.peers { + fmt.Printf("📡 Broadcasting file '%s' to peer: %s\n", msg.Key, addr) + if err := peer.Send(payloadBytes); err != nil { + fmt.Printf("Failed to broadcast to peer %s: %s\n", addr, err) + } + } + return nil } diff --git a/store/my_test_network/cbc5c/fb24f/2efb5/a3df0/8afc3/9ac80/18ea3/b8290/cbc5cfb24f2efb5a3df08afc39ac8018ea3b8290 b/store/my_test_network/cbc5c/fb24f/2efb5/a3df0/8afc3/9ac80/18ea3/b8290/cbc5cfb24f2efb5a3df08afc39ac8018ea3b8290 deleted file mode 100644 index 04967e4..0000000 --- a/store/my_test_network/cbc5c/fb24f/2efb5/a3df0/8afc3/9ac80/18ea3/b8290/cbc5cfb24f2efb5a3df08afc39ac8018ea3b8290 +++ /dev/null @@ -1 +0,0 @@ -These are some fake image bytes. Pretend this is a cool PNG! \ No newline at end of file diff --git a/store/store.go b/store/store.go index c114b3b..da7d835 100644 --- a/store/store.go +++ b/store/store.go @@ -3,8 +3,10 @@ package store import ( "crypto/sha1" "encoding/hex" + "errors" "fmt" "io" + "io/fs" "log" "os" "strings" @@ -122,3 +124,40 @@ func (s *Store) WriteStream(key string, r io.Reader) error { log.Printf("Written (%d) bytes to the disk: %s", n, fullPathWithRoot) return nil } + +// ReadStream locates the file by its key and returns a stream to read it +// Note: it returns an io.ReadCloser so the caller is responsible for calling close() when finished! +func (s *Store) ReadStream(key string) (io.ReadCloser, error) { + //1 Calculate the exact path using CAS function + pathKey := s.PathTransformFunc(key) + + //2 Combine it with root directory + fullPathWithRoot := fmt.Sprintf("%s/%s", s.Root, pathKey.FullPath()) + + // Open the file and return Both the stream and the error + return os.Open(fullPathWithRoot) +} + +// Delete completely removes the file and its nested CAS directories from the disk +func (s *Store) Delete(key string) error { + pathKey := s.PathTransformFunc(key) + + //Delete from the root of the hash to clean up all empty folders + firstPathWithRoot := fmt.Sprintf("%s/%s", s.Root, pathKey.FirstPathName()) + + err := os.RemoveAll(firstPathWithRoot) + if err != nil { + return err + } + log.Printf("Deleted file and cleaned up directories: %s", firstPathWithRoot) + return nil +} + +// Has checks if a file already exists in the local CAS storage +func (s *Store) Has(key string) bool { + pathKey := s.PathTransformFunc(key) + fullPathWithRoot := fmt.Sprintf("%s/%s", s.Root, pathKey.FullPath()) + + _, err := os.Stat(fullPathWithRoot) + return !errors.Is(err, fs.ErrNotExist) +} diff --git a/store/store_test.go b/store/store_test.go index f4a9187..a08ec48 100644 --- a/store/store_test.go +++ b/store/store_test.go @@ -2,29 +2,75 @@ package store import ( "bytes" + "io" "testing" ) -// ... (keep your existing TestPathTransformFunc) ... +func TestPathTransformFunc(t *testing.T) { + key := "my_best_picture.png" + pathKey := CASPathTransfromFunc(key) + + expectedFilename := "e09bba016f88f2b92e824d9a4142a9c81c6440ae" + expectedPathName := "e09bb/a016f/88f2b/92e82/4d9a4/142a9/c81c6/440ae" + + if pathKey.PathName != expectedPathName { + t.Errorf("have %s want %s", pathKey.PathName, expectedPathName) + } + + if pathKey.FileName != expectedFilename { + t.Errorf("have %s want %s", pathKey.FileName, expectedFilename) + } +} func TestStore(t *testing.T) { - // 1. Initialize our Store opts := StoreOpts{ Root: "my_test_network", PathTransformFunc: CASPathTransfromFunc, } s := NewStore(opts) - // 2. Pretend this is a file a user is uploading key := "my_special_picture.png" data := []byte("These are some fake image bytes. Pretend this is a cool PNG!") - - // Create an io.Reader out of our bytes + sourceStream := bytes.NewReader(data) - // 3. Write it to disk! err := s.WriteStream(key, sourceStream) if err != nil { t.Errorf("expected no error, got %s", err) } -} \ No newline at end of file +} + +func TestStoreRead(t *testing.T) { + opts := StoreOpts{ + Root: "my_test_network", + PathTransformFunc: CASPathTransfromFunc, + } + s := NewStore(opts) + + key := "test_read_file.txt" + data := []byte("We need to make sure we can read this back from the disk!") + + // 1. Write the file + err := s.WriteStream(key, bytes.NewReader(data)) + if err != nil { + t.Fatalf("failed to write file: %s", err) + } + + // 2. Read the file back + r, err := s.ReadStream(key) + if err != nil { + t.Fatalf("failed to read file: %s", err) + } + defer r.Close() + + // 3. Read the bytes from the stream + b, err := io.ReadAll(r) + if err != nil { + t.Fatalf("failed to read bytes from stream: %s", err) + } + + // 4. Verify it matches what we wrote! + if string(b) != string(data) { + t.Errorf("want %s, got %s", string(data), string(b)) + } +}