📁 File Processor

A concurrent file processing engine built in Go — powered by goroutines, channels, and a worker pool pattern.

---

Overview

File Processor watches a directory for incoming files and processes them concurrently using a configurable pool of workers. It is designed to be simple, observable, and gracefully shut down — making it a practical demonstration of real-world Go concurrency patterns.

This project was built as a deliberate learning exercise toward understanding Kubernetes controller internals — the watch → enqueue → reconcile pattern used in controller-runtime maps directly to the architecture here.

---

Architecture

┌─────────────────────────────────────────────────────────┐
│                        main.go                          │
│  Creates context, wires watcher + worker pool together  │
└────────────────┬───────────────────────┬────────────────┘
                 │                       │
                 ▼                       ▼
        ┌────────────────┐     ┌──────────────────────┐
        │   watcher.go   │     │      worker.go        │
        │                │     │                       │
        │ Polls ./incoming│    │  Worker Pool (N=5)    │
        │ every 1 second │     │                       │
        │ Detects new    │     │  ┌─────┐  ┌─────┐    │
        │ files          │     │  │ W-1 │  │ W-2 │    │
        └───────┬────────┘     │  └─────┘  └─────┘    │
                │              │  ┌─────┐  ┌─────┐    │
                │  jobs chan   │  │ W-3 │  │ W-4 │    │
                └─────────────►│  └─────┘  └─────┘    │
                               │       ┌─────┐         │
                               │       │ W-5 │         │
                               │       └─────┘         │
                               └──────────┬────────────┘
                                          │
                                          ▼
                               ┌──────────────────────┐
                               │    processor.go       │
                               │  Reads file, counts   │
                               │  lines & words        │
                               └──────────────────────┘

Key design decisions:

• A buffered jobs channel (cap 100) decouples the watcher from workers — the watcher never blocks waiting for a free worker
• Context propagation ensures every goroutine shuts down cleanly on SIGINT / SIGTERM
• The results channel is closed only after all workers finish, preventing data loss on shutdown

---

Project Structure

file-processor/
│
├── main.go               # Entry point — wires everything, handles OS signals
│
├── watcher/
│   └── watcher.go        # Polls directory every 1s, enqueues new files
│
├── worker/
│   └── worker.go         # Worker pool — goroutines consuming the jobs channel
│
├── processor/
│   └── processor.go      # Core logic — reads file, returns ProcessResult
│
├── go.mod
└── README.md

---

Getting Started

Prerequisites

• Go 1.21+
• Linux / macOS / WSL / Docker

Clone & Run

# Clone the repository
git clone https://github.com/incodi404/file-processor.git
cd file-processor

# Run the processor
go run .

Run with Docker

# Build image
docker build -t file-processor .

# Run container
docker run --rm -it -v $(pwd)/incoming:/home file-processor

---

Usage

Once running, drop any text files into the watched directory:

# Drop a single file
echo "hello world" > ./incoming/test.txt

# Drop multiple files at once to see all workers activate
for i in {1..5}; do
  echo "file content $i with some words here" > ./incoming/file$i.txt
done

Example Output

[MAIN] File processor has been started
[MAIN] Watching: ./incoming, Workers: 5
[MAIN] Press CTRL+C to stop
===================== OUTPUT =====================
[INFO] Worker has been started :: Worker ID: 3
[INFO] Worker has been started :: Worker ID: 1
[INFO] Worker has been started :: Worker ID: 5
[INFO] Worker has been started :: Worker ID: 2
[INFO] Worker has been started :: Worker ID: 4
[INFO] Watching directory: ./incoming
[INFO] New file detected: file1.txt
[INFO] File is processing :: Worker ID: 2
[INFO] New file detected: file2.txt
[INFO] File is processing :: Worker ID: 5
Status: OK | Filename: ./incoming/file1.txt | Message: lines=1 words=5 | Duration: 500ms

Note: Workers start in non-deterministic order — this is expected and correct behaviour. Go's scheduler decides when each goroutine runs. The pool's correctness does not depend on start order.

Graceful Shutdown

Press Ctrl+C at any time. The program will:

1. Receive SIGINT
2. Cancel the context — watcher and all workers stop cleanly
3. Drain and close the results channel
4. Exit

^C
[MAIN] Terminating signal received: interrupt
[MAIN] Shutting down...
[MAIN] Successfully terminated

---

Configuration

Edit the constants in main.go:

Variable	Default	Description
workDir	./home	Directory to watch for new files
workerCount	5	Number of concurrent worker goroutines

---

What I Learned

This project was intentionally chosen as a bridge toward Kubernetes development. The core concepts it covers:

Concept	Applied Here	Relevance to K8s
goroutines	Each worker runs as a goroutine	K8s controllers run reconcile loops concurrently
channels	Jobs and results flow through channels	K8s work queues use the same pattern
context.Context	Propagated to every goroutine for cancellation	Every K8s API call uses context
sync.WaitGroup	Ensures all workers finish before exit	Controller manager waits for all controllers
select statement	Workers listen for jobs or shutdown signal	Reconcilers watch for resource events or stop signals
SIGTERM handling	Graceful shutdown on container stop	K8s sends SIGTERM before killing pods
Separation of concerns	watcher / worker / processor split	K8s splits informers / reconcilers / clients

The watch → enqueue → reconcile pattern in this project directly mirrors how controller-runtime works internally. Building this first made reading K8s controller code significantly easier.

---

Author

incodi404