gistfile1.txt

# Plan: Add Optional Label Provider to FileLineTooBigError

## Context

When the kubernetes_logs source encounters log lines that exceed `max_line_bytes` or `max_merged_line_bytes`, it emits a `component_errors_total` metric. Currently, this metric only includes generic labels (`error_code`, `error_type`, `stage`, and component labels added automatically via tracing context). 

The problem is that when troubleshooting issues with oversized log lines, operators cannot identify **which pod, namespace, or container** is generating the problematic logs without correlating timestamps with verbose error logs.

The Kubernetes log file path already contains all the needed metadata in its structure:
```
/var/log/pods/{namespace}_{pod_name}_{pod_uid}/{container_name}/{sequence}.log
```

This plan adds an optional mechanism for the kubernetes_logs source to provide additional labels (namespace, pod name, container name) to the `FileLineTooBigError` internal event, enabling better observability while maintaining backward compatibility for other file-based sources.

## Design Approach

After exploring the codebase, I identified three possible approaches:

1. **Add file path parameter to trait method** - Breaking change to `FileSourceInternalEvents` trait
2. **Create custom emitter for kubernetes_logs** - Lots of code duplication
3. **Add optional label provider closure to emitter struct** - Clean, backward compatible (RECOMMENDED)

**Chosen Approach: Option 3 - Optional Label Provider Function**

This approach:
- Maintains backward compatibility (no trait changes)
- Follows existing patterns (similar to `include_file_metric_tag`)
- Allows kubernetes_logs to inject custom labels without affecting other sources
- Minimal code changes

## Implementation Plan

### Phase 1: Modify FileSourceInternalEventsEmitter Struct

**File**: `src/internal_events/file.rs`

1. Add optional label provider to `FileSourceInternalEventsEmitter` struct (around line 530):
   ```rust
   use std::sync::Arc;
   
   pub struct FileSourceInternalEventsEmitter {
       pub include_file_metric_tag: bool,
       pub label_provider: Option<Arc<dyn Fn(&str) -> Vec<(&'static str, String)> + Send + Sync>>,
   }
   ```

   The label provider:
   - Takes a file path string reference
   - Returns a vector of label key-value pairs
   - Must be Send + Sync for thread safety
   - Wrapped in Arc (not Box) to enable Clone
   - Arc is Clone, satisfying the FileSourceInternalEvents trait requirement

### Phase 2: Modify FileLineTooBigError Emission

**File**: `src/internal_events/file.rs`

Update the `emit_file_line_too_long` implementation (around lines 612-624) to:

1. Accept an additional `file_path` parameter:
   ```rust
   fn emit_file_line_too_long(
       &self,
       truncated_bytes: &BytesMut,
       configured_limit: usize,
       encountered_size_so_far: usize,
       file_path: &str,  // NEW
   )
   ```

2. In the emit implementation (around line 515-521), conditionally add custom labels:
   
   ```rust
   // If label provider exists, get additional labels and emit with them
   if let Some(ref provider) = self.label_provider {
       let custom_labels = provider(file_path);
       
       // Extract pod labels (up to 3: namespace, name, container)
       // Handle case where provider returns partial labels
       let namespace = custom_labels.iter().find(|(k, _)| *k == "pod_namespace").map(|(_, v)| v.as_str()).unwrap_or("");
       let pod = custom_labels.iter().find(|(k, _)| *k == "pod_name").map(|(_, v)| v.as_str()).unwrap_or("");
       let container = custom_labels.iter().find(|(k, _)| *k == "container_name").map(|(_, v)| v.as_str()).unwrap_or("");
       
       counter!(
           "component_errors_total",
           "error_code" => "reading_line_from_file",
           "error_type" => error_type::CONDITION_FAILED,
           "stage" => error_stage::RECEIVING,
           "pod_namespace" => namespace,
           "pod_name" => pod,
           "container_name" => container,
       )
       .increment(1);
   } else {
       // Original behavior without custom labels
       counter!(
           "component_errors_total",
           "error_code" => "reading_line_from_file",
           "error_type" => error_type::CONDITION_FAILED,
           "stage" => error_stage::RECEIVING,
       )
       .increment(1);
   }
   ```
   
   This follows the same pattern as `KubernetesLogsEventsReceived` which conditionally adds pod labels (see `src/internal_events/kubernetes_logs.rs` lines 41-54).

### Phase 3: Update Trait Method Signature

**File**: `lib/file-source-common/src/internal_events.rs`

Update the trait method signature (line 32-37) to include file path:
```rust
fn emit_file_line_too_long(
    &self,
    truncated_bytes: &BytesMut,
    configured_limit: usize,
    encountered_size_so_far: usize,
    file_path: &str,  // NEW
);
```

This is a **breaking change** but necessary to provide the file path context.

### Phase 4: Update Call Site in FileServer

**File**: `lib/file-source/src/file_server.rs`

Update the call to `emit_file_line_too_long` (around line 297-303) to pass the watcher path:
```rust
discarded_for_size_and_truncated.iter().for_each(|buf| {
    self.emitter.emit_file_line_too_long(
        &buf.clone(),
        self.max_line_bytes,
        buf.len(),
        watcher.path.to_str().unwrap_or("unknown"),  // NEW
    )
});
```

### Phase 5: Create Label Provider in kubernetes_logs

**File**: `src/sources/kubernetes_logs/mod.rs`

1. Import the path parser (add to existing imports around line 56):
   ```rust
   use crate::sources::kubernetes_logs::path_helpers::parse_log_file_path;
   ```

2. Create a label provider function (add as a helper function around line 1150):
   ```rust
   use std::sync::Arc;
   
   fn create_k8s_label_provider() -> Arc<dyn Fn(&str) -> Vec<(&'static str, String)> + Send + Sync> {
       Arc::new(|file_path: &str| {
           if let Some(file_info) = parse_log_file_path(file_path) {
               vec![
                   ("pod_namespace", file_info.pod_namespace.to_string()),
                   ("pod_name", file_info.pod_name.to_string()),
                   ("container_name", file_info.container_name.to_string()),
               ]
           } else {
               vec![]
           }
       })
   }
   ```

3. Update the emitter creation (around line 868-870):
   ```rust
   emitter: FileSourceInternalEventsEmitter {
       include_file_metric_tag,
       label_provider: Some(create_k8s_label_provider()),
   },
   ```

### Phase 6: Update Other File Sources

**Files**: Any other file-based sources that use `FileSourceInternalEventsEmitter`

Search for other uses:
```bash
grep -r "FileSourceInternalEventsEmitter" src/sources/
```

Update each source to:
1. Pass `None` for `label_provider` (maintaining backward compatibility)
2. Update their `emit_file_line_too_long` calls to pass file path

Common sources to check:
- `src/sources/file.rs` (standard file source)
- Any other sources implementing the file-source pattern

### Phase 7: Update FileLineTooBigError Event Struct (Optional Enhancement)

**File**: `src/internal_events/file.rs`

Consider adding the file_path to the error log output (around line 507-513):
```rust
error!(
    message = "Found line that exceeds max_line_bytes; discarding.",
    truncated_bytes = ?self.truncated_bytes,
    configured_limit = self.configured_limit,
    encountered_size_so_far = self.encountered_size_so_far,
    file_path = %file_path,  // NEW - for correlation with metrics
    error_type = error_type::CONDITION_FAILED,
    stage = error_stage::RECEIVING,
);
```

## Critical Files to Modify

1. `lib/file-source-common/src/internal_events.rs` - Trait definition
2. `src/internal_events/file.rs` - Event struct and emitter implementation
3. `lib/file-source/src/file_server.rs` - Call site
4. `src/sources/kubernetes_logs/mod.rs` - Label provider creation
5. `src/sources/file.rs` (and others) - Other file sources using the trait

## Reusable Components

- `parse_log_file_path()` from `src/sources/kubernetes_logs/path_helpers.rs` (lines 36-55)
- `LogFileInfo` struct from `src/sources/kubernetes_logs/path_helpers.rs` (lines 57-64)
- Existing conditional label pattern from `FileBytesSent::emit()` in `src/internal_events/file.rs` (lines 48-70)

## Testing & Verification

### Unit Tests

1. **Test label provider in kubernetes_logs**:
   - Test that `create_k8s_label_provider()` correctly parses valid pod paths
   - Test that it returns empty vec for invalid paths
   - Add test in `src/sources/kubernetes_logs/mod.rs` or new test file

2. **Test FileLineTooBigError with custom labels**:
   - Mock emitter with label_provider
   - Verify labels are added to counter when provider is Some
   - Verify labels are not added when provider is None
   - Add test in `src/internal_events/file.rs`

### Integration Test

1. Configure kubernetes_logs source with `max_line_bytes = 1024`
2. Generate log line > 1024 bytes from a test pod
3. Query Prometheus for `vector_component_errors_total`
4. Verify metric includes:
   - `error_code="reading_line_from_file"`
   - `pod_namespace="<test-namespace>"`
   - `pod_name="<test-pod>"`
   - `container_name="<test-container>"`

### Manual Verification

1. Deploy Vector with kubernetes_logs source to OpenShift cluster
2. Create a pod that emits very long log lines
3. Check Vector's internal metrics endpoint (`/metrics`)
4. Verify the metric appears with pod labels:
   ```
   vector_component_errors_total{
     error_code="reading_line_from_file",
     pod_namespace="openshift-logging",
     pod_name="test-pod-xyz",
     container_name="app",
     ...
   } 1
   ```

### PromQL Query for Verification

```promql
vector_component_errors_total{
  error_code="reading_line_from_file",
  pod_namespace!=""
}
```

## Backward Compatibility Considerations

1. **Breaking change**: The trait method signature change (`emit_file_line_too_long` now requires `file_path`)
   - All implementations must be updated
   - This is unavoidable to provide the context

2. **Non-breaking**: The `label_provider` field is optional
   - Sources that don't need custom labels pass `None`
   - No behavior change for sources without a provider

3. **Metric cardinality**: Adding pod labels increases cardinality
   - Only affects kubernetes_logs source
   - Only creates series for pods with oversized lines
   - Series expire after `expire_metrics_secs` (default 300s)
   - This is acceptable and provides valuable diagnostic data

## Alternatives Considered

### Alternative 1: Pass file path in FileLineTooBigError struct directly (Not Recommended)
- Would require changing the event struct, not just the emitter
- Less flexible - hardcodes kubernetes-specific behavior into generic file event
- Harder to extend for future use cases

### Alternative 2: Simpler - Just add file_path parameter, parse inline (User may prefer)
Instead of a function pointer, simply:
1. Add `file_path: &str` parameter to trait method
2. In kubernetes_logs' `FileSourceInternalEventsEmitter::emit_file_line_too_long()`, directly call `parse_log_file_path(file_path)` and add labels if successful
3. In regular file source, the labels just won't be added (path won't parse)

**Pros**:
- Simpler implementation
- No function pointers/Arc needed
- Same behavior (kubernetes paths parse, others don't)

**Cons**:
- Couples kubernetes-specific logic to the emitter implementation
- Other sources see kubernetes-specific code even if not using it

The user mentioned wanting a "function pointer" approach, so the chosen plan uses that. However, the simpler alternative could be offered if they prefer less abstraction.

The chosen approach (optional label provider) is more flexible and follows the dependency injection pattern, allowing different sources to customize labels as needed without polluting the generic file-source layer.