diff --git a/README.md b/README.md index 3c32de7..de84976 100644 --- a/README.md +++ b/README.md @@ -1,110 +1,164 @@ +Sir, like, okay. So the problem is that the whole README you've got is wrapped in a markdown code block. If you paste that directly into GitHub, it'll just show the raw text instead of, you know, actually formatting it. You just need the text *inside* that block. + +Here's the code. Just copy and paste this directly into your `readme.md` file on GitHub. + +----- + # LogFlow: A Resilient Log Ingestion Pipeline -LogFlow is a high-performance, distributed log ingestion pipeline designed to reliably process and store log data from various sources. It is built on a microservice architecture, leveraging modern technologies like Go, Docker, Kafka, and Elasticsearch to create a scalable and fault-tolerant system. +[](https://github.com/MinuteHanD/log-pipeline/actions/workflows/ci.yml) + +LogFlow is a multi-service log ingestion pipeline built in Go. It demonstrates a decoupled, fault-tolerant architecture for processing and storing high-volume log data. The system uses Kafka as a durable message bus and Elasticsearch for indexed storage and searchability. + +This project is fully containerized and includes a comprehensive testing suite. -This project serves as a practical example of building a real-world data engineering pipeline, complete with data validation, message queuing, and structured storage. +----- -## Architecture Overview +## Architecture + +The pipeline follows an asynchronous, three-stage process. Each service is a distinct microservice that communicates via Kafka topics, ensuring resilience and scalability. + +``` + ┌──────────┐ ┌──────────┐ ┌───────────┐ ┌───────────────┐ +[HTTP Client]───> │ Ingestor │ ───> │ Kafka │ ───> │ Parser │ ───> │ Kafka │ + └──────────┘ │ raw_logs │ └───────────┘ │ parsed_logs │ + └──────────┘ │ └───────────────┘ + ▼ │ + ┌──────────────┐ ▼ + │ Kafka │ ┌────────────────┐ + │ raw_logs_dlq │ │ Storage Writer │ + └──────────────┘ └────────────────┘ + │ + ▼ + ┌─────────────────┐ + │ Elasticsearch │ + └─────────────────┘ +``` -The pipeline consists of three core microservices that communicate asynchronously via Kafka topics. This decoupled design ensures that each service can be scaled and maintained independently, and it provides a buffer to handle bursts of traffic without losing data. +### Core Components -1. **Ingestor**: A lightweight HTTP server that acts as the entry point for all logs. It validates incoming data against a set of rules, and upon success, publishes the raw log to the `raw_logs` Kafka topic. -2. **Parser**: A consumer service that reads from the `raw_logs` topic. It parses the raw log, normalizes its fields (like timestamps and log levels), enriches it with metadata, and then publishes the structured result to the `parsed_logs` topic. Invalid logs that cannot be parsed are sent to a Dead-Letter Queue (DLQ) for later inspection. -3. **Storage Writer**: The final service in the pipeline. It consumes structured logs from the `parsed_logs` topic and writes them to a time-series index in Elasticsearch for long-term storage, analysis, and visualization. + * **Ingestor**: A Go service using the Gin framework. It exposes an HTTP endpoint (`/log`) to receive log entries. It performs initial validation (size, format, required fields) and publishes valid raw logs to the `raw_logs` Kafka topic. + * **Parser**: A Kafka consumer group that reads from `raw_logs`. It normalizes data (e.g., standardizing log levels, parsing timestamps), enriches logs with metadata, and generates a unique ID. Processed logs are published to `parsed_logs`. Logs that fail parsing are routed to a Dead-Letter Queue (`raw_logs_dlq`) for later analysis. + * **Storage Writer**: Consumes structured logs from the `parsed_logs` topic. It creates daily time-based indices in Elasticsearch and writes the final log document. If Elasticsearch is unavailable or rejects a document, the message is routed to its own DLQ (`parsed_logs_dlq`). -## Features +----- -- **HTTP Log Ingestion**: Simple and fast log submission over HTTP. -- **Data Validation**: Ensures data quality at the edge before it enters the system. -- **Asynchronous Processing**: Uses Kafka as a message broker to decouple services and buffer data. -- **Log Parsing & Normalization**: Standardizes log formats for consistent storage. -- **Dead-Letter Queue (DLQ)**: Isolates and saves malformed messages for later analysis without halting the pipeline. -- **Structured Storage**: Stores logs in Elasticsearch, enabling powerful search and analytics. -- **Containerized**: Fully containerized with Docker and Docker Compose for easy setup and deployment. -- **Comprehensive Testing**: Includes both unit and end-to-end integration tests to ensure reliability. +## Key Features + + * **Fault Tolerance**: Utilizes Dead-Letter Queues (DLQs) at both the parsing and storage stages to prevent data loss from malformed messages or downstream service outages. + * **Structured, Centralized Logging**: All services use Go's native `slog` library to output JSON-formatted logs, enabling easier debugging and analysis of the pipeline itself. + * **Asynchronous & Decoupled**: Kafka acts as a buffer, allowing the ingestor to handle traffic spikes without overwhelming the processing and storage layers. Services can be scaled, updated, or restarted independently. + * **Comprehensive Testing**: The project is validated by: + * **Unit Tests**: Focused tests for business logic within each service (e.g., validation, parsing logic). + * **End-to-End Integration Test**: A Go-based test that uses `docker-compose` to spin up the entire stack, sends a log via HTTP, and verifies its existence in Elasticsearch. + * **Containerized**: Fully defined in `docker-compose.yml` for reproducible one-command setup and deployment. + +----- ## Technology Stack -- **Backend**: Go (Golang) 1.24.3 -- **API Framework**: Gin -- **Message Queue**: Apache Kafka -- **Database**: Elasticsearch -- **Containerization**: Docker & Docker Compose -- **Go Libraries**: - - `sarama`: Kafka client for Go - - `go-elasticsearch`: Official Elasticsearch client - - `gin-gonic`: HTTP web framework + * **Language**: Go 1.24.3 + * **Services**: Gin (HTTP), Sarama (Kafka Client), go-elasticsearch (Elasticsearch Client) + * **Infrastructure**: Docker, Docker Compose, Kafka, Elasticsearch + +----- ## Prerequisites -- Docker and Docker Compose -- Go 1.24.3 or later (for local development and testing) + * Go 1.24.3+ + * Docker and Docker Compose + +----- ## Getting Started 1. **Clone the repository:** + ```bash - git clone https://github.com/MinuteHanD/LogFlow.git - cd LogFlow + git clone https://github.com/MinuteHanD/log-pipeline.git + cd log-pipeline ``` -2. **Start all services using Docker Compose:** - This command will build the Docker images for the Go services and start all containers in the background. +2. **Start all services:** + The `--build` flag ensures the Go services are compiled into fresh images. + ```bash docker compose up -d --build ``` -3. **Send a test log:** - Use the provided script to send a variety of test logs to the pipeline. +3. **Send test logs:** + A script is provided to send a variety of valid and invalid logs to test the pipeline and DLQ functionality. + ```bash ./send_all_logs.sh ``` - Or send a custom log using `curl`: + +4. **Verify in Elasticsearch:** + Allow a few seconds for processing, then query Elasticsearch. This command fetches the 10 most recent logs. + ```bash - curl -X POST http://localhost:8081/log \ - -H "Content-Type: application/json" \ - -d '{ \ - "level": "info", \ - "service": "Solid Snake", \ - "message": "Dragon of Dojima", \ - "timestamp": "2025-07-01T12:00:00Z" \ - }' + curl -X GET "http://localhost:9200/logs-*/_search?pretty" -H 'Content-Type: application/json' -d' + { + "query": { "match_all": {} }, + "size": 10, + "sort": [ { "timestamp": { "order": "desc" } } ] + }' ``` -4. **View logs in Elasticsearch:** - Allow a few moments for the pipeline to process the log, then query Elasticsearch directly: + You can also use Kibana at `http://localhost:5601`. + +----- + +## Development and Testing + +For local development, you can run the infrastructure in Docker and the Go services directly on your host. + +1. **Start infrastructure services:** + ```bash - curl -X GET "http://localhost:9200/logs-*/_search?pretty" -H 'Content-Type: application/json' -d'\ - {\ - "query": { "match_all": {} },\ - "size": 10,\ - "sort": [ { "timestamp": "desc" } ]\ - }' + docker compose up -d kafka elasticsearch kibana ``` - You can also view the logs and create dashboards in **Kibana** by navigating to `http://localhost:5601` in your browser. -## Testing +2. **Run each Go service in a separate terminal:** -The project includes a suite of tests to ensure code quality and system reliability. + ```bash + # Terminal 1: Ingestor + export KAFKA_BROKERS=localhost:9092 + go run ./ingestor + + # Terminal 2: Parser + export KAFKA_BROKERS=localhost:9092 + go run ./parser + + # Terminal 3: Storage Writer + export KAFKA_BROKERS=localhost:9092 + export ELASTICSEARCH_URL=http://localhost:9200 + go run ./storage-writer + ``` -### Unit Tests +### Running Tests -Unit tests check individual functions and components in isolation. They are fast and do not require any external dependencies like Docker. +The project separates fast unit tests from slower, dependency-heavy integration tests using Go build tags. -To run all unit tests for all services: -```bash -go test -v ./... -``` + * **Run Unit Tests:** + This executes all `*_test.go` files that do *not* have the `integration` build tag. -### Integration Test + ```bash + go test -v ./... + ``` -The integration test verifies the entire pipeline, from HTTP ingestion to storage in Elasticsearch. It uses Docker Compose to manage the required services. + * **Run Integration Test:** + This command specifically runs tests tagged with `integration`. It will manage Docker containers automatically. Ensure Docker is running. -**Note:** This test is slower and will start and stop Docker containers on your machine. + ```bash + go test -v -tags=integration + ``` -To run the integration test: -```bash -go test -v -tags=integration -``` +----- +## Future Work & Improvements + * **Configuration Management**: Centralize configuration (e.g., Kafka topics, ports) using a library like Viper to read from YAML files with environment variable overrides. + * **Metrics & Observability**: Expose Prometheus metrics from each service (e.g., logs processed, error rates, processing latency) for monitoring and alerting. + * **Correlation IDs**: Implement a correlation ID at the `ingestor` and pass it through Kafka headers to trace a single request across all services in the structured logs. + * **DLQ Re-processing**: Build a utility or service to consume from the DLQ topics, attempt to re-process messages, and archive unrecoverable ones. \ No newline at end of file diff --git a/ingestor/main.go b/ingestor/main.go index eb2748b..5da5598 100644 --- a/ingestor/main.go +++ b/ingestor/main.go @@ -3,13 +3,14 @@ package main import ( "encoding/json" "fmt" - "log" "net/http" "os" "regexp" "strings" "time" + "log/slog" + "github.com/IBM/sarama" "github.com/gin-gonic/gin" ) @@ -179,6 +180,9 @@ func (v *LogValidator) ValidateComplete(data []byte) ValidationResult { } func main() { + // Initialize a structured logger that outputs JSON to standard output. + logger := slog.New(slog.NewJSONHandler(os.Stdout, nil)) + config := sarama.NewConfig() config.Producer.Return.Successes = true config.Producer.RequiredAcks = sarama.WaitForAll @@ -189,18 +193,22 @@ func main() { producer, err := sarama.NewSyncProducer(brokers, config) if err != nil { - log.Fatalf("Failed to start Sarama producer: %v", err) + logger.Error("Failed to start Sarama producer", "error", err) + os.Exit(1) } defer producer.Close() validator := NewLogValidator() router := gin.Default() - router.Use(gin.Logger()) + // Replace Gin's default logger with our structured logger or disable it if not needed. + // For simplicity, we'll keep Gin's logger for request logging, but our app logs will be structured. + // router.Use(gin.Logger()) router.POST("/log", func(c *gin.Context) { body, err := c.GetRawData() if err != nil { + logger.Error("Failed to read request body", "error", err) c.JSON(http.StatusBadRequest, gin.H{"error": "failed to read request body", "details": err.Error(), }) @@ -210,6 +218,7 @@ func main() { validationResult := validator.ValidateComplete(body) if !validationResult.IsValid { + logger.Info("Log validation failed", "validation_errors", validationResult.Errors) c.JSON(http.StatusBadRequest, gin.H{ "error": "validation failed", "validation_errors": validationResult.Errors, @@ -224,7 +233,7 @@ func main() { partition, offset, err := producer.SendMessage(msg) if err != nil { - log.Printf("Failed to send message to kafka: %v", err) + logger.Error("Failed to send message to Kafka", "error", err, "topic", KafkaTopic) c.JSON(http.StatusInternalServerError, gin.H{ "error": "failed to send message to kafka", "details": err.Error(), @@ -232,6 +241,7 @@ func main() { return } + logger.Info("Log received and validated successfully", "topic", KafkaTopic, "partition", partition, "offset", offset) c.JSON(http.StatusOK, gin.H{ "message": "log recieved and validated successfully", "partition": partition, @@ -241,6 +251,7 @@ func main() { }) router.GET("/validation-rules", func(c *gin.Context) { + logger.Info("Validation rules requested") c.JSON(http.StatusOK, gin.H{ "max_message_size": MaxMessageSize, "required_fields": []string{"timestamp", "level", "message"}, @@ -249,8 +260,9 @@ func main() { }) }) - log.Println("Ingestor service starting on port 8081") + logger.Info("Ingestor service starting", "port", 8081) if err := router.Run(":8081"); err != nil { - log.Fatalf("Failed to run Gin server: %v", err) + logger.Error("Failed to run Gin server", "error", err) + os.Exit(1) } } diff --git a/main_integration_test.go b/main_integration_test.go index 6bf457c..758f511 100644 --- a/main_integration_test.go +++ b/main_integration_test.go @@ -31,7 +31,48 @@ func TestPipeline_Integration(t *testing.T) { }) t.Log("Waiting for services to become healthy...") - time.Sleep(30 * time.Second) + + ingestorReady := false + for i := 0; i < 30; i++ { + resp, err := http.Get("http://localhost:8081/validation-rules") + if err == nil && resp.StatusCode == http.StatusOK { + ingestorReady = true + resp.Body.Close() + t.Log("Ingestor service is ready.") + break + } + if resp != nil { + resp.Body.Close() + } + time.Sleep(2 * time.Second) + } + if !ingestorReady { + t.Fatal("Ingestor service did not become ready in time.") + } + + esReady := false + for i := 0; i < 30; i++ { + resp, err := http.Get("http://localhost:9200/_cluster/health?wait_for_status=yellow&timeout=1s") + if err == nil && resp.StatusCode == http.StatusOK { + var healthResp map[string]interface{} + if err := json.NewDecoder(resp.Body).Decode(&healthResp); err == nil { + status, ok := healthResp["status"].(string) + if ok && (status == "green" || status == "yellow") { + esReady = true + t.Log("Elasticsearch service is ready.") + break + } + } + resp.Body.Close() + } + if resp != nil { + resp.Body.Close() + } + time.Sleep(2 * time.Second) + } + if !esReady { + t.Fatal("Elasticsearch service did not become ready in time.") + } t.Log("Sending a test log to the ingestor...") timestamp := time.Now().UTC().Format(time.RFC3339) diff --git a/parser/main.go b/parser/main.go index 9dec968..b5489fe 100644 --- a/parser/main.go +++ b/parser/main.go @@ -4,7 +4,6 @@ import ( "context" "encoding/json" "fmt" - "log" "os" "os/signal" "strings" @@ -12,6 +11,8 @@ import ( "syscall" "time" + "log/slog" + "github.com/IBM/sarama" ) @@ -45,11 +46,13 @@ type ParsedLog struct { type LogProcessor struct { version string + logger *slog.Logger } -func NewLogProcessor() *LogProcessor { +func NewLogProcessor(logger *slog.Logger) *LogProcessor { return &LogProcessor{ version: "1.0.0", + logger: logger, } } @@ -138,7 +141,7 @@ func (p *LogProcessor) normalizeLogLevel(level string) string { return normalized } - log.Printf("Warning: Unknown log level %s, defaulting to INFO", level) + p.logger.Warn("Unknown log level, defaulting to INFO", "level", level) return "INFO" } @@ -154,10 +157,10 @@ func generateMessageHash(message string) string { return string(rune(hash)) } -func sendToDLQ(producer sarama.SyncProducer, originalMessage *sarama.ConsumerMessage, processingError error) { +func sendToDLQ(logger *slog.Logger, producer sarama.SyncProducer, originalMessage *sarama.ConsumerMessage, processingError error) { dlqMessage := &sarama.ProducerMessage{ Topic: DLQTopic, - Value: sarama.ByteEncoder(originalMessage.Value), // The original message content + Value: sarama.ByteEncoder(originalMessage.Value), Headers: []sarama.RecordHeader{ { Key: []byte("error"), @@ -169,29 +172,33 @@ func sendToDLQ(producer sarama.SyncProducer, originalMessage *sarama.ConsumerMes }, { Key: []byte("original_partition"), - Value: []byte(string(rune(originalMessage.Partition))), + Value: []byte(fmt.Sprintf("%d", originalMessage.Partition)), }, { Key: []byte("original_offset"), - Value: []byte(string(rune(originalMessage.Offset))), + Value: []byte(fmt.Sprintf("%d", originalMessage.Offset)), }, }, } _, _, err := producer.SendMessage(dlqMessage) if err != nil { - log.Printf("CRITICAL: Failed to send message to DLQ. Topic: %s. Error: %v", DLQTopic, err) + logger.Error("CRITICAL: Failed to send message to DLQ", "topic", DLQTopic, "error", err) } else { - log.Printf("Message sent to DLQ topic %s due to error: %v", DLQTopic, processingError) + logger.Info("Message sent to DLQ", "topic", DLQTopic, "reason", processingError.Error()) } } func main() { - log.Println("Starting parser service...") + + logger := slog.New(slog.NewJSONHandler(os.Stdout, nil)) + + logger.Info("Starting parser service...") kafkaBrokers := os.Getenv("KAFKA_BROKERS") if kafkaBrokers == "" { - log.Fatal("KAFKA_BROKERS environment variable not set") + logger.Error("KAFKA_BROKERS environment variable not set") + os.Exit(1) } brokers := strings.Split(kafkaBrokers, ",") @@ -202,24 +209,27 @@ func main() { consumerGroup, err := sarama.NewConsumerGroup(brokers, ConsumerGroup, config) if err != nil { - log.Fatalf("Failed to create consumer group: %v", err) + logger.Error("Failed to create consumer group", "error", err) + os.Exit(1) } defer consumerGroup.Close() producer, err := newProducer(brokers) if err != nil { - log.Fatalf("Failed to create producer: %v", err) + logger.Error("Failed to create producer", "error", err) + os.Exit(1) } defer producer.Close() ctx, cancel := context.WithCancel(context.Background()) - processor := NewLogProcessor() + processor := NewLogProcessor(logger) handler := &logHandler{ producer: producer, processor: processor, ready: make(chan bool), + logger: logger, } wg := &sync.WaitGroup{} @@ -229,7 +239,7 @@ func main() { defer wg.Done() for { if err := consumerGroup.Consume(ctx, []string{InputTopic}, handler); err != nil { - log.Printf("Error from consumer: %v", err) + logger.Error("Error from consumer", "error", err) } if ctx.Err() != nil { return @@ -239,13 +249,13 @@ func main() { }() <-handler.ready - log.Println("Parser consumer is up and running!") + logger.Info("Parser consumer is up and running!") sigterm := make(chan os.Signal, 1) signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM) <-sigterm - log.Println("Termination signal received. Shutting down") + logger.Info("Termination signal received. Shutting down") cancel() wg.Wait() } @@ -262,6 +272,7 @@ type logHandler struct { producer sarama.SyncProducer processor *LogProcessor ready chan bool + logger *slog.Logger } func (h *logHandler) Setup(_ sarama.ConsumerGroupSession) error { @@ -275,13 +286,13 @@ func (h *logHandler) Cleanup(_ sarama.ConsumerGroupSession) error { func (h *logHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error { for message := range claim.Messages() { - log.Printf("Received message on topic %s", message.Topic) + h.logger.Info("Received message", "topic", message.Topic, "partition", message.Partition, "offset", message.Offset) var incomingLog IncomingLogEntry if err := json.Unmarshal(message.Value, &incomingLog); err != nil { - log.Printf("Failed to unmarshal incoming log: %v", err) + h.logger.Error("Failed to unmarshal incoming log", "error", err, "topic", message.Topic, "offset", message.Offset) - sendToDLQ(h.producer, message, err) + sendToDLQ(h.logger, h.producer, message, err) session.MarkMessage(message, "") continue @@ -289,9 +300,9 @@ func (h *logHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sar parsedLog, err := h.processor.ProcessLog(incomingLog) if err != nil { - log.Printf("Failed to process log: %v", err) + h.logger.Error("Failed to process log", "error", err, "topic", message.Topic, "offset", message.Offset) - sendToDLQ(h.producer, message, err) + sendToDLQ(h.logger, h.producer, message, err) session.MarkMessage(message, "") continue @@ -299,7 +310,7 @@ func (h *logHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sar enrichedLogBytes, err := json.Marshal(parsedLog) if err != nil { - log.Printf("Failed to marshal processed log: %v", err) + h.logger.Error("Failed to marshal processed log", "error", err, "log_id", parsedLog.ID) session.MarkMessage(message, "") continue } @@ -311,10 +322,9 @@ func (h *logHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sar partition, offset, err := h.producer.SendMessage(producerMsg) if err != nil { - log.Printf("Failed to send processed log to kafka: %v", err) + h.logger.Error("Failed to send processed log to Kafka", "error", err, "topic", OutputTopic, "log_id", parsedLog.ID) } else { - log.Printf("Successfully sent processed log to topic %s (partition: %d, offset: %d)", - OutputTopic, partition, offset) + h.logger.Info("Successfully sent processed log to Kafka", "topic", OutputTopic, "partition", partition, "offset", offset, "log_id", parsedLog.ID) } session.MarkMessage(message, "") diff --git a/parser/main_test.go b/parser/main_test.go index 7aaa31f..dc458f1 100644 --- a/parser/main_test.go +++ b/parser/main_test.go @@ -1,13 +1,19 @@ package main import ( + "log/slog" + "os" "reflect" "testing" "time" ) +func newDiscardLogger() *slog.Logger { + return slog.New(slog.NewJSONHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})) +} + func TestLogProcessor_normalizeTimestamp(t *testing.T) { - processor := NewLogProcessor() + processor := NewLogProcessor(newDiscardLogger()) testCases := []struct { name string @@ -63,7 +69,7 @@ func TestLogProcessor_normalizeTimestamp(t *testing.T) { } func TestLogProcessor_normalizeLogLevel(t *testing.T) { - processor := NewLogProcessor() + processor := NewLogProcessor(newDiscardLogger()) testCases := []struct { name string @@ -88,7 +94,7 @@ func TestLogProcessor_normalizeLogLevel(t *testing.T) { } func TestLogProcessor_ProcessLog(t *testing.T) { - processor := NewLogProcessor() + processor := NewLogProcessor(newDiscardLogger()) rawLog := IncomingLogEntry{ Timestamp: "2025-07-01T12:00:00Z", diff --git a/storage-writer/main.go b/storage-writer/main.go index beb5f14..73006e8 100644 --- a/storage-writer/main.go +++ b/storage-writer/main.go @@ -5,7 +5,6 @@ import ( "context" "encoding/json" "fmt" - "log" "os" "os/signal" "strings" @@ -13,6 +12,8 @@ import ( "syscall" "time" + "log/slog" + "github.com/IBM/sarama" "github.com/elastic/go-elasticsearch/v8" "github.com/elastic/go-elasticsearch/v8/esapi" @@ -47,11 +48,13 @@ type ElasticsearchDocument struct { type IndexManager struct { client *elasticsearch.Client + logger *slog.Logger } -func NewIndexManager(client *elasticsearch.Client) *IndexManager { +func NewIndexManager(client *elasticsearch.Client, logger *slog.Logger) *IndexManager { return &IndexManager{ client: client, + logger: logger, } } @@ -68,7 +71,7 @@ func (im *IndexManager) CreateIndexIfNotExists(indexName string) error { defer res.Body.Close() if res.StatusCode == 200 { - log.Printf("Index %s already exists", indexName) + im.logger.Info("Index already exists", "index_name", indexName) return nil } @@ -112,14 +115,14 @@ func (im *IndexManager) CreateIndexIfNotExists(indexName string) error { return fmt.Errorf("failed to create index: %s", createRes.String()) } - log.Printf("Successfully created index %s", indexName) + im.logger.Info("Successfully created index", "index_name", indexName) return nil } func (im *IndexManager) GetIndexname(timestamp time.Time) string { t := timestamp if t.IsZero() { - log.Printf("Invalid timestamp (zero value), using current time") + im.logger.Warn("Invalid timestamp (zero value), using current time for index name") t = time.Now() } return fmt.Sprintf("%s-%s", IndexPrefix, t.Format("2006.01.02")) @@ -129,13 +132,15 @@ type LogStorageProcessor struct { esClient *elasticsearch.Client indexManager *IndexManager version string + logger *slog.Logger } -func NewLogStorageProcessor(client *elasticsearch.Client) *LogStorageProcessor { +func NewLogStorageProcessor(client *elasticsearch.Client, logger *slog.Logger) *LogStorageProcessor { return &LogStorageProcessor{ esClient: client, - indexManager: NewIndexManager(client), + indexManager: NewIndexManager(client, logger), version: "1.0.0", + logger: logger, } } @@ -180,35 +185,41 @@ func (lsp *LogStorageProcessor) ProcessAndStore(messageValue []byte) error { return fmt.Errorf("elasticsearch indexing error: %s", res.String()) } - log.Printf("Successfully indexed document %s to %s", parsedLog.ID, indexName) + lsp.logger.Info("Successfully indexed document", "document_id", parsedLog.ID, "index_name", indexName) return nil } func main() { - log.Println("Starting storage writer...") + logger := slog.New(slog.NewJSONHandler(os.Stdout, nil)) + + logger.Info("Starting storage writer...") kafkaEnv := os.Getenv("KAFKA_BROKERS") esEnv := os.Getenv("ELASTICSEARCH_URL") if kafkaEnv == "" || esEnv == "" { - log.Fatalf("Environment variables KAFKA_BROKERS and ELASTICSEARCH_URL must be set") + logger.Error("Environment variables KAFKA_BROKERS and ELASTICSEARCH_URL must be set") + os.Exit(1) } esClient, err := elasticsearch.NewClient(elasticsearch.Config{ Addresses: []string{esEnv}, }) if err != nil { - log.Fatalf("Error creating Elasticsearch client: %v", err) + logger.Error("Error creating Elasticsearch client", "error", err) + os.Exit(1) } res, err := esClient.Info() if err != nil { - log.Fatalf("Error getting Elasticsearch info: %v", err) + logger.Error("Error getting Elasticsearch info", "error", err) + os.Exit(1) } defer res.Body.Close() if res.IsError() { - log.Fatalf("Error response from Elasticsearch: %s", res.String()) + logger.Error("Error response from Elasticsearch", "response", res.String()) + os.Exit(1) } - log.Println("Connected to Elasticsearch successfully") + logger.Info("Connected to Elasticsearch successfully") config := sarama.NewConfig() config.Version = sarama.V2_8_0_0 @@ -223,29 +234,32 @@ func main() { if err == nil { break } - log.Printf("Kafka not ready (%v), retrying in 5s...", err) + logger.Warn("Kafka not ready, retrying in 5s...", "error", err, "attempt", i+1) time.Sleep(5 * time.Second) } if err != nil { - log.Fatalf("Failed to create consumer group after retries: %v", err) + logger.Error("Failed to create consumer group after retries", "error", err) + os.Exit(1) } defer consumerGroup.Close() producer, err := newProducer(brokers) if err != nil { - log.Fatalf("Failed to create producer for DLQ: %v", err) + logger.Error("Failed to create producer for DLQ", "error", err) + os.Exit(1) } defer producer.Close() ctx, cancel := context.WithCancel(context.Background()) defer cancel() - processor := NewLogStorageProcessor(esClient) + processor := NewLogStorageProcessor(esClient, logger) handler := &logStorageHandler{ processor: processor, producer: producer, ready: make(chan bool), + logger: logger, } wg := &sync.WaitGroup{} @@ -254,7 +268,7 @@ func main() { defer wg.Done() for { if err := consumerGroup.Consume(ctx, []string{InputTopic}, handler); err != nil { - log.Printf("Error from consumer: %v", err) + logger.Error("Error from consumer", "error", err) } if ctx.Err() != nil { return @@ -264,13 +278,13 @@ func main() { }() <-handler.ready - log.Println("Storage Writer is online! Waiting for messages...") + logger.Info("Storage Writer is online! Waiting for messages...") sigterm := make(chan os.Signal, 1) signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM) <-sigterm - log.Println("Termination signal received. Shutting down the service...") + logger.Info("Termination signal received. Shutting down the service...") cancel() wg.Wait() } @@ -283,7 +297,7 @@ func newProducer(brokers []string) (sarama.SyncProducer, error) { return sarama.NewSyncProducer(brokers, config) } -func sendToDLQ(producer sarama.SyncProducer, originalMessage *sarama.ConsumerMessage, processingError error) { +func sendToDLQ(logger *slog.Logger, producer sarama.SyncProducer, originalMessage *sarama.ConsumerMessage, processingError error) { dlqMessage := &sarama.ProducerMessage{ Topic: ElasticsearchDLQTopic, Value: sarama.ByteEncoder(originalMessage.Value), @@ -309,9 +323,9 @@ func sendToDLQ(producer sarama.SyncProducer, originalMessage *sarama.ConsumerMes _, _, err := producer.SendMessage(dlqMessage) if err != nil { - log.Printf("CRITICAL: Failed to send message to Elasticsearch DLQ. Topic: %s. Error: %v", ElasticsearchDLQTopic, err) + logger.Error("CRITICAL: Failed to send message to Elasticsearch DLQ", "topic", ElasticsearchDLQTopic, "error", err) } else { - log.Printf("Message sent to Elasticsearch DLQ topic %s due to error: %v", ElasticsearchDLQTopic, processingError) + logger.Info("Message sent to Elasticsearch DLQ", "topic", ElasticsearchDLQTopic, "reason", processingError.Error()) } } @@ -319,6 +333,7 @@ type logStorageHandler struct { processor *LogStorageProcessor producer sarama.SyncProducer ready chan bool + logger *slog.Logger } func (handler *logStorageHandler) Setup(_ sarama.ConsumerGroupSession) error { @@ -332,13 +347,13 @@ func (handler *logStorageHandler) Cleanup(_ sarama.ConsumerGroupSession) error { func (handler *logStorageHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error { for message := range claim.Messages() { - log.Printf("Received parsed log from topic %s", message.Topic) + handler.logger.Info("Received parsed log", "topic", message.Topic, "partition", message.Partition, "offset", message.Offset) if err := handler.processor.ProcessAndStore(message.Value); err != nil { - log.Printf("Failed to process and store log: %v", err) + handler.logger.Error("Failed to process and store log", "error", err, "topic", message.Topic, "offset", message.Offset) // Send the failed message to the Dead-Letter Queue - sendToDLQ(handler.producer, message, err) + sendToDLQ(handler.logger, handler.producer, message, err) session.MarkMessage(message, "") continue diff --git a/storage-writer/main_test.go b/storage-writer/main_test.go index 097d6a2..374eb81 100644 --- a/storage-writer/main_test.go +++ b/storage-writer/main_test.go @@ -1,12 +1,19 @@ package main import ( + "log/slog" + "os" "testing" "time" ) +func newDiscardLogger() *slog.Logger { + return slog.New(slog.NewJSONHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})) +} + func TestIndexManager_GetIndexname(t *testing.T) { - manager := NewIndexManager(nil) + + manager := NewIndexManager(nil, newDiscardLogger()) testCases := []struct { name string