goosetav/K6_GO_IMPORTER_README.md

## docker-compose.yml
services:
  influxdb:
    image: influxdb:1.8
    container_name: k6-influxdb
    ports:
      - "8086:8086"
    environment:
      - INFLUXDB_DB=k6
      - INFLUXDB_ADMIN_USER=admin
      - INFLUXDB_ADMIN_PASSWORD=admin
      - INFLUXDB_HTTP_AUTH_ENABLED=true
    volumes:
      - influxdb-data:/var/lib/influxdb
    networks:
      - k6-network
    deploy:
      resources:
        limits:
          cpus: '8.0'
          memory: 8G
        reservations:
          cpus: '4.0'
          memory: 4G

  grafana:
    image: grafana/grafana:latest
    container_name: k6-grafana
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_USER=admin
      - GF_SECURITY_ADMIN_PASSWORD=admin
      - GF_AUTH_ANONYMOUS_ENABLED=true
      - GF_AUTH_ANONYMOUS_ORG_ROLE=Viewer
    volumes:
      - grafana-data:/var/lib/grafana
    depends_on:
      - influxdb
    networks:
      - k6-network
    deploy:
      resources:
        limits:
          cpus: '2.0'
          memory: 2G
        reservations:
          cpus: '0.5'
          memory: 512M

volumes:
  influxdb-data:
  grafana-data:

networks:
  k6-network:
    driver: bridge

## go.mod
module k6-to-influxdb

go 1.25.3

require (
	github.com/davecgh/go-spew v1.1.1 // indirect
	github.com/json-iterator/go v1.1.12 // indirect
	github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421 // indirect
	github.com/modern-go/reflect2 v1.0.2 // indirect
	github.com/pmezard/go-difflib v1.0.0 // indirect
	github.com/stretchr/objx v0.5.0 // indirect
	github.com/stretchr/testify v1.8.4 // indirect
	gopkg.in/yaml.v3 v3.0.1 // indirect
)

## importer.go
package main

import (
	"bufio"
	"bytes"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"os"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	jsoniter "github.com/json-iterator/go"
)

var json = jsoniter.ConfigCompatibleWithStandardLibrary

// K6Point represents a single k6 metric point from JSON
type K6Point struct {
	Type   string `json:"type"`
	Metric string `json:"metric"`
	Data   K6Data `json:"data"`
}

// K6Data represents the data field in a k6 point
type K6Data struct {
	Time  string             `json:"time"`
	Value float64            `json:"value"`
	Tags  map[string]string  `json:"tags"`
}

// Config holds the importer configuration
type Config struct {
	FilePath  string
	Host      string
	Port      int
	Protocol  string // "udp" or "http"
	Database  string
	Username  string
	Password  string
	BatchSize int
	Workers   int
	Verbose   bool
}

// Validate checks if the configuration is valid
func (c *Config) Validate() error {
	if c.FilePath == "" {
		return errors.New("file path is required")
	}
	if c.BatchSize <= 0 {
		return errors.New("batch size must be positive")
	}
	if c.Workers <= 0 {
		return errors.New("number of workers must be positive")
	}
	if c.Database == "" {
		return errors.New("database name is required")
	}
	return nil
}

// Stats tracks import statistics using atomic counters
type Stats struct {
	pointsProcessed atomic.Int64
	batchesWritten  atomic.Int64
	bytesRead       atomic.Int64
	writeTimeNs     atomic.Int64
	startTime       time.Time
}

// NewStats creates a new Stats instance
func NewStats() *Stats {
	return &Stats{
		startTime: time.Now(),
	}
}

// AddPoints increments the points processed counter
func (s *Stats) AddPoints(n int64) {
	s.pointsProcessed.Add(n)
}

// AddBatches increments the batches written counter
func (s *Stats) AddBatches(n int64) {
	s.batchesWritten.Add(n)
}

// AddBytes increments the bytes read counter
func (s *Stats) AddBytes(n int64) {
	s.bytesRead.Add(n)
}

// AddWriteTime adds to the total write time in nanoseconds
func (s *Stats) AddWriteTime(duration time.Duration) {
	s.writeTimeNs.Add(duration.Nanoseconds())
}

// PointsProcessed returns the current points processed count
func (s *Stats) PointsProcessed() int64 {
	return s.pointsProcessed.Load()
}

// BatchesWritten returns the current batches written count
func (s *Stats) BatchesWritten() int64 {
	return s.batchesWritten.Load()
}

// BytesRead returns the current bytes read count
func (s *Stats) BytesRead() int64 {
	return s.bytesRead.Load()
}

// ElapsedTime returns the elapsed time since start
func (s *Stats) ElapsedTime() time.Duration {
	return time.Since(s.startTime)
}

// Throughput returns points per second
func (s *Stats) Throughput() float64 {
	elapsed := s.ElapsedTime().Seconds()
	if elapsed == 0 {
		return 0
	}
	return float64(s.PointsProcessed()) / elapsed
}

// Writer interface for writing batches to InfluxDB
type Writer interface {
	Write(batch []K6Point) error
	Close() error
}

// HTTPWriter writes to InfluxDB via HTTP API
type HTTPWriter struct {
	client   *http.Client
	writeURL string
	username string
	password string
}

// NewHTTPWriter creates a new HTTP writer
func NewHTTPWriter(host string, port int, database, username, password string) (*HTTPWriter, error) {
	// Build the write URL
	writeURL := fmt.Sprintf("http://%s:%d/write?db=%s", host, port, url.QueryEscape(database))

	return &HTTPWriter{
		client: &http.Client{
			Timeout: 30 * time.Second,
		},
		writeURL: writeURL,
		username: username,
		password: password,
	}, nil
}

// Write sends a batch of points via HTTP
func (w *HTTPWriter) Write(batch []K6Point) error {
	// Convert batch to line protocol
	lineProtocol, err := formatBatchAsLineProtocol(batch)
	if err != nil {
		return fmt.Errorf("failed to format batch: %w", err)
	}

	// Create HTTP request
	req, err := http.NewRequest("POST", w.writeURL, bytes.NewBufferString(lineProtocol))
	if err != nil {
		return fmt.Errorf("failed to create HTTP request: %w", err)
	}

	// Add basic auth if credentials provided
	if w.username != "" {
		req.SetBasicAuth(w.username, w.password)
	}

	req.Header.Set("Content-Type", "application/octet-stream")

	// Send HTTP POST request
	resp, err := w.client.Do(req)
	if err != nil {
		return fmt.Errorf("failed to send HTTP request: %w", err)
	}
	defer resp.Body.Close()

	// Check response status
	if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusNoContent {
		body, _ := io.ReadAll(resp.Body)
		return fmt.Errorf("HTTP write failed with status %d: %s", resp.StatusCode, string(body))
	}

	return nil
}

// Close closes any resources
func (w *HTTPWriter) Close() error {
	return nil
}

// parseTimeToNanos parses RFC3339 time string to nanoseconds since epoch
func parseTimeToNanos(timeStr string) (int64, error) {
	t, err := time.Parse(time.RFC3339Nano, timeStr)
	if err != nil {
		// Try without nanoseconds
		t, err = time.Parse(time.RFC3339, timeStr)
		if err != nil {
			return 0, fmt.Errorf("failed to parse time: %w", err)
		}
	}
	return t.UnixNano(), nil
}

// escapeTag escapes special characters in tag keys and values
func escapeTag(s string) string {
	s = strings.ReplaceAll(s, ",", "\\,")
	s = strings.ReplaceAll(s, "=", "\\=")
	s = strings.ReplaceAll(s, " ", "\\ ")
	return s
}

// escapeFieldValue escapes field values (strings)
func escapeFieldValue(s string) string {
	s = strings.ReplaceAll(s, "\\", "\\\\")
	s = strings.ReplaceAll(s, "\"", "\\\"")
	return fmt.Sprintf("\"%s\"", s)
}

// convertToLineProtocol converts a K6Point to InfluxDB line protocol format
func convertToLineProtocol(point K6Point) (string, error) {
	// Parse timestamp
	nanos, err := parseTimeToNanos(point.Data.Time)
	if err != nil {
		return "", err
	}

	// Build tags string, filtering out empty values
	var tags []string
	for k, v := range point.Data.Tags {
		// Skip tags with empty values - InfluxDB doesn't accept them
		if v != "" {
			tags = append(tags, fmt.Sprintf("%s=%s", escapeTag(k), escapeTag(v)))
		}
	}
	tagsStr := ""
	if len(tags) > 0 {
		tagsStr = "," + strings.Join(tags, ",")
	}

	// Build line protocol: measurement,tag1=value1,tag2=value2 field1=value1 timestamp
	line := fmt.Sprintf("%s%s value=%v %d",
		escapeTag(point.Metric),
		tagsStr,
		point.Data.Value,
		nanos,
	)

	return line, nil
}

// formatBatchAsLineProtocol converts a batch of K6Points to line protocol
func formatBatchAsLineProtocol(batch []K6Point) (string, error) {
	var lines []string

	for _, point := range batch {
		line, err := convertToLineProtocol(point)
		if err != nil {
			return "", err
		}
		lines = append(lines, line)
	}

	return strings.Join(lines, "\n"), nil
}

// processFile reads and processes the JSON file
func processFile(config Config, writer Writer, stats *Stats) error {
	file, err := os.Open(config.FilePath)
	if err != nil {
		return fmt.Errorf("failed to open file: %w", err)
	}
	defer file.Close()

	// Get file size for progress tracking
	fileInfo, err := file.Stat()
	if err != nil {
		return fmt.Errorf("failed to stat file: %w", err)
	}
	fileSize := fileInfo.Size()

	// Create batch channel
	batchChan := make(chan []K6Point, 64)

	// Start writer workers
	var wg sync.WaitGroup
	errorChan := make(chan error, config.Workers)

	for i := 0; i < config.Workers; i++ {
		wg.Add(1)
		go writerWorker(writer, batchChan, &wg, stats, errorChan)
	}

	// Start progress reporter
	stopProgress := make(chan bool)
	go progressReporter(stats, fileSize, stopProgress, config.Verbose)

	// Read and batch points
	scanner := bufio.NewScanner(file)
	buf := make([]byte, 0, 64*1024)
	scanner.Buffer(buf, 10*1024*1024) // 10MB max line size

	currentBatch := make([]K6Point, 0, config.BatchSize)
	lineNumber := 0

	for scanner.Scan() {
		lineNumber++
		line := scanner.Bytes()
		stats.AddBytes(int64(len(line) + 1)) // +1 for newline

		var point K6Point
		if err := json.Unmarshal(line, &point); err != nil {
			// Skip invalid JSON lines
			if config.Verbose {
				fmt.Fprintf(os.Stderr, "\nWarning: Failed to parse line %d: %v\n", lineNumber, err)
			}
			continue
		}

		// Only process "Point" type entries
		if point.Type == "Point" {
			currentBatch = append(currentBatch, point)

			// Send batch when full
			if len(currentBatch) >= config.BatchSize {
				// Check for errors from workers
				select {
				case err := <-errorChan:
					close(batchChan)
					stopProgress <- true
					return err
				default:
				}

				batchChan <- currentBatch
				currentBatch = make([]K6Point, 0, config.BatchSize)
			}
		}
	}

	if err := scanner.Err(); err != nil {
		close(batchChan)
		stopProgress <- true
		return fmt.Errorf("error reading file: %w", err)
	}

	// Send final batch
	if len(currentBatch) > 0 {
		batchChan <- currentBatch
	}

	// Close channel and wait for workers
	close(batchChan)
	wg.Wait()

	// Check for any errors from workers
	close(errorChan)
	for err := range errorChan {
		stopProgress <- true
		return err
	}

	stopProgress <- true
	return nil
}

// writerWorker consumes batches and writes them to InfluxDB
func writerWorker(writer Writer, batches <-chan []K6Point, wg *sync.WaitGroup, stats *Stats, errorChan chan<- error) {
	defer wg.Done()

	for batch := range batches {
		start := time.Now()

		if err := writer.Write(batch); err != nil {
			errorChan <- fmt.Errorf("failed to write batch: %w", err)
			return
		}

		stats.AddWriteTime(time.Since(start))
		stats.AddBatches(1)
		stats.AddPoints(int64(len(batch)))
	}
}

// progressReporter displays progress updates
func progressReporter(stats *Stats, fileSize int64, stop <-chan bool, verbose bool) {
	if !verbose {
		return
	}

	ticker := time.NewTicker(500 * time.Millisecond)
	defer ticker.Stop()

	for {
		select {
		case <-stop:
			// Final update
			printProgress(stats, fileSize)
			fmt.Println() // New line after progress
			return
		case <-ticker.C:
			printProgress(stats, fileSize)
		}
	}
}

// printProgress prints current progress
func printProgress(stats *Stats, fileSize int64) {
	bytesRead := stats.BytesRead()
	pointsProcessed := stats.PointsProcessed()
	throughput := stats.Throughput()

	var progress float64
	if fileSize > 0 {
		progress = float64(bytesRead) / float64(fileSize) * 100
	}

	elapsed := stats.ElapsedTime()
	var eta time.Duration
	if progress > 0 {
		totalTime := time.Duration(float64(elapsed) / progress * 100)
		eta = totalTime - elapsed
	}

	fmt.Printf("\rProgress: %.1f%% | %d points | %.0f pts/sec | Elapsed: %s | ETA: %s",
		progress,
		pointsProcessed,
		throughput,
		formatDuration(elapsed),
		formatDuration(eta),
	)
}

// formatDuration formats a duration nicely
func formatDuration(d time.Duration) string {
	if d < 0 {
		return "N/A"
	}
	if d < time.Minute {
		return fmt.Sprintf("%ds", int(d.Seconds()))
	}
	minutes := int(d.Minutes())
	seconds := int(d.Seconds()) % 60
	return fmt.Sprintf("%dm%ds", minutes, seconds)
}

## importer_test.go
package main

import (
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

// Test K6Point JSON unmarshaling
func TestK6PointUnmarshal(t *testing.T) {
	jsonStr := `{
		"type": "Point",
		"metric": "http_req_duration",
		"data": {
			"time": "2024-01-22T10:00:00Z",
			"value": 123.45,
			"tags": {
				"name": "test",
				"status": "200"
			}
		}
	}`

	var point K6Point
	err := json.Unmarshal([]byte(jsonStr), &point)

	require.NoError(t, err)
	assert.Equal(t, "Point", point.Type)
	assert.Equal(t, "http_req_duration", point.Metric)
	assert.Equal(t, "2024-01-22T10:00:00Z", point.Data.Time)
	assert.Equal(t, 123.45, point.Data.Value)
	assert.Equal(t, "test", point.Data.Tags["name"])
	assert.Equal(t, "200", point.Data.Tags["status"])
}

// Test converting K6Point to InfluxDB line protocol
func TestConvertToLineProtocol(t *testing.T) {
	k6Point := K6Point{
		Type:   "Point",
		Metric: "http_req_duration",
		Data: K6Data{
			Time:  "2024-01-22T10:00:00Z",
			Value: 123.45,
			Tags: map[string]string{
				"name":   "test",
				"status": "200",
			},
		},
	}

	line, err := convertToLineProtocol(k6Point)

	require.NoError(t, err)
	assert.Contains(t, line, "http_req_duration")
	assert.Contains(t, line, "name=test")
	assert.Contains(t, line, "status=200")
	assert.Contains(t, line, "value=123.45")
}

// Test batch formatting to line protocol
func TestFormatBatchAsLineProtocol(t *testing.T) {
	batch := []K6Point{
		{
			Type:   "Point",
			Metric: "metric1",
			Data: K6Data{
				Time:  "2024-01-22T10:00:00Z",
				Value: 100.0,
				Tags:  map[string]string{"tag1": "value1"},
			},
		},
		{
			Type:   "Point",
			Metric: "metric2",
			Data: K6Data{
				Time:  "2024-01-22T10:00:01Z",
				Value: 200.0,
				Tags:  map[string]string{"tag2": "value2"},
			},
		},
	}

	result, err := formatBatchAsLineProtocol(batch)

	require.NoError(t, err)
	assert.Contains(t, result, "metric1")
	assert.Contains(t, result, "metric2")
	assert.Contains(t, result, "\n")
}

// Test time parsing and conversion to nanoseconds
func TestParseTime(t *testing.T) {
	tests := []struct {
		name     string
		timeStr  string
		wantErr  bool
	}{
		{
			name:    "RFC3339 format",
			timeStr: "2024-01-22T10:00:00Z",
			wantErr: false,
		},
		{
			name:    "RFC3339 with microseconds",
			timeStr: "2024-01-22T10:00:00.123456Z",
			wantErr: false,
		},
		{
			name:    "Invalid format",
			timeStr: "not a time",
			wantErr: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			nanos, err := parseTimeToNanos(tt.timeStr)

			if tt.wantErr {
				assert.Error(t, err)
			} else {
				require.NoError(t, err)
				assert.Greater(t, nanos, int64(0))
			}
		})
	}
}

// Test tag escaping for InfluxDB line protocol
func TestEscapeTag(t *testing.T) {
	tests := []struct {
		input    string
		expected string
	}{
		{input: "simple", expected: "simple"},
		{input: "with space", expected: "with\\ space"},
		{input: "with,comma", expected: "with\\,comma"},
		{input: "with=equals", expected: "with\\=equals"},
		{input: "multiple spaces  here", expected: "multiple\\ spaces\\ \\ here"},
	}

	for _, tt := range tests {
		t.Run(tt.input, func(t *testing.T) {
			result := escapeTag(tt.input)
			assert.Equal(t, tt.expected, result)
		})
	}
}

// Test field value escaping
func TestEscapeFieldValue(t *testing.T) {
	tests := []struct {
		input    string
		expected string
	}{
		{input: "simple", expected: "\"simple\""},
		{input: "with\"quote", expected: "\"with\\\"quote\""},
		{input: "with\\backslash", expected: "\"with\\\\backslash\""},
	}

	for _, tt := range tests {
		t.Run(tt.input, func(t *testing.T) {
			result := escapeFieldValue(tt.input)
			assert.Equal(t, tt.expected, result)
		})
	}
}

// Test stats tracking
func TestStatsTracking(t *testing.T) {
	stats := NewStats()

	stats.AddPoints(100)
	stats.AddBatches(1)
	stats.AddBytes(1000)

	assert.Equal(t, int64(100), stats.PointsProcessed())
	assert.Equal(t, int64(1), stats.BatchesWritten())
	assert.Equal(t, int64(1000), stats.BytesRead())
}

// Test config validation
func TestConfigValidation(t *testing.T) {
	tests := []struct {
		name    string
		config  Config
		wantErr bool
	}{
		{
			name: "Valid config",
			config: Config{
				FilePath:  "/path/to/file.json",
				Host:      "localhost",
				Port:      8086,
				Database:  "k6",
				BatchSize: 2000,
				Workers:   4,
			},
			wantErr: false,
		},
		{
			name: "Missing file path",
			config: Config{
				Host:      "localhost",
				Port:      8086,
				Database:  "k6",
				BatchSize: 2000,
				Workers:   4,
			},
			wantErr: true,
		},
		{
			name: "Invalid batch size",
			config: Config{
				FilePath:  "/path/to/file.json",
				Host:      "localhost",
				Port:      8086,
				Database:  "k6",
				BatchSize: 0,
				Workers:   4,
			},
			wantErr: true,
		},
		{
			name: "Invalid workers",
			config: Config{
				FilePath:  "/path/to/file.json",
				Host:      "localhost",
				Port:      8086,
				Database:  "k6",
				BatchSize: 2000,
				Workers:   0,
			},
			wantErr: true,
		},
		{
			name: "Missing database",
			config: Config{
				FilePath:  "/path/to/file.json",
				Host:      "localhost",
				Port:      8086,
				BatchSize: 2000,
				Workers:   4,
			},
			wantErr: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			err := tt.config.Validate()

			if tt.wantErr {
				assert.Error(t, err)
			} else {
				assert.NoError(t, err)
			}
		})
	}
}

// Test duration formatting
func TestFormatDuration(t *testing.T) {
	tests := []struct {
		name     string
		duration time.Duration
		expected string
	}{
		{
			name:     "Negative duration",
			duration: -1 * time.Second,
			expected: "N/A",
		},
		{
			name:     "Less than a minute",
			duration: 45 * time.Second,
			expected: "45s",
		},
		{
			name:     "Exactly one minute",
			duration: 60 * time.Second,
			expected: "1m0s",
		},
		{
			name:     "Multiple minutes",
			duration: 3*time.Minute + 25*time.Second,
			expected: "3m25s",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			result := formatDuration(tt.duration)
			assert.Equal(t, tt.expected, result)
		})
	}
}

// Test stats methods
func TestStatsElapsedAndThroughput(t *testing.T) {
	stats := NewStats()

	stats.AddPoints(1000)
	time.Sleep(10 * time.Millisecond)

	elapsed := stats.ElapsedTime()
	assert.Greater(t, elapsed, time.Duration(0))

	throughput := stats.Throughput()
	assert.Greater(t, throughput, 0.0)
}

// Test line protocol with no tags
func TestConvertToLineProtocolNoTags(t *testing.T) {
	k6Point := K6Point{
		Type:   "Point",
		Metric: "simple_metric",
		Data: K6Data{
			Time:  "2024-01-22T10:00:00Z",
			Value: 42.0,
			Tags:  map[string]string{},
		},
	}

	line, err := convertToLineProtocol(k6Point)

	require.NoError(t, err)
	assert.Contains(t, line, "simple_metric")
	assert.Contains(t, line, "value=42")
	assert.NotContains(t, line, ",,")
}

// Test line protocol with special characters in metric name
func TestConvertToLineProtocolSpecialChars(t *testing.T) {
	k6Point := K6Point{
		Type:   "Point",
		Metric: "metric with spaces",
		Data: K6Data{
			Time:  "2024-01-22T10:00:00Z",
			Value: 100.0,
			Tags: map[string]string{
				"tag with space": "value with space",
			},
		},
	}

	line, err := convertToLineProtocol(k6Point)

	require.NoError(t, err)
	assert.Contains(t, line, "metric\\ with\\ spaces")
	assert.Contains(t, line, "tag\\ with\\ space=value\\ with\\ space")
}

// Test line protocol with invalid time
func TestConvertToLineProtocolInvalidTime(t *testing.T) {
	k6Point := K6Point{
		Type:   "Point",
		Metric: "test",
		Data: K6Data{
			Time:  "invalid time",
			Value: 100.0,
			Tags:  map[string]string{},
		},
	}

	_, err := convertToLineProtocol(k6Point)
	assert.Error(t, err)
}

// Test batch formatting with error
func TestFormatBatchAsLineProtocolWithError(t *testing.T) {
	batch := []K6Point{
		{
			Type:   "Point",
			Metric: "good_metric",
			Data: K6Data{
				Time:  "2024-01-22T10:00:00Z",
				Value: 100.0,
				Tags:  map[string]string{},
			},
		},
		{
			Type:   "Point",
			Metric: "bad_metric",
			Data: K6Data{
				Time:  "invalid",
				Value: 200.0,
				Tags:  map[string]string{},
			},
		},
	}

	_, err := formatBatchAsLineProtocol(batch)
	assert.Error(t, err)
}

// Test HTTP writer creation
func TestNewHTTPWriter(t *testing.T) {
	writer, err := NewHTTPWriter("localhost", 8086, "testdb", "admin", "admin")

	require.NoError(t, err)
	assert.NotNil(t, writer)

	err = writer.Close()
	assert.NoError(t, err)
}

// Test stats write time tracking
func TestStatsWriteTime(t *testing.T) {
	stats := NewStats()

	stats.AddWriteTime(100 * time.Millisecond)
	stats.AddWriteTime(200 * time.Millisecond)

	// We don't expose write time directly, but we can verify it doesn't panic
	assert.Greater(t, stats.PointsProcessed(), int64(-1))
}

## K6_GO_IMPORTER_README.md

      
    Raw
  

              K6_GO_IMPORTER_README.md
            
          
    k6-to-influxdb Go Importer

High-performance Go implementation for importing k6 JSON output to InfluxDB via HTTP.
Overview

This Go-based importer replaces the Python converter with a much faster implementation that achieves 6x better performance through:

True parallelism (no GIL limitation)
Fast JSON parsing with jsoniter (2-3x faster than stdlib)
Producer-consumer architecture with goroutines
HTTP protocol with authentication support
Optimized batch size (2,000 points per batch)
Efficient batching and concurrent writers

Performance

Small File Test (LS-001S.3.json - 3MB, 11,409 points)

Go Implementation:

Throughput: ~181,000 points/sec
Import time: <0.1 seconds
Memory: Constant (~50MB)

Python (Multi-threaded):

Throughput: ~30,000 points/sec
Import time: ~0.4 seconds
Memory: Growing (484MB → 1.2GB)

Improvement: 6x faster
Expected Large File Performance (LS-004.json - 14GB, 52M points)

Estimated Performance:

Throughput: 180,000-200,000 points/sec
Import time: 4-5 minutes (vs 25+ minutes with Python)
Memory: Constant (~100-200 MB)

Batch Size Optimization

Extensive testing revealed that smaller batches are faster for this use case:


Batch Size
Throughput
Notes


1,000
130,566 pts/sec
Too many HTTP requests


2,000
180,935 pts/sec
✅ Optimal


5,000
142,377 pts/sec
Good balance


10,000
117,450 pts/sec
Default for many tools


25,000
53,313 pts/sec
Too large


50,000
43,720 pts/sec
Network overhead dominates


Why 2K is optimal:

Smaller HTTP payloads = faster network transmission
More frequent completion = better parallelism across workers
Less time blocked waiting for large batch to complete
Sweet spot between request overhead and throughput

Architecture

Producer-Consumer Model

Main Goroutine (Producer)
├── Read file line-by-line
├── Parse JSON (jsoniter)
├── Convert to InfluxDB point
├── Accumulate into batches (2,000 points)
└── Send batches to channel
        │
        ▼
    [Buffered Channel]
    (64 batch queue)
        │
        ▼
Worker Pool (8 Consumers)
├── Format as InfluxDB line protocol
├── Write via HTTP with auth
└── Track stats atomically

Key Components


Fast JSON Parsing: Uses jsoniter (2-3x faster than stdlib)
Concurrent Workers: 8 goroutines writing in parallel
Optimized Batching: 2,000 points per batch (tested and proven)
Progress Tracking: Real-time stats with atomic counters
Error Handling: Fail-fast on first error
Empty Tag Filtering: Automatically removes empty tag values

Installation

Build from Source

go build -o k6-to-influxdb
Test Coverage

go test -v -cover
Current coverage: 29.2% (core functionality tested)
Usage

Basic Usage

./k6-to-influxdb \
  --file=k6-output.json \
  --username=admin \
  --password=admin
All Options

./k6-to-influxdb \
  --file=k6-output.json \
  --host=localhost \
  --port=8086 \
  --database=k6 \
  --username=admin \
  --password=admin \
  --batch-size=2000 \
  --workers=8 \
  --verbose
Command-Line Options


Option
Default
Description


--file
required
Path to k6 JSON file


--host
localhost
InfluxDB host


--port
8086
InfluxDB HTTP port


--database
k6
InfluxDB database name


--username
""
InfluxDB username


--password
""
InfluxDB password


--batch-size
2000
Points per batch (optimal: 2000)


--workers
8
Number of concurrent workers


--verbose
true
Enable progress output


Integration with Setup Script

The setup-k6-grafana.sh script automatically:

Detects if Go binary exists
Uses Go importer if available (6x faster)
Falls back to Python converter if not

Manual Integration

Copy the binary to the project root:
cp k6-to-influxdb /path/to/gm-perf-k6-tests/
The setup script will automatically use it on the next run.
Implementation Details

Line Protocol Format

The importer converts k6 JSON to InfluxDB line protocol:
metric_name,tag1=value1,tag2=value2 value=123.45 1234567890000000000

Key features:

Filters out tags with empty values (InfluxDB requirement)
Escapes special characters in tags and metrics
Preserves nanosecond timestamp precision
Handles all k6 metric types

HTTP Protocol

Uses InfluxDB HTTP write API with:

Basic authentication
Connection reuse
30-second timeout per request
Error response parsing

Error Handling


Fail-fast: Stops on first error
Error reporting: Shows HTTP status and response body
Validation: Pre-validates config before starting

Memory Management


Streaming: Reads file line-by-line (no full file in memory)
Pre-allocated batches: Batch slices created with capacity
Constant memory: ~50-200 MB regardless of file size
No memory leaks: Proper resource cleanup with defer

Testing

Unit Tests

go test -v
Tests cover:

JSON unmarshaling
Line protocol formatting
Tag escaping (including empty values)
Time parsing
Config validation
Stats tracking

Integration Test

# Clean database
docker exec k6-influxdb influx -username admin -password admin \
  -execute "DROP DATABASE k6; CREATE DATABASE k6"

# Import test file
./k6-to-influxdb \
  --file=LS-001S.3.json \
  --username=admin \
  --password=admin \
  --verbose

# Verify import
docker exec k6-influxdb influx -username admin -password admin \
  -database k6 -execute "SELECT COUNT(value) FROM http_req_duration"
Troubleshooting

"Missing tag value" error

Cause: Tags with empty values in k6 output
Solution: ✅ Fixed - automatically filters out empty tags
"HTTP write failed with status 401"

Cause: Missing or incorrect authentication credentials
Solution: Add --username=admin --password=admin
Slow import speed

Possible causes:

Batch size too large (use 2000, not 10000+)
Too few workers (increase to 8)
Network latency (ensure InfluxDB is local/fast connection)

Performance Comparison

Go vs Python


Metric
Python (Multi-threaded)
Go (HTTP, 2K batch)
Improvement


Throughput
~30,000 pts/sec
~181,000 pts/sec
6.0x faster


14GB file
25+ minutes
4-5 minutes
~5-6x faster


CPU usage
99.6% (1 core)
400-800% (4-8 cores)
True parallelism


Memory
484MB → 1.2GB
~50-200 MB
Constant memory


Concurrency
Limited by GIL
True parallelism
All cores used


Why HTTP-Only?

UDP was initially considered for speed, but:

Packet size limits: k6 metrics are verbose (400-600 bytes each)
No fragmentation support: Large batches fail with "message too long"
No error feedback: Silent data loss on packet drops
No authentication: Security requirement not met
HTTP is actually faster: Better batching and connection reuse

Testing showed HTTP with 2K batches outperforms UDP attempts.
Future Improvements

Potential Enhancements


Compression: Gzip compression for HTTP requests
Retry Logic: Automatic retry on transient errors
Progress Persistence: Resume interrupted imports
Multi-file Import: Import multiple files in sequence
Schema Validation: Validate k6 JSON structure before import

Performance Optimizations


Memory pooling: Reuse allocated batches
Buffer tuning: Optimize channel buffer sizes
Connection pooling: HTTP connection pool tuning
Adaptive batch sizing: Adjust based on response times

Dependencies

require (
    github.com/json-iterator/go v1.1.12  // Fast JSON parsing
    github.com/stretchr/testify v1.8.4   // Testing framework
)
Contributing

To contribute improvements:

Run tests: go test -v
Check coverage: go test -cover
Format code: go fmt ./...
Build binary: go build -o k6-to-influxdb
Test with sample file

Technical Notes

Why Smaller Batches Win

Counter-intuitively, smaller batches (2K) outperform larger ones (10K+):

Network transmission time: Smaller payloads transfer faster
Parallelism window: Workers can grab new batches sooner
Reduced blocking: Less time waiting for large writes to complete
Better load distribution: More even work distribution across workers

The 2K batch size was empirically determined through systematic testing.
Architecture Decisions


HTTP over UDP: Reliability and auth requirements
8 workers: Optimal for multi-core CPUs
Fail-fast: Data integrity over partial imports
No retries: User should re-run if errors occur
Atomic stats: Lock-free performance tracking

License

Same as parent project.

  
## main.go
package main

import (
	"flag"
	"fmt"
	"os"
)

func main() {
	// Parse command line flags
	config := Config{
		Host:      "localhost",
		Port:      8086,
		Database:  "k6",
		Username:  "",
		Password:  "",
		BatchSize: 2000,
		Workers:   8,
		Verbose:   true,
	}

	flag.StringVar(&config.FilePath, "file", "", "Path to k6 JSON file (required)")
	flag.StringVar(&config.Host, "host", config.Host, "InfluxDB host")
	flag.IntVar(&config.Port, "port", config.Port, "InfluxDB port")
	flag.StringVar(&config.Database, "database", config.Database, "InfluxDB database name")
	flag.StringVar(&config.Username, "username", config.Username, "InfluxDB username")
	flag.StringVar(&config.Password, "password", config.Password, "InfluxDB password")
	flag.IntVar(&config.BatchSize, "batch-size", config.BatchSize, "Number of points per batch (recommended: 2000)")
	flag.IntVar(&config.Workers, "workers", config.Workers, "Number of concurrent writer workers")
	flag.BoolVar(&config.Verbose, "verbose", config.Verbose, "Enable verbose output")

	flag.Usage = func() {
		fmt.Fprintf(os.Stderr, "Usage: %s [options]\n\n", os.Args[0])
		fmt.Fprintf(os.Stderr, "k6-to-influxdb imports k6 JSON output to InfluxDB via HTTP\n\n")
		fmt.Fprintf(os.Stderr, "Options:\n")
		flag.PrintDefaults()
		fmt.Fprintf(os.Stderr, "\nExample:\n")
		fmt.Fprintf(os.Stderr, "  %s --file=k6-output.json --username=admin --password=admin\n", os.Args[0])
	}

	flag.Parse()

	// Validate configuration
	if err := config.Validate(); err != nil {
		fmt.Fprintf(os.Stderr, "Configuration error: %v\n\n", err)
		flag.Usage()
		os.Exit(1)
	}

	// Check if file exists
	if _, err := os.Stat(config.FilePath); os.IsNotExist(err) {
		fmt.Fprintf(os.Stderr, "Error: File does not exist: %s\n", config.FilePath)
		os.Exit(1)
	}

	// Create HTTP writer
	if config.Verbose {
		fmt.Printf("Starting k6-to-influxdb importer\n")
		fmt.Printf("File: %s\n", config.FilePath)
		fmt.Printf("Target: http://%s:%d/%s\n", config.Host, config.Port, config.Database)
		fmt.Printf("Batch size: %d points\n", config.BatchSize)
		fmt.Printf("Workers: %d\n\n", config.Workers)
	}

	writer, err := NewHTTPWriter(config.Host, config.Port, config.Database, config.Username, config.Password)
	if err != nil {
		fmt.Fprintf(os.Stderr, "Failed to create HTTP writer: %v\n", err)
		os.Exit(1)
	}
	defer writer.Close()

	// Create stats tracker
	stats := NewStats()

	// Process file
	if err := processFile(config, writer, stats); err != nil {
		fmt.Fprintf(os.Stderr, "\nError: %v\n", err)
		os.Exit(1)
	}

	// Print final summary
	if config.Verbose {
		fmt.Printf("\n\nImport completed successfully!\n")
		fmt.Printf("Total points imported: %d\n", stats.PointsProcessed())
		fmt.Printf("Total batches written: %d\n", stats.BatchesWritten())
		fmt.Printf("Total time: %s\n", formatDuration(stats.ElapsedTime()))
		fmt.Printf("Average throughput: %.0f points/sec\n", stats.Throughput())
		fmt.Printf("Total data read: %.2f MB\n", float64(stats.BytesRead())/(1024*1024))
	}
}

## setup-k6-grafana.sh
#!/bin/bash

set -e

# Colors for output
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
BLUE='\033[0;34m'
NC='\033[0m' # No Color

echo -e "${BLUE}========================================${NC}"
echo -e "${BLUE}k6 + InfluxDB + Grafana Setup Script${NC}"
echo -e "${BLUE}========================================${NC}"
echo ""

# Function to check if command exists
command_exists() {
    command -v "$1" >/dev/null 2>&1
}

# Check prerequisites
echo -e "${YELLOW}Checking prerequisites...${NC}"

if ! command_exists docker; then
    echo -e "${RED}Error: docker is not installed${NC}"
    exit 1
fi

if ! command_exists docker-compose && ! docker compose version >/dev/null 2>&1; then
    echo -e "${RED}Error: docker-compose is not installed${NC}"
    exit 1
fi

if ! command_exists python3; then
    echo -e "${RED}Error: python3 is not installed${NC}"
    exit 1
fi

if ! command_exists git; then
    echo -e "${RED}Error: git is not installed${NC}"
    exit 1
fi

if ! command_exists jq; then
    echo -e "${YELLOW}Warning: jq is not installed. Dashboard import may fail.${NC}"
    echo -e "${YELLOW}Install jq with: brew install jq (macOS) or apt-get install jq (Linux)${NC}"
    JQ_AVAILABLE=false
else
    JQ_AVAILABLE=true
fi

echo -e "${GREEN}✓ All prerequisites met${NC}"
echo ""

# Get the script directory
SCRIPT_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"

# Prompt for JSON file
echo -e "${YELLOW}Enter the path to your k6 JSON results file:${NC}"
echo -e "${YELLOW}(You can use a filename, relative path, absolute path, or ~/path)${NC}"
read -e -r JSON_FILE_INPUT

# Sanitize input - remove any dangerous characters
# Allow: alphanumeric, dash, underscore, dot, slash, tilde, space
JSON_FILE_INPUT=$(echo "$JSON_FILE_INPUT" | sed 's/[^a-zA-Z0-9._/~[:space:]-]//g')

# Trim leading/trailing whitespace
JSON_FILE_INPUT=$(echo "$JSON_FILE_INPUT" | sed -e 's/^[[:space:]]*//' -e 's/[[:space:]]*$//')

if [ -z "$JSON_FILE_INPUT" ]; then
    echo -e "${RED}Error: No file path provided${NC}"
    exit 1
fi

# Expand tilde to home directory
JSON_FILE_INPUT="${JSON_FILE_INPUT/#\~/$HOME}"

# Try to find the file in multiple locations
JSON_FILE=""

# 1. Try as absolute or relative path from current directory
if [ -f "$JSON_FILE_INPUT" ]; then
    JSON_FILE="$JSON_FILE_INPUT"
    echo -e "${GREEN}✓ Found JSON file in current directory${NC}"
# 2. Try relative to script directory
elif [ -f "$SCRIPT_DIR/$JSON_FILE_INPUT" ]; then
    JSON_FILE="$SCRIPT_DIR/$JSON_FILE_INPUT"
    echo -e "${GREEN}✓ Found JSON file in script directory${NC}"
# 3. Check if it's just a filename and search for it in script directory
elif [[ "$JSON_FILE_INPUT" != *"/"* ]]; then
    # It's just a filename, try to find it
    FOUND_FILES=$(find "$SCRIPT_DIR" -maxdepth 2 -type f -name "$JSON_FILE_INPUT" 2>/dev/null)
    FILE_COUNT=$(echo "$FOUND_FILES" | grep -c .)

    if [ -n "$FOUND_FILES" ] && [ "$FILE_COUNT" -eq 1 ]; then
        JSON_FILE="$FOUND_FILES"
        echo -e "${GREEN}✓ Found JSON file: $(basename "$JSON_FILE")${NC}"
    elif [ "$FILE_COUNT" -gt 1 ]; then
        echo -e "${RED}Error: Multiple files found matching '$JSON_FILE_INPUT':${NC}"
        echo "$FOUND_FILES"
        echo -e "${YELLOW}Please provide a more specific path${NC}"
        exit 1
    fi
fi

# If still not found, show error with helpful message
if [ -z "$JSON_FILE" ] || [ ! -f "$JSON_FILE" ]; then
    echo -e "${RED}Error: File not found: $JSON_FILE_INPUT${NC}"
    echo ""
    echo -e "${YELLOW}Searched in:${NC}"
    echo -e "  • Current directory: $(pwd)"
    echo -e "  • Script directory: $SCRIPT_DIR"
    echo ""
    echo -e "${YELLOW}Available JSON files in script directory:${NC}"
    find "$SCRIPT_DIR" -maxdepth 2 -type f -name "*.json" 2>/dev/null | head -10 | while read -r file; do
        echo -e "  • $(basename "$file")"
    done
    exit 1
fi

# Convert to absolute path
JSON_FILE=$(cd "$(dirname "$JSON_FILE")" && pwd)/$(basename "$JSON_FILE")
echo -e "${GREEN}✓ Using file: $JSON_FILE${NC}"
echo ""

# Ask for project directory
echo -e "${YELLOW}Enter directory name for the setup (default: k6-grafana-stack):${NC}"
read -r PROJECT_DIR
PROJECT_DIR=${PROJECT_DIR:-k6-grafana-stack}

# Check if directory exists and clean up any existing setup
if [ -d "$PROJECT_DIR" ]; then
    echo -e "${YELLOW}Found existing setup directory. Cleaning up...${NC}"
    cd "$PROJECT_DIR"

    # Stop and remove any running containers
    if [ -f "docker-compose.yml" ]; then
        docker-compose down -v > /dev/null 2>&1
        echo -e "${GREEN}✓ Stopped and removed existing containers${NC}"
    fi

    # Remove old converter directory if it exists
    if [ -d "k6-json-to-influxdb-line-protocol" ]; then
        rm -rf k6-json-to-influxdb-line-protocol
    fi

    cd ..
    echo -e "${GREEN}✓ Cleaned up existing setup${NC}"
else
    echo -e "${YELLOW}Creating new setup directory...${NC}"
fi

# Create/recreate project directory
mkdir -p "$PROJECT_DIR"
cd "$PROJECT_DIR"
echo -e "${GREEN}✓ Project directory ready: $PROJECT_DIR${NC}"
echo ""

# Create docker-compose.yml
echo -e "${YELLOW}Creating docker-compose.yml...${NC}"
cat > docker-compose.yml <<'EOF'
services:
  influxdb:
    image: influxdb:1.8
    container_name: k6-influxdb
    ports:
      - "8086:8086"
    environment:
      - INFLUXDB_DB=k6
      - INFLUXDB_ADMIN_USER=admin
      - INFLUXDB_ADMIN_PASSWORD=admin
      - INFLUXDB_HTTP_AUTH_ENABLED=true
    volumes:
      - influxdb-data:/var/lib/influxdb
    networks:
      - k6-network
    deploy:
      resources:
        limits:
          cpus: '8.0'
          memory: 8G
        reservations:
          cpus: '4.0'
          memory: 4G

  grafana:
    image: grafana/grafana:latest
    container_name: k6-grafana
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_USER=admin
      - GF_SECURITY_ADMIN_PASSWORD=admin
      - GF_AUTH_ANONYMOUS_ENABLED=true
      - GF_AUTH_ANONYMOUS_ORG_ROLE=Viewer
    volumes:
      - grafana-data:/var/lib/grafana
    depends_on:
      - influxdb
    networks:
      - k6-network
    deploy:
      resources:
        limits:
          cpus: '2.0'
          memory: 2G
        reservations:
          cpus: '0.5'
          memory: 512M

volumes:
  influxdb-data:
  grafana-data:

networks:
  k6-network:
    driver: bridge
EOF

echo -e "${GREEN}✓ docker-compose.yml created${NC}"
echo ""

# Start Docker containers
echo -e "${YELLOW}Starting Docker containers...${NC}"
docker-compose up -d

echo -e "${GREEN}✓ Containers started${NC}"
echo ""

# Wait for services to be ready
echo -e "${YELLOW}Waiting for services to be ready...${NC}"
sleep 10

# Wait for InfluxDB
echo -n "Waiting for InfluxDB"
INFLUX_READY=false
for i in {1..30}; do
    if curl -s --max-time 2 http://localhost:8086/ping > /dev/null 2>&1; then
        echo -e " ${GREEN}✓${NC}"
        INFLUX_READY=true
        break
    fi
    echo -n "."
    sleep 2
done

if [ "$INFLUX_READY" = false ]; then
    echo -e " ${RED}✗${NC}"
    echo -e "${RED}Error: InfluxDB did not start properly${NC}"
    echo -e "${YELLOW}Check logs with: cd $PROJECT_DIR && docker-compose logs influxdb${NC}"
    exit 1
fi

# Wait for Grafana
echo -n "Waiting for Grafana"
GRAFANA_READY=false
for i in {1..60}; do
    if curl -s --max-time 2 http://localhost:3000/api/health > /dev/null 2>&1; then
        echo -e " ${GREEN}✓${NC}"
        GRAFANA_READY=true
        break
    fi
    echo -n "."
    sleep 2

    # Check if we've been waiting too long
    if [ $i -eq 30 ]; then
        echo ""
        echo -e "${YELLOW}⚠ Grafana is taking longer than expected to start...${NC}"
        echo -n "Still waiting"
    fi
done

if [ "$GRAFANA_READY" = false ]; then
    echo -e " ${RED}✗${NC}"
    echo -e "${RED}Error: Grafana did not start properly${NC}"
    echo -e "${YELLOW}Check logs with: cd $PROJECT_DIR && docker-compose logs grafana${NC}"
    exit 1
fi

echo ""

# Configure Grafana data source via API
echo -e "${YELLOW}Configuring InfluxDB data source in Grafana...${NC}"

# Wait a bit more for Grafana to fully initialize
sleep 5

# Try to configure data source with timeout and error capture
DS_RESPONSE=$(curl -s --max-time 10 -w "\n%{http_code}" -X POST \
  -H "Content-Type: application/json" \
  -d '{
    "name": "InfluxDB-k6",
    "type": "influxdb",
    "url": "http://influxdb:8086",
    "access": "proxy",
    "database": "k6",
    "basicAuth": false,
    "isDefault": true,
    "jsonData": {
      "httpMode": "GET"
    },
    "secureJsonData": {
      "password": "admin"
    },
    "user": "admin"
  }' \
  http://admin:admin@localhost:3000/api/datasources 2>&1)

HTTP_CODE=$(echo "$DS_RESPONSE" | tail -n1)
DS_BODY=$(echo "$DS_RESPONSE" | head -n-1)

if [ "$HTTP_CODE" = "200" ] || [ "$HTTP_CODE" = "201" ]; then
    echo -e "${GREEN}✓ Data source configured${NC}"
elif [ "$HTTP_CODE" = "409" ]; then
    echo -e "${YELLOW}⚠ Data source already exists (this is fine)${NC}"
else
    echo -e "${YELLOW}⚠ Data source configuration returned HTTP $HTTP_CODE${NC}"
    echo -e "${YELLOW}  You may need to manually configure it in Grafana${NC}"
    if [ -n "$DS_BODY" ]; then
        echo -e "${YELLOW}  Response: $DS_BODY${NC}"
    fi
fi
echo ""

# Import k6 Grafana dashboard (14801 - modern version)
echo -e "${YELLOW}Importing K6 Dashboard (ID: 14801)...${NC}"

if [ "$JQ_AVAILABLE" = true ]; then
    # Download dashboard JSON
    if curl -s --max-time 10 https://grafana.com/api/dashboards/14801/revisions/1/download -o k6-dashboard.json 2>&1; then
        # Check if file was downloaded
        if [ -s k6-dashboard.json ]; then
            # Create proper import payload using jq
            # Dashboard expects DS_DUMMY as the input name
            IMPORT_PAYLOAD=$(jq -n --argjson dashboard "$(cat k6-dashboard.json)" '{
              dashboard: $dashboard,
              overwrite: true,
              inputs: [
                {
                  name: "DS_DUMMY",
                  type: "datasource",
                  pluginId: "influxdb",
                  value: "InfluxDB-k6"
                }
              ]
            }' 2>&1)

            if [ $? -eq 0 ]; then
                # Import dashboard via API
                IMPORT_RESPONSE=$(curl -s --max-time 10 -w "\n%{http_code}" -X POST \
                  -H "Content-Type: application/json" \
                  -d "$IMPORT_PAYLOAD" \
                  http://admin:admin@localhost:3000/api/dashboards/import 2>&1)

                IMPORT_CODE=$(echo "$IMPORT_RESPONSE" | tail -n1)
                IMPORT_BODY=$(echo "$IMPORT_RESPONSE" | head -n-1)

                if [ "$IMPORT_CODE" = "200" ] || [ "$IMPORT_CODE" = "201" ]; then
                    echo -e "${GREEN}✓ K6 Dashboard imported${NC}"
                    # Extract dashboard URL if available
                    DASH_URL=$(echo "$IMPORT_BODY" | jq -r '.importedUrl // empty' 2>/dev/null)
                    if [ -n "$DASH_URL" ]; then
                        echo -e "${GREEN}  Dashboard URL: http://localhost:3000${DASH_URL}${NC}"
                    fi
                elif [ "$IMPORT_CODE" = "412" ]; then
                    echo -e "${YELLOW}⚠ Dashboard already exists (this is fine)${NC}"
                else
                    echo -e "${YELLOW}⚠ Dashboard import returned HTTP $IMPORT_CODE${NC}"
                    if echo "$IMPORT_BODY" | grep -q "name-exists"; then
                        echo -e "${YELLOW}  Dashboard may already exist - continuing${NC}"
                    else
                        echo -e "${YELLOW}  You can manually import dashboard ID 14801 from Grafana${NC}"
                    fi
                fi
            else
                echo -e "${YELLOW}⚠ Failed to create import payload${NC}"
                echo -e "${YELLOW}  You can manually import dashboard ID 14801 from Grafana${NC}"
            fi
        else
            echo -e "${YELLOW}⚠ Failed to download dashboard${NC}"
            echo -e "${YELLOW}  You can manually import dashboard ID 14801 from Grafana${NC}"
        fi
    else
        echo -e "${YELLOW}⚠ Failed to download dashboard from Grafana.com${NC}"
        echo -e "${YELLOW}  You can manually import dashboard ID 14801 from Grafana${NC}"
    fi
else
    echo -e "${YELLOW}⚠ Skipping automatic dashboard import (jq not installed)${NC}"
    echo -e "${YELLOW}  Manually import dashboard: Go to Grafana → Dashboards → Import → Use ID: 14801${NC}"
fi
echo ""

# Clone the k6-json-to-influxdb converter
echo -e "${YELLOW}Setting up JSON to InfluxDB converter...${NC}"

if [ ! -d "k6-json-to-influxdb-line-protocol" ]; then
    if git clone https://github.com/yooap/k6-json-to-influxdb-line-protocol.git > /dev/null 2>&1; then
        echo -e "${GREEN}✓ Converter cloned${NC}"
    else
        echo -e "${RED}✗ Failed to clone converter repository${NC}"
        echo -e "${YELLOW}Check your internet connection or try again later${NC}"
        exit 1
    fi
else
    echo -e "${GREEN}✓ Converter already exists${NC}"
fi

cd k6-json-to-influxdb-line-protocol

# Copy optimized converters if available
if [ -f "$SCRIPT_DIR/convert_and_write_to_db_multithreaded.py" ]; then
    cp "$SCRIPT_DIR/convert_and_write_to_db_multithreaded.py" .
    chmod +x convert_and_write_to_db_multithreaded.py
    echo -e "${GREEN}✓ Multi-threaded converter installed${NC}"
elif [ -f "$SCRIPT_DIR/convert_and_write_to_db_streaming.py" ]; then
    cp "$SCRIPT_DIR/convert_and_write_to_db_streaming.py" .
    chmod +x convert_and_write_to_db_streaming.py
    echo -e "${GREEN}✓ Streaming converter installed${NC}"
fi

# Create and activate virtual environment
echo -e "${YELLOW}Creating Python virtual environment...${NC}"
if python3 -m venv venv 2>&1; then
    echo -e "${GREEN}✓ Virtual environment created${NC}"
else
    echo -e "${RED}✗ Failed to create virtual environment${NC}"
    cd ..
    exit 1
fi

# Activate virtual environment
source venv/bin/activate

# Install Python dependencies
echo -e "${YELLOW}Installing Python dependencies in venv...${NC}"
if [ -f "requirements.txt" ]; then
    if python3 -m pip install -q -r requirements.txt 2>&1; then
        echo -e "${GREEN}✓ Dependencies installed${NC}"
    else
        echo -e "${YELLOW}⚠ Some dependencies may have failed to install${NC}"
    fi
else
    # Install manually if requirements.txt doesn't exist
    if python3 -m pip install -q influxdb 2>&1; then
        echo -e "${GREEN}✓ Dependencies installed (influxdb)${NC}"
    else
        echo -e "${RED}✗ Failed to install influxdb package${NC}"
        deactivate
        cd ..
        exit 1
    fi
fi
echo ""

# Import JSON data to InfluxDB using optimized converter
echo -e "${YELLOW}Importing k6 JSON data to InfluxDB...${NC}"

# Get file size for progress indication
FILE_SIZE=$(ls -lh "$JSON_FILE" | awk '{print $5}')
FILE_SIZE_BYTES=$(stat -f%z "$JSON_FILE" 2>/dev/null || stat -c%s "$JSON_FILE" 2>/dev/null)
FILE_SIZE_MB=$(echo "scale=1; $FILE_SIZE_BYTES / 1024 / 1024" | bc 2>/dev/null || echo "?")

echo -e "${BLUE}File size: ${FILE_SIZE} (${FILE_SIZE_MB} MB)${NC}"

# Check for Go binary first (fastest option)
USE_GO=false
GO_BINARY=""

# Check in script directory
if [ -f "$SCRIPT_DIR/k6-to-influxdb" ]; then
    GO_BINARY="$SCRIPT_DIR/k6-to-influxdb"
    USE_GO=true
elif [ -f "./k6-to-influxdb" ]; then
    GO_BINARY="./k6-to-influxdb"
    USE_GO=true
fi

if [ "$USE_GO" = true ]; then
    echo -e "${BLUE}Using Go importer (8 workers, 2k batch size)${NC}"
    echo -e "${GREEN}⚡ High-performance mode enabled${NC}"
    echo ""

    # Run the Go binary with optimized batch size
    if "$GO_BINARY" \
      --file="$JSON_FILE" \
      --host=localhost \
      --port=8086 \
      --database=k6 \
      --username=admin \
      --password=admin \
      --batch-size=2000 \
      --workers=8 \
      --verbose 2>&1; then
        echo ""
        echo -e "${GREEN}✓ JSON data imported successfully${NC}"

        # Get final count
        FINAL_COUNT=$(docker exec k6-influxdb influx -username admin -password admin -database k6 -execute "SELECT COUNT(*) FROM http_req_duration" 2>/dev/null | tail -1 | awk '{print $2}')
        if [ -n "$FINAL_COUNT" ] && [ "$FINAL_COUNT" != "count_value" ]; then
            echo -e "${GREEN}  Total requests in database: ${FINAL_COUNT}${NC}"
        fi
    else
        IMPORT_EXIT_CODE=$?
        echo ""
        echo -e "${RED}✗ Failed to import JSON data (exit code: $IMPORT_EXIT_CODE)${NC}"
        echo -e "${YELLOW}Possible issues:${NC}"
        echo -e "${YELLOW}  • JSON file may not be in k6 NDJSON format${NC}"
        echo -e "${YELLOW}  • InfluxDB may not be ready${NC}"
        echo -e "${YELLOW}  • Check the error messages above${NC}"
        echo ""
        echo -e "${YELLOW}Try checking InfluxDB logs:${NC}"
        echo -e "${YELLOW}  cd .. && docker-compose logs influxdb${NC}"
        cd ..
        exit 1
    fi
else
    # Fallback to Python converter
    echo -e "${BLUE}Go importer not found, using Python converter${NC}"
    echo -e "${YELLOW}Tip: Build Go binary for 50-100x faster imports!${NC}"
    echo ""

    # Select best Python converter
    if [ -f "convert_and_write_to_db_multithreaded.py" ]; then
        CONVERTER_SCRIPT="convert_and_write_to_db_multithreaded.py"
        # Use 100k batch size - sweet spot between performance and HTTP limits
        CONVERTER_ARGS="--verbose --workers=4 --batch-size=100000"
        echo -e "${BLUE}Using multi-threaded import (4 workers, 100k batch size)${NC}"
    elif [ -f "convert_and_write_to_db_streaming.py" ]; then
        CONVERTER_SCRIPT="convert_and_write_to_db_streaming.py"
        CONVERTER_ARGS="--verbose"
        echo -e "${BLUE}Using streaming import${NC}"
    else
        CONVERTER_SCRIPT="convert_and_write_to_db.py"
        CONVERTER_ARGS=""
        echo -e "${BLUE}Using standard import${NC}"
    fi
    echo ""

    # Run the import (in venv) with live progress output
    if python3 $CONVERTER_SCRIPT "$JSON_FILE" \
      --host=localhost \
      --port=8086 \
      --username=admin \
      --password=admin \
      --db=k6 \
      $CONVERTER_ARGS 2>&1; then
        echo ""
        echo -e "${GREEN}✓ JSON data imported successfully${NC}"

        # Get final count
        FINAL_COUNT=$(docker exec k6-influxdb influx -username admin -password admin -database k6 -execute "SELECT COUNT(*) FROM http_req_duration" 2>/dev/null | tail -1 | awk '{print $2}')
        if [ -n "$FINAL_COUNT" ] && [ "$FINAL_COUNT" != "count_value" ]; then
            echo -e "${GREEN}  Total requests in database: ${FINAL_COUNT}${NC}"
        fi
    else
        IMPORT_EXIT_CODE=$?
        echo ""
        echo -e "${RED}✗ Failed to import JSON data (exit code: $IMPORT_EXIT_CODE)${NC}"
        echo -e "${YELLOW}Possible issues:${NC}"
        echo -e "${YELLOW}  • JSON file may not be in k6 NDJSON format${NC}"
        echo -e "${YELLOW}  • InfluxDB may not be ready${NC}"
        echo -e "${YELLOW}  • Check the error messages above${NC}"
        echo ""
        echo -e "${YELLOW}Try checking InfluxDB logs:${NC}"
        echo -e "${YELLOW}  cd .. && docker-compose logs influxdb${NC}"
        deactivate
        cd ..
        exit 1
    fi
fi
echo ""

# Deactivate venv (only if Python was used)
if [ "$USE_GO" = false ]; then
    deactivate
fi

cd ..

# Final instructions
echo -e "${GREEN}========================================${NC}"
echo -e "${GREEN}Setup Complete! 🎉${NC}"
echo -e "${GREEN}========================================${NC}"
echo ""
echo -e "${BLUE}Services:${NC}"
echo -e "  • Grafana:   ${GREEN}http://localhost:3000${NC} (admin/admin)"
echo -e "  • InfluxDB:  ${GREEN}http://localhost:8086${NC} (admin/admin)"
echo ""
echo -e "${BLUE}Next Steps:${NC}"
echo -e "  1. Grafana opened automatically with the K6 Dashboard"
echo -e "  2. Time range is preset to match your test data"
echo -e "  3. View your test results!"
echo ""
echo -e "${YELLOW}Important:${NC}"
echo -e "  • Time range is automatically calculated from your test data"
echo -e "  • If some panels show 'No data', try adjusting the time range slightly"
echo -e "  • Dashboard data source is configured as 'InfluxDB-k6'"
echo -e "  • Dashboard: K6 Dashboard (ID: 14801) - Modern, community-maintained"
echo ""
echo -e "${BLUE}To stop the services:${NC}"
echo -e "  cd $(pwd) && docker-compose down"
echo ""
echo -e "${BLUE}To restart the services:${NC}"
echo -e "  cd $(pwd) && docker-compose up -d"
echo ""
echo -e "${BLUE}To remove everything (including data):${NC}"
echo -e "  cd $(pwd) && docker-compose down -v"
echo ""

# Extract time range from JSON file
echo -e "${YELLOW}Calculating time range from test data...${NC}"

# Get first and last timestamps from JSON
FIRST_TIME=$(grep -o '"time":"[^"]*"' "$JSON_FILE" | head -1 | sed 's/"time":"//;s/"//')
LAST_TIME=$(grep -o '"time":"[^"]*"' "$JSON_FILE" | tail -1 | sed 's/"time":"//;s/"//')

if [ -n "$FIRST_TIME" ] && [ -n "$LAST_TIME" ]; then
    # Convert to milliseconds for Grafana URL (with 5 minute buffer before/after)
    FROM_MS=$(python3 -c "
import datetime
dt = datetime.datetime.fromisoformat('$FIRST_TIME'.replace('Z', '+00:00'))
# Subtract 5 minutes for buffer
dt = dt - datetime.timedelta(minutes=5)
print(int(dt.timestamp() * 1000))
" 2>/dev/null)

    TO_MS=$(python3 -c "
import datetime
dt = datetime.datetime.fromisoformat('$LAST_TIME'.replace('Z', '+00:00'))
# Add 5 minutes for buffer
dt = dt + datetime.timedelta(minutes=5)
print(int(dt.timestamp() * 1000))
" 2>/dev/null)

    if [ -n "$FROM_MS" ] && [ -n "$TO_MS" ]; then
        echo -e "${GREEN}✓ Time range detected: $FIRST_TIME to $LAST_TIME${NC}"
        DASHBOARD_URL="http://localhost:3000/d/9lcthCWnk/k6-dashboard?from=${FROM_MS}&to=${TO_MS}"
    else
        echo -e "${YELLOW}⚠ Could not calculate time range, using default${NC}"
        DASHBOARD_URL="http://localhost:3000"
    fi
else
    echo -e "${YELLOW}⚠ Could not extract timestamps from JSON file${NC}"
    DASHBOARD_URL="http://localhost:3000"
fi

# Open Grafana in browser
echo -e "${YELLOW}Opening Grafana dashboard...${NC}"
sleep 2

# Detect OS and open browser
if [[ "$OSTYPE" == "darwin"* ]]; then
    # macOS
    open "$DASHBOARD_URL"
elif [[ "$OSTYPE" == "linux-gnu"* ]]; then
    # Linux
    if command_exists xdg-open; then
        xdg-open "$DASHBOARD_URL" 2>/dev/null
    elif command_exists gnome-open; then
        gnome-open "$DASHBOARD_URL" 2>/dev/null
    else
        echo -e "${YELLOW}⚠ Could not auto-open browser. Please manually navigate to ${DASHBOARD_URL}${NC}"
    fi
elif [[ "$OSTYPE" == "msys" || "$OSTYPE" == "cygwin" ]]; then
    # Windows (Git Bash or Cygwin)
    start "$DASHBOARD_URL"
else
    echo -e "${YELLOW}⚠ Could not auto-open browser. Please manually navigate to ${DASHBOARD_URL}${NC}"
fi

echo -e "${GREEN}✓ Browser opened with time range preset${NC}"
echo ""
	services:
	influxdb:
	image: influxdb:1.8
	container_name: k6-influxdb
	ports:
	- "8086:8086"
	environment:
	- INFLUXDB_DB=k6
	- INFLUXDB_ADMIN_USER=admin
	- INFLUXDB_ADMIN_PASSWORD=admin
	- INFLUXDB_HTTP_AUTH_ENABLED=true
	volumes:
	- influxdb-data:/var/lib/influxdb
	networks:
	- k6-network
	deploy:
	resources:
	limits:
	cpus: '8.0'
	memory: 8G
	reservations:
	cpus: '4.0'
	memory: 4G

	grafana:
	image: grafana/grafana:latest
	container_name: k6-grafana
	ports:
	- "3000:3000"
	environment:
	- GF_SECURITY_ADMIN_USER=admin
	- GF_SECURITY_ADMIN_PASSWORD=admin
	- GF_AUTH_ANONYMOUS_ENABLED=true
	- GF_AUTH_ANONYMOUS_ORG_ROLE=Viewer
	volumes:
	- grafana-data:/var/lib/grafana
	depends_on:
	- influxdb
	networks:
	- k6-network
	deploy:
	resources:
	limits:
	cpus: '2.0'
	memory: 2G
	reservations:
	cpus: '0.5'
	memory: 512M

	volumes:
	influxdb-data:
	grafana-data:

	networks:
	k6-network:
	driver: bridge
	module k6-to-influxdb

	go 1.25.3

	require (
	github.com/davecgh/go-spew v1.1.1 // indirect
	github.com/json-iterator/go v1.1.12 // indirect
	github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421 // indirect
	github.com/modern-go/reflect2 v1.0.2 // indirect
	github.com/pmezard/go-difflib v1.0.0 // indirect
	github.com/stretchr/objx v0.5.0 // indirect
	github.com/stretchr/testify v1.8.4 // indirect
	gopkg.in/yaml.v3 v3.0.1 // indirect
	)
	package main

	import (
	"bufio"
	"bytes"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"os"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	jsoniter "github.com/json-iterator/go"
	)

	var json = jsoniter.ConfigCompatibleWithStandardLibrary

	// K6Point represents a single k6 metric point from JSON
	type K6Point struct {
	Type string `json:"type"`
	Metric string `json:"metric"`
	Data K6Data `json:"data"`
	}

	// K6Data represents the data field in a k6 point
	type K6Data struct {
	Time string `json:"time"`
	Value float64 `json:"value"`
	Tags map[string]string `json:"tags"`
	}

	// Config holds the importer configuration
	type Config struct {
	FilePath string
	Host string
	Port int
	Protocol string // "udp" or "http"
	Database string
	Username string
	Password string
	BatchSize int
	Workers int
	Verbose bool
	}

	// Validate checks if the configuration is valid
	func (c *Config) Validate() error {
	if c.FilePath == "" {
	return errors.New("file path is required")
	}
	if c.BatchSize <= 0 {
	return errors.New("batch size must be positive")
	}
	if c.Workers <= 0 {
	return errors.New("number of workers must be positive")
	}
	if c.Database == "" {
	return errors.New("database name is required")
	}
	return nil
	}

	// Stats tracks import statistics using atomic counters
	type Stats struct {
	pointsProcessed atomic.Int64
	batchesWritten atomic.Int64
	bytesRead atomic.Int64
	writeTimeNs atomic.Int64
	startTime time.Time
	}

	// NewStats creates a new Stats instance
	func NewStats() *Stats {
	return &Stats{
	startTime: time.Now(),
	}
	}

	// AddPoints increments the points processed counter
	func (s *Stats) AddPoints(n int64) {
	s.pointsProcessed.Add(n)
	}

	// AddBatches increments the batches written counter
	func (s *Stats) AddBatches(n int64) {
	s.batchesWritten.Add(n)
	}

	// AddBytes increments the bytes read counter
	func (s *Stats) AddBytes(n int64) {
	s.bytesRead.Add(n)
	}

	// AddWriteTime adds to the total write time in nanoseconds
	func (s *Stats) AddWriteTime(duration time.Duration) {
	s.writeTimeNs.Add(duration.Nanoseconds())
	}

	// PointsProcessed returns the current points processed count
	func (s *Stats) PointsProcessed() int64 {
	return s.pointsProcessed.Load()
	}

	// BatchesWritten returns the current batches written count
	func (s *Stats) BatchesWritten() int64 {
	return s.batchesWritten.Load()
	}

	// BytesRead returns the current bytes read count
	func (s *Stats) BytesRead() int64 {
	return s.bytesRead.Load()
	}

	// ElapsedTime returns the elapsed time since start
	func (s *Stats) ElapsedTime() time.Duration {
	return time.Since(s.startTime)
	}

	// Throughput returns points per second
	func (s *Stats) Throughput() float64 {
	elapsed := s.ElapsedTime().Seconds()
	if elapsed == 0 {
	return 0
	}
	return float64(s.PointsProcessed()) / elapsed
	}

	// Writer interface for writing batches to InfluxDB
	type Writer interface {
	Write(batch []K6Point) error
	Close() error
	}

	// HTTPWriter writes to InfluxDB via HTTP API
	type HTTPWriter struct {
	client *http.Client
	writeURL string
	username string
	password string
	}

	// NewHTTPWriter creates a new HTTP writer
	func NewHTTPWriter(host string, port int, database, username, password string) (*HTTPWriter, error) {
	// Build the write URL
	writeURL := fmt.Sprintf("http://%s:%d/write?db=%s", host, port, url.QueryEscape(database))

	return &HTTPWriter{
	client: &http.Client{
	Timeout: 30 * time.Second,
	},
	writeURL: writeURL,
	username: username,
	password: password,
	}, nil
	}

	// Write sends a batch of points via HTTP
	func (w *HTTPWriter) Write(batch []K6Point) error {
	// Convert batch to line protocol
	lineProtocol, err := formatBatchAsLineProtocol(batch)
	if err != nil {
	return fmt.Errorf("failed to format batch: %w", err)
	}

	// Create HTTP request
	req, err := http.NewRequest("POST", w.writeURL, bytes.NewBufferString(lineProtocol))
	if err != nil {
	return fmt.Errorf("failed to create HTTP request: %w", err)
	}

	// Add basic auth if credentials provided
	if w.username != "" {
	req.SetBasicAuth(w.username, w.password)
	}

	req.Header.Set("Content-Type", "application/octet-stream")

	// Send HTTP POST request
	resp, err := w.client.Do(req)
	if err != nil {
	return fmt.Errorf("failed to send HTTP request: %w", err)
	}
	defer resp.Body.Close()

	// Check response status
	if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusNoContent {
	body, _ := io.ReadAll(resp.Body)
	return fmt.Errorf("HTTP write failed with status %d: %s", resp.StatusCode, string(body))
	}

	return nil
	}

	// Close closes any resources
	func (w *HTTPWriter) Close() error {
	return nil
	}

	// parseTimeToNanos parses RFC3339 time string to nanoseconds since epoch
	func parseTimeToNanos(timeStr string) (int64, error) {
	t, err := time.Parse(time.RFC3339Nano, timeStr)
	if err != nil {
	// Try without nanoseconds
	t, err = time.Parse(time.RFC3339, timeStr)
	if err != nil {
	return 0, fmt.Errorf("failed to parse time: %w", err)
	}
	}
	return t.UnixNano(), nil
	}

	// escapeTag escapes special characters in tag keys and values
	func escapeTag(s string) string {
	s = strings.ReplaceAll(s, ",", "\\,")
	s = strings.ReplaceAll(s, "=", "\\=")
	s = strings.ReplaceAll(s, " ", "\\ ")
	return s
	}

	// escapeFieldValue escapes field values (strings)
	func escapeFieldValue(s string) string {
	s = strings.ReplaceAll(s, "\\", "\\\\")
	s = strings.ReplaceAll(s, "\"", "\\\"")
	return fmt.Sprintf("\"%s\"", s)
	}

	// convertToLineProtocol converts a K6Point to InfluxDB line protocol format
	func convertToLineProtocol(point K6Point) (string, error) {
	// Parse timestamp
	nanos, err := parseTimeToNanos(point.Data.Time)
	if err != nil {
	return "", err
	}

	// Build tags string, filtering out empty values
	var tags []string
	for k, v := range point.Data.Tags {
	// Skip tags with empty values - InfluxDB doesn't accept them
	if v != "" {
	tags = append(tags, fmt.Sprintf("%s=%s", escapeTag(k), escapeTag(v)))
	}
	}
	tagsStr := ""
	if len(tags) > 0 {
	tagsStr = "," + strings.Join(tags, ",")
	}

	// Build line protocol: measurement,tag1=value1,tag2=value2 field1=value1 timestamp
	line := fmt.Sprintf("%s%s value=%v %d",
	escapeTag(point.Metric),
	tagsStr,
	point.Data.Value,
	nanos,
	)

	return line, nil
	}

	// formatBatchAsLineProtocol converts a batch of K6Points to line protocol
	func formatBatchAsLineProtocol(batch []K6Point) (string, error) {
	var lines []string

	for _, point := range batch {
	line, err := convertToLineProtocol(point)
	if err != nil {
	return "", err
	}
	lines = append(lines, line)
	}

	return strings.Join(lines, "\n"), nil
	}

	// processFile reads and processes the JSON file
	func processFile(config Config, writer Writer, stats *Stats) error {
	file, err := os.Open(config.FilePath)
	if err != nil {
	return fmt.Errorf("failed to open file: %w", err)
	}
	defer file.Close()

	// Get file size for progress tracking
	fileInfo, err := file.Stat()
	if err != nil {
	return fmt.Errorf("failed to stat file: %w", err)
	}
	fileSize := fileInfo.Size()

	// Create batch channel
	batchChan := make(chan []K6Point, 64)

	// Start writer workers
	var wg sync.WaitGroup
	errorChan := make(chan error, config.Workers)

	for i := 0; i < config.Workers; i++ {
	wg.Add(1)
	go writerWorker(writer, batchChan, &wg, stats, errorChan)
	}

	// Start progress reporter
	stopProgress := make(chan bool)
	go progressReporter(stats, fileSize, stopProgress, config.Verbose)

	// Read and batch points
	scanner := bufio.NewScanner(file)
	buf := make([]byte, 0, 64*1024)
	scanner.Buffer(buf, 1010241024) // 10MB max line size

	currentBatch := make([]K6Point, 0, config.BatchSize)
	lineNumber := 0

	for scanner.Scan() {
	lineNumber++
	line := scanner.Bytes()
	stats.AddBytes(int64(len(line) + 1)) // +1 for newline

	var point K6Point
	if err := json.Unmarshal(line, &point); err != nil {
	// Skip invalid JSON lines
	if config.Verbose {
	fmt.Fprintf(os.Stderr, "\nWarning: Failed to parse line %d: %v\n", lineNumber, err)
	}
	continue
	}

	// Only process "Point" type entries
	if point.Type == "Point" {
	currentBatch = append(currentBatch, point)

	// Send batch when full
	if len(currentBatch) >= config.BatchSize {
	// Check for errors from workers
	select {
	case err := <-errorChan:
	close(batchChan)
	stopProgress <- true
	return err
	default:
	}

	batchChan <- currentBatch
	currentBatch = make([]K6Point, 0, config.BatchSize)
	}
	}
	}

	if err := scanner.Err(); err != nil {
	close(batchChan)
	stopProgress <- true
	return fmt.Errorf("error reading file: %w", err)
	}

	// Send final batch
	if len(currentBatch) > 0 {
	batchChan <- currentBatch
	}

	// Close channel and wait for workers
	close(batchChan)
	wg.Wait()

	// Check for any errors from workers
	close(errorChan)
	for err := range errorChan {
	stopProgress <- true
	return err
	}

	stopProgress <- true
	return nil
	}

	// writerWorker consumes batches and writes them to InfluxDB
	func writerWorker(writer Writer, batches <-chan []K6Point, wg sync.WaitGroup, stats Stats, errorChan chan<- error) {
	defer wg.Done()

	for batch := range batches {
	start := time.Now()

	if err := writer.Write(batch); err != nil {
	errorChan <- fmt.Errorf("failed to write batch: %w", err)
	return
	}

	stats.AddWriteTime(time.Since(start))
	stats.AddBatches(1)
	stats.AddPoints(int64(len(batch)))
	}
	}

	// progressReporter displays progress updates
	func progressReporter(stats *Stats, fileSize int64, stop <-chan bool, verbose bool) {
	if !verbose {
	return
	}

	ticker := time.NewTicker(500 * time.Millisecond)
	defer ticker.Stop()

	for {
	select {
	case <-stop:
	// Final update
	printProgress(stats, fileSize)
	fmt.Println() // New line after progress
	return
	case <-ticker.C:
	printProgress(stats, fileSize)
	}
	}
	}

	// printProgress prints current progress
	func printProgress(stats *Stats, fileSize int64) {
	bytesRead := stats.BytesRead()
	pointsProcessed := stats.PointsProcessed()
	throughput := stats.Throughput()

	var progress float64
	if fileSize > 0 {
	progress = float64(bytesRead) / float64(fileSize) * 100
	}

	elapsed := stats.ElapsedTime()
	var eta time.Duration
	if progress > 0 {
	totalTime := time.Duration(float64(elapsed) / progress * 100)
	eta = totalTime - elapsed
	}

	fmt.Printf("\rProgress: %.1f%% \| %d points \| %.0f pts/sec \| Elapsed: %s \| ETA: %s",
	progress,
	pointsProcessed,
	throughput,
	formatDuration(elapsed),
	formatDuration(eta),
	)
	}

	// formatDuration formats a duration nicely
	func formatDuration(d time.Duration) string {
	if d < 0 {
	return "N/A"
	}
	if d < time.Minute {
	return fmt.Sprintf("%ds", int(d.Seconds()))
	}
	minutes := int(d.Minutes())
	seconds := int(d.Seconds()) % 60
	return fmt.Sprintf("%dm%ds", minutes, seconds)
	}
	package main

	import (
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	)

	// Test K6Point JSON unmarshaling
	func TestK6PointUnmarshal(t *testing.T) {
	jsonStr := `{
	"type": "Point",
	"metric": "http_req_duration",
	"data": {
	"time": "2024-01-22T10:00:00Z",
	"value": 123.45,
	"tags": {
	"name": "test",
	"status": "200"
	}
	}
	}`

	var point K6Point
	err := json.Unmarshal([]byte(jsonStr), &point)

	require.NoError(t, err)
	assert.Equal(t, "Point", point.Type)
	assert.Equal(t, "http_req_duration", point.Metric)
	assert.Equal(t, "2024-01-22T10:00:00Z", point.Data.Time)
	assert.Equal(t, 123.45, point.Data.Value)
	assert.Equal(t, "test", point.Data.Tags["name"])
	assert.Equal(t, "200", point.Data.Tags["status"])
	}

	// Test converting K6Point to InfluxDB line protocol
	func TestConvertToLineProtocol(t *testing.T) {
	k6Point := K6Point{
	Type: "Point",
	Metric: "http_req_duration",
	Data: K6Data{
	Time: "2024-01-22T10:00:00Z",
	Value: 123.45,
	Tags: map[string]string{
	"name": "test",
	"status": "200",
	},
	},
	}

	line, err := convertToLineProtocol(k6Point)

	require.NoError(t, err)
	assert.Contains(t, line, "http_req_duration")
	assert.Contains(t, line, "name=test")
	assert.Contains(t, line, "status=200")
	assert.Contains(t, line, "value=123.45")
	}

	// Test batch formatting to line protocol
	func TestFormatBatchAsLineProtocol(t *testing.T) {
	batch := []K6Point{
	{
	Type: "Point",
	Metric: "metric1",
	Data: K6Data{
	Time: "2024-01-22T10:00:00Z",
	Value: 100.0,
	Tags: map[string]string{"tag1": "value1"},
	},
	},
	{
	Type: "Point",
	Metric: "metric2",
	Data: K6Data{
	Time: "2024-01-22T10:00:01Z",
	Value: 200.0,
	Tags: map[string]string{"tag2": "value2"},
	},
	},
	}

	result, err := formatBatchAsLineProtocol(batch)

	require.NoError(t, err)
	assert.Contains(t, result, "metric1")
	assert.Contains(t, result, "metric2")
	assert.Contains(t, result, "\n")
	}

	// Test time parsing and conversion to nanoseconds
	func TestParseTime(t *testing.T) {
	tests := []struct {
	name string
	timeStr string
	wantErr bool
	}{
	{
	name: "RFC3339 format",
	timeStr: "2024-01-22T10:00:00Z",
	wantErr: false,
	},
	{
	name: "RFC3339 with microseconds",
	timeStr: "2024-01-22T10:00:00.123456Z",
	wantErr: false,
	},
	{
	name: "Invalid format",
	timeStr: "not a time",
	wantErr: true,
	},
	}

	for _, tt := range tests {
	t.Run(tt.name, func(t *testing.T) {
	nanos, err := parseTimeToNanos(tt.timeStr)

	if tt.wantErr {
	assert.Error(t, err)
	} else {
	require.NoError(t, err)
	assert.Greater(t, nanos, int64(0))
	}
	})
	}
	}

	// Test tag escaping for InfluxDB line protocol
	func TestEscapeTag(t *testing.T) {
	tests := []struct {
	input string
	expected string
	}{
	{input: "simple", expected: "simple"},
	{input: "with space", expected: "with\\ space"},
	{input: "with,comma", expected: "with\\,comma"},
	{input: "with=equals", expected: "with\\=equals"},
	{input: "multiple spaces here", expected: "multiple\\ spaces\\ \\ here"},
	}

	for _, tt := range tests {
	t.Run(tt.input, func(t *testing.T) {
	result := escapeTag(tt.input)
	assert.Equal(t, tt.expected, result)
	})
	}
	}

	// Test field value escaping
	func TestEscapeFieldValue(t *testing.T) {
	tests := []struct {
	input string
	expected string
	}{
	{input: "simple", expected: "\"simple\""},
	{input: "with\"quote", expected: "\"with\\\"quote\""},
	{input: "with\\backslash", expected: "\"with\\\\backslash\""},
	}

	for _, tt := range tests {
	t.Run(tt.input, func(t *testing.T) {
	result := escapeFieldValue(tt.input)
	assert.Equal(t, tt.expected, result)
	})
	}
	}

	// Test stats tracking
	func TestStatsTracking(t *testing.T) {
	stats := NewStats()

	stats.AddPoints(100)
	stats.AddBatches(1)
	stats.AddBytes(1000)

	assert.Equal(t, int64(100), stats.PointsProcessed())
	assert.Equal(t, int64(1), stats.BatchesWritten())
	assert.Equal(t, int64(1000), stats.BytesRead())
	}

	// Test config validation
	func TestConfigValidation(t *testing.T) {
	tests := []struct {
	name string
	config Config
	wantErr bool
	}{
	{
	name: "Valid config",
	config: Config{
	FilePath: "/path/to/file.json",
	Host: "localhost",
	Port: 8086,
	Database: "k6",
	BatchSize: 2000,
	Workers: 4,
	},
	wantErr: false,
	},
	{
	name: "Missing file path",
	config: Config{
	Host: "localhost",
	Port: 8086,
	Database: "k6",
	BatchSize: 2000,
	Workers: 4,
	},
	wantErr: true,
	},
	{
	name: "Invalid batch size",
	config: Config{
	FilePath: "/path/to/file.json",
	Host: "localhost",
	Port: 8086,
	Database: "k6",
	BatchSize: 0,
	Workers: 4,
	},
	wantErr: true,
	},
	{
	name: "Invalid workers",
	config: Config{
	FilePath: "/path/to/file.json",
	Host: "localhost",
	Port: 8086,
	Database: "k6",
	BatchSize: 2000,
	Workers: 0,
	},
	wantErr: true,
	},
	{
	name: "Missing database",
	config: Config{
	FilePath: "/path/to/file.json",
	Host: "localhost",
	Port: 8086,
	BatchSize: 2000,
	Workers: 4,
	},
	wantErr: true,
	},
	}

	for _, tt := range tests {
	t.Run(tt.name, func(t *testing.T) {
	err := tt.config.Validate()

	if tt.wantErr {
	assert.Error(t, err)
	} else {
	assert.NoError(t, err)
	}
	})
	}
	}

	// Test duration formatting
	func TestFormatDuration(t *testing.T) {
	tests := []struct {
	name string
	duration time.Duration
	expected string
	}{
	{
	name: "Negative duration",
	duration: -1 * time.Second,
	expected: "N/A",
	},
	{
	name: "Less than a minute",
	duration: 45 * time.Second,
	expected: "45s",
	},
	{
	name: "Exactly one minute",
	duration: 60 * time.Second,
	expected: "1m0s",
	},
	{
	name: "Multiple minutes",
	duration: 3time.Minute + 25time.Second,
	expected: "3m25s",
	},
	}

	for _, tt := range tests {
	t.Run(tt.name, func(t *testing.T) {
	result := formatDuration(tt.duration)
	assert.Equal(t, tt.expected, result)
	})
	}
	}

	// Test stats methods
	func TestStatsElapsedAndThroughput(t *testing.T) {
	stats := NewStats()

	stats.AddPoints(1000)
	time.Sleep(10 * time.Millisecond)

	elapsed := stats.ElapsedTime()
	assert.Greater(t, elapsed, time.Duration(0))

	throughput := stats.Throughput()
	assert.Greater(t, throughput, 0.0)
	}

	// Test line protocol with no tags
	func TestConvertToLineProtocolNoTags(t *testing.T) {
	k6Point := K6Point{
	Type: "Point",
	Metric: "simple_metric",
	Data: K6Data{
	Time: "2024-01-22T10:00:00Z",
	Value: 42.0,
	Tags: map[string]string{},
	},
	}

	line, err := convertToLineProtocol(k6Point)

	require.NoError(t, err)
	assert.Contains(t, line, "simple_metric")
	assert.Contains(t, line, "value=42")
	assert.NotContains(t, line, ",,")
	}

	// Test line protocol with special characters in metric name
	func TestConvertToLineProtocolSpecialChars(t *testing.T) {
	k6Point := K6Point{
	Type: "Point",
	Metric: "metric with spaces",
	Data: K6Data{
	Time: "2024-01-22T10:00:00Z",
	Value: 100.0,
	Tags: map[string]string{
	"tag with space": "value with space",
	},
	},
	}

	line, err := convertToLineProtocol(k6Point)

	require.NoError(t, err)
	assert.Contains(t, line, "metric\\ with\\ spaces")
	assert.Contains(t, line, "tag\\ with\\ space=value\\ with\\ space")
	}

	// Test line protocol with invalid time
	func TestConvertToLineProtocolInvalidTime(t *testing.T) {
	k6Point := K6Point{
	Type: "Point",
	Metric: "test",
	Data: K6Data{
	Time: "invalid time",
	Value: 100.0,
	Tags: map[string]string{},
	},
	}

	_, err := convertToLineProtocol(k6Point)
	assert.Error(t, err)
	}

	// Test batch formatting with error
	func TestFormatBatchAsLineProtocolWithError(t *testing.T) {
	batch := []K6Point{
	{
	Type: "Point",
	Metric: "good_metric",
	Data: K6Data{
	Time: "2024-01-22T10:00:00Z",
	Value: 100.0,
	Tags: map[string]string{},
	},
	},
	{
	Type: "Point",
	Metric: "bad_metric",
	Data: K6Data{
	Time: "invalid",
	Value: 200.0,
	Tags: map[string]string{},
	},
	},
	}

	_, err := formatBatchAsLineProtocol(batch)
	assert.Error(t, err)
	}

	// Test HTTP writer creation
	func TestNewHTTPWriter(t *testing.T) {
	writer, err := NewHTTPWriter("localhost", 8086, "testdb", "admin", "admin")

	require.NoError(t, err)
	assert.NotNil(t, writer)

	err = writer.Close()
	assert.NoError(t, err)
	}

	// Test stats write time tracking
	func TestStatsWriteTime(t *testing.T) {
	stats := NewStats()

	stats.AddWriteTime(100 * time.Millisecond)
	stats.AddWriteTime(200 * time.Millisecond)

	// We don't expose write time directly, but we can verify it doesn't panic
	assert.Greater(t, stats.PointsProcessed(), int64(-1))
	}
Batch Size	Throughput	Notes
1,000	130,566 pts/sec	Too many HTTP requests
2,000	180,935 pts/sec	✅ Optimal
5,000	142,377 pts/sec	Good balance
10,000	117,450 pts/sec	Default for many tools
25,000	53,313 pts/sec	Too large
50,000	43,720 pts/sec	Network overhead dominates
Option	Default	Description
`--file`	required	Path to k6 JSON file
`--host`	localhost	InfluxDB host
`--port`	8086	InfluxDB HTTP port
`--database`	k6	InfluxDB database name
`--username`	""	InfluxDB username
`--password`	""	InfluxDB password
`--batch-size`	2000	Points per batch (optimal: 2000)
`--workers`	8	Number of concurrent workers
`--verbose`	true	Enable progress output
Metric	Python (Multi-threaded)	Go (HTTP, 2K batch)	Improvement
Throughput	~30,000 pts/sec	~181,000 pts/sec	6.0x faster
14GB file	25+ minutes	4-5 minutes	~5-6x faster
CPU usage	99.6% (1 core)	400-800% (4-8 cores)	True parallelism
Memory	484MB → 1.2GB	~50-200 MB	Constant memory
Concurrency	Limited by GIL	True parallelism	All cores used
	package main

	import (
	"flag"
	"fmt"
	"os"
	)

	func main() {
	// Parse command line flags
	config := Config{
	Host: "localhost",
	Port: 8086,
	Database: "k6",
	Username: "",
	Password: "",
	BatchSize: 2000,
	Workers: 8,
	Verbose: true,
	}

	flag.StringVar(&config.FilePath, "file", "", "Path to k6 JSON file (required)")
	flag.StringVar(&config.Host, "host", config.Host, "InfluxDB host")
	flag.IntVar(&config.Port, "port", config.Port, "InfluxDB port")
	flag.StringVar(&config.Database, "database", config.Database, "InfluxDB database name")
	flag.StringVar(&config.Username, "username", config.Username, "InfluxDB username")
	flag.StringVar(&config.Password, "password", config.Password, "InfluxDB password")
	flag.IntVar(&config.BatchSize, "batch-size", config.BatchSize, "Number of points per batch (recommended: 2000)")
	flag.IntVar(&config.Workers, "workers", config.Workers, "Number of concurrent writer workers")
	flag.BoolVar(&config.Verbose, "verbose", config.Verbose, "Enable verbose output")

	flag.Usage = func() {
	fmt.Fprintf(os.Stderr, "Usage: %s [options]\n\n", os.Args[0])
	fmt.Fprintf(os.Stderr, "k6-to-influxdb imports k6 JSON output to InfluxDB via HTTP\n\n")
	fmt.Fprintf(os.Stderr, "Options:\n")
	flag.PrintDefaults()
	fmt.Fprintf(os.Stderr, "\nExample:\n")
	fmt.Fprintf(os.Stderr, " %s --file=k6-output.json --username=admin --password=admin\n", os.Args[0])
	}

	flag.Parse()

	// Validate configuration
	if err := config.Validate(); err != nil {
	fmt.Fprintf(os.Stderr, "Configuration error: %v\n\n", err)
	flag.Usage()
	os.Exit(1)
	}

	// Check if file exists
	if _, err := os.Stat(config.FilePath); os.IsNotExist(err) {
	fmt.Fprintf(os.Stderr, "Error: File does not exist: %s\n", config.FilePath)
	os.Exit(1)
	}

	// Create HTTP writer
	if config.Verbose {
	fmt.Printf("Starting k6-to-influxdb importer\n")
	fmt.Printf("File: %s\n", config.FilePath)
	fmt.Printf("Target: http://%s:%d/%s\n", config.Host, config.Port, config.Database)
	fmt.Printf("Batch size: %d points\n", config.BatchSize)
	fmt.Printf("Workers: %d\n\n", config.Workers)
	}

	writer, err := NewHTTPWriter(config.Host, config.Port, config.Database, config.Username, config.Password)
	if err != nil {
	fmt.Fprintf(os.Stderr, "Failed to create HTTP writer: %v\n", err)
	os.Exit(1)
	}
	defer writer.Close()

	// Create stats tracker
	stats := NewStats()

	// Process file
	if err := processFile(config, writer, stats); err != nil {
	fmt.Fprintf(os.Stderr, "\nError: %v\n", err)
	os.Exit(1)
	}

	// Print final summary
	if config.Verbose {
	fmt.Printf("\n\nImport completed successfully!\n")
	fmt.Printf("Total points imported: %d\n", stats.PointsProcessed())
	fmt.Printf("Total batches written: %d\n", stats.BatchesWritten())
	fmt.Printf("Total time: %s\n", formatDuration(stats.ElapsedTime()))
	fmt.Printf("Average throughput: %.0f points/sec\n", stats.Throughput())
	fmt.Printf("Total data read: %.2f MB\n", float64(stats.BytesRead())/(1024*1024))
	}
	}
	#!/bin/bash

	set -e

	# Colors for output
	RED='\033[0;31m'
	GREEN='\033[0;32m'
	YELLOW='\033[1;33m'
	BLUE='\033[0;34m'
	NC='\033[0m' # No Color

	echo -e "${BLUE}========================================${NC}"
	echo -e "${BLUE}k6 + InfluxDB + Grafana Setup Script${NC}"
	echo -e "${BLUE}========================================${NC}"
	echo ""

	# Function to check if command exists
	command_exists() {
	command -v "$1" >/dev/null 2>&1
	}

	# Check prerequisites
	echo -e "${YELLOW}Checking prerequisites...${NC}"

	if ! command_exists docker; then
	echo -e "${RED}Error: docker is not installed${NC}"
	exit 1
	fi

	if ! command_exists docker-compose && ! docker compose version >/dev/null 2>&1; then
	echo -e "${RED}Error: docker-compose is not installed${NC}"
	exit 1
	fi

	if ! command_exists python3; then
	echo -e "${RED}Error: python3 is not installed${NC}"
	exit 1
	fi

	if ! command_exists git; then
	echo -e "${RED}Error: git is not installed${NC}"
	exit 1
	fi

	if ! command_exists jq; then
	echo -e "${YELLOW}Warning: jq is not installed. Dashboard import may fail.${NC}"
	echo -e "${YELLOW}Install jq with: brew install jq (macOS) or apt-get install jq (Linux)${NC}"
	JQ_AVAILABLE=false
	else
	JQ_AVAILABLE=true
	fi

	echo -e "${GREEN}✓ All prerequisites met${NC}"
	echo ""

	# Get the script directory
	SCRIPT_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"

	# Prompt for JSON file
	echo -e "${YELLOW}Enter the path to your k6 JSON results file:${NC}"
	echo -e "${YELLOW}(You can use a filename, relative path, absolute path, or ~/path)${NC}"
	read -e -r JSON_FILE_INPUT

	# Sanitize input - remove any dangerous characters
	# Allow: alphanumeric, dash, underscore, dot, slash, tilde, space
	JSON_FILE_INPUT=$(echo "$JSON_FILE_INPUT" \| sed 's/[^a-zA-Z0-9._/~[:space:]-]//g')

	# Trim leading/trailing whitespace
	JSON_FILE_INPUT=$(echo "$JSON_FILE_INPUT" \| sed -e 's/^[[:space:]]//' -e 's/[[:space:]]$//')

	if [ -z "$JSON_FILE_INPUT" ]; then
	echo -e "${RED}Error: No file path provided${NC}"
	exit 1
	fi

	# Expand tilde to home directory
	JSON_FILE_INPUT="${JSON_FILE_INPUT/#\~/$HOME}"

	# Try to find the file in multiple locations
	JSON_FILE=""

	# 1. Try as absolute or relative path from current directory
	if [ -f "$JSON_FILE_INPUT" ]; then
	JSON_FILE="$JSON_FILE_INPUT"
	echo -e "${GREEN}✓ Found JSON file in current directory${NC}"
	# 2. Try relative to script directory
	elif [ -f "$SCRIPT_DIR/$JSON_FILE_INPUT" ]; then
	JSON_FILE="$SCRIPT_DIR/$JSON_FILE_INPUT"
	echo -e "${GREEN}✓ Found JSON file in script directory${NC}"
	# 3. Check if it's just a filename and search for it in script directory
	elif [[ "$JSON_FILE_INPUT" != "/" ]]; then
	# It's just a filename, try to find it
	FOUND_FILES=$(find "$SCRIPT_DIR" -maxdepth 2 -type f -name "$JSON_FILE_INPUT" 2>/dev/null)
	FILE_COUNT=$(echo "$FOUND_FILES" \| grep -c .)

	if [ -n "$FOUND_FILES" ] && [ "$FILE_COUNT" -eq 1 ]; then
	JSON_FILE="$FOUND_FILES"
	echo -e "${GREEN}✓ Found JSON file: $(basename "$JSON_FILE")${NC}"
	elif [ "$FILE_COUNT" -gt 1 ]; then
	echo -e "${RED}Error: Multiple files found matching '$JSON_FILE_INPUT':${NC}"
	echo "$FOUND_FILES"
	echo -e "${YELLOW}Please provide a more specific path${NC}"
	exit 1
	fi
	fi

	# If still not found, show error with helpful message
	if [ -z "$JSON_FILE" ] \|\| [ ! -f "$JSON_FILE" ]; then
	echo -e "${RED}Error: File not found: $JSON_FILE_INPUT${NC}"
	echo ""
	echo -e "${YELLOW}Searched in:${NC}"
	echo -e " • Current directory: $(pwd)"
	echo -e " • Script directory: $SCRIPT_DIR"
	echo ""
	echo -e "${YELLOW}Available JSON files in script directory:${NC}"
	find "$SCRIPT_DIR" -maxdepth 2 -type f -name "*.json" 2>/dev/null \| head -10 \| while read -r file; do
	echo -e " • $(basename "$file")"
	done
	exit 1
	fi

	# Convert to absolute path
	JSON_FILE=$(cd "$(dirname "$JSON_FILE")" && pwd)/$(basename "$JSON_FILE")
	echo -e "${GREEN}✓ Using file: $JSON_FILE${NC}"
	echo ""

	# Ask for project directory
	echo -e "${YELLOW}Enter directory name for the setup (default: k6-grafana-stack):${NC}"
	read -r PROJECT_DIR
	PROJECT_DIR=${PROJECT_DIR:-k6-grafana-stack}

	# Check if directory exists and clean up any existing setup
	if [ -d "$PROJECT_DIR" ]; then
	echo -e "${YELLOW}Found existing setup directory. Cleaning up...${NC}"
	cd "$PROJECT_DIR"

	# Stop and remove any running containers
	if [ -f "docker-compose.yml" ]; then
	docker-compose down -v > /dev/null 2>&1
	echo -e "${GREEN}✓ Stopped and removed existing containers${NC}"
	fi

	# Remove old converter directory if it exists
	if [ -d "k6-json-to-influxdb-line-protocol" ]; then
	rm -rf k6-json-to-influxdb-line-protocol
	fi

	cd ..
	echo -e "${GREEN}✓ Cleaned up existing setup${NC}"
	else
	echo -e "${YELLOW}Creating new setup directory...${NC}"
	fi

	# Create/recreate project directory
	mkdir -p "$PROJECT_DIR"
	cd "$PROJECT_DIR"
	echo -e "${GREEN}✓ Project directory ready: $PROJECT_DIR${NC}"
	echo ""

	# Create docker-compose.yml
	echo -e "${YELLOW}Creating docker-compose.yml...${NC}"
	cat > docker-compose.yml <<'EOF'
	services:
	influxdb:
	image: influxdb:1.8
	container_name: k6-influxdb
	ports:
	- "8086:8086"
	environment:
	- INFLUXDB_DB=k6
	- INFLUXDB_ADMIN_USER=admin
	- INFLUXDB_ADMIN_PASSWORD=admin
	- INFLUXDB_HTTP_AUTH_ENABLED=true
	volumes:
	- influxdb-data:/var/lib/influxdb
	networks:
	- k6-network
	deploy:
	resources:
	limits:
	cpus: '8.0'
	memory: 8G
	reservations:
	cpus: '4.0'
	memory: 4G

	grafana:
	image: grafana/grafana:latest
	container_name: k6-grafana
	ports:
	- "3000:3000"
	environment:
	- GF_SECURITY_ADMIN_USER=admin
	- GF_SECURITY_ADMIN_PASSWORD=admin
	- GF_AUTH_ANONYMOUS_ENABLED=true
	- GF_AUTH_ANONYMOUS_ORG_ROLE=Viewer
	volumes:
	- grafana-data:/var/lib/grafana
	depends_on:
	- influxdb
	networks:
	- k6-network
	deploy:
	resources:
	limits:
	cpus: '2.0'
	memory: 2G
	reservations:
	cpus: '0.5'
	memory: 512M

	volumes:
	influxdb-data:
	grafana-data:

	networks:
	k6-network:
	driver: bridge
	EOF

	echo -e "${GREEN}✓ docker-compose.yml created${NC}"
	echo ""

	# Start Docker containers
	echo -e "${YELLOW}Starting Docker containers...${NC}"
	docker-compose up -d

	echo -e "${GREEN}✓ Containers started${NC}"
	echo ""

	# Wait for services to be ready
	echo -e "${YELLOW}Waiting for services to be ready...${NC}"
	sleep 10

	# Wait for InfluxDB
	echo -n "Waiting for InfluxDB"
	INFLUX_READY=false
	for i in {1..30}; do
	if curl -s --max-time 2 http://localhost:8086/ping > /dev/null 2>&1; then
	echo -e " ${GREEN}✓${NC}"
	INFLUX_READY=true
	break
	fi
	echo -n "."
	sleep 2
	done

	if [ "$INFLUX_READY" = false ]; then
	echo -e " ${RED}✗${NC}"
	echo -e "${RED}Error: InfluxDB did not start properly${NC}"
	echo -e "${YELLOW}Check logs with: cd $PROJECT_DIR && docker-compose logs influxdb${NC}"
	exit 1
	fi

	# Wait for Grafana
	echo -n "Waiting for Grafana"
	GRAFANA_READY=false
	for i in {1..60}; do
	if curl -s --max-time 2 http://localhost:3000/api/health > /dev/null 2>&1; then
	echo -e " ${GREEN}✓${NC}"
	GRAFANA_READY=true
	break
	fi
	echo -n "."
	sleep 2

	# Check if we've been waiting too long
	if [ $i -eq 30 ]; then
	echo ""
	echo -e "${YELLOW}⚠ Grafana is taking longer than expected to start...${NC}"
	echo -n "Still waiting"
	fi
	done

	if [ "$GRAFANA_READY" = false ]; then
	echo -e " ${RED}✗${NC}"
	echo -e "${RED}Error: Grafana did not start properly${NC}"
	echo -e "${YELLOW}Check logs with: cd $PROJECT_DIR && docker-compose logs grafana${NC}"
	exit 1
	fi

	echo ""

	# Configure Grafana data source via API
	echo -e "${YELLOW}Configuring InfluxDB data source in Grafana...${NC}"

	# Wait a bit more for Grafana to fully initialize
	sleep 5

	# Try to configure data source with timeout and error capture
	DS_RESPONSE=$(curl -s --max-time 10 -w "\n%{http_code}" -X POST \
	-H "Content-Type: application/json" \
	-d '{
	"name": "InfluxDB-k6",
	"type": "influxdb",
	"url": "http://influxdb:8086",
	"access": "proxy",
	"database": "k6",
	"basicAuth": false,
	"isDefault": true,
	"jsonData": {
	"httpMode": "GET"
	},
	"secureJsonData": {
	"password": "admin"
	},
	"user": "admin"
	}' \
	http://admin:admin@localhost:3000/api/datasources 2>&1)

	HTTP_CODE=$(echo "$DS_RESPONSE" \| tail -n1)
	DS_BODY=$(echo "$DS_RESPONSE" \| head -n-1)

	if [ "$HTTP_CODE" = "200" ] \|\| [ "$HTTP_CODE" = "201" ]; then
	echo -e "${GREEN}✓ Data source configured${NC}"
	elif [ "$HTTP_CODE" = "409" ]; then
	echo -e "${YELLOW}⚠ Data source already exists (this is fine)${NC}"
	else
	echo -e "${YELLOW}⚠ Data source configuration returned HTTP $HTTP_CODE${NC}"
	echo -e "${YELLOW} You may need to manually configure it in Grafana${NC}"
	if [ -n "$DS_BODY" ]; then
	echo -e "${YELLOW} Response: $DS_BODY${NC}"
	fi
	fi
	echo ""

	# Import k6 Grafana dashboard (14801 - modern version)
	echo -e "${YELLOW}Importing K6 Dashboard (ID: 14801)...${NC}"

	if [ "$JQ_AVAILABLE" = true ]; then
	# Download dashboard JSON
	if curl -s --max-time 10 https://grafana.com/api/dashboards/14801/revisions/1/download -o k6-dashboard.json 2>&1; then
	# Check if file was downloaded
	if [ -s k6-dashboard.json ]; then
	# Create proper import payload using jq
	# Dashboard expects DS_DUMMY as the input name
	IMPORT_PAYLOAD=$(jq -n --argjson dashboard "$(cat k6-dashboard.json)" '{
	dashboard: $dashboard,
	overwrite: true,
	inputs: [
	{
	name: "DS_DUMMY",
	type: "datasource",
	pluginId: "influxdb",
	value: "InfluxDB-k6"
	}
	]
	}' 2>&1)

	if [ $? -eq 0 ]; then
	# Import dashboard via API
	IMPORT_RESPONSE=$(curl -s --max-time 10 -w "\n%{http_code}" -X POST \
	-H "Content-Type: application/json" \
	-d "$IMPORT_PAYLOAD" \
	http://admin:admin@localhost:3000/api/dashboards/import 2>&1)

	IMPORT_CODE=$(echo "$IMPORT_RESPONSE" \| tail -n1)
	IMPORT_BODY=$(echo "$IMPORT_RESPONSE" \| head -n-1)

	if [ "$IMPORT_CODE" = "200" ] \|\| [ "$IMPORT_CODE" = "201" ]; then
	echo -e "${GREEN}✓ K6 Dashboard imported${NC}"
	# Extract dashboard URL if available
	DASH_URL=$(echo "$IMPORT_BODY" \| jq -r '.importedUrl // empty' 2>/dev/null)
	if [ -n "$DASH_URL" ]; then
	echo -e "${GREEN} Dashboard URL: http://localhost:3000${DASH_URL}${NC}"
	fi
	elif [ "$IMPORT_CODE" = "412" ]; then
	echo -e "${YELLOW}⚠ Dashboard already exists (this is fine)${NC}"
	else
	echo -e "${YELLOW}⚠ Dashboard import returned HTTP $IMPORT_CODE${NC}"
	if echo "$IMPORT_BODY" \| grep -q "name-exists"; then
	echo -e "${YELLOW} Dashboard may already exist - continuing${NC}"
	else
	echo -e "${YELLOW} You can manually import dashboard ID 14801 from Grafana${NC}"
	fi
	fi
	else
	echo -e "${YELLOW}⚠ Failed to create import payload${NC}"
	echo -e "${YELLOW} You can manually import dashboard ID 14801 from Grafana${NC}"
	fi
	else
	echo -e "${YELLOW}⚠ Failed to download dashboard${NC}"
	echo -e "${YELLOW} You can manually import dashboard ID 14801 from Grafana${NC}"
	fi
	else
	echo -e "${YELLOW}⚠ Failed to download dashboard from Grafana.com${NC}"
	echo -e "${YELLOW} You can manually import dashboard ID 14801 from Grafana${NC}"
	fi
	else
	echo -e "${YELLOW}⚠ Skipping automatic dashboard import (jq not installed)${NC}"
	echo -e "${YELLOW} Manually import dashboard: Go to Grafana → Dashboards → Import → Use ID: 14801${NC}"
	fi
	echo ""

	# Clone the k6-json-to-influxdb converter
	echo -e "${YELLOW}Setting up JSON to InfluxDB converter...${NC}"

	if [ ! -d "k6-json-to-influxdb-line-protocol" ]; then
	if git clone https://github.com/yooap/k6-json-to-influxdb-line-protocol.git > /dev/null 2>&1; then
	echo -e "${GREEN}✓ Converter cloned${NC}"
	else
	echo -e "${RED}✗ Failed to clone converter repository${NC}"
	echo -e "${YELLOW}Check your internet connection or try again later${NC}"
	exit 1
	fi
	else
	echo -e "${GREEN}✓ Converter already exists${NC}"
	fi

	cd k6-json-to-influxdb-line-protocol

	# Copy optimized converters if available
	if [ -f "$SCRIPT_DIR/convert_and_write_to_db_multithreaded.py" ]; then
	cp "$SCRIPT_DIR/convert_and_write_to_db_multithreaded.py" .
	chmod +x convert_and_write_to_db_multithreaded.py
	echo -e "${GREEN}✓ Multi-threaded converter installed${NC}"
	elif [ -f "$SCRIPT_DIR/convert_and_write_to_db_streaming.py" ]; then
	cp "$SCRIPT_DIR/convert_and_write_to_db_streaming.py" .
	chmod +x convert_and_write_to_db_streaming.py
	echo -e "${GREEN}✓ Streaming converter installed${NC}"
	fi

	# Create and activate virtual environment
	echo -e "${YELLOW}Creating Python virtual environment...${NC}"
	if python3 -m venv venv 2>&1; then
	echo -e "${GREEN}✓ Virtual environment created${NC}"
	else
	echo -e "${RED}✗ Failed to create virtual environment${NC}"
	cd ..
	exit 1
	fi

	# Activate virtual environment
	source venv/bin/activate

	# Install Python dependencies
	echo -e "${YELLOW}Installing Python dependencies in venv...${NC}"
	if [ -f "requirements.txt" ]; then
	if python3 -m pip install -q -r requirements.txt 2>&1; then
	echo -e "${GREEN}✓ Dependencies installed${NC}"
	else
	echo -e "${YELLOW}⚠ Some dependencies may have failed to install${NC}"
	fi
	else
	# Install manually if requirements.txt doesn't exist
	if python3 -m pip install -q influxdb 2>&1; then
	echo -e "${GREEN}✓ Dependencies installed (influxdb)${NC}"
	else
	echo -e "${RED}✗ Failed to install influxdb package${NC}"
	deactivate
	cd ..
	exit 1
	fi
	fi
	echo ""

	# Import JSON data to InfluxDB using optimized converter
	echo -e "${YELLOW}Importing k6 JSON data to InfluxDB...${NC}"

	# Get file size for progress indication
	FILE_SIZE=$(ls -lh "$JSON_FILE" \| awk '{print $5}')
	FILE_SIZE_BYTES=$(stat -f%z "$JSON_FILE" 2>/dev/null \|\| stat -c%s "$JSON_FILE" 2>/dev/null)
	FILE_SIZE_MB=$(echo "scale=1; $FILE_SIZE_BYTES / 1024 / 1024" \| bc 2>/dev/null \|\| echo "?")

	echo -e "${BLUE}File size: ${FILE_SIZE} (${FILE_SIZE_MB} MB)${NC}"

	# Check for Go binary first (fastest option)
	USE_GO=false
	GO_BINARY=""

	# Check in script directory
	if [ -f "$SCRIPT_DIR/k6-to-influxdb" ]; then
	GO_BINARY="$SCRIPT_DIR/k6-to-influxdb"
	USE_GO=true
	elif [ -f "./k6-to-influxdb" ]; then
	GO_BINARY="./k6-to-influxdb"
	USE_GO=true
	fi

	if [ "$USE_GO" = true ]; then
	echo -e "${BLUE}Using Go importer (8 workers, 2k batch size)${NC}"
	echo -e "${GREEN}⚡ High-performance mode enabled${NC}"
	echo ""

	# Run the Go binary with optimized batch size
	if "$GO_BINARY" \
	--file="$JSON_FILE" \
	--host=localhost \
	--port=8086 \
	--database=k6 \
	--username=admin \
	--password=admin \
	--batch-size=2000 \
	--workers=8 \
	--verbose 2>&1; then
	echo ""
	echo -e "${GREEN}✓ JSON data imported successfully${NC}"

	# Get final count
	FINAL_COUNT=$(docker exec k6-influxdb influx -username admin -password admin -database k6 -execute "SELECT COUNT(*) FROM http_req_duration" 2>/dev/null \| tail -1 \| awk '{print $2}')
	if [ -n "$FINAL_COUNT" ] && [ "$FINAL_COUNT" != "count_value" ]; then
	echo -e "${GREEN} Total requests in database: ${FINAL_COUNT}${NC}"
	fi
	else
	IMPORT_EXIT_CODE=$?
	echo ""
	echo -e "${RED}✗ Failed to import JSON data (exit code: $IMPORT_EXIT_CODE)${NC}"
	echo -e "${YELLOW}Possible issues:${NC}"
	echo -e "${YELLOW} • JSON file may not be in k6 NDJSON format${NC}"
	echo -e "${YELLOW} • InfluxDB may not be ready${NC}"
	echo -e "${YELLOW} • Check the error messages above${NC}"
	echo ""
	echo -e "${YELLOW}Try checking InfluxDB logs:${NC}"
	echo -e "${YELLOW} cd .. && docker-compose logs influxdb${NC}"
	cd ..
	exit 1
	fi
	else
	# Fallback to Python converter
	echo -e "${BLUE}Go importer not found, using Python converter${NC}"
	echo -e "${YELLOW}Tip: Build Go binary for 50-100x faster imports!${NC}"
	echo ""

	# Select best Python converter
	if [ -f "convert_and_write_to_db_multithreaded.py" ]; then
	CONVERTER_SCRIPT="convert_and_write_to_db_multithreaded.py"
	# Use 100k batch size - sweet spot between performance and HTTP limits
	CONVERTER_ARGS="--verbose --workers=4 --batch-size=100000"
	echo -e "${BLUE}Using multi-threaded import (4 workers, 100k batch size)${NC}"
	elif [ -f "convert_and_write_to_db_streaming.py" ]; then
	CONVERTER_SCRIPT="convert_and_write_to_db_streaming.py"
	CONVERTER_ARGS="--verbose"
	echo -e "${BLUE}Using streaming import${NC}"
	else
	CONVERTER_SCRIPT="convert_and_write_to_db.py"
	CONVERTER_ARGS=""
	echo -e "${BLUE}Using standard import${NC}"
	fi
	echo ""

	# Run the import (in venv) with live progress output
	if python3 $CONVERTER_SCRIPT "$JSON_FILE" \
	--host=localhost \
	--port=8086 \
	--username=admin \
	--password=admin \
	--db=k6 \
	$CONVERTER_ARGS 2>&1; then
	echo ""
	echo -e "${GREEN}✓ JSON data imported successfully${NC}"

	# Get final count
	FINAL_COUNT=$(docker exec k6-influxdb influx -username admin -password admin -database k6 -execute "SELECT COUNT(*) FROM http_req_duration" 2>/dev/null \| tail -1 \| awk '{print $2}')
	if [ -n "$FINAL_COUNT" ] && [ "$FINAL_COUNT" != "count_value" ]; then
	echo -e "${GREEN} Total requests in database: ${FINAL_COUNT}${NC}"
	fi
	else
	IMPORT_EXIT_CODE=$?
	echo ""
	echo -e "${RED}✗ Failed to import JSON data (exit code: $IMPORT_EXIT_CODE)${NC}"
	echo -e "${YELLOW}Possible issues:${NC}"
	echo -e "${YELLOW} • JSON file may not be in k6 NDJSON format${NC}"
	echo -e "${YELLOW} • InfluxDB may not be ready${NC}"
	echo -e "${YELLOW} • Check the error messages above${NC}"
	echo ""
	echo -e "${YELLOW}Try checking InfluxDB logs:${NC}"
	echo -e "${YELLOW} cd .. && docker-compose logs influxdb${NC}"
	deactivate
	cd ..
	exit 1
	fi
	fi
	echo ""

	# Deactivate venv (only if Python was used)
	if [ "$USE_GO" = false ]; then
	deactivate
	fi

	cd ..

	# Final instructions
	echo -e "${GREEN}========================================${NC}"
	echo -e "${GREEN}Setup Complete! 🎉${NC}"
	echo -e "${GREEN}========================================${NC}"
	echo ""
	echo -e "${BLUE}Services:${NC}"
	echo -e " • Grafana: ${GREEN}http://localhost:3000${NC} (admin/admin)"
	echo -e " • InfluxDB: ${GREEN}http://localhost:8086${NC} (admin/admin)"
	echo ""
	echo -e "${BLUE}Next Steps:${NC}"
	echo -e " 1. Grafana opened automatically with the K6 Dashboard"
	echo -e " 2. Time range is preset to match your test data"
	echo -e " 3. View your test results!"
	echo ""
	echo -e "${YELLOW}Important:${NC}"
	echo -e " • Time range is automatically calculated from your test data"
	echo -e " • If some panels show 'No data', try adjusting the time range slightly"
	echo -e " • Dashboard data source is configured as 'InfluxDB-k6'"
	echo -e " • Dashboard: K6 Dashboard (ID: 14801) - Modern, community-maintained"
	echo ""
	echo -e "${BLUE}To stop the services:${NC}"
	echo -e " cd $(pwd) && docker-compose down"
	echo ""
	echo -e "${BLUE}To restart the services:${NC}"
	echo -e " cd $(pwd) && docker-compose up -d"
	echo ""
	echo -e "${BLUE}To remove everything (including data):${NC}"
	echo -e " cd $(pwd) && docker-compose down -v"
	echo ""

	# Extract time range from JSON file
	echo -e "${YELLOW}Calculating time range from test data...${NC}"

	# Get first and last timestamps from JSON
	FIRST_TIME=$(grep -o '"time":"[^"]*"' "$JSON_FILE" \| head -1 \| sed 's/"time":"//;s/"//')
	LAST_TIME=$(grep -o '"time":"[^"]*"' "$JSON_FILE" \| tail -1 \| sed 's/"time":"//;s/"//')

	if [ -n "$FIRST_TIME" ] && [ -n "$LAST_TIME" ]; then
	# Convert to milliseconds for Grafana URL (with 5 minute buffer before/after)
	FROM_MS=$(python3 -c "
	import datetime
	dt = datetime.datetime.fromisoformat('$FIRST_TIME'.replace('Z', '+00:00'))
	# Subtract 5 minutes for buffer
	dt = dt - datetime.timedelta(minutes=5)
	print(int(dt.timestamp() * 1000))
	" 2>/dev/null)

	TO_MS=$(python3 -c "
	import datetime
	dt = datetime.datetime.fromisoformat('$LAST_TIME'.replace('Z', '+00:00'))
	# Add 5 minutes for buffer
	dt = dt + datetime.timedelta(minutes=5)
	print(int(dt.timestamp() * 1000))
	" 2>/dev/null)

	if [ -n "$FROM_MS" ] && [ -n "$TO_MS" ]; then
	echo -e "${GREEN}✓ Time range detected: $FIRST_TIME to $LAST_TIME${NC}"
	DASHBOARD_URL="http://localhost:3000/d/9lcthCWnk/k6-dashboard?from=${FROM_MS}&to=${TO_MS}"
	else
	echo -e "${YELLOW}⚠ Could not calculate time range, using default${NC}"
	DASHBOARD_URL="http://localhost:3000"
	fi
	else
	echo -e "${YELLOW}⚠ Could not extract timestamps from JSON file${NC}"
	DASHBOARD_URL="http://localhost:3000"
	fi

	# Open Grafana in browser
	echo -e "${YELLOW}Opening Grafana dashboard...${NC}"
	sleep 2

	# Detect OS and open browser
	if [[ "$OSTYPE" == "darwin"* ]]; then
	# macOS
	open "$DASHBOARD_URL"
	elif [[ "$OSTYPE" == "linux-gnu"* ]]; then
	# Linux
	if command_exists xdg-open; then
	xdg-open "$DASHBOARD_URL" 2>/dev/null
	elif command_exists gnome-open; then
	gnome-open "$DASHBOARD_URL" 2>/dev/null
	else
	echo -e "${YELLOW}⚠ Could not auto-open browser. Please manually navigate to ${DASHBOARD_URL}${NC}"
	fi
	elif [[ "$OSTYPE" == "msys" \|\| "$OSTYPE" == "cygwin" ]]; then
	# Windows (Git Bash or Cygwin)
	start "$DASHBOARD_URL"
	else
	echo -e "${YELLOW}⚠ Could not auto-open browser. Please manually navigate to ${DASHBOARD_URL}${NC}"
	fi

	echo -e "${GREEN}✓ Browser opened with time range preset${NC}"
	echo ""