From 4ac4fd57446df1422761c35c4f476331dd68f492 Mon Sep 17 00:00:00 2001 From: tehbooom Date: Tue, 14 Oct 2025 09:34:15 -0400 Subject: [PATCH 1/4] fix: Change client version based on application version --- go.mod | 1 + go.sum | 2 + internal/elasticsearch/elasticsearch.go | 106 +++++++++++++++++++++--- internal/kibana/kibana.go | 3 + 4 files changed, 102 insertions(+), 10 deletions(-) diff --git a/go.mod b/go.mod index 00a4352..e49e38d 100644 --- a/go.mod +++ b/go.mod @@ -9,6 +9,7 @@ require ( github.com/charmbracelet/lipgloss v1.1.1-0.20250404203927-76690c660834 github.com/charmbracelet/log v0.4.1 github.com/elastic/beats/v7 v7.17.28 + github.com/elastic/go-elasticsearch/v8 v8.17.0 github.com/elastic/go-elasticsearch/v9 v9.0.0 github.com/muesli/termenv v0.16.0 github.com/spf13/cobra v1.9.1 diff --git a/go.sum b/go.sum index d8aa363..c934009 100644 --- a/go.sum +++ b/go.sum @@ -54,6 +54,8 @@ github.com/elastic/elastic-agent-libs v0.20.0 h1:MPjenwuEr+QfMeQRV4BK817ZbiNS38S github.com/elastic/elastic-agent-libs v0.20.0/go.mod h1:1HNxREH8C27kGrJCtKZh/ot8pV8joH8VREP21+FrH5s= github.com/elastic/elastic-transport-go/v8 v8.7.0 h1:OgTneVuXP2uip4BA658Xi6Hfw+PeIOod2rY3GVMGoVE= github.com/elastic/elastic-transport-go/v8 v8.7.0/go.mod h1:YLHer5cj0csTzNFXoNQ8qhtGY1GTvSqPnKWKaqQE3Hk= +github.com/elastic/go-elasticsearch/v8 v8.17.0 h1:e9cWksE/Fr7urDRmGPGp47Nsp4/mvNOrU8As1l2HQQ0= +github.com/elastic/go-elasticsearch/v8 v8.17.0/go.mod h1:lGMlgKIbYoRvay3xWBeKahAiJOgmFDsjZC39nmO3H64= github.com/elastic/go-elasticsearch/v9 v9.0.0 h1:krpgPeJ2lC8apkaw6B58gKDYJq5eUhP8AMwpPt01Q/U= github.com/elastic/go-elasticsearch/v9 v9.0.0/go.mod h1:2PB5YQPpY5tWbF65MRqzEXA31PZOdXCkloQSOZtU14I= github.com/elastic/go-licenser v0.3.1/go.mod h1:D8eNQk70FOCVBl3smCGQt/lv7meBeQno2eI1S5apiHQ= diff --git a/internal/elasticsearch/elasticsearch.go b/internal/elasticsearch/elasticsearch.go index 8be8380..4512d07 100644 --- a/internal/elasticsearch/elasticsearch.go +++ b/internal/elasticsearch/elasticsearch.go @@ -6,11 +6,12 @@ import ( "fmt" "net/http" "os" + "strings" "time" "github.com/charmbracelet/log" - "github.com/elastic/go-elasticsearch/v9" - "github.com/elastic/go-elasticsearch/v9/typedapi/types" + "github.com/elastic/go-elasticsearch/v8" + "github.com/elastic/go-elasticsearch/v8/typedapi/types" "github.com/tehbooom/elastic-data/internal/config" ) @@ -38,11 +39,72 @@ func (c *Config) TestConnection() error { return nil } +// detectElasticsearchVersion makes a simple request to detect the ES version +func detectElasticsearchVersion(cfg config.ConfigConnection) (string, error) { + tempConfig := elasticsearch.Config{ + Addresses: cfg.ElasticsearchEndpoints, + Header: http.Header{ + "Content-Type": []string{"application/json"}, + "Accept": []string{"application/json"}, + }, + } + + if cfg.APIKey != "" { + tempConfig.APIKey = cfg.APIKey + } else { + tempConfig.Username = cfg.Username + tempConfig.Password = cfg.Password + } + + if cfg.Unsafe { + tempConfig.Transport = &http.Transport{ + TLSClientConfig: &tls.Config{ + InsecureSkipVerify: true, + }, + } + } + + if cfg.CACert != "" { + cert, err := os.ReadFile(cfg.CACert) + if err != nil { + return "", err + } + tempConfig.CACert = cert + } + + tempClient, err := elasticsearch.NewTypedClient(tempConfig) + if err != nil { + return "", err + } + + resp, err := tempClient.Core.Info().Do(context.Background()) + if err != nil { + return "", err + } + + return resp.Version.Int, nil +} + func SetClient(cfg config.ConfigConnection) (*elasticsearch.TypedClient, error) { esConfig := elasticsearch.Config{ Addresses: cfg.ElasticsearchEndpoints, } + version, err := detectElasticsearchVersion(cfg) + if err != nil { + log.Warn("Could not detect Elasticsearch version, using default headers", "error", err) + } else { + log.Info("Detected Elasticsearch version", "version", version) + + // If version is 8.x, set compatible headers + if strings.HasPrefix(version, "8.") { + esConfig.Header = http.Header{ + "Content-Type": []string{"application/json"}, + "Accept": []string{"application/json"}, + } + } + } + if cfg.APIKey != "" { esConfig.APIKey = cfg.APIKey } else { @@ -50,14 +112,23 @@ func SetClient(cfg config.ConfigConnection) (*elasticsearch.TypedClient, error) esConfig.Password = cfg.Password } + // Configure HTTP transport with optimized settings for high throughput + transport := &http.Transport{ + MaxIdleConns: 100, // Increased connection pool + MaxIdleConnsPerHost: 100, // Allow more connections per host + IdleConnTimeout: 90 * time.Second, // Keep connections alive longer + DisableCompression: false, // Enable compression + DisableKeepAlives: false, // Enable keep-alives + } + if cfg.Unsafe { - esConfig.Transport = &http.Transport{ - TLSClientConfig: &tls.Config{ - InsecureSkipVerify: true, - }, + transport.TLSClientConfig = &tls.Config{ + InsecureSkipVerify: true, } } + esConfig.Transport = transport + if cfg.CACert != "" { cert, err := os.ReadFile(cfg.CACert) if err != nil { @@ -77,29 +148,44 @@ func SetClient(cfg config.ConfigConnection) (*elasticsearch.TypedClient, error) } func (c *Config) BulkRequest(index string, events []map[string]interface{}) (time.Duration, error) { - var duration time.Duration + if len(events) == 0 { + return 0, nil + } + bulk := c.Client.Bulk().Index(index) + + // Batch add operations for better performance for _, event := range events { err := bulk.CreateOp(types.CreateOperation{}, event) if err != nil { log.Debug(err) - return duration, err + return 0, err } } start := time.Now() resp, err := bulk.Do(c.Ctx) if err != nil { - return duration, err + return 0, err } - duration = time.Since(start) + duration := time.Since(start) + // Only log first few errors to avoid performance impact if resp.Errors { + errorCount := 0 for _, item := range resp.Items { + if errorCount >= 5 { + log.Printf("... and more errors (suppressed for performance)") + break + } for opType, respItem := range item { if respItem.Error != nil { log.Printf("Error in %s operation for document %s: %s", opType, *respItem.Id_, *respItem.Error.Reason) + errorCount++ + if errorCount >= 5 { + break + } } } } diff --git a/internal/kibana/kibana.go b/internal/kibana/kibana.go index 8501330..281edc8 100644 --- a/internal/kibana/kibana.go +++ b/internal/kibana/kibana.go @@ -80,6 +80,9 @@ func (c *Config) InstallPackage(pkgName string) error { Params: kbapi.FleetEPMInstallPackageRegistryRequestParams{ Prerelease: kbapi.BoolPtr(true), }, + Body: kbapi.FleetEPMInstallPackageRegistryRequestBody{ + Force: kbapi.BoolPtr(false), + }, }) if err != nil { From 55f3bff641b400ae71d10418da1b480264fc6219 Mon Sep 17 00:00:00 2001 From: tehbooom Date: Tue, 14 Oct 2025 09:34:39 -0400 Subject: [PATCH 2/4] feat: More effecient generation and bulk requests --- ui/tabs/run/data_generator.go | 95 +++++++++++++++++++++++++---------- ui/ui.go | 2 +- 2 files changed, 70 insertions(+), 27 deletions(-) diff --git a/ui/tabs/run/data_generator.go b/ui/tabs/run/data_generator.go index 6841ee1..452b407 100644 --- a/ui/tabs/run/data_generator.go +++ b/ui/tabs/run/data_generator.go @@ -5,6 +5,7 @@ import ( "encoding/json" "fmt" "math/rand" + "strings" "sync" "time" @@ -62,13 +63,19 @@ func (dg *DataGenerator) startEPS() { batchSize := dg.calculateOptimalBatchSize() batchInterval := time.Duration(batchSize) * time.Second / time.Duration(targetEPS) - log.Debug(fmt.Sprintf("Batch interval set to %d for %s", batchInterval, dg.config.Name)) + log.Debug(fmt.Sprintf("Batch interval set to %v for %s", batchInterval, dg.config.Name)) ticker := time.NewTicker(batchInterval) defer ticker.Stop() log.Debug("Starting EPS generation for %s: %d EPS (batch size: %d, interval: %v)", dg.config.Name, targetEPS, batchSize, batchInterval) + // Send first batch immediately instead of waiting for first tick + if err := dg.sendEPS(); err != nil { + log.Debug(err) + log.Debug("Error sending initial EPS batch for %s: %v", dg.config.Name, err) + } + for { select { case <-dg.ctx.Done(): @@ -84,14 +91,19 @@ func (dg *DataGenerator) startEPS() { } func (dg *DataGenerator) sendEPS() error { - dg.mu.Lock() - defer dg.mu.Unlock() - + // Get batch configuration without holding lock batchSize := dg.calculateOptimalBatchSize() + + dg.mu.Lock() selectedTemplates := dg.selectTemplatesAdaptive(batchSize) + preserveOriginal := dg.config.PreserveEventOriginal + dg.mu.Unlock() - var events []map[string]interface{} + // Generate events outside of lock + events := make([]map[string]interface{}, 0, batchSize) var batchBytes int + timestamp := time.Now().UTC().Format(time.RFC3339) + timestampOverhead := 50 // Approximate overhead for @timestamp and other metadata for i := 0; i < batchSize; i++ { template := selectedTemplates[i%len(selectedTemplates)] @@ -103,21 +115,24 @@ func (dg *DataGenerator) sendEPS() error { return err } + messageLen := len(message) var event map[string]interface{} + if template.IsJSON { - if err := json.Unmarshal([]byte(message), &event); err != nil { + decoder := json.NewDecoder(strings.NewReader(message)) + if err := decoder.Decode(&event); err != nil { log.Debug("Failed to parse JSON message:", err) return err } - event["@timestamp"] = time.Now().UTC().Format(time.RFC3339) + event["@timestamp"] = timestamp } else { event = map[string]interface{}{ "message": message, - "@timestamp": time.Now().UTC().Format(time.RFC3339), + "@timestamp": timestamp, } } - if dg.config.PreserveEventOriginal { + if preserveOriginal { event["tags"] = []string{"preserve_original_event"} } @@ -125,7 +140,8 @@ func (dg *DataGenerator) sendEPS() error { events = append(events, event) - eventBytes := dg.calculateEventSize(event) + // Approximate size without re-marshaling + eventBytes := messageLen + timestampOverhead batchBytes += eventBytes } @@ -133,31 +149,42 @@ func (dg *DataGenerator) sendEPS() error { return nil } + // Send bulk request without lock duration, err := dg.sendBulkRequest(events) if err != nil { log.Debug(err) return err } + // Only lock for updating stats + dg.mu.Lock() dg.bytesSent += batchBytes dg.updateStats(len(events), duration) + dg.mu.Unlock() + return nil } func (dg *DataGenerator) sendBytes() error { + // Get current state and batch configuration with minimal lock time dg.mu.Lock() - defer dg.mu.Unlock() - batchSize := dg.calculateOptimalBatchSize() - log.Debug(fmt.Sprintf("Batch size is %d for %s", batchSize, dg.config.Name)) - + currentBytesSent := dg.bytesSent + threshold := dg.config.Threshold selectedTemplates := dg.selectTemplatesAdaptive(batchSize) + preserveOriginal := dg.config.PreserveEventOriginal + dg.mu.Unlock() + + log.Debug(fmt.Sprintf("Batch size is %d for %s", batchSize, dg.config.Name)) - var events []map[string]interface{} + // Generate events outside of lock + events := make([]map[string]interface{}, 0, batchSize) var batchBytes int + timestamp := time.Now().UTC().Format(time.RFC3339) + timestampOverhead := 50 // Approximate overhead for @timestamp and other metadata for i := 0; i < batchSize; i++ { - if dg.config.Unit == "bytes" && dg.bytesSent >= dg.config.Threshold { + if dg.config.Unit == "bytes" && currentBytesSent >= threshold { log.Debug(fmt.Sprintf("Threshold %d met for %s", batchSize, dg.config.Name)) break } @@ -170,27 +197,31 @@ func (dg *DataGenerator) sendBytes() error { return err } + messageLen := len(message) var event map[string]interface{} + if template.IsJSON { - if err := json.Unmarshal([]byte(message), &event); err != nil { + decoder := json.NewDecoder(strings.NewReader(message)) + if err := decoder.Decode(&event); err != nil { log.Debug("Failed to parse JSON message:", err) return err } - event["@timestamp"] = time.Now().UTC().Format(time.RFC3339) + event["@timestamp"] = timestamp } else { event = map[string]interface{}{ "message": message, - "@timestamp": time.Now().UTC().Format(time.RFC3339), + "@timestamp": timestamp, } } - if dg.config.PreserveEventOriginal { + if preserveOriginal { event["tags"] = []string{"preserve_original_event"} } - eventBytes := dg.calculateEventSize(event) + // Approximate size without re-marshaling + eventBytes := messageLen + timestampOverhead - if dg.config.Unit == "bytes" && (dg.bytesSent+batchBytes+eventBytes) > dg.config.Threshold { + if dg.config.Unit == "bytes" && (currentBytesSent+batchBytes+eventBytes) > threshold { break } @@ -202,14 +233,19 @@ func (dg *DataGenerator) sendBytes() error { return nil } + // Send bulk request without lock duration, err := dg.sendBulkRequest(events) if err != nil { log.Debug(err) return err } + // Only lock for updating stats + dg.mu.Lock() dg.bytesSent += batchBytes dg.updateStats(len(events), duration) + dg.mu.Unlock() + return nil } @@ -293,14 +329,19 @@ func (dg *DataGenerator) selectRandomTemplates(templates []*generator.LogTemplat func (dg *DataGenerator) calculateOptimalBatchSize() int { if dg.config.Unit == "eps" { target := dg.config.Threshold + // Larger batches for higher EPS to reduce overhead if target <= 10 { return 1 - } else if target <= 100 { + } else if target <= 50 { return 10 + } else if target <= 200 { + return 50 } else if target <= 1000 { - return 100 + return 200 + } else if target <= 5000 { + return 1000 } else { - return 500 + return 2000 } } else { remainingBytes := dg.config.Threshold - dg.bytesSent @@ -317,8 +358,10 @@ func (dg *DataGenerator) calculateOptimalBatchSize() int { return int(estimatedEvents) } else if estimatedEvents <= 100 { return 100 + } else if estimatedEvents <= 1000 { + return 1000 } else { - return 500 + return 2000 } } } diff --git a/ui/ui.go b/ui/ui.go index bcb32d6..af67e38 100644 --- a/ui/ui.go +++ b/ui/ui.go @@ -9,7 +9,7 @@ import ( tea "github.com/charmbracelet/bubbletea" "github.com/charmbracelet/lipgloss" "github.com/charmbracelet/log" - es "github.com/elastic/go-elasticsearch/v9" + es "github.com/elastic/go-elasticsearch/v8" "github.com/tehbooom/elastic-data/internal/config" "github.com/tehbooom/elastic-data/internal/elasticsearch" "github.com/tehbooom/elastic-data/internal/integrations" From 78fcb804e0e16da8b7b58245e013459915af2c09 Mon Sep 17 00:00:00 2001 From: tehbooom Date: Tue, 14 Oct 2025 09:40:48 -0400 Subject: [PATCH 3/4] fix: Remove calculateEventSize func --- ui/tabs/run/data_generator.go | 5 ----- 1 file changed, 5 deletions(-) diff --git a/ui/tabs/run/data_generator.go b/ui/tabs/run/data_generator.go index 452b407..660e20c 100644 --- a/ui/tabs/run/data_generator.go +++ b/ui/tabs/run/data_generator.go @@ -366,11 +366,6 @@ func (dg *DataGenerator) calculateOptimalBatchSize() int { } } -func (dg *DataGenerator) calculateEventSize(event map[string]interface{}) int { - jsonBytes, _ := json.Marshal(event) - return int(len(jsonBytes)) -} - func (dg *DataGenerator) sendBulkRequest(events []map[string]interface{}) (time.Duration, error) { index := "logs-" + dg.integrationName + "." + dg.config.Name + "-default" duration, err := dg.client.BulkRequest(index, events) From 3157b3c470bde7027b9b8ce72ecb67169e8b25cf Mon Sep 17 00:00:00 2001 From: tehbooom Date: Tue, 14 Oct 2025 09:46:11 -0400 Subject: [PATCH 4/4] ci: Build only on tags --- .github/workflows/build.yml | 1 - 1 file changed, 1 deletion(-) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index c40f57c..0863cd1 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -4,7 +4,6 @@ on: push: tags: - 'v*' - pull_request: jobs: build: