diff --git a/.gitignore b/.gitignore index d8fb1c5..f4247ec 100644 --- a/.gitignore +++ b/.gitignore @@ -42,6 +42,8 @@ go.sum # Logs *.log +logs/ +*.log.gz # Dependency directories (remove the comment below if you want to ignore them) #vendor/ diff --git a/configs/config.dev.toml b/configs/config.dev.toml index 1985196..88df636 100644 --- a/configs/config.dev.toml +++ b/configs/config.dev.toml @@ -50,6 +50,25 @@ monitor_interval = "30s" [log] level = "info" format = "console" +output = ["stdout", "file"] +file = "logs/caatsm.log" + +# File rotation settings +max_size = 100 # MB +max_backups = 7 # Keep 7 rotated files +max_age = 30 # Keep logs for 30 days +compress = true # Compress old log files + +# Advanced options +disable_caller = false +disable_stacktrace = false +development = false + +# Sampling configuration (optional, for high-volume scenarios) +# [log.sampling] +# initial = 100 # Log first 100 messages +# thereafter = 100 # Then log every 100th message +# tick = "1s" # Per second [telemetry] enabled = true diff --git a/go.mod b/go.mod index c99cd55..879cd19 100644 --- a/go.mod +++ b/go.mod @@ -107,4 +107,5 @@ require ( google.golang.org/genproto/googleapis/rpc v0.0.0-20251111163417-95abcf5c77ba // indirect google.golang.org/grpc v1.76.0 // indirect google.golang.org/protobuf v1.36.10 // indirect + gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect ) diff --git a/internal/app/processor.go b/internal/app/processor.go index 70bd01b..9cc53c1 100644 --- a/internal/app/processor.go +++ b/internal/app/processor.go @@ -150,6 +150,12 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) latency := parsed.ParsedAt.Sub(receivedAt) p.telemetry.RecordFailure("repository") p.telemetry.RecordProcessingResult(ctx, string(parsed.Status), parsed.Category, latency) + // Log business layer failure with message context (Repository layer already logged technical error) + msgLogger.Error("Failed to persist parsed message", + zap.String("status", string(parsed.Status)), + zap.Duration("latency", latency), + zap.Error(err), + ) return fmt.Errorf("failed to insert message: %w", err) } @@ -202,11 +208,11 @@ func (p *MessageProcessor) persistRaw(ctx context.Context, msg *dto.ParsedTelegr msg.ReceivedAt = time.Now() } if err := p.repository.InsertRaw(ctx, msg); err != nil { - p.logger.Error("Failed to persist raw telegram", - zap.String("message_id", msg.MessageID), - zap.String("status", string(msg.Status)), - zap.Error(err), - ) + // Error already logged in Repository.InsertRaw, no need to log again + // Just add event to span if recording + if span := trace.SpanFromContext(ctx); span.IsRecording() { + span.RecordError(err) + } } else { if span := trace.SpanFromContext(ctx); span.IsRecording() { span.AddEvent("raw telegram persisted", diff --git a/internal/infra/config/config.go b/internal/infra/config/config.go index e1d762c..32be97a 100644 --- a/internal/infra/config/config.go +++ b/internal/infra/config/config.go @@ -80,6 +80,29 @@ type AppConfig struct { type LogConfig struct { Level string `koanf:"level"` Format string `koanf:"format"` // json or console + + // Output configuration + Output []string `koanf:"output"` // stdout, stderr, or file path + File string `koanf:"file"` // Log file path (if output includes file) + + // File rotation settings + MaxSize int `koanf:"max_size"` // Max size in MB before rotation + MaxBackups int `koanf:"max_backups"` // Max number of old log files to keep + MaxAge int `koanf:"max_age"` // Max days to retain old log files + Compress bool `koanf:"compress"` // Compress rotated log files + + // Advanced zap options + DisableCaller bool `koanf:"disable_caller"` // Disable caller information + DisableStacktrace bool `koanf:"disable_stacktrace"` // Disable stacktrace + Development bool `koanf:"development"` // Enable development mode + Sampling *LogSamplingConfig `koanf:"sampling"` +} + +// LogSamplingConfig configures log sampling to reduce high-volume logs +type LogSamplingConfig struct { + Initial int `koanf:"initial"` // Log first N messages per tick + Thereafter int `koanf:"thereafter"` // Then log every Nth message + Tick time.Duration `koanf:"tick"` // Sampling tick interval } // PublisherConfig holds publisher configuration @@ -182,6 +205,30 @@ func LoadConfig() (*Config, error) { if cfg.Log.Format == "" { cfg.Log.Format = "json" } + if len(cfg.Log.Output) == 0 { + cfg.Log.Output = []string{"stdout"} + } else { + // If output contains "file" but file path is empty, set default path + hasFile := false + for _, out := range cfg.Log.Output { + if out == "file" { + hasFile = true + break + } + } + if hasFile && cfg.Log.File == "" { + cfg.Log.File = "logs/caatsm.log" + } + } + if cfg.Log.MaxSize == 0 { + cfg.Log.MaxSize = 100 // 100 MB + } + if cfg.Log.MaxBackups == 0 { + cfg.Log.MaxBackups = 7 + } + if cfg.Log.MaxAge == 0 { + cfg.Log.MaxAge = 30 // 30 days + } if cfg.NATS.Mode == "" { cfg.NATS.Mode = "jetstream" } else { diff --git a/internal/infra/log/logger.go b/internal/infra/log/logger.go index 2d8c8c5..95149e4 100644 --- a/internal/infra/log/logger.go +++ b/internal/infra/log/logger.go @@ -2,44 +2,145 @@ package log import ( "caatsm/internal/infra/config" + "os" + "path/filepath" + "time" + "go.uber.org/zap" "go.uber.org/zap/zapcore" + "gopkg.in/natefinch/lumberjack.v2" ) // ProvideLogger creates a zap logger based on configuration func ProvideLogger(cfg *config.Config) (*zap.Logger, error) { var zapConfig zap.Config - - if cfg.Log.Format == "console" { + + if cfg.Log.Development || cfg.Log.Format == "console" { zapConfig = zap.NewDevelopmentConfig() } else { zapConfig = zap.NewProductionConfig() } - + // Set log level + var level zapcore.Level switch cfg.Log.Level { case "debug": - zapConfig.Level = zap.NewAtomicLevelAt(zapcore.DebugLevel) + level = zapcore.DebugLevel case "info": - zapConfig.Level = zap.NewAtomicLevelAt(zapcore.InfoLevel) + level = zapcore.InfoLevel case "warn": - zapConfig.Level = zap.NewAtomicLevelAt(zapcore.WarnLevel) + level = zapcore.WarnLevel case "error": - zapConfig.Level = zap.NewAtomicLevelAt(zapcore.ErrorLevel) + level = zapcore.ErrorLevel default: - zapConfig.Level = zap.NewAtomicLevelAt(zapcore.InfoLevel) + level = zapcore.InfoLevel } - + zapConfig.Level = zap.NewAtomicLevelAt(level) + + // Determine outputs + outputs := cfg.Log.Output + if len(outputs) == 0 { + outputs = []string{"stdout"} // Default to stdout + } + + // Create encoder + var encoder zapcore.Encoder + if cfg.Log.Format == "json" { + encoder = zapcore.NewJSONEncoder(zapConfig.EncoderConfig) + } else { + encoder = zapcore.NewConsoleEncoder(zapConfig.EncoderConfig) + } + + // Build cores for different outputs + var cores []zapcore.Core + + // Create cores for each output + for _, output := range outputs { + var writeSyncer zapcore.WriteSyncer + + switch output { + case "stdout": + writeSyncer = zapcore.AddSync(os.Stdout) + case "stderr": + writeSyncer = zapcore.AddSync(os.Stderr) + case "file": + filePath := cfg.Log.File + if filePath == "" { + // Use default path if file path is not specified + filePath = "logs/caatsm.log" + } + // Ensure log directory exists + dir := filepath.Dir(filePath) + if dir != "" && dir != "." { + if err := os.MkdirAll(dir, 0755); err != nil { + return nil, err + } + } + + // Configure log rotation + lj := &lumberjack.Logger{ + Filename: filePath, + MaxSize: cfg.Log.MaxSize, // MB + MaxBackups: cfg.Log.MaxBackups, + MaxAge: cfg.Log.MaxAge, // days + Compress: cfg.Log.Compress, + } + writeSyncer = zapcore.AddSync(lj) + default: + // Treat as file path + dir := filepath.Dir(output) + if dir != "" && dir != "." { + if err := os.MkdirAll(dir, 0755); err != nil { + return nil, err + } + } + lj := &lumberjack.Logger{ + Filename: output, + MaxSize: cfg.Log.MaxSize, + MaxBackups: cfg.Log.MaxBackups, + MaxAge: cfg.Log.MaxAge, + Compress: cfg.Log.Compress, + } + writeSyncer = zapcore.AddSync(lj) + } + + cores = append(cores, zapcore.NewCore(encoder, writeSyncer, zapConfig.Level)) + } + + // Combine cores + core := zapcore.NewTee(cores...) + + // Build options + opts := []zap.Option{} + if !cfg.Log.DisableCaller { + opts = append(opts, zap.AddCaller()) + } + if !cfg.Log.DisableStacktrace { + opts = append(opts, zap.AddStacktrace(zapcore.ErrorLevel)) + } + + // Add sampling if configured + if cfg.Log.Sampling != nil && cfg.Log.Sampling.Initial > 0 { + tick := cfg.Log.Sampling.Tick + if tick == 0 { + tick = time.Second // Default to 1 second + } + opts = append(opts, zap.WrapCore(func(core zapcore.Core) zapcore.Core { + return zapcore.NewSamplerWithOptions( + core, + tick, + cfg.Log.Sampling.Initial, + cfg.Log.Sampling.Thereafter, + ) + })) + } + // Build logger - logger, err := zapConfig.Build(zap.AddCaller(), zap.AddStacktrace(zapcore.ErrorLevel)) - if err != nil { - return nil, err - } - + logger := zap.New(core, opts...) + // Note: We still replace global logger for backward compatibility with parsers package // This will be removed once parsers are fully migrated to use dependency injection zap.ReplaceGlobals(logger) - + return logger, nil } - diff --git a/internal/infra/nats/consumer.go b/internal/infra/nats/consumer.go index 18e7304..157a8b3 100644 --- a/internal/infra/nats/consumer.go +++ b/internal/infra/nats/consumer.go @@ -713,6 +713,12 @@ func (c *Consumer) routeToDLQ(ctx context.Context, msg *nats.Msg, cause error) e data, err := json.Marshal(payload) if err != nil { + c.logger.Error("failed to marshal DLQ payload", + zap.String("stream", c.streamName), + zap.String("consumer", c.consumerName), + zap.String("dlq_subject", c.dlqSubject), + zap.Error(err), + ) return fmt.Errorf("marshal dlq payload: %w", err) } @@ -722,9 +728,23 @@ func (c *Consumer) routeToDLQ(ctx context.Context, msg *nats.Msg, cause error) e // unavailable. Surface this explicitly to make operational diagnosis // easier. if errors.Is(err, nats.ErrNoResponders) { + c.logger.Error("transient DLQ publish error (no responders)", + zap.String("stream", c.streamName), + zap.String("consumer", c.consumerName), + zap.String("dlq_subject", c.dlqSubject), + zap.Int("payload_size", len(data)), + zap.Error(err), + ) c.telemetry.RecordDLQPublishFailure(ctx, c.streamName, c.consumerName) return fmt.Errorf("publish to dlq subject %s: no JetStream stream found for subject or JetStream unavailable: %w", c.dlqSubject, err) } + c.logger.Error("failed to publish to DLQ", + zap.String("stream", c.streamName), + zap.String("consumer", c.consumerName), + zap.String("dlq_subject", c.dlqSubject), + zap.Int("payload_size", len(data)), + zap.Error(err), + ) c.telemetry.RecordDLQPublishFailure(ctx, c.streamName, c.consumerName) return fmt.Errorf("publish to dlq subject %s: %w", c.dlqSubject, err) } diff --git a/internal/infra/nats/jetstream.go b/internal/infra/nats/jetstream.go index ad4acf5..8a2654b 100644 --- a/internal/infra/nats/jetstream.go +++ b/internal/infra/nats/jetstream.go @@ -30,6 +30,12 @@ func ProvideNATSConn(cfg *config.Config, logger *zap.Logger) (*nats.Conn, error) }), ) if err != nil { + logger.Error("failed to connect to NATS", + zap.String("url", cfg.NATS.URL), + zap.Duration("timeout", cfg.Timeouts.Server), + zap.Duration("reconnect_wait", cfg.Timeouts.ReconnectWait), + zap.Error(err), + ) return nil, fmt.Errorf("failed to connect to NATS: %w", err) } @@ -41,6 +47,10 @@ func ProvideJetStream(nc *nats.Conn, cfg *config.Config, logger *zap.Logger) (na // Get JetStream context js, err := nc.JetStream() if err != nil { + logger.Error("failed to get JetStream context", + zap.String("url", cfg.NATS.URL), + zap.Error(err), + ) nc.Close() return nil, fmt.Errorf("failed to get JetStream context: %w", err) } @@ -67,6 +77,11 @@ func EnsureStream(js nats.JetStreamContext, cfg *config.Config, logger *zap.Logg streamSubjects := dedupeSubjects([]string{consumerSubject, publisherSubject}) if len(streamSubjects) == 0 { + logger.Error("no subjects configured for JetStream stream", + zap.String("stream", streamName), + zap.String("consumer_subject", consumerSubject), + zap.String("publisher_subject", publisherSubject), + ) return fmt.Errorf("no subjects configured for JetStream stream %s", streamName) } @@ -101,6 +116,11 @@ func EnsureStream(js nats.JetStreamContext, cfg *config.Config, logger *zap.Logg if errors.Is(err, nats.ErrStreamNotFound) { if shouldBootstrapStream() { if _, err = js.AddStream(streamConfig); err != nil { + logger.Error("failed to create stream", + zap.String("stream", streamName), + zap.Strings("subjects", streamSubjects), + zap.Error(err), + ) return fmt.Errorf("failed to create stream %s: %w", streamName, err) } logger.Info("Created JetStream stream", @@ -109,8 +129,16 @@ func EnsureStream(js nats.JetStreamContext, cfg *config.Config, logger *zap.Logg ) return nil } + logger.Error("stream not found and auto-creation disabled", + zap.String("stream", streamName), + zap.Strings("expected_subjects", streamSubjects), + ) return fmt.Errorf("stream %s not found and auto-creation disabled", streamName) } + logger.Error("failed to fetch stream info", + zap.String("stream", streamName), + zap.Error(err), + ) return fmt.Errorf("failed to fetch stream info for %s: %w", streamName, err) } diff --git a/internal/infra/nats/publisher.go b/internal/infra/nats/publisher.go index a5dffcd..080594a 100644 --- a/internal/infra/nats/publisher.go +++ b/internal/infra/nats/publisher.go @@ -37,12 +37,17 @@ func ProvidePublisher( func (p *Publisher) Publish(message interface{}) error { topic := p.cfg.Publisher.Topic if topic == "" { + p.logger.Error("publisher topic is not configured") return fmt.Errorf("publisher topic is not configured") } // Marshal message to JSON messageBytes, err := json.Marshal(message) if err != nil { + p.logger.Error("failed to marshal message", + zap.String("topic", topic), + zap.Error(err), + ) return fmt.Errorf("failed to marshal message: %w", err) } @@ -66,8 +71,18 @@ func (p *Publisher) Publish(message interface{}) error { if err != nil { // Distinguish temporary JetStream unavailability from permanent config errors. if errors.Is(err, nats.ErrNoResponders) { + p.logger.Error("transient publish error (no responders)", + zap.String("topic", topic), + zap.Int("message_size", len(messageBytes)), + zap.Error(err), + ) return fmt.Errorf("transient publish error (no responders): %w", err) } + p.logger.Error("failed to publish message", + zap.String("topic", topic), + zap.Int("message_size", len(messageBytes)), + zap.Error(err), + ) return fmt.Errorf("failed to publish message: %w", err) } diff --git a/internal/infra/postgres/db.go b/internal/infra/postgres/db.go index 38d4379..e0e8227 100644 --- a/internal/infra/postgres/db.go +++ b/internal/infra/postgres/db.go @@ -7,32 +7,48 @@ import ( "time" "github.com/jackc/pgx/v5/pgxpool" + "go.uber.org/zap" ) // ProvideDB creates a PostgreSQL connection pool -func ProvideDB(cfg *config.Config) (*pgxpool.Pool, error) { +func ProvideDB(cfg *config.Config, logger *zap.Logger) (*pgxpool.Pool, error) { ctx := context.Background() - + poolConfig, err := pgxpool.ParseConfig(cfg.Postgres.URL) if err != nil { + logger.Error("failed to parse postgres URL", + zap.String("url", cfg.Postgres.URL), + zap.Error(err), + ) return nil, fmt.Errorf("failed to parse postgres URL: %w", err) } - + poolConfig.MaxConns = int32(cfg.Postgres.MaxConns) poolConfig.MinConns = int32(cfg.Postgres.MinConns) poolConfig.MaxConnLifetime = time.Hour poolConfig.MaxConnIdleTime = time.Minute * 30 - + pool, err := pgxpool.NewWithConfig(ctx, poolConfig) if err != nil { + logger.Error("failed to create connection pool", + zap.String("url", cfg.Postgres.URL), + zap.Int32("max_conns", cfg.Postgres.MaxConns), + zap.Int32("min_conns", cfg.Postgres.MinConns), + zap.Error(err), + ) return nil, fmt.Errorf("failed to create connection pool: %w", err) } - + // Test connection if err := pool.Ping(ctx); err != nil { + logger.Error("failed to ping database", + zap.String("url", cfg.Postgres.URL), + zap.Error(err), + ) return nil, fmt.Errorf("failed to ping database: %w", err) } - + + // Log the connection pool + logger.Info("Connected to PostgreSQL", zap.Int32("max_conns", cfg.Postgres.MaxConns), zap.Int32("min_conns", cfg.Postgres.MinConns), zap.String("url", cfg.Postgres.URL)) return pool, nil } - diff --git a/internal/infra/postgres/repository.go b/internal/infra/postgres/repository.go index 45742df..cfa2c35 100644 --- a/internal/infra/postgres/repository.go +++ b/internal/infra/postgres/repository.go @@ -1,10 +1,10 @@ package postgres import ( - "caatsm/internal/adapter/mapper" "caatsm/internal/adapter/dto" - "caatsm/internal/port" + "caatsm/internal/adapter/mapper" obsmetrics "caatsm/internal/infra/metrics" + "caatsm/internal/port" "context" "encoding/json" "fmt" @@ -49,6 +49,11 @@ func (r *Repository) InsertOne(ctx context.Context, msg *dto.ParsedTelegram) err if err != nil { span.RecordError(err) span.SetStatus(codes.Error, err.Error()) + r.logger.Error("failed to check existing message", + zap.String("message_id", msg.MessageID), + zap.String("date_time", msg.DateTime), + zap.Error(err), + ) return fmt.Errorf("failed to check existing message: %w", err) } if exists { @@ -64,6 +69,8 @@ func (r *Repository) InsertOne(ctx context.Context, msg *dto.ParsedTelegram) err if err != nil { span.RecordError(err) span.SetStatus(codes.Error, err.Error()) + + r.logger.Error("failed to map message to DB row", zap.Error(err)) return fmt.Errorf("failed to map message to DB row: %w", err) } @@ -92,6 +99,12 @@ func (r *Repository) InsertOne(ctx context.Context, msg *dto.ParsedTelegram) err span.RecordError(err) span.SetStatus(codes.Error, err.Error()) obsmetrics.RecordDBQuery("insert_one", result, elapsed) + r.logger.Error("failed to insert message", + zap.String("uuid", msg.Uuid), + zap.String("message_id", msg.MessageID), + zap.Duration("elapsed", elapsed), + zap.Error(err), + ) return fmt.Errorf("failed to insert message: %w", err) } @@ -128,6 +141,12 @@ func (r *Repository) InsertBatch(ctx context.Context, msgs []*dto.ParsedTelegram for i, msg := range msgs { row, err := r.mapper.ToDBRow(msg) if err != nil { + r.logger.Error("failed to map message to DB row in batch", + zap.Int("index", i), + zap.Int("total", len(msgs)), + zap.String("message_id", msg.MessageID), + zap.Error(err), + ) return fmt.Errorf("failed to map message %d to DB row: %w", i, err) } rows[i] = row @@ -154,6 +173,11 @@ func (r *Repository) InsertBatch(ctx context.Context, msgs []*dto.ParsedTelegram span.RecordError(err) span.SetStatus(codes.Error, err.Error()) obsmetrics.RecordDBQuery("insert_batch", result, elapsed) + r.logger.Error("failed to batch insert messages", + zap.Int("attempted", len(msgs)), + zap.Duration("elapsed", elapsed), + zap.Error(err), + ) return fmt.Errorf("failed to batch insert messages: %w", err) } @@ -178,6 +202,7 @@ func (r *Repository) InsertRaw(ctx context.Context, msg *dto.ParsedTelegram) err err := fmt.Errorf("message is nil") span.RecordError(err) span.SetStatus(codes.Error, err.Error()) + r.logger.Error("message is nil in InsertRaw") return fmt.Errorf("message is nil") } if msg.Uuid == "" { @@ -194,6 +219,10 @@ func (r *Repository) InsertRaw(ctx context.Context, msg *dto.ParsedTelegram) err } metadataJSON, err := json.Marshal(metadata) if err != nil { + r.logger.Error("failed to marshal metadata", + zap.String("uuid", msg.Uuid), + zap.Error(err), + ) return fmt.Errorf("failed to marshal metadata: %w", err) } @@ -227,6 +256,12 @@ func (r *Repository) InsertRaw(ctx context.Context, msg *dto.ParsedTelegram) err span.RecordError(err) span.SetStatus(codes.Error, err.Error()) obsmetrics.RecordDBQuery("insert_raw", result, elapsed) + r.logger.Error("failed to insert raw telegram", + zap.String("uuid", msg.Uuid), + zap.String("status", string(msg.Status)), + zap.Duration("elapsed", elapsed), + zap.Error(err), + ) return fmt.Errorf("failed to insert raw telegram: %w", err) } @@ -261,6 +296,11 @@ func (r *Repository) messageExists(ctx context.Context, messageID, dateTime stri if err == pgx.ErrNoRows { return false, nil } + r.logger.Error("failed to check message existence", + zap.String("message_id", messageID), + zap.String("date_time", dateTime), + zap.Error(err), + ) return false, err } diff --git a/test/integration/jetstream_to_timescale_test.go b/test/integration/jetstream_to_timescale_test.go index a040d20..665f510 100644 --- a/test/integration/jetstream_to_timescale_test.go +++ b/test/integration/jetstream_to_timescale_test.go @@ -50,7 +50,7 @@ func TestJetStreamToTimescaleFlow(t *testing.T) { } defer logger.Sync() - pool, err := postgresinfra.ProvideDB(cfg) + pool, err := postgresinfra.ProvideDB(cfg, logger) if err != nil { t.Fatalf("failed to init postgres: %v", err) }