K8s-ingestion

README.md

@@ -45,6 +45,7 @@ Here are some references to get you started:
 ### 🎯 Real-Time Analysis
 
 - **Live streaming** - Process logs as they arrive from stdin, files, or network
+- **Kubernetes native** - Direct integration with Kubernetes clusters for pod log streaming
 - **OTLP native** - First-class support for OpenTelemetry log format
 - **OTLP receiver** - Built-in gRPC server to receive logs via OpenTelemetry protocol
 - **Format detection** - Automatically detects JSON, logfmt, and plain text
@@ -68,6 +69,7 @@ Here are some references to get you started:
 - **Regex support** - Filter logs with regular expressions
 - **Attribute search** - Find logs by specific attribute values
 - **Severity filtering** - Interactive modal to select specific log levels (Ctrl+f)
+- **Kubernetes filtering** - Filter by namespace and pod with interactive selection (Ctrl+k)
 - **Multi-level selection** - Enable/disable multiple severity levels at once
 - **Interactive selection** - Click or keyboard navigate to explore logs
 
@@ -144,7 +146,12 @@ gonzo -f "/var/log/*.log" --follow
 # Analyze logs from stdin (traditional way)
 cat application.log | gonzo
 
-# Stream logs from kubectl
+# Stream logs directly from Kubernetes clusters
+gonzo --k8s-enabled=true --k8s-namespace=default
+gonzo --k8s-enabled=true --k8s-namespace=production --k8s-namespace=staging
+gonzo --k8s-enabled=true --k8s-selector="app=my-app"
+
+# Stream logs from kubectl (traditional way)
 kubectl logs -f deployment/my-app | gonzo
 
 # Follow system logs
@@ -311,8 +318,9 @@ cat logs.json | gonzo --ai-model="gpt-4"
 | `/`            | Enter filter mode (regex supported)       |
 | `s`            | Search and highlight text in logs         |
 | `Ctrl+f`       | Open severity filter modal                |
+| `Ctrl+k`       | Open Kubernetes filter modal (k8s mode)   |
 | `f`            | Open fullscreen log viewer modal          |
-| `c`            | Toggle Host/Service columns in log view   |
+| `c`            | Toggle Namespace/Pod or Host/Service cols |
 | `r`            | Reset all data (manual reset)             |
 | `u` / `U`      | Cycle update intervals (forward/backward) |
 | `i`            | AI analysis (in detail view)              |
@@ -415,6 +423,16 @@ Flags:
   --ai-model string                AI model for analysis (auto-selects best available if not specified)
   -s, --skin string                Color scheme/skin to use (default, or name of a skin file)
   --stop-words strings             Additional stop words to filter out from analysis (adds to built-in list)
+
+Kubernetes Flags:
+  --k8s-enabled=true               Enable Kubernetes log streaming mode
+  --k8s-namespace stringArray      Kubernetes namespace(s) to watch (can specify multiple, default: all)
+  --k8s-selector string            Kubernetes label selector for filtering pods
+  --k8s-tail int                   Number of previous log lines to retrieve (default: 10)
+  --k8s-since int                  Only return logs newer than relative duration in seconds
+  --k8s-kubeconfig string          Path to kubeconfig file (default: $HOME/.kube/config)
+  --k8s-context string             Kubernetes context to use
+
   -t, --test-mode                  Run without TTY for testing
   -v, --version                    Print version information
   --config string                  Config file (default: $HOME/.config/gonzo/config.yml)
@@ -669,7 +687,7 @@ internal/
 
 ### Prerequisites
 
-- Go 1.21 or higher
+- Go 1.24 or higher
 - Make (optional, for convenience)
 
 ### Building
@@ -768,6 +786,7 @@ This project is licensed under the MIT License - see the [LICENSE](LICENSE) file
 
 - [Docs](https://docs.controltheory.com/) - Complete user guide, integration examples, advanced features (AI, OTel, custom log formats)
 - [Usage Guide](USAGE_GUIDE.md) - Detailed usage instructions
+- [Kubernetes Integration Guide](guides/KUBERNETES_USAGE.md) - Direct Kubernetes cluster integration, filtering, and usage examples
 - [AWS CloudWatch Logs Usage Guide](guides/CLOUDWATCH_USAGE_GUIDE.md) - Usage instructions for AWS CLI log tail and live tail with Gonzo
 - [Stern Usage Guide](guides/STERN_USAGE_GUIDE.md) - Usage and examples for using Stern with Gonzo
 - [Victoria Logs Integration](guides/VICTORIA_LOGS_USAGE.md) - Using Gonzo with Victoria Logs API

cmd/gonzo/app.go

@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"context"
 	"fmt"
+	"io"
 	"log"
 	"os"
 	"strings"
@@ -12,6 +13,7 @@ import (
 	"github.com/control-theory/gonzo/internal/analyzer"
 	"github.com/control-theory/gonzo/internal/filereader"
 	"github.com/control-theory/gonzo/internal/formats"
+	"github.com/control-theory/gonzo/internal/k8s"
 	"github.com/control-theory/gonzo/internal/memory"
 	"github.com/control-theory/gonzo/internal/otlplog"
 	"github.com/control-theory/gonzo/internal/otlpreceiver"
@@ -21,6 +23,7 @@ import (
 
 	tea "github.com/charmbracelet/bubbletea"
 	"github.com/spf13/cobra"
+	"k8s.io/klog/v2"
 )
 
 // runApp initializes and runs the application
@@ -31,6 +34,15 @@ func runApp(cmd *cobra.Command, args []string) error {
 		return nil
 	}
 
+	// Redirect log output to discard to avoid messing up the TUI
+	// All log.Printf calls will be silently discarded
+	log.SetOutput(io.Discard)
+
+	// Suppress klog output from Kubernetes client-go
+	// This prevents errors like "request.go:752" from appearing in the TUI
+	klog.SetOutput(io.Discard)
+	klog.LogToStderr(false)
+
 	// Start version checking in background (if not disabled)
 	var versionChecker *versioncheck.Checker
 	if !cfg.DisableVersionCheck {
@@ -181,6 +193,10 @@ type simpleTuiModel struct {
 	vmlogsReceiver *vmlogs.Receiver // Victoria Logs receiver for streaming logs
 	hasVmlogsInput bool             // Whether we're receiving Victoria Logs data
 
+	// Kubernetes receiver support
+	k8sReceiver *k8s.KubernetesLogSource // Kubernetes log source for streaming pod logs
+	hasK8sInput bool                     // Whether we're receiving Kubernetes logs
+
 	// JSON accumulation for multi-line OTLP support
 	jsonBuffer   strings.Builder // Buffer for accumulating multi-line JSON
 	jsonDepth    int             // Track JSON object/array nesting depth
@@ -195,8 +211,45 @@ func (m *simpleTuiModel) Init() tea.Cmd {
 	// Initialize frequency reset timer
 	m.lastFreqReset = time.Now()
 
-	// Check if Victoria Logs receiver is enabled
-	if cfg.VmlogsURL != "" {
+	// Check if Kubernetes receiver is enabled
+	if cfg.K8sEnabled {
+		// Kubernetes input mode
+		m.hasK8sInput = true
+		m.inputChan = make(chan string, 100)
+
+		// Create kubernetes config
+		k8sConfig := &k8s.Config{
+			Kubeconfig: cfg.K8sKubeconfig,
+			Context:    cfg.K8sContext,
+			Namespaces: cfg.K8sNamespaces,
+			Selector:   cfg.K8sSelector,
+			Since:      cfg.K8sSince,
+			TailLines:  cfg.K8sTailLines,
+		}
+
+		// Create and start Kubernetes log source
+		k8sSource, err := k8s.NewKubernetesLogSource(k8sConfig)
+		if err != nil {
+			log.Printf("Error creating Kubernetes log source: %v", err)
+			// Fall back to other input methods if Kubernetes fails
+			m.hasK8sInput = false
+		} else {
+			m.k8sReceiver = k8sSource
+			if err := m.k8sReceiver.Start(); err != nil {
+				log.Printf("Error starting Kubernetes log source: %v", err)
+				// Fall back to other input methods if Kubernetes fails
+				m.hasK8sInput = false
+			} else {
+				// Wire K8s source to the dashboard for namespace/pod listing
+				m.dashboard.SetK8sSource(k8sSource)
+				// Start reading from Kubernetes receiver in the background
+				go m.readK8sAsync()
+			}
+		}
+	}
+
+	// Check if Victoria Logs receiver is enabled (only if Kubernetes is not enabled)
+	if !m.hasK8sInput && cfg.VmlogsURL != "" {
 		// Victoria Logs input mode
 		m.hasVmlogsInput = true
 		m.inputChan = make(chan string, 100)
@@ -215,8 +268,8 @@ func (m *simpleTuiModel) Init() tea.Cmd {
 		}
 	}
 
-	// Check if OTLP receiver is enabled (only if Victoria Logs is not enabled)
-	if !m.hasVmlogsInput && cfg.OTLPEnabled {
+	// Check if OTLP receiver is enabled (only if Kubernetes and Victoria Logs are not enabled)
+	if !m.hasK8sInput && !m.hasVmlogsInput && cfg.OTLPEnabled {
 		// OTLP input mode
 		m.hasOTLPInput = true
 		m.inputChan = make(chan string, 100)
@@ -233,8 +286,8 @@ func (m *simpleTuiModel) Init() tea.Cmd {
 		}
 	}
 
-	// Check if we have file inputs specified (only if Victoria Logs and OTLP are not enabled)
-	if !m.hasVmlogsInput && !m.hasOTLPInput && len(cfg.Files) > 0 {
+	// Check if we have file inputs specified (only if Kubernetes, Victoria Logs and OTLP are not enabled)
+	if !m.hasK8sInput && !m.hasVmlogsInput && !m.hasOTLPInput && len(cfg.Files) > 0 {
 		// File input mode
 		m.hasFileInput = true
 		m.inputChan = make(chan string, 100)
@@ -252,8 +305,8 @@ func (m *simpleTuiModel) Init() tea.Cmd {
 		}
 	}
 
-	// If no Victoria Logs, no OTLP, no file input or file input failed, check stdin
-	if !m.hasVmlogsInput && !m.hasOTLPInput && !m.hasFileInput {
+	// If no Kubernetes, no Victoria Logs, no OTLP, no file input or file input failed, check stdin
+	if !m.hasK8sInput && !m.hasVmlogsInput && !m.hasOTLPInput && !m.hasFileInput {
 		// Check if stdin has data available (not a terminal)
 		stat, _ := os.Stdin.Stat()
 		if (stat.Mode() & os.ModeCharDevice) == 0 {
@@ -274,13 +327,46 @@ func (m *simpleTuiModel) Init() tea.Cmd {
 	cmds = append(cmds, m.periodicUpdate())
 
 	// Start checking for input data if we have any input source
-	if m.hasStdinData || m.hasFileInput || m.hasOTLPInput || m.hasVmlogsInput {
+	if m.hasStdinData || m.hasFileInput || m.hasOTLPInput || m.hasVmlogsInput || m.hasK8sInput {
 		cmds = append(cmds, m.checkInputChannel())
 	}
 
 	return tea.Batch(cmds...)
 }
 
+// readK8sAsync reads from the Kubernetes log source
+func (m *simpleTuiModel) readK8sAsync() {
+	defer close(m.inputChan)
+
+	if m.k8sReceiver == nil {
+		return
+	}
+
+	// Get the channel from Kubernetes receiver
+	k8sLineChan := m.k8sReceiver.GetLineChan()
+
+	// Forward lines from Kubernetes receiver to input channel
+	for {
+		select {
+		case <-m.ctx.Done():
+			m.k8sReceiver.Stop()
+			return
+		case line, ok := <-k8sLineChan:
+			if !ok {
+				// Kubernetes receiver finished
+				return
+			}
+			if line != "" {
+				select {
+				case m.inputChan <- line:
+				case <-m.ctx.Done():
+					return
+				}
+			}
+		}
+	}
+}
+
 // readVmlogsAsync reads from the Victoria Logs receiver
 func (m *simpleTuiModel) readVmlogsAsync() {
 	defer close(m.inputChan)
@@ -495,7 +581,7 @@ func (m *simpleTuiModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
 		m.processLogLine(string(msg))
 
 		// Continue checking for more data if we have input sources
-		if (m.hasStdinData || m.hasFileInput || m.hasOTLPInput || m.hasVmlogsInput) && !m.finished {
+		if (m.hasStdinData || m.hasFileInput || m.hasOTLPInput || m.hasVmlogsInput || m.hasK8sInput) && !m.finished {
 			cmds = append(cmds, m.checkInputChannel())
 		}
 

cmd/gonzo/main.go

@@ -41,6 +41,13 @@ type Config struct {
 	VmlogsUser           string        `mapstructure:"vmlogs-user"`
 	VmlogsPassword       string        `mapstructure:"vmlogs-password"`
 	VmlogsQuery          string        `mapstructure:"vmlogs-query"`
+	K8sEnabled           bool          `mapstructure:"k8s-enabled"`
+	K8sKubeconfig        string        `mapstructure:"k8s-kubeconfig"`
+	K8sContext           string        `mapstructure:"k8s-context"`
+	K8sNamespaces        []string      `mapstructure:"k8s-namespaces"`
+	K8sSelector          string        `mapstructure:"k8s-selector"`
+	K8sSince             int64         `mapstructure:"k8s-since"`
+	K8sTailLines         int64         `mapstructure:"k8s-tail-lines"`
 	Skin                 string        `mapstructure:"skin"`
 	StopWords            []string      `mapstructure:"stop-words"`
 	Format               string        `mapstructure:"format"`
@@ -71,9 +78,18 @@ Supports OTLP (OpenTelemetry) format natively, with automatic detection of JSON,
   # Use glob patterns to read multiple files
   gonzo -f "/var/log/*.log" --follow
 
-  # Stream logs from kubectl  
+  # Stream logs from kubectl
   kubectl logs -f deployment/my-app | gonzo
-
+
+  # Stream logs directly from Kubernetes
+  gonzo --k8s-enabled
+
+  # Stream logs from specific namespaces
+  gonzo --k8s-enabled --k8s-namespaces=default,production
+
+  # Stream logs with label selector
+  gonzo --k8s-enabled --k8s-selector="app=myapp,env=prod"
+
   # With custom settings
   gonzo -f logs.json --update-interval=2s --log-buffer=2000
 
@@ -153,6 +169,13 @@ func init() {
 	rootCmd.Flags().String("vmlogs-user", "", "Victoria Logs basic auth username (can also use GONZO_VMLOGS_USER env var)")
 	rootCmd.Flags().String("vmlogs-password", "", "Victoria Logs basic auth password (can also use GONZO_VMLOGS_PASSWORD env var)")
 	rootCmd.Flags().String("vmlogs-query", "*", "Victoria Logs query (LogsQL) to use for streaming (default: '*' for all logs)")
+	rootCmd.Flags().Bool("k8s-enabled", false, "Enable Kubernetes log streaming from pods")
+	rootCmd.Flags().String("k8s-kubeconfig", "", "Path to kubeconfig file (default: $KUBECONFIG or ~/.kube/config)")
+	rootCmd.Flags().String("k8s-context", "", "Kubernetes context to use (default: current context)")
+	rootCmd.Flags().StringSlice("k8s-namespaces", []string{}, "Kubernetes namespaces to watch (default: all namespaces)")
+	rootCmd.Flags().String("k8s-selector", "", "Label selector to filter pods (e.g., 'app=myapp,env=prod')")
+	rootCmd.Flags().Int64("k8s-since", 0, "Only show logs newer than this many seconds (default: 0 = all)")
+	rootCmd.Flags().Int64("k8s-tail-lines", 10, "Lines of recent logs to show initially per pod (default: 10, use -1 for all)")
 	rootCmd.Flags().StringP("skin", "s", "default", "Color scheme/skin to use (default, or name of a skin file in ~/.config/gonzo/skins/)")
 	rootCmd.Flags().StringSlice("stop-words", []string{}, "Additional stop words to filter out from analysis (adds to built-in list)")
 	rootCmd.Flags().String("format", "", "Log format to use (auto-detect if not specified). Can be: otlp, json, text, or a custom format name from ~/.config/gonzo/formats/")
@@ -174,6 +197,13 @@ func init() {
 	viper.BindPFlag("vmlogs-user", rootCmd.Flags().Lookup("vmlogs-user"))
 	viper.BindPFlag("vmlogs-password", rootCmd.Flags().Lookup("vmlogs-password"))
 	viper.BindPFlag("vmlogs-query", rootCmd.Flags().Lookup("vmlogs-query"))
+	viper.BindPFlag("k8s-enabled", rootCmd.Flags().Lookup("k8s-enabled"))
+	viper.BindPFlag("k8s-kubeconfig", rootCmd.Flags().Lookup("k8s-kubeconfig"))
+	viper.BindPFlag("k8s-context", rootCmd.Flags().Lookup("k8s-context"))
+	viper.BindPFlag("k8s-namespaces", rootCmd.Flags().Lookup("k8s-namespaces"))
+	viper.BindPFlag("k8s-selector", rootCmd.Flags().Lookup("k8s-selector"))
+	viper.BindPFlag("k8s-since", rootCmd.Flags().Lookup("k8s-since"))
+	viper.BindPFlag("k8s-tail-lines", rootCmd.Flags().Lookup("k8s-tail-lines"))
 	viper.BindPFlag("skin", rootCmd.Flags().Lookup("skin"))
 	viper.BindPFlag("stop-words", rootCmd.Flags().Lookup("stop-words"))
 	viper.BindPFlag("format", rootCmd.Flags().Lookup("format"))