feat: Change client version based on application version, bulk index and generation efficiencies

.github/workflows/build.yml

@@ -4,7 +4,6 @@ on:
   push:
     tags:
       - 'v*'
-  pull_request:
 
 jobs:
   build:

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

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=

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,26 +39,96 @@ 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 {
 		esConfig.Username = cfg.Username
 		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
+					}
 				}
 			}
 		}

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 {