refactor(v2): improve code quality with comments and file reorganization

- Add comprehensive package documentation explaining the plugin architecture
- Reorganize internal package into logical files:
  - s3.go: S3 client creation, validation, and upload functions
  - ingestion.go: Ingestion context creation and management
  - context.go: Plugin context, configuration, and type definitions
  - flush_manager.go: Dual-timer flush strategy implementation
- Add detailed block comment explaining the dual-timer flush strategy
- Add godoc comments to all exported types and functions
- Document Fluent Bit plugin callbacks and data flow
- Fix potential bug: use m.hardTimeout.IsZero() instead of nextHardTimeout.IsZero()
- Extract magic values into named constants
- Improve error messages and logging consistency

Author: jackluo923

plugins/out_clp_s3_v2/internal/context.go

@@ -13,52 +13,128 @@ import (
 	"github.com/y-scope/clp-ffi-go/ir"
 )
 
-// FlushConfigContext stores the flush control configurations
+// Default configuration values.
+const (
+	// defaultLogLevelKey is the JSON key used to extract log severity from records.
+	defaultLogLevelKey = "level"
+	// defaultFlushDelta is the default time between flushes for all log levels.
+	defaultFlushDelta = 3 * time.Second
+)
+
+// FlushConfigContext stores configuration for the dual-timer flush strategy.
+//
+// The flush strategy uses two timers per log stream:
+//   - Hard timer: Guarantees logs are uploaded within a maximum time window
+//   - Soft timer: Uploads after a period of inactivity (no new logs)
+//
+// Each log level can have different timer values, allowing critical logs
+// (ERROR, FATAL) to trigger faster uploads than debug logs.
 type FlushConfigContext struct {
-	LogLevelKey     string
+	// LogLevelKey is the JSON key used to extract log severity from records.
+	// Common values: "level", "severity", "log_level"
+	LogLevelKey string
+
+	// defaultLogLevel is the log level index to use when level cannot be determined.
+	// Maps to the index in hardDeltas/softDeltas arrays.
 	defaultLogLevel int
-	hardDeltas      []time.Duration
-	softDeltas      []time.Duration
+
+	// hardDeltas contains the hard flush deadline for each log level.
+	// Index corresponds to log level (0=debug, 1=info, 2=warn, 3=error, 4=fatal).
+	hardDeltas []time.Duration
+
+	// softDeltas contains the soft flush delay for each log level.
+	// Soft timer resets on each log event, triggering upload after inactivity.
+	softDeltas []time.Duration
 }
 
-// flushContext manages timing and callback logic for log flushing.
+// flushContext manages the dual-timer flush strategy for a single log stream.
+//
+// Thread-safety: All public methods acquire the Mutex before modifying state.
+// The timers run in separate goroutines but callbacks also acquire the Mutex.
 type flushContext struct {
-	HardTimer    *time.Timer
-	hardTimeout  time.Time
-	softDelta    time.Duration
-	SoftTimer    *time.Timer
+	// HardTimer fires at the hard deadline, guaranteeing upload within a time window.
+	HardTimer *time.Timer
+	// hardTimeout tracks when the hard timer will fire (for comparison with new events).
+	hardTimeout time.Time
+
+	// SoftTimer fires after a period of inactivity (no new log events).
+	SoftTimer *time.Timer
+	// softDelta tracks the current soft timer duration (minimum seen for this batch).
+	softDelta time.Duration
+
+	// userCallback is invoked when either timer fires, triggering S3 upload.
 	userCallback func()
-	Mutex        sync.Mutex
+
+	// Mutex protects all fields from concurrent access.
+	Mutex sync.Mutex
 }
 
-// compressionContext encapsulates file and compression writers.
+// compressionContext encapsulates the compression pipeline resources.
+//
+// The pipeline processes log events as:
+//
+//	Log Event → IR Writer → Zstd Writer → File
 type compressionContext struct {
-	File       *os.File
+	// File is the temporary file storing compressed logs before S3 upload.
+	File *os.File
+	// ZstdWriter compresses data using Zstandard algorithm.
 	ZstdWriter *zstd.Encoder
-	IRWriter   *ir.Writer
+	// IRWriter encodes log events into CLP's Intermediate Representation format.
+	IRWriter *ir.Writer
 }
 
-// IngestionContext contains compression and flush contexts for a particular log path.
+// IngestionContext contains all resources for processing a single log stream.
+//
+// Each unique log path (derived from Fluent Bit tag) has its own IngestionContext,
+// allowing independent compression and flush timing per stream.
 type IngestionContext struct {
+	// Compression holds the file and encoder resources for this stream.
 	Compression *compressionContext
-	Flush       *flushContext
+	// Flush manages the upload timing for this stream.
+	Flush *flushContext
 }
 
-// s3Context holds AWS S3 configuration and client.
+// s3Context holds the S3 client and bucket configuration.
 type s3Context struct {
+	// Client is the configured AWS S3 client.
 	Client *s3.Client
+	// Bucket is the target S3 bucket for log uploads.
 	Bucket string
 }
 
-// PluginContext is the top-level context for the plugin.
+// PluginContext is the top-level context for the out_clp_s3_v2 plugin.
+//
+// A single PluginContext is created during FLBPluginInit and shared across
+// all flush callbacks. It contains:
+//   - S3 configuration and client
+//   - Map of ingestion contexts (one per log stream/tag)
+//   - Flush timing configuration
 type PluginContext struct {
-	S3          *s3Context
-	Ingestion   map[string]*IngestionContext
+	// S3 holds the S3 client and bucket configuration.
+	S3 *s3Context
+	// Ingestion maps log paths to their ingestion contexts.
+	// Key is typically the Fluent Bit tag or file_path from log records.
+	Ingestion map[string]*IngestionContext
+	// FlushConfig contains the dual-timer flush strategy configuration.
 	FlushConfig *FlushConfigContext
 }
 
-// NewPluginContext initializes a new PluginContext
+// NewPluginContext creates and initializes a new PluginContext from Fluent Bit configuration.
+//
+// This function:
+//  1. Creates an S3 client using AWS credentials from the environment
+//  2. Validates the target S3 bucket exists and is accessible
+//  3. Loads flush timing configuration from plugin settings
+//
+// Configuration keys read from Fluent Bit:
+//   - log_bucket: Target S3 bucket name (required)
+//   - log_level_key: JSON key for log severity (default: "level")
+//   - flush_hard_delta_*: Hard timer durations per log level
+//   - flush_soft_delta_*: Soft timer durations per log level
+//
+// Returns an error if S3 client creation or bucket validation fails.
 func NewPluginContext(plugin unsafe.Pointer) (*PluginContext, error) {
+	// Create and validate S3 client
 	client, err := S3CreateClient()
 	if err != nil {
 		log.Printf("[error] Failed to create S3 client: %v", err)
@@ -72,60 +148,60 @@ func NewPluginContext(plugin unsafe.Pointer) (*PluginContext, error) {
 	}
 	log.Printf("[info] Logs are configured to be uploaded to s3://%s", bucket)
 
-	logLevelKey := getConfigWithDefaultString(plugin, "log_level_key", "level")
+	// Load log level key configuration
+	logLevelKey := getConfigWithDefault(plugin, "log_level_key", defaultLogLevelKey)
 	log.Printf("[info] Log level key is configured to: %q", logLevelKey)
 
-	// Flush behavior control - use very aggressive defaults for now
+	// Load flush timing configuration for each log level
+	// Index order: 0=debug, 1=info, 2=warn, 3=error, 4=fatal
 	hardDeltas := []time.Duration{
-		getConfigWithDefaultTimeDuration(plugin, "flush_hard_delta_debug", 3*time.Second),
-		getConfigWithDefaultTimeDuration(plugin, "flush_hard_delta_info", 3*time.Second),
-		getConfigWithDefaultTimeDuration(plugin, "flush_hard_delta_warn", 3*time.Second),
-		getConfigWithDefaultTimeDuration(plugin, "flush_hard_delta_error", 3*time.Second),
-		getConfigWithDefaultTimeDuration(plugin, "flush_hard_delta_fatal", 3*time.Second),
+		getConfigDuration(plugin, "flush_hard_delta_debug", defaultFlushDelta),
+		getConfigDuration(plugin, "flush_hard_delta_info", defaultFlushDelta),
+		getConfigDuration(plugin, "flush_hard_delta_warn", defaultFlushDelta),
+		getConfigDuration(plugin, "flush_hard_delta_error", defaultFlushDelta),
+		getConfigDuration(plugin, "flush_hard_delta_fatal", defaultFlushDelta),
 	}
 	softDeltas := []time.Duration{
-		getConfigWithDefaultTimeDuration(plugin, "flush_soft_delta_debug", 3*time.Second),
-		getConfigWithDefaultTimeDuration(plugin, "flush_soft_delta_info", 3*time.Second),
-		getConfigWithDefaultTimeDuration(plugin, "flush_soft_delta_warn", 3*time.Second),
-		getConfigWithDefaultTimeDuration(plugin, "flush_soft_delta_error", 3*time.Second),
-		getConfigWithDefaultTimeDuration(plugin, "flush_soft_delta_fatal", 3*time.Second),
+		getConfigDuration(plugin, "flush_soft_delta_debug", defaultFlushDelta),
+		getConfigDuration(plugin, "flush_soft_delta_info", defaultFlushDelta),
+		getConfigDuration(plugin, "flush_soft_delta_warn", defaultFlushDelta),
+		getConfigDuration(plugin, "flush_soft_delta_error", defaultFlushDelta),
+		getConfigDuration(plugin, "flush_soft_delta_fatal", defaultFlushDelta),
 	}
 
-	pluginCtx := &PluginContext{
+	return &PluginContext{
 		S3: &s3Context{
 			Client: client,
 			Bucket: bucket,
 		},
 		Ingestion: make(map[string]*IngestionContext),
 		FlushConfig: &FlushConfigContext{
 			LogLevelKey:     logLevelKey,
-			defaultLogLevel: 0,
+			defaultLogLevel: 0, // Default to debug level
 			hardDeltas:      hardDeltas,
 			softDeltas:      softDeltas,
 		},
-	}
-
-	return pluginCtx, nil
+	}, nil
 }
 
-func getConfigWithDefaultTimeDuration(
-	plugin unsafe.Pointer,
-	key string,
-	defaultVal time.Duration,
-) time.Duration {
-	duration, err := time.ParseDuration(output.FLBPluginConfigKey(plugin, key))
+// getConfigDuration reads a duration configuration value with a default fallback.
+func getConfigDuration(plugin unsafe.Pointer, key string, defaultVal time.Duration) time.Duration {
+	rawValue := output.FLBPluginConfigKey(plugin, key)
+	if rawValue == "" {
+		return defaultVal
+	}
+
+	duration, err := time.ParseDuration(rawValue)
 	if err != nil {
-		log.Printf("[error] Failed to parse duration %q: %v", key, err)
+		log.Printf("[warn] Failed to parse duration for %q (%q): %v; using default %v",
+			key, rawValue, err, defaultVal)
 		return defaultVal
 	}
 	return duration
 }
 
-func getConfigWithDefaultString(
-	plugin unsafe.Pointer,
-	key,
-	defaultVal string,
-) string {
+// getConfigWithDefault reads a string configuration value with a default fallback.
+func getConfigWithDefault(plugin unsafe.Pointer, key, defaultVal string) string {
 	val := output.FLBPluginConfigKey(plugin, key)
 	if val == "" {
 		return defaultVal